from __future__ import annotations import re import hashlib from typing import Any from uuid import uuid4 from ..models import User from ..schemas.context_workspace import normalize_research_workspace_payload from ..schemas.rag import RAGAskRequest from ..services.ai_pmo.project_operating_context_builder import ProjectOperatingContextBuilder def resolve_effective_tenant_id(current_user: User, fallback_tenant_id: int) -> int: try: return int(getattr(current_user, "tenant_id") or fallback_tenant_id or 1) except Exception: return int(fallback_tenant_id or 1) def build_answer_question_kwargs(request: RAGAskRequest, current_user: User) -> dict[str, Any]: asked_by = int(getattr(current_user, "id")) return { "question": request.question, "asked_by": asked_by, "tenant_id": _safe_int(getattr(current_user, "tenant_id", None)), "asked_via": request.asked_via, "project_id": request.project_id, "source_chat_id": request.source_chat_id, "session_id": request.session_id, "force_new_session": bool(request.new_session), "thinking_mode": request.thinking_mode, "action_mode": request.action_mode, "sync_requirement": bool(request.sync_requirement), "message_intent": request.message_intent, "mixed_signal_part": request.mixed_signal_part, "consultant_mode_preference": request.consultant_mode_preference, "surface_context": _normalize_optional_dict(request.surface_context), } def _safe_int(value: Any) -> int | None: try: if value is None: return None return int(value) except Exception: return None def _safe_float(value: Any, *, default: float = 0.0) -> float: try: return float(value) except Exception: return float(default) def _clamp_float(value: Any, *, default: float = 0.0) -> float: return max(0.0, min(1.0, _safe_float(value, default=default))) def _resolve_sender_name(current_user: User) -> str: sender = str( getattr(current_user, "name", None) or getattr(current_user, "username", None) or "PMFlow用户" ).strip() return sender or "PMFlow用户" def _normalize_dict_list(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, dict)] def _normalize_optional_dict(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, dict) else {} def _clean_text(value: Any, *, limit: int = 240) -> str: return " ".join(str(value or "").strip().split())[:limit] def ensure_customer_interaction_request(payload: dict[str, Any]) -> dict[str, Any]: """Attach a customer-safe follow-up form when the payload declares an input need. This is the shared PMFlow AI equivalent of a Codex-style input request: downstream services declare missing information through structured packets, and this builder converts that into a UI contract without parsing assistant prose or enumerating product scenarios. """ working = dict(payload or {}) if bool(working.get("interaction_request_suppressed") or working.get("suppress_interaction_request")): working["interaction_request"] = None working["answer"] = _strip_interaction_prompt_section(working.get("answer")) working["reply_text"] = _strip_interaction_prompt_section(working.get("reply_text") or working.get("answer")) return working existing = _normalize_interaction_request(working.get("interaction_request")) if existing: working["interaction_request"] = existing working["answer"] = _strip_interaction_prompt_section(working.get("answer")) working["reply_text"] = _strip_interaction_prompt_section(working.get("reply_text") or working.get("answer")) return working return working def apply_turn_presentation_contract(payload: dict[str, Any]) -> dict[str, Any]: """Decide the single primary interaction surface for a PMFlow AI turn. The answer payload may contain both an input request and suggested next actions because they are produced by different services. This arbiter makes the Codex-style runtime decision explicit: blocking input wins over action suggestions, and the frontend renders by contract instead of by field existence. """ working = dict(payload or {}) interaction = _normalize_interaction_request(working.get("interaction_request")) if interaction: working["interaction_request"] = interaction packet = working.get("customer_evidence_packet") if isinstance(working.get("customer_evidence_packet"), dict) else {} answer_type = _clean_text(packet.get("answer_type") or working.get("answer_type"), limit=80) grounding = _clean_text(working.get("grounding_status"), limit=80) processing_stage = _clean_text(working.get("processing_stage"), limit=80) if interaction: blocking = interaction.get("mode") != "optional" turn_state = "selection_required" if _interaction_has_choice_questions(interaction) else "input_required" contract = _turn_presentation_contract( turn_state=turn_state, primary_surface="interaction_request", blocking=blocking, reason_code="blocking_interaction_request" if blocking else "optional_interaction_request", answer_can_show=True, next_actions_can_show=not blocking, interaction_request_can_show=True, ) working["turn_presentation_contract"] = contract if blocking: working = _suppress_next_actions_for_blocking_interaction(working) return working if grounding == "blocked" or answer_type == "protected_scope": working["turn_presentation_contract"] = _turn_presentation_contract( turn_state="blocked", primary_surface="blocked_notice", blocking=True, reason_code="blocked_or_protected_scope", answer_can_show=True, next_actions_can_show=False, interaction_request_can_show=False, ) return _clear_next_action_surfaces(working) if grounding == "insufficient_evidence" or answer_type == "insufficient_evidence": working["turn_presentation_contract"] = _turn_presentation_contract( turn_state="blocked", primary_surface="blocked_notice", blocking=True, reason_code="insufficient_evidence_without_input_request", answer_can_show=True, next_actions_can_show=False, interaction_request_can_show=False, ) return _clear_next_action_surfaces(working) working["turn_presentation_contract"] = _turn_presentation_contract( turn_state="answer_ready", primary_surface="answer_canvas", blocking=False, reason_code="answer_ready", answer_can_show=True, next_actions_can_show=True, interaction_request_can_show=False, ) return working 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, } def _normalize_interaction_request(value: Any) -> dict[str, Any]: raw = dict(value) if isinstance(value, dict) else {} request_type = _clean_text(raw.get("type"), limit=40) if request_type not in {"confirm_and_supplement", "clarify", "selection"}: return {} questions = _normalize_interaction_questions(raw.get("questions")) supplement_raw = raw.get("supplement") if isinstance(raw.get("supplement"), dict) else {} if not questions and not bool(supplement_raw.get("enabled", False)): return {} mode = _clean_text(raw.get("mode"), limit=24) or "blocking" if mode not in {"blocking", "optional"}: mode = "blocking" request_id = _clean_text(raw.get("id"), limit=100) if not request_id: request_id = _stable_interaction_id(questions=questions, prefix="pmflow-ai-interaction") return { "id": request_id, "type": "confirm_and_supplement", "title": _clean_text(raw.get("title"), limit=80) or "还要你确认的点", "mode": mode, "presentation": _clean_text(raw.get("presentation"), limit=32) or "stepper", "questions": questions[:8], "supplement": { "enabled": bool(supplement_raw.get("enabled", True)), "required": bool(supplement_raw.get("required", False)), "placeholder": _clean_text(supplement_raw.get("placeholder"), limit=180) or "也可以直接补充背景、限制或你的判断...", }, "submit_label": _clean_text(raw.get("submit_label"), limit=40) or "确认并继续", "cancel_label": _clean_text(raw.get("cancel_label"), limit=40) or "先不补充", } def _turn_presentation_contract( *, turn_state: str, primary_surface: str, blocking: bool, reason_code: str, answer_can_show: bool, next_actions_can_show: bool, interaction_request_can_show: bool, ) -> dict[str, Any]: return { "version": "pmflow_ai_turn_presentation.v1", "turn_state": turn_state, "primary_surface": primary_surface, "blocking": bool(blocking), "reason_code": reason_code, "answer_can_show": bool(answer_can_show), "next_actions_can_show": bool(next_actions_can_show), "interaction_request_can_show": bool(interaction_request_can_show), } def _interaction_has_choice_questions(interaction: dict[str, Any]) -> bool: for question in list(interaction.get("questions") or []): if not isinstance(question, dict): continue input_type = _clean_text(question.get("input_type"), limit=40) options = _normalize_interaction_options(question.get("options")) if input_type in {"single_choice", "multi_choice"} and options: return True return False def _suppress_next_actions_for_blocking_interaction(payload: dict[str, Any]) -> dict[str, Any]: working = _clear_next_action_surfaces(payload) answer = _strip_next_step_section(_strip_interaction_prompt_section(working.get("answer"))) if not answer: answer = "先确认下面信息后，我再继续。" working["answer"] = answer working["reply_text"] = answer working["answer_canvas"] = None return working def _clear_next_action_surfaces(payload: dict[str, Any]) -> dict[str, Any]: working = dict(payload or {}) canvas = working.get("answer_canvas") if isinstance(working.get("answer_canvas"), dict) else None if canvas is not None: canvas = dict(canvas) canvas["next_actions"] = [] working["answer_canvas"] = canvas working["quick_actions"] = [] return working def _normalize_interaction_questions(value: Any) -> list[dict[str, Any]]: rows = list(value or []) if isinstance(value, list) else [] out: list[dict[str, Any]] = [] for index, item in enumerate(rows, start=1): if not isinstance(item, dict): continue label = _clean_text(item.get("label") or item.get("question"), limit=180) if not label: continue input_type = _clean_text(item.get("input_type") or item.get("type"), limit=32) or "text" options = _normalize_interaction_options(item.get("options")) if input_type not in {"single_choice", "multi_choice", "text"}: input_type = "text" if input_type in {"single_choice", "multi_choice"} and not options: input_type = "text" question = { "id": _clean_text(item.get("id"), limit=80) or f"question_{index}", "label": label, "input_type": input_type, "required": bool(item.get("required", True)), "placeholder": _clean_text(item.get("placeholder"), limit=160) or None, "options": options, } out.append(_ensure_question_choice_options(question)) return out def _normalize_interaction_options(value: Any) -> list[dict[str, Any]]: rows = list(value or []) if isinstance(value, list) else [] out: list[dict[str, Any]] = [] for index, option in enumerate(rows[:10], start=1): if isinstance(option, dict): label = _clean_text(option.get("label") or option.get("title") or option.get("name"), limit=120) option_id = _clean_text(option.get("id") or option.get("value") or option.get("key"), limit=80) description = _clean_text(option.get("description") or option.get("summary") or option.get("reason"), limit=180) else: label = _clean_text(option, limit=120) option_id = "" description = "" if not label: continue out.append({"id": option_id or f"option_{index}", "label": label, "description": description or None}) return out def _ensure_question_choice_options(question: dict[str, Any]) -> dict[str, Any]: """Prefer Codex-style guided choices; final supplement handles custom input.""" out = dict(question or {}) options = _normalize_interaction_options(out.get("options")) if options: out["options"] = options return out return out def _stable_interaction_id(*, questions: list[dict[str, Any]], prefix: str) -> str: seed = "|".join(_clean_text(item.get("label"), limit=120) for item in questions) or prefix digest = hashlib.sha1(seed.encode("utf-8")).hexdigest()[:10] return f"{prefix}-{digest}" def _strip_interaction_prompt_section(value: Any) -> str: text = str(value or "").strip() if not text: return "" pattern = re.compile( r"(?ims)\n*\s*(?:\*\*)?\s*还要你确认的点\s*(?:\*\*)?\s*\n+.*?(?=\n+\s*(?:\*\*)?\s*下一步\s*(?:\*\*)?\s*\n+|$)" ) stripped = pattern.sub("\n\n", text).strip() return re.sub(r"\n{3,}", "\n\n", stripped).strip() def _strip_next_step_section(value: Any) -> str: text = str(value or "").strip() if not text: return "" pattern = re.compile(r"(?ims)\n*\s*(?:\*\*)?\s*(?:下一步|建议下一步)\s*(?:\*\*)?\s*\n+.*$") stripped = pattern.sub("", text).strip() return re.sub(r"\n{3,}", "\n\n", stripped).strip() def _normalize_next_action(item: dict[str, Any]) -> dict[str, Any]: anchor = _clean_object_anchor(item.get("object_anchor") or item.get("objectAnchor")) object_ref = _clean_text( item.get("object_ref") or item.get("objectRef") or (anchor or {}).get("object_ref"), limit=120, ) return { "title": _clean_text(item.get("title"), limit=120), "project": _clean_text(item.get("project"), limit=80), "priority": _clean_text(item.get("priority"), limit=40), "href": _clean_text(item.get("href"), limit=160), "object_ref": object_ref or None, "object_anchor": anchor, } def _normalize_surface_context(value: Any) -> dict[str, Any]: surface = _normalize_optional_dict(value) if not surface: return {} workbench = _normalize_optional_dict(surface.get("workbench")) ai_readiness = _normalize_optional_dict(surface.get("ai_readiness")) counts = _normalize_optional_dict(workbench.get("counts")) recent_projects = list( workbench.get("recent_projects") or workbench.get("recentProjects") or workbench.get("projects") or [] ) return { "surface": _clean_text(surface.get("surface") or "pmflow", limit=80), "page": _clean_text(surface.get("page") or surface.get("route") or "", limit=120) or None, "ai_readiness": { "status": _clean_text(ai_readiness.get("status"), limit=40) or None, "can_generate_project_draft": bool(ai_readiness.get("can_generate_project_draft")), "can_manage": bool(ai_readiness.get("can_manage")), }, "workbench": { "counts": { key: int(counts.get(key) or 0) for key in ( "nextActions", "pendingConfirmations", "aiProgress", "projects", "savedIdeas", "drafts", "releaseAttention", "feedbackAttention", ) if str(counts.get(key) or "").strip() or counts.get(key) == 0 }, "top_next_actions": [ _normalize_next_action(item) for item in list(workbench.get("top_next_actions") or workbench.get("topNextActions") or [])[:6] if isinstance(item, dict) ], "recent_projects": [ { "id": _safe_int(item.get("id") or item.get("project_id") or item.get("projectId")), "project_id": _safe_int(item.get("project_id") or item.get("projectId") or item.get("id")), "object_ref": _clean_text( item.get("object_ref") or item.get("objectRef") or ( f"project:{item.get('id') or item.get('project_id') or item.get('projectId')}" if (item.get("id") or item.get("project_id") or item.get("projectId")) is not None else "" ), limit=120, ) or None, "title": _clean_text(item.get("title"), limit=120), "stage": _clean_text(item.get("stage"), limit=60), "status": _clean_text(item.get("status"), limit=60), "next_action": _clean_text(item.get("next_action") or item.get("nextAction"), limit=160), "href": _clean_text(item.get("href"), limit=160), } for item in recent_projects[:8] if isinstance(item, dict) ], "warnings": [_clean_text(item, limit=160) for item in list(workbench.get("warnings") or [])[:6]], }, } def _project_id_from_conversation_context(conversation_context: dict[str, Any] | None) -> int | None: context = _normalize_optional_dict(conversation_context) workspace = _normalize_optional_dict(context.get("active_workspace")) primary_object_ref = str(workspace.get("primary_object_ref") or "").strip().lower() if primary_object_ref.startswith("project:"): project_id = _safe_int(primary_object_ref.split(":", 1)[1]) if project_id is not None: return project_id project_id = _safe_int(workspace.get("primary_object_id")) if project_id is not None: return project_id for value in list(context.get("last_entity_ids") or []): project_id = _safe_int(value) if project_id is not None: return project_id return None def _project_id_from_surface_context(surface_context: dict[str, Any] | None) -> int | None: surface = _normalize_optional_dict(surface_context) workbench = _normalize_optional_dict(surface.get("workbench")) for key in ("recent_projects", "recentProjects", "projects"): for item in list(workbench.get(key) or []): if not isinstance(item, dict): continue project_id = _safe_int(item.get("project_id") or item.get("projectId") or item.get("id")) if project_id is not None: return project_id object_ref = str(item.get("object_ref") or item.get("objectRef") or "").strip() if object_ref.startswith("project:"): project_id = _safe_int(object_ref.split(":", 1)[1]) if project_id is not None: return project_id operating_packet = _normalize_optional_dict(surface.get("space_operating_packet")) for item in list(operating_packet.get("visible_projects") or []): if not isinstance(item, dict): continue object_ref = str(item.get("object_ref") or item.get("objectRef") or "").strip() if object_ref.startswith("project:"): project_id = _safe_int(object_ref.split(":", 1)[1]) if project_id is not None: return project_id return None def build_tenant_project_context_packet( request: RAGAskRequest, current_user: User, *, current_project_id_override: int | None = None, ) -> dict[str, Any]: surface_context = _normalize_surface_context(request.surface_context) tenant_id = getattr(current_user, "tenant_id", None) if tenant_id in (None, ""): tenant_id = 1 effective_project_id = _safe_int(request.project_id) or _safe_int(current_project_id_override) return { "packet_version": "tenant_project_context.v1", "visibility": "customer", "tenant_id": tenant_id, "user_id": _safe_int(getattr(current_user, "id", None)), "current_project_id": effective_project_id, "surface_context": surface_context, "allowed_answer_scopes": [ "pmflow_product_help", "tenant_project_memory", "personal_space_status", "external_research_with_sources", "next_action_guidance", ], "protected_scopes": [ "source_code", "database_schema", "runtime_trace", "prompt_keys", "model_provider_diagnostics", "internal_architecture", ], } def build_customer_direct_answer(request: RAGAskRequest, *, route_plan: dict[str, Any] | None = None) -> dict[str, Any] | None: """Render direct answers only from a structured route-plan contract.""" return build_planner_direct_answer(request=request, route_plan=route_plan or {}) def build_planner_direct_answer( *, request: RAGAskRequest, route_plan: dict[str, Any] | None, response_time_ms: int = 0, ) -> dict[str, Any] | None: plan = _normalize_optional_dict(route_plan) if str(plan.get("source_strategy") or "").strip().lower() != "direct": return None executors = { str(item or "").strip().lower() for item in list(plan.get("allowed_executors") or []) if str(item or "").strip() } if "direct_answer" not in executors: return None goal_kind = str(plan.get("goal_kind") or plan.get("intent") or "").strip().lower() answer_contract = str(plan.get("answer_contract") or "").strip().lower() if goal_kind == "protected_scope" or answer_contract == "protected_scope": payload = _build_customer_direct_payload( answer=( "这个入口只回答 PMFlow 产品使用、业务对象、当前空间/项目状态和外部公开信息。" "系统代码、内部架构、路由、prompt、模型诊断和 trace 不在用户侧问答范围内。" "如果你想了解某个功能怎么用，我可以按产品操作路径解释。" ), reason_code="protected_internal_scope", ) elif goal_kind == "casual_chat": payload = _build_customer_direct_payload( answer=( "我在。你可以直接问 PMFlow 里的项目进展、下一步、待确认事项、草案、AI 能力状态，" "也可以让我基于外部公开信息帮你判断某个方向是否适合当前项目。" ), reason_code="planner_casual_chat", ) elif goal_kind in {"pmflow_product_help", "pmflow_operation_help"} or answer_contract == "pmflow_product_help": payload = _build_customer_direct_payload( answer=_customer_capability_answer(), reason_code="planner_product_capability", ) else: return None payload["response_time_ms"] = max(0, int(response_time_ms or 0)) payload["pmflow_ai_route_plan"] = dict(plan) payload["processing_stage"] = "planner_direct" payload["answer_mode"] = "planner_direct" payload["policy_version"] = "customer-answer-policy-v2|route:pmflow_ai_orchestrator" payload["answer_process_packet"] = attach_answer_process_timing( payload.get("answer_process_packet"), total_duration_ms=max(0, int(response_time_ms or 0)), route_key="planner_direct", ) return payload def build_surface_direct_answer( *, request: RAGAskRequest, current_user: User, presentation_service: Any, pmflow_ai_route_plan: dict[str, Any] | None, response_time_ms: int = 0, ) -> dict[str, Any] | None: """Answer simple Personal Space questions directly from structured surface context. This path is only for customer-visible read-only facts that are already in `surface_context` (project inventory and current space status). It avoids both route-planner and workflow-planner model calls. """ surface_context = _normalize_surface_context(request.surface_context) if not surface_context: return None route_plan = _normalize_optional_dict(pmflow_ai_route_plan) is_quick_surface_prompt = str(request.message_intent or "").strip().lower().startswith("space_quick_") if not is_quick_surface_prompt and not _route_plan_allows_surface_direct(route_plan): return None packet = presentation_service.build_customer_evidence_packet( request=request, current_user=current_user, payload={ "pmflow_ai_route_plan": route_plan, "sources": [{"source": "pmflow", "label": "当前空间状态", "score": 1.0}], "sources_used": [{"display_label": "当前空间状态"}], "grounding_status": "grounded", }, ) answer = "" render = getattr(presentation_service, "_render_deterministic_answer", None) if callable(render): answer = str(render(packet) or "").strip() if not answer: return None interaction = presentation_service.build_customer_interaction_context_packet(packet) process = presentation_service.build_answer_process_packet(packet=packet, verification={"ok": True}) process = attach_answer_process_timing( process, total_duration_ms=max(0, int(response_time_ms or 0)), route_key="surface_direct", ) answer_canvas = {} build_canvas = getattr(presentation_service, "build_answer_canvas", None) if callable(build_canvas): answer_canvas = build_canvas(packet=packet, answer=answer) sources = [{"source": "pmflow", "label": "当前空间状态", "score": 1.0}] try: from ..services.customer_answer_presentation_service import _customer_anchor_sources_from_packet sources.extend(_customer_anchor_sources_from_packet(packet)) except Exception: pass return { "answer": answer, "reply_text": answer, "sources": sources, "sources_used": [{"display_label": "当前空间状态"}], "response_time_ms": max(0, int(response_time_ms or 0)), "thinking_mode": request.thinking_mode or "auto", "processing_stage": "surface_direct", "answer_mode": "surface_direct", "grounding_status": "grounded", "policy_version": "surface-direct-v1|customer-evidence-packet-v1", "pmflow_ai_route_plan": route_plan, "customer_evidence_packet": packet, "customer_interaction_context_packet": interaction, "answer_process_packet": process, "answer_canvas": answer_canvas, "customer_answer_policy": {"visibility": "customer", "protected": True}, } def build_active_workspace_followup_answer( *, request: RAGAskRequest, current_user: User, presentation_service: Any, active_workspace: dict[str, Any] | None, response_time_ms: int = 0, ) -> dict[str, Any] | None: workspace = normalize_research_workspace_payload(active_workspace) if not workspace or not _is_short_workspace_continue_reply(request.question): return None workspace_kind = str(workspace.get("workspace_kind") or "").strip().lower() if workspace_kind not in {"external_research", "hybrid_research"}: return None candidates = [ item for item in list(workspace.get("candidate_items") or [])[:6] if isinstance(item, dict) and str(item.get("display_name") or "").strip() ] next_actions = [str(item or "").strip() for item in list(workspace.get("next_recommended_actions") or []) if str(item or "").strip()] wants_mapping_checklist = any("映射" in item and "验证" in item for item in next_actions) if not wants_mapping_checklist and not candidates: return None topic = _clean_text(workspace.get("topic") or "上一轮外部调研", limit=120) primary_title = _clean_text(workspace.get("primary_object_title"), limit=120) dimensions = [ _clean_text(item, limit=40) for item in list(workspace.get("analysis_dimensions") or []) if _clean_text(item, limit=40) ][:6] or ["目标用户", "核心问题", "AI 能力形态", "数据/隐私", "部署形态", "差异化机会"] lines = [f"可以。我基于上一轮“{topic}”先整理一份映射验证清单。"] if primary_title: lines.append(f"当前项目锚点：{primary_title}") lines.append("") lines.append("验证维度：" + "、".join(dimensions)) if candidates: lines.append("") lines.append("候选对象：") for index, item in enumerate(candidates, start=1): name = _clean_text(item.get("display_name"), limit=80) status = _clean_text(item.get("status"), limit=40) or "待确认" confidence = _clean_text(item.get("confidence"), limit=40) or "unknown" summary = _clean_text(item.get("summary"), limit=140) source_labels = [ _clean_text(label, limit=60) for label in list(item.get("source_labels") or [])[:3] if _clean_text(label, limit=60) ] evidence = f"；来源：{'、'.join(source_labels)}" if source_labels else "" lines.append(f"{index}. {name}：状态 {status}，置信度 {confidence}。{summary}{evidence}") lines.append("") lines.append("下一步建议：先逐项补齐直接竞品证据、差异点证据和当前项目适配风险，再决定是否进入变更评审或需求拆解。") answer = "\n".join(lines).strip() route_plan = { "source_strategy": "hybrid_external_first" if workspace_kind == "hybrid_research" else "external_only", "primary_source_policy": "external", "is_hybrid": workspace_kind == "hybrid_research", "confidence": 0.86 if candidates else 0.72, "intent": "external_research_followup", "goal_kind": "project_fit_research", "answer_contract": "mapping_verification_checklist", "route": "external_query", "surface_direct_eligible": False, } research_contract = { "contract_version": "customer_research_followup.v1", "effective_lane": "external", "external_sources_required": True, "project_fit_required": bool(primary_title), "async_handoff": "none", "completion_surface": "space_ai_concierge", } payload = { "answer": answer, "reply_text": answer, "sources": [{"source": "external", "label": "上一轮外部公开来源", "score": 0.9}], "sources_used": [{"display_label": "上一轮外部公开来源"}], "response_time_ms": max(0, int(response_time_ms or 0)), "thinking_mode": request.thinking_mode or "auto", "processing_stage": "active_workspace_followup", "answer_mode": "active_workspace_followup", "grounding_status": "partially_grounded", "policy_version": "customer-answer-policy-v2|route:active_workspace_followup", "pmflow_ai_route_plan": route_plan, "research_workspace": workspace, "research_contract": research_contract, "source_policy": "external", "query_type": "external_followup", "customer_answer_policy": {"visibility": "customer", "protected": True}, } packet = presentation_service.build_customer_evidence_packet( request=request, current_user=current_user, payload=payload, ) payload["customer_evidence_packet"] = packet payload["customer_interaction_context_packet"] = presentation_service.build_customer_interaction_context_packet( packet, research_workspace=workspace, ) payload["answer_process_packet"] = attach_answer_process_timing( presentation_service.build_answer_process_packet(packet=packet, verification={"ok": True}), total_duration_ms=max(0, int(response_time_ms or 0)), route_key="active_workspace_followup", ) build_canvas = getattr(presentation_service, "build_answer_canvas", None) if callable(build_canvas): payload["answer_canvas"] = build_canvas( packet=packet, answer=answer, research_workspace=workspace, ) return payload def _is_short_workspace_continue_reply(value: Any) -> bool: text = re.sub(r"[\s。！？!?.，,；;：:、]+", "", str(value or "").strip().lower()) if not text or len(text) > 12: return False return text in { "需要", "要", "可以", "继续", "好的", "好", "行", "嗯", "嗯嗯", "对", "是", "yes", "y", "ok", "okay", } def _route_plan_allows_surface_direct(route_plan: dict[str, Any]) -> bool: if not route_plan: return False if str(route_plan.get("source_strategy") or "").strip().lower() != "internal_only": return False if route_plan.get("surface_direct_allowed") is False: return False analysis_task = str(route_plan.get("analysis_task") or "").strip().lower() business_frame = str(route_plan.get("business_frame") or "").strip().lower() if analysis_task and analysis_task not in {"status", "list", "count"}: return False if business_frame and business_frame not in {"general", "space_status"}: return False if not bool(route_plan.get("surface_direct_eligible")): return False executors = { str(item or "").strip().lower() for item in list(route_plan.get("allowed_executors") or []) if str(item or "").strip() } return "surface_direct" in executors def attach_answer_process_timing( process_packet: Any, *, total_duration_ms: int, wall_clock_duration_ms: int | None = None, route_key: str = "", ) -> dict[str, Any]: payload = dict(process_packet or {}) if isinstance(process_packet, dict) else {} steps = [dict(item) for item in list(payload.get("steps") or []) if isinstance(item, dict)] if steps: assigned = False for step in steps: key = str(step.get("key") or "").strip() if key in {"read_projects", "read_space_status", "collect_context", route_key}: step.setdefault("duration_ms", max(0, int(total_duration_ms or 0))) assigned = True if not assigned: if steps[-1].get("duration_ms") in {None, ""}: steps[-1]["duration_ms"] = max(0, int(total_duration_ms or 0)) payload["steps"] = steps payload["total_duration_ms"] = max(0, int(total_duration_ms or 0)) if wall_clock_duration_ms is not None: payload["wall_clock_duration_ms"] = max(0, int(wall_clock_duration_ms or 0)) return payload def _customer_capability_answer() -> str: return ( "我是 PMFlow AI，面向当前空间和项目工作的 AI 助手。" "我可以回答 PMFlow 平台的业务和操作问题，结合当前租户下的项目、待确认事项、草案、执行状态和 AI 能力状态给你下一步建议，" "也可以基于外部公开信息做调研并说明这些信息对当前项目是否有用。" "我不会在用户侧回答系统代码、内部架构、数据库表、prompt、trace 或模型诊断这类实现细节。" ) def _build_customer_direct_payload(*, answer: str, reason_code: str) -> dict[str, Any]: protected = reason_code == "protected_internal_scope" return { "answer": answer, "reply_text": answer, "sources": [{"source": "pmflow", "label": "PMFlow 产品说明", "score": 1.0}], "sources_used": [{"display_label": "PMFlow 产品说明"}], "response_time_ms": 0, "thinking_mode": "normal", "processing_stage": "customer_policy", "answer_mode": "fast_path", "grounding_status": "grounded", "policy_version": "customer-answer-policy-v1", "customer_answer_policy": _customer_answer_policy(), "answer_process_packet": { "packet_version": "customer_answer_process.v1", "visibility": "customer", "answer_type": "protected_scope" if protected else "pmflow_product_help", "steps": [], }, "reason_code": reason_code, } def _customer_answer_policy() -> dict[str, Any]: return { "visibility": "customer", "source_labels": "display_only", "protected_internal_details": True, "allowed_scopes": [ "PMFlow 产品和操作说明", "当前空间和项目状态", "当前租户项目记忆与待确认事项", "带来源的外部公开信息", "下一步行动建议", ], } def _default_evidence_pack(*, answer: str, tool_trace: list[dict[str, Any]]) -> dict[str, Any]: return { "summary": { "conclusion": str(answer or "").strip(), "evidence": [], "uncertainties": [], }, "tool_trace": tool_trace, } def build_kernel_ingress_kwargs( request: RAGAskRequest, current_user: User, *, pmflow_ai_route_plan: dict[str, Any] | None = None, conversation_context: dict[str, Any] | None = None, ) -> dict[str, Any]: rag_request_context = build_answer_question_kwargs(request=request, current_user=current_user) normalized_conversation_context = _normalize_optional_dict(conversation_context) active_workspace = _normalize_optional_dict(normalized_conversation_context.get("active_workspace")) if active_workspace: rag_request_context["active_workspace"] = active_workspace inferred_project_id = ( _safe_int(request.project_id) or _project_id_from_conversation_context(normalized_conversation_context) ) if inferred_project_id is not None: rag_request_context["project_id"] = inferred_project_id if isinstance(pmflow_ai_route_plan, dict) and pmflow_ai_route_plan: rag_request_context["pmflow_ai_route_plan"] = dict(pmflow_ai_route_plan) surface_context = _normalize_surface_context(request.surface_context) tenant_project_context = build_tenant_project_context_packet( request=request, current_user=current_user, current_project_id_override=inferred_project_id, ) project_operating_context = ProjectOperatingContextBuilder().build_from_ingress( project_hint_id=inferred_project_id, rag_request_context=rag_request_context, ) ingress_options: dict[str, Any] = { "rag_request_context": rag_request_context, "surface_context": surface_context, "tenant_project_context": tenant_project_context, "project_operating_context": project_operating_context.model_dump(), } semantic_bundle = build_readonly_query_semantic_bundle( request=request, current_user=current_user, pmflow_ai_route_plan=pmflow_ai_route_plan, ) if semantic_bundle: ingress_options["semantic_bundle"] = semantic_bundle return { "source_system": str(request.asked_via or "pmflow").strip() or "pmflow", "source_message_id": f"rag-ask-{uuid4().hex}", "message_text": request.question, "sender": _resolve_sender_name(current_user), "channel_ref": request.source_chat_id, "source_sender_open_id": None, "captured_at": None, "chat_type": "manual", "project_hint_id": request.project_id, "ingress_options": ingress_options, } def build_readonly_query_semantic_bundle( *, request: RAGAskRequest, current_user: User, pmflow_ai_route_plan: dict[str, Any] | None, ) -> dict[str, Any] | None: """Build a lightweight query_flow bundle when the Space route planner already resolved read-only QA. This prevents the downstream AI PMO kernel from calling the heavy `PMFLOW_SEMANTIC_CONTEXT_DECISION_PROMPT` again for ordinary question answering. Write/draft/execute workflows still go through the normal semantic planner. """ plan = dict(pmflow_ai_route_plan or {}) primary = str(plan.get("primary_source_policy") or "").strip().lower() route = str(plan.get("route") or "").strip().lower() intent = str(plan.get("intent") or "").strip().lower() if primary not in {"internal", "external"}: return None if "query" not in route and route not in {"internal_query", "external_query", "answer", "query_answer"}: return None if any(token in intent for token in ("draft", "create", "execute", "write", "confirm", "requirement")): return None confidence = max(0.0, min(1.0, _safe_float(plan.get("confidence"), default=0.0))) if confidence < 0.65: return None digest_basis = "|".join( [ str(request.asked_via or "pmflow").strip().lower(), str(request.question or "").strip(), primary, intent or "query", ] ) bundle_id = "pmflow-ai-route-" + hashlib.sha1( digest_basis.encode("utf-8"), usedforsecurity=False, ).hexdigest()[:16] source_system = str(request.asked_via or "pmflow").strip().lower() or "pmflow" return { "bundle_id": bundle_id, "semantic_version": "2026-03-27.workflow-v1", "source_system": source_system, "primary_workflow": "query_flow", "workflows": [ { "workflow_type": "query_flow", "workflow_action": "answer", "target_scope": primary, "operation_codes": ["query.answer"], "requires_confirmation": False, "confidence": confidence, "slots": [ {"name": "source_policy", "value": primary, "source": "pmflow_ai_route_planner"}, {"name": "source_strategy", "value": str(plan.get("source_strategy") or ""), "source": "pmflow_ai_route_planner"}, ], "clarification_needed": False, "metadata": { "message_intent": "query.answer", "project_id": _safe_int(request.project_id), "pmflow_ai_route_plan": plan, "semantic_shortcut": "pmflow_ai_route_plan_readonly_query", }, } ], "message_gate": "auto_execute", "reply_contract": { "mode": "aggregate_text", "template_key": "workflow_bundle", "summary": "query_flow", }, "diagnostics": { "message_intent": "query.answer", "project_id": _safe_int(request.project_id), "source_system": source_system, "source_message_context": "pmflow_ai_route_plan", "context_source": "pmflow_ai_route_plan_readonly_query", "semantic_shortcut": True, "pmflow_ai_route_plan": plan, "user_id": _safe_int(getattr(current_user, "id", None)), "tenant_id": _safe_int(getattr(current_user, "tenant_id", None)), }, } def build_rag_response_payload(*, ingress_result: dict[str, Any]) -> dict[str, Any]: result = ingress_result if isinstance(ingress_result, dict) else {} rag_result = result.get("rag_result") if isinstance(result.get("rag_result"), dict) else {} payload = dict(rag_result) consultant_dispatch = ( payload.get("consultant_dispatch") if isinstance(payload.get("consultant_dispatch"), dict) else (result.get("consultant_dispatch") if isinstance(result.get("consultant_dispatch"), dict) else None) ) if isinstance(consultant_dispatch, dict): payload["consultant_dispatch"] = consultant_dispatch answer = str(payload.get("answer") or result.get("direct_reply_text") or "").strip() payload["answer"] = answer if "reply_required" not in payload: payload["reply_required"] = bool(result.get("direct_reply_required")) if "reply_text" not in payload: payload["reply_text"] = str(result.get("direct_reply_text") or "").strip() or None if "trace_id" not in payload: payload["trace_id"] = str(result.get("ai_trace_id") or "").strip() or None if "processing_stage" not in payload: payload["processing_stage"] = "qa" if "tool_trace" not in payload: payload["tool_trace"] = [] payload["tool_trace"] = _normalize_dict_list(payload.get("tool_trace")) payload["route_decision"] = _normalize_optional_dict(payload.get("route_decision")) payload["clarification_code"] = str(payload.get("clarification_code") or "").strip() or None payload["clarification_candidates"] = _normalize_dict_list(payload.get("clarification_candidates")) payload["citations"] = _normalize_dict_list(payload.get("citations")) payload["sources_used"] = _normalize_dict_list(payload.get("sources_used")) payload["answer_mode"] = str(payload.get("answer_mode") or "retrieval_normal").strip() or "retrieval_normal" payload["grounding_status"] = str(payload.get("grounding_status") or "partially_grounded").strip() or "partially_grounded" payload["policy_version"] = ( str(payload.get("policy_version") or "retrieval-v2.1|routing:unknown|prompt:unknown").strip() or "retrieval-v2.1|routing:unknown|prompt:unknown" ) evidence_pack = payload.get("evidence_pack") if isinstance(payload.get("evidence_pack"), dict) else None payload["evidence_pack"] = evidence_pack or _default_evidence_pack( answer=answer, tool_trace=payload["tool_trace"], ) payload["work_item_id"] = _safe_int(result.get("work_item_id")) payload["ingress_status"] = str(result.get("status") or "").strip() or None payload["intent"] = str(result.get("intent") or "").strip() or None route_plan = None if isinstance(rag_result.get("pmflow_ai_route_plan"), dict): route_plan = rag_result.get("pmflow_ai_route_plan") elif isinstance(result.get("pmflow_ai_route_plan"), dict): route_plan = result.get("pmflow_ai_route_plan") if isinstance(route_plan, dict): payload["pmflow_ai_route_plan"] = dict(route_plan) return payload def sanitize_customer_rag_payload( payload: dict[str, Any], *, request: RAGAskRequest | None = None, current_user: User | None = None, apply_presentation_contract: bool = True, ) -> dict[str, Any]: cleaned = dict(payload or {}) if apply_presentation_contract: cleaned = apply_turn_presentation_contract(ensure_customer_interaction_request(cleaned)) answer = _normalize_customer_answer_markdown( _sanitize_customer_answer_text(cleaned.get("reply_text") or cleaned.get("answer")) ) if not answer: answer = _normalize_customer_answer_markdown( _sanitize_customer_answer_text(_answer_from_customer_truth_packet(cleaned)) ) cleaned["answer"] = answer cleaned["reply_text"] = answer cleaned["sources"] = _sanitize_customer_sources(cleaned.get("sources")) cleaned["evidence"] = _sanitize_customer_sources(cleaned.get("evidence")) cleaned["sources_used"] = _sanitize_customer_sources_used(cleaned.get("sources_used") or cleaned.get("citations")) cleaned["citations"] = _sanitize_customer_sources_used(cleaned.get("citations")) cleaned["customer_answer_policy"] = _customer_answer_policy() if isinstance(cleaned.get("pmflow_ai_route_plan"), dict): cleaned["pmflow_ai_route_plan"] = sanitize_customer_route_plan(cleaned.get("pmflow_ai_route_plan")) if isinstance(cleaned.get("customer_evidence_packet"), dict): cleaned["customer_evidence_packet"] = _sanitize_customer_display_packet(cleaned.get("customer_evidence_packet")) if isinstance(cleaned.get("customer_interaction_context_packet"), dict): cleaned["customer_interaction_context_packet"] = _sanitize_customer_display_packet( cleaned.get("customer_interaction_context_packet") ) if isinstance(cleaned.get("research_workspace"), dict): cleaned["research_workspace"] = _sanitize_customer_display_packet(cleaned.get("research_workspace")) if isinstance(cleaned.get("answer_process_packet"), dict): cleaned["answer_process_packet"] = _sanitize_customer_display_packet(cleaned.get("answer_process_packet")) if isinstance(cleaned.get("terminal_answer_state"), dict): cleaned["terminal_answer_state"] = _sanitize_customer_display_packet(cleaned.get("terminal_answer_state")) if isinstance(cleaned.get("deliverable_packet"), dict): cleaned["deliverable_packet"] = _sanitize_customer_display_packet(cleaned.get("deliverable_packet")) if isinstance(cleaned.get("pmflow_agent_run"), dict): cleaned["pmflow_agent_run"] = _sanitize_customer_display_packet(cleaned.get("pmflow_agent_run")) if isinstance(cleaned.get("agent_os_environment"), dict): cleaned["agent_os_environment"] = _sanitize_customer_display_packet(cleaned.get("agent_os_environment")) if isinstance(cleaned.get("agent_os_specialists"), dict): cleaned["agent_os_specialists"] = _sanitize_customer_display_packet(cleaned.get("agent_os_specialists")) if isinstance(cleaned.get("answer_canvas"), dict): cleaned["answer_canvas"] = _sanitize_customer_display_packet(cleaned.get("answer_canvas")) if answer: cleaned["answer_canvas"]["direct_answer"] = answer if isinstance(cleaned.get("pmflow_ai_turn_state"), dict): cleaned["pmflow_ai_turn_state"] = _sanitize_customer_display_packet(cleaned.get("pmflow_ai_turn_state")) if isinstance(cleaned.get("interaction_request"), dict): cleaned["interaction_request"] = _sanitize_customer_display_packet(cleaned.get("interaction_request")) if isinstance(cleaned.get("turn_presentation_contract"), dict): cleaned["turn_presentation_contract"] = _sanitize_customer_display_packet(cleaned.get("turn_presentation_contract")) if isinstance(cleaned.get("object_ledger"), dict): cleaned["object_ledger"] = _sanitize_customer_display_packet(cleaned.get("object_ledger")) for key in ( "operating_mission", "project_world_state_packet", "customer_safe_claim_ledger", "operating_run_packet", "operating_plan", "agent_council_review", "autonomy_gate", "operating_cycle", "consumed_learning_ledger", "learning_ledger", "command_payload", "agent_os_command_payload", "agent_os_agent_council_review", "agent_os_operating_plan", "agent_os_autonomy_gate", "agent_os_operating_cycle", "agent_os_consumed_learning_ledger", "agent_os_learning_ledger", ): if isinstance(cleaned.get(key), dict): cleaned[key] = _sanitize_customer_display_packet(cleaned.get(key)) for key in ( "agent_assignments", "agent_work_products", "evidence_ledger", "conflict_ledger", "prepared_action_packets", "confirmation_requests", "execution_preparations", "product_object_projections", "outcome_observations", "learning_updates", "next_cycle_triggers", "agent_os_steps", "agent_os_observations", "agent_os_artifacts", "agent_os_verifications", "agent_os_memories", "agent_os_specialist_assignments", "agent_os_work_products", "agent_os_prepared_action_packets", "agent_os_confirmation_requests", "agent_os_execution_preparations", "agent_os_product_object_projections", "agent_os_outcome_observations", "agent_os_learning_updates", "agent_os_next_cycle_triggers", ): if isinstance(cleaned.get(key), list): cleaned[key] = [ _sanitize_customer_display_packet(item) for item in list(cleaned.get(key) or [])[:30] if isinstance(item, dict) ] cleaned["tool_trace"] = [] cleaned["trace_id"] = None cleaned["policy_version"] = str(cleaned.get("policy_version") or "customer-answer-policy-v1").replace("prompt:unknown", "customer") cleaned["processing_stage"] = _customer_processing_stage(cleaned.get("processing_stage")) if request is not None and current_user is not None: cleaned["tenant_project_context_packet"] = build_tenant_project_context_packet( request=request, current_user=current_user, ) cleaned["tenant_project_context_packet"] = _customer_tenant_project_context_packet( cleaned.get("tenant_project_context_packet") ) cleaned["external_project_fit_packet"] = _build_external_project_fit_packet(cleaned) cleaned["research_contract"] = _build_customer_research_contract( request=request, payload=cleaned, ) return _strip_customer_internal_payload_fields(cleaned) def _answer_from_customer_truth_packet(payload: dict[str, Any]) -> str: deliverable = payload.get("deliverable_packet") if isinstance(payload.get("deliverable_packet"), dict) else {} canvas = payload.get("answer_canvas") if isinstance(payload.get("answer_canvas"), dict) else {} evidence = payload.get("customer_evidence_packet") if isinstance(payload.get("customer_evidence_packet"), dict) else {} for value in ( deliverable.get("answer_text"), deliverable.get("summary"), canvas.get("direct_answer"), evidence.get("fallback_summary"), ): text = str(value or "").strip() if text: return text return "" _CUSTOMER_INTERNAL_TOP_LEVEL_KEYS = { "answer_runtime_packet", "evidence_findings", "project_operating_answer_verification_packet", "pmflow_ai_turn_state", "pmflow_ai_route_plan", "route_decision", "runtime_observation", "runtime_evidence_tool_result", "routing_snapshot", "fit_brief", "execution_recipe", "tool_trace_debug", "tool_trace", "trace_id", "debug", "raw_payload", "prompt_keys", "ai_call_logs", "agent_workspace", "operating_summary", } def _strip_customer_internal_payload_fields(payload: dict[str, Any]) -> dict[str, Any]: cleaned = dict(payload or {}) for key in _CUSTOMER_INTERNAL_TOP_LEVEL_KEYS: cleaned.pop(key, None) return cleaned def _customer_tenant_project_context_packet(value: Any) -> dict[str, Any]: raw = _normalize_optional_dict(value) surface = _sanitize_customer_display_packet(raw.get("surface_context")) return { "packet_version": "tenant_project_context.v1", "visibility": "customer", "current_project_id": _safe_int(raw.get("current_project_id")), "surface_context": surface, "allowed_answer_scopes": [ "PMFlow 产品能力说明", "当前空间与项目状态", "项目记忆和客户可见业务内容", "带来源的外部公开信息", "下一步建议", ], "protected_scopes": [ "源码与实现细节", "数据库结构", "运行诊断轨迹", "内部提示配置", "模型供应商诊断", "内部技术架构", ], } def _sanitize_customer_display_packet(value: Any) -> dict[str, Any]: if not isinstance(value, dict): return {} text = _sanitize_customer_answer_text literal_value_keys = { "id", "key", "type", "mode", "status", "workspace_kind", "workspace_version", "packet_version", "object_type", "object_ref", "object_id", "binding_strength", "active_goal", "research_goal", "relevance_to_goal", "confidence", "confidence_state", "source_type", } internal_value_keys = { "tenant_id", "user_id", "trace_id", "prompt_key", "prompt_keys", "raw_source_label", "tool_trace", "routing_meta", "route_plan_input", "legacy_runtime_decision", "answer_objective", "answer_plan", "answer_quality_findings", "answer_runtime_packet", "project_operating_answer_verification_packet", "verification_failures", "verification_delta", "evidence_findings", "structured_evidence", "runtime_observation", "runtime_evidence_tool_result", "model_provider", "provider", "raw_payload", "internal_diagnostic", "debug", } def clean(raw: Any, *, parent_key: str = "") -> Any: if isinstance(raw, dict): return { str(key): clean(item, parent_key=str(key)) for key, item in raw.items() if str(key).strip() not in internal_value_keys if str(key).strip().lower() not in internal_value_keys } if isinstance(raw, list): return [clean(item, parent_key=parent_key) for item in raw[:30]] if isinstance(raw, str): if parent_key in literal_value_keys or parent_key.endswith("_kind") or parent_key.endswith("_version"): return str(raw).strip() return text(raw) return raw cleaned = clean(value) return cleaned if isinstance(cleaned, dict) else {} def _customer_processing_stage(value: Any) -> str: normalized = str(value or "").strip().lower() if normalized in { "strategic_judgement", "status_answer", "clarification", "execution_guidance", "fact_list", "composite_answer", "general_answer", "multi_agent_operating_runtime", "agent_runtime", "pmflow_agent_runtime", }: return normalized return "answer_ready" def sanitize_customer_route_plan(value: Any) -> dict[str, Any]: raw = _normalize_optional_dict(value) allowed_strategy = { "direct", "internal_only", "external_only", "hybrid_internal_first", "hybrid_external_first", "clarify", } strategy = str(raw.get("source_strategy") or "").strip().lower() if strategy not in allowed_strategy: strategy = "clarify" return { "source_strategy": strategy, "primary_source_policy": str(raw.get("primary_source_policy") or "").strip().lower() or None, "is_hybrid": bool(raw.get("is_hybrid")), "confidence": _safe_score(raw.get("confidence")), "intent": _clean_text(raw.get("intent"), limit=80) or None, "goal_kind": _clean_text(raw.get("goal_kind"), limit=80) or None, "answer_contract": _clean_text(raw.get("answer_contract"), limit=80) or None, "route": _clean_text(raw.get("route"), limit=80) or None, "analysis_task": _clean_text(raw.get("analysis_task"), limit=80) or None, "business_frame": _clean_text(raw.get("business_frame"), limit=80) or None, "evidence_plan": [ _clean_text(item, limit=80) for item in list(raw.get("evidence_plan") or [])[:12] if _clean_text(item, limit=80) ], "surface_direct_allowed": bool(raw.get("surface_direct_allowed")), "needs_clarification": bool(raw.get("needs_clarification")), "clarification_question": _clean_text(raw.get("clarification_question"), limit=240) or None, "surface_direct_eligible": bool(raw.get("surface_direct_eligible")), "decision_origin": "pmflow_ai_orchestrator", } def _sanitize_customer_answer_text(value: Any) -> str: text = str(value or "").strip() if not text: return "" text = text.replace("\r\n", "\n").replace("\r", "\n") leak_markers = ( "Traceback", "NameError", "PROMPT_", "AICallLog", "backend/", "frontend/", "app/services", "SemanticDecision", "AdaptiveRAG", ) if any(marker.lower() in text.lower() for marker in leak_markers): # semantic-lint: ignore return "这次回答生成失败，系统没有拿到可展示的业务结论。你可以换一种问法，或稍后重试。" replacements = { "System Snapshot": "当前空间状态", "User Snapshot": "当前用户状态", "Session State": "当前对话状态", "Project database": "项目数据", "Knowledge Graph": "项目记忆", "Feishu history": "飞书记录", "PMFlow records": "PMFlow 记录", "project_object_required_for_world_state": "需要明确项目对象后才能读取项目运行状态", "world_state": "项目运行状态", "route_plan": "回答计划", "function_name": "能力模块", "ResearchNote claim": "研究记录里的可用证据判断", "ResearchNote": "研究记录", "usable claim": "可用证据判断", "claim_ledger": "可用证据判断", "claim ledger": "可用证据判断", "gap_ledger": "证据缺口", "gap ledger": "证据缺口", "blocking gaps": "阻塞性证据缺口", "evidence_normalization_gap": "证据整理缺口", "normalization gap": "证据整理缺口", } for raw, display in replacements.items(): text = text.replace(raw, display) text = _sanitize_customer_answer_internal_terms(text) text = "\n".join(re.sub(r"[ \t]+", " ", line).strip() for line in text.split("\n")) text = re.sub(r"(?<=[\u4e00-\u9fff])[ \t]+(?=[\u4e00-\u9fff])", "", text) text = re.sub(r"\n{3,}", "\n\n", text).strip() return text def _normalize_customer_answer_markdown(value: Any) -> str: text = str(value or "").strip() if not text: return "" text = _repair_numbered_customer_markdown_table_continuations(text) text = _repair_fragmented_customer_markdown_tables(text) text = _remove_orphan_customer_markdown_pipe_lines(text) text = _repair_inline_customer_bold_boundaries(text) text = _repair_inline_customer_markdown_tables(text) text = _repair_raw_customer_markdown_headings(text) text = _repair_inline_customer_markdown_lists(text) text = _repair_inline_customer_section_breaks(text) if "\n\n" in text or _has_markdown_block_structure(text): return text paragraphs = [block.strip() for block in re.split(r"\n\s*\n+", text) if block.strip()] if len(paragraphs) > 1: return "\n\n".join(paragraphs) if len(text) <= 420 and _sentence_count(text) < 4: return text repaired: list[str] = [] for paragraph in paragraphs or [text]: repaired.extend(_split_dense_customer_answer_paragraph(paragraph)) return "\n\n".join(item for item in repaired if item).strip() def _has_markdown_block_structure(text: str) -> bool: return any( re.match(r"^\s*(?:[-*+]|\d+[.)、])\s+", line) or (line.strip().startswith("|") and line.strip().endswith("|")) for line in str(text or "").splitlines() ) def _remove_orphan_customer_markdown_pipe_lines(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value: return value lines = [line for line in value.split("\n") if not re.fullmatch(r"[ \t]*\|+[ \t]*", line)] return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip() def _repair_inline_customer_bold_boundaries(text: str) -> str: value = str(text or "").strip() if "**" not in value: return value def repl(match: re.Match[str]) -> str: body = match.group(1).strip() before = value[match.start() - 1] if match.start() > 0 else "" after = value[match.end()] if match.end() < len(value) else "" prefix = " " if before and not re.fullmatch(r"\s|[(（\[【{《“]", before) else "" suffix = " " if after and not re.fullmatch(r"\s|[)，)\]】}。；：，、！？,.!?:》”]", after) else "" return f"{prefix}**{body}**{suffix}" value = re.sub(r"\*\*([^*\n]+?)\*\*", repl, value) return re.sub(r"[ \t]{2,}", " ", value).strip() def _repair_fragmented_customer_markdown_tables(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value or "---" not in value: return value lines = value.split("\n") out: list[str] = [] index = 0 while index < len(lines): current = lines[index].strip() next_line = lines[index + 1].strip() if index + 1 < len(lines) else "" if not _is_customer_markdown_pipe_line(current) or not _is_customer_markdown_table_separator_line(next_line): out.append(lines[index]) index += 1 continue first_pipe = current.find("|") lead = current[:first_pipe].strip() if first_pipe > 0 else "" header_line = current[first_pipe:].strip() if first_pipe > 0 else current segment = [header_line, next_line] cursor = index + 2 crossed_blank_line = False while cursor < len(lines): candidate = lines[cursor].strip() if not candidate: next_table_index = _next_non_empty_customer_line_index(lines, cursor + 1) if next_table_index >= 0 and _is_customer_markdown_pipe_line(lines[next_table_index].strip()): crossed_blank_line = True cursor += 1 continue break if not _is_customer_markdown_pipe_line(candidate): break segment.append(candidate) cursor += 1 reconstructed = _reconstruct_fragmented_customer_markdown_table(segment) if reconstructed and (crossed_blank_line or reconstructed != "\n".join(segment)): if lead: out.append(lead) out.append(reconstructed) index = cursor continue out.append(lines[index]) index += 1 return "\n".join(out).strip() def _repair_numbered_customer_markdown_table_continuations(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value or "---" not in value: return value lines = value.split("\n") out: list[str] = [] index = 0 while index < len(lines): current = lines[index].strip() next_line = lines[index + 1].strip() if index + 1 < len(lines) else "" if not _is_customer_markdown_pipe_line(current) or not _is_customer_markdown_table_separator_line(next_line): out.append(lines[index]) index += 1 continue header = _customer_markdown_pipe_cells(current, preserve_internal_empty=False) separators = _customer_markdown_pipe_cells(next_line, preserve_internal_empty=False) column_count = len(header) if column_count < 2 or len(separators) != column_count: out.append(lines[index]) index += 1 continue segment: list[str] = [] cursor = index + 2 saw_bare_continuation = False while cursor < len(lines): candidate = lines[cursor].strip() if not candidate: next_table_index = _next_non_empty_customer_line_index(lines, cursor + 1) next_candidate = lines[next_table_index].strip() if next_table_index >= 0 else "" if next_candidate and ( _is_customer_markdown_pipe_line(next_candidate) or next_candidate.startswith("|") or _is_numbered_customer_table_continuation_line(next_candidate) or (saw_bare_continuation and not _looks_like_new_customer_markdown_block(next_candidate)) ): segment.append("") cursor += 1 continue break if _is_standalone_customer_markdown_heading(candidate): break if _is_customer_markdown_pipe_line(candidate) or (saw_bare_continuation and candidate.startswith("|")): segment.append(candidate) cursor += 1 continue if _is_numbered_customer_table_continuation_line(candidate) or (segment and "|" in candidate): saw_bare_continuation = True segment.append(candidate) cursor += 1 continue if saw_bare_continuation and segment and not _looks_like_new_customer_markdown_block(candidate): segment.append(candidate) cursor += 1 continue break if not saw_bare_continuation or not segment: out.append(lines[index]) index += 1 continue repaired = _reconstruct_numbered_customer_continuation_table(header, separators, segment) if not repaired: out.append(lines[index]) index += 1 continue out.append(repaired) index = cursor return "\n".join(out).strip() def _is_standalone_customer_markdown_heading(line: str) -> bool: value = str(line or "").strip() return bool(re.fullmatch(r"#{1,6}\s+\S.*", value) or re.fullmatch(r"\*\*[^*\n]{2,80}\*\*", value)) def _looks_like_new_customer_markdown_block(line: str) -> bool: value = str(line or "").strip() return _is_standalone_customer_markdown_heading(value) or bool(re.fullmatch(r"[-*+]\s+\S.*", value)) def _is_numbered_customer_table_continuation_line(line: str) -> bool: value = str(line or "").strip() return bool(re.match(r"^\|?\s*\d+[.)、]\s+\S", value) or re.match(r"^\|{2,}\s*\d+[.)、]\s+\S", value)) def _reconstruct_numbered_customer_continuation_table( header: list[str], separators: list[str], lines: list[str], ) -> str: column_count = len(header) normalized = "\n".join(lines) normalized = normalized.replace("||", "\n| ") normalized = re.sub(r"\n{2,}", "\n", normalized).strip() chunks = [ chunk.strip() for chunk in re.split(r"(?=(?:^|\n)\s*\|?\s*\d+[.)、]\s+\S)", normalized) if chunk.strip() ] rows: list[list[str]] = [] malformed = False for chunk in chunks: row = _customer_row_from_numbered_table_chunk(chunk, column_count) if not row: malformed = True continue if not chunk.strip().startswith("|") or len(_customer_markdown_pipe_cells(chunk, preserve_internal_empty=False)) != column_count: malformed = True rows.append(row) if not malformed or len(rows) < 2: return "" return "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) def _customer_row_from_numbered_table_chunk(chunk: str, column_count: int) -> list[str] | None: cleaned = re.sub(r"^\|+", "", str(chunk or "")) cleaned = re.sub(r"\|+$", "", cleaned).strip() cells = [_normalize_customer_markdown_table_cell(cell) for cell in cleaned.split("|")] cells = [cell for cell in cells if cell] if len(cells) >= column_count: return [*cells[: column_count - 1], "；".join(cells[column_count - 1 :])][:column_count] if len(cells) == column_count - 1 and column_count >= 3: split = _split_numbered_customer_table_lead_cell(cells[0]) if split: cells = [split[0], split[1], *cells[1:]] if len(cells) == column_count: return cells return None def _split_numbered_customer_table_lead_cell(value: str) -> tuple[str, str] | None: text = _normalize_customer_markdown_table_cell(value) match = re.match(r"^((?:\d+[.)、]\s*)?\*\*[^*]{2,60}\*\*)\s+(.+)$", text) if match: return match.group(1).strip(), match.group(2).strip() match = re.match(r"^(\d+[.)、]\s+[^，。；：:、|]{2,28}?)[ \t]+(.+)$", text) if match: return match.group(1).strip(), match.group(2).strip() return None def _next_non_empty_customer_line_index(lines: list[str], start: int) -> int: for index in range(start, len(lines)): if lines[index].strip(): return index return -1 def _is_customer_markdown_pipe_line(line: str) -> bool: value = str(line or "").strip() return value.startswith("|") and value.count("|") >= 2 def _is_customer_markdown_table_separator_line(line: str) -> bool: cells = _customer_markdown_pipe_cells(line, preserve_internal_empty=False) return len(cells) >= 2 and all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells) def _customer_markdown_pipe_cells(line: str, *, preserve_internal_empty: bool) -> list[str]: cells = [_normalize_customer_markdown_table_cell(cell) for cell in str(line or "").split("|")] if cells and cells[0] == "": cells = cells[1:] if cells and cells[-1] == "": cells = cells[:-1] if preserve_internal_empty: return cells return [cell for cell in cells if cell] def _normalize_customer_markdown_table_cell(value: str) -> str: cell = re.sub(r"", "；", str(value or ""), flags=re.IGNORECASE) cell = re.sub(r"\s+", " ", cell) cell = re.sub(r"；{2,}", "；", cell) return cell.strip() def _reconstruct_fragmented_customer_markdown_table(lines: list[str]) -> str: if len(lines) < 3: return "" if any(re.search(r"#{2,6}\s*\S", line) for line in lines[2:]): return "" header = _customer_markdown_pipe_cells(lines[0], preserve_internal_empty=False) separators = _customer_markdown_pipe_cells(lines[1], preserve_internal_empty=False) column_count = len(header) if column_count < 2 or len(separators) != column_count: return "" if not all(re.fullmatch(r":?-{3,}:?", cell) for cell in separators): return "" rows: list[list[str]] = [] buffer: list[str] = [] for line in lines[2:]: cells = _customer_markdown_pipe_cells(line, preserve_internal_empty=True) non_empty_cells = [cell for cell in cells if cell] if not non_empty_cells: continue can_use_whole_row = ( not buffer and len(cells) == column_count and bool(cells[0]) and len(non_empty_cells) >= max(2, column_count - 1) ) if can_use_whole_row: rows.append(cells) continue buffer.extend(non_empty_cells) while len(buffer) >= column_count: rows.append(buffer[:column_count]) buffer = buffer[column_count:] if not rows: return "" table = "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) trailing = " ".join(buffer).strip() return "\n\n".join(part for part in [table, trailing] if part).strip() def _repair_inline_customer_markdown_tables(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value: return value paragraphs = re.split(r"\n\s*\n+", value) repaired: list[str] = [] for paragraph in paragraphs: item = paragraph.strip() if not item: continue if "|" in item and "---" not in item and item.count("|") >= 4: recovered = _recover_malformed_customer_pipe_rows(item) if recovered: repaired.append(recovered) continue if "\n" in item or item.count("|") < 8 or not re.search(r"\|\s*:?-{3,}:?\s*\|", item): repaired.append(item) continue first_pipe = item.find("|") if first_pipe > 0: lead = item[:first_pipe].strip() if lead: repaired.append(lead) item = item[first_pipe:].strip() reconstructed = _reconstruct_inline_customer_markdown_table(item) if reconstructed: repaired.append(reconstructed) continue table = re.sub(r"\s*\|\|\s*", " |\n| ", item) table = re.sub(r"\s+(\|\s*:?-{3,}:?\s*\|)", r"\n\1", table) table = re.sub(r"(\|\s*:?-{3,}:?\s*\|[ \t]*(?:\|[ \t]*:?-{3,}:?[ \t]*)*)\s+\|", r"\1\n|", table) table = re.sub(r"\|\s+(:?-{3,}:?)", r"|\1", table) table = re.sub(r"(:?-{3,}:?)\s+\|", r"\1|", table) repaired.append("\n".join(line.strip() for line in table.splitlines() if line.strip())) return "\n\n".join(repaired).strip() def _recover_malformed_customer_pipe_rows(item: str) -> str: first_pipe = item.find("|") lead = item[:first_pipe].strip() if first_pipe > 0 else "" body = item[first_pipe:] if first_pipe >= 0 else item cells, trailing = _split_customer_heading_from_pipe_cells(body) rows = _customer_rows_from_pipe_cells(cells) if not rows: return "" table = "\n".join( [ "| 优先级 | 项目 | 说明 |", "| :--- | :--- | :--- |", *[f"| {row['priority']} | {row['title']} | {row['summary']} |" for row in rows], ] ) return "\n\n".join(part for part in [lead, table, trailing] if part).strip() def _split_customer_heading_from_pipe_cells(body: str) -> tuple[list[str], str]: cells = [cell.strip() for cell in str(body or "").split("|")] cells = [cell for cell in cells if cell] heading_index = next((index for index, cell in enumerate(cells) if re.search(r"#{2,6}\s*\S", cell)), -1) if heading_index < 0: return cells, "" heading = re.sub(r"^#{2,6}\s*", "", cells[heading_index]).strip() return cells[:heading_index], _format_customer_markdown_heading_body(heading) if heading else "" def _customer_rows_from_pipe_cells(cells: list[str]) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] priority_re = re.compile(r"^(?:P\d+|高|中|低|High|Medium|Low)$", re.IGNORECASE) index = 0 while index + 2 < len(cells): if priority_re.fullmatch(cells[index]): rows.append({"priority": cells[index], "title": cells[index + 1], "summary": cells[index + 2]}) index += 3 continue if priority_re.fullmatch(cells[index + 2]): rows.append({"priority": cells[index + 2], "title": cells[index], "summary": cells[index + 1]}) index += 3 continue break return [row for row in rows if row["title"] and row["summary"]] def _reconstruct_inline_customer_markdown_table(item: str) -> str: cells = [cell.strip() for cell in str(item or "").split("|")] cells = [cell for cell in cells if cell] sep_index = next((index for index, cell in enumerate(cells) if re.fullmatch(r":?-{3,}:?", cell)), -1) if sep_index < 2: return "" column_count = sep_index separators = cells[sep_index : sep_index + column_count] if len(separators) < column_count or not all(re.fullmatch(r":?-{3,}:?", cell) for cell in separators): return "" header = cells[:column_count] rest = cells[sep_index + column_count :] rows = [rest[index : index + column_count] for index in range(0, len(rest), column_count)] rows = [row for row in rows if len(row) == column_count and any(cell for cell in row)] if not rows: return "" return "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) def _repair_inline_customer_markdown_lists(text: str) -> str: value = str(text or "").strip() if not value: return "" value = re.sub(r"([：:])\s+((?:[-*+–—]|\d+[.)、])\s+)", r"\1\n\n\2", value) value = re.sub(r"([。！？；）)])\s*((?:[-*+–—]|\d+[.)、])\s+)", r"\1\n\n\2", value) value = re.sub(r"(? str: value = str(text or "").strip() if not value: return "" value = re.sub(r"^(?![ \t]*#{2,6}\s)([^。！？\n|]{3,36}?)[ \t]+[—–-][ \t]+(?=\S{2,24}[：:])", r"**\1**\n\n", value) value = re.sub(r"([。！？])[ \t]+([^。！？\n|]{3,36}?)[ \t]+[—–-][ \t]+(?=\S{2,24}[：:])", r"\1\n\n**\2**\n\n", value) return re.sub(r"\n{3,}", "\n\n", value).strip() def _repair_raw_customer_markdown_headings(text: str) -> str: value = str(text or "").replace(r"\#", "#") value = re.sub(r"([^\n])\s+(#{2,6}\s+\S)", r"\1\n\n\2", value) return re.sub( r"^[ \t]*#{2,6}\s+(.+)$", lambda match: _format_customer_markdown_heading_body(match.group(1)), value, flags=re.MULTILINE, ).strip() def _format_customer_markdown_heading_body(value: str) -> str: body = re.sub(r"^#{2,6}\s+", "", str(value or "").strip()) if not body: return "" separator = re.search(r"\s+[—–-]\s+", body) if not separator: return f"**{body}**" title = body[: separator.start()].strip() rest = body[separator.end() :].strip() if not title: return rest if not rest: return f"**{title}**" return f"**{title}**\n\n{rest}" def _sentence_count(text: str) -> int: return len([item for item in re.split(r"[。！？!?]\s*", str(text or "")) if item.strip()]) def _split_dense_customer_answer_paragraph(paragraph: str) -> list[str]: value = re.sub(r"[ \t]+", " ", str(paragraph or "").strip()) if not value: return [] if len(value) <= 420 and _sentence_count(value) < 4: return [value] parts = [item.strip() for item in re.split(r"(?<=[。！？!?])\s*", value) if item.strip()] if len(parts) <= 1: return [value] out: list[str] = [] current: list[str] = [] current_len = 0 for part in parts: current.append(part) current_len += len(part) if current_len >= 280 or len(current) >= 2: out.append("".join(current).strip()) current = [] current_len = 0 if current: out.append("".join(current).strip()) return out or [value] _CUSTOMER_FIELD_REPLACEMENTS = { "projects": "项目列表", "team_members.assigned_projects": "成员项目关联", "team_members": "成员信息", "assigned_projects": "关联项目", "tenant_id": "空间编号", "project_id": "项目编号", "source_strategy": "来源策略", "primary_source_policy": "主要来源", } _RAW_INTERNAL_EVIDENCE_LINE_RE = re.compile( r"^(依据|根据).*(字段|数组|快照|上下文|明确列出|一致指向|无其他项目|出现在).*$", re.IGNORECASE, ) _RAW_SOURCE_LINE_RE = re.compile( # semantic-lint: ignore r"^来源[：:].*(snapshot|字段|数组|projects|team_members|assigned_projects|当前用户状态|当前空间状态).*$", re.IGNORECASE, ) def _sanitize_customer_answer_internal_terms(text: str) -> str: lines: list[str] = [] for raw_line in str(text or "").splitlines(): line = raw_line.rstrip() compact = line.strip() if compact and (_RAW_INTERNAL_EVIDENCE_LINE_RE.search(compact) or _RAW_SOURCE_LINE_RE.search(compact)): continue line = _replace_customer_internal_fields(line) line = _replace_customer_internal_status(line) lines.append(line) cleaned = "\n".join(lines) cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip() return cleaned or str(text or "").strip() def _replace_customer_internal_fields(text: str) -> str: out = str(text or "") for raw, display in sorted(_CUSTOMER_FIELD_REPLACEMENTS.items(), key=lambda item: len(item[0]), reverse=True): out = re.sub(rf"`?{re.escape(raw)}`?", display, out) return out def _replace_customer_internal_status(text: str) -> str: out = str(text or "") out = re.sub(r"状态[：:]\s*active", "状态：进行中", out, flags=re.IGNORECASE) out = re.sub(r"状态[：:]\s*archived", "状态：已归档", out, flags=re.IGNORECASE) out = re.sub(r"状态[：:]\s*draft", "状态：草稿", out, flags=re.IGNORECASE) return out def _customer_source_label(source: Any, label: Any) -> str: source_text = str(source or "").strip().lower() label_text = str(label or "").strip() label_lower = label_text.lower() if source_text in {"project_status", "project_operating", "project_execution"}: return "项目运行态证据" if source_text in {"internal_projects", "project_object_resolver"}: return "项目数据" if source_text in {"url", "mcp_search", "web", "external"} or label_lower.startswith(("url[", "http")): return "外部公开来源" if source_text == "doc" or "document" in label_lower: return "知识库文档" if source_text == "feishu" or "feishu" in label_lower: return "飞书记录" if source_text == "graph" or "knowledge graph" in label_lower: return "项目记忆" if source_text in {"db", "pmflow"} or "project database" in label_lower or "pmflow" in label_lower: return "项目数据" if source_text == "system" or any(token in label_lower for token in ("system snapshot", "user snapshot", "session state")): return "当前空间状态" if label_text in {"个人空间", "PMFlow 产品说明"}: return label_text return "" def _safe_customer_metadata(value: Any) -> dict[str, Any]: raw = _normalize_optional_dict(value) allowed = {} for key in ("target_url", "url", "updated_at", "project_id", "source_type"): if raw.get(key) not in (None, "", [], {}): allowed[key] = raw.get(key) return allowed def _sanitize_customer_sources(value: Any) -> list[dict[str, Any]]: rows = _normalize_dict_list(value) if not rows: return [] out: list[dict[str, Any]] = [] seen: set[str] = set() for item in rows: source = str(item.get("source") or "").strip().lower() or "pmflow" label = _customer_source_label(source, item.get("label") or item.get("display_label")) if not label: continue key = f"{source}:{label}" if key in seen: continue seen.add(key) out.append( { "source": "external" if label == "外部公开来源" else ("pmflow" if source in {"system", "db", "graph"} else source), "label": label, "score": _safe_score(item.get("score")), "metadata": _safe_customer_metadata(item.get("metadata")), } ) return out[:6] def _safe_score(value: Any) -> float: try: return round(max(0.0, min(1.0, float(value))), 4) except Exception: return 1.0 def _sanitize_customer_sources_used(value: Any) -> list[dict[str, Any]]: rows = _normalize_dict_list(value) out: list[dict[str, Any]] = [] seen: set[str] = set() for item in rows: label = _customer_source_label(item.get("source") or item.get("source_type"), item.get("display_label") or item.get("label") or item.get("raw_source_label")) if not label: continue row = {"display_label": label} url = item.get("target_url") or item.get("url") dedupe_key = str(url or label) if dedupe_key in seen: continue seen.add(dedupe_key) if url: row["target_url"] = str(url) out.append(row) return out[:6] def _build_customer_research_contract(*, request: RAGAskRequest, payload: dict[str, Any]) -> dict[str, Any]: route_decision = _normalize_optional_dict(payload.get("route_decision")) execution_recipe = _normalize_optional_dict(payload.get("execution_recipe")) effective_lane = str( execution_recipe.get("primary_lane") or route_decision.get("effective_source_policy") or route_decision.get("route_mode") or "internal" ).strip().lower() thinking_mode = str(request.thinking_mode or "auto").strip().lower() route_plan = _normalize_optional_dict(payload.get("pmflow_ai_route_plan")) return { "contract_version": "customer_research.v1", "requested_mode": thinking_mode, "effective_lane": effective_lane if effective_lane in {"internal", "external", "plan"} else "internal", "source_strategy": str(route_plan.get("source_strategy") or "").strip().lower() or None, "hybrid_required": bool(route_plan.get("is_hybrid")), "external_sources_required": effective_lane == "external", "project_fit_required": bool(execution_recipe.get("fit_brief_required")) or effective_lane == "external", "foreground_wait_seconds": 10 if thinking_mode == "deep" or effective_lane == "external" else 4, "async_handoff": "supported_by_contract", "completion_surface": "ai_history", } def _build_external_project_fit_packet(payload: dict[str, Any]) -> dict[str, Any]: fit_brief = _normalize_optional_dict(payload.get("fit_brief")) recommendation = _normalize_optional_dict(payload.get("external_recommendation")) if not fit_brief and not recommendation: return {} return { "packet_version": "external_project_fit.v1", "project_summary": _clean_text(fit_brief.get("project_summary"), limit=300) or None, "repo_summary": _clean_text(fit_brief.get("repo_summary"), limit=300) or None, "current_capabilities": [ _clean_text(item, limit=120) for item in list(fit_brief.get("current_capabilities") or [])[:8] if _clean_text(item, limit=120) ], "constraints": [ _clean_text(item, limit=120) for item in list(fit_brief.get("constraints") or [])[:8] if _clean_text(item, limit=120) ], "external_summary": _clean_text(recommendation.get("external_summary"), limit=500) or None, "external_impact": _clean_text(recommendation.get("external_impact"), limit=500) or None, "external_action": _clean_text(recommendation.get("external_action"), limit=240) or None, "affected_signal_sources": [ _clean_text(item, limit=120) for item in list(recommendation.get("affected_signal_sources") or [])[:8] if _clean_text(item, limit=120) ], }