"""Evidence absorption for the PMFlow AI turn runtime. The absorber is the only layer that turns tool evidence into findings, implications, remaining gaps, and judgement state. Runtime owns the loop; tools only return evidence; renderer only expresses the final state. """ from __future__ import annotations from typing import Any from ..schemas.pmflow_ai_turn_runtime import ( EvidenceConfidence, PMFlowAIEvidenceAbsorptionResult, PMFlowAIEvidenceFinding, PMFlowAIJudgementState, PMFlowAIRuntimeAction, PMFlowAITurnState, ) class PMFlowAIEvidenceAbsorber: """Convert bounded tool payloads into TurnState evidence semantics.""" def absorb( self, *, turn_state: PMFlowAITurnState, action: PMFlowAIRuntimeAction, payload: dict[str, Any] | None, ) -> PMFlowAIEvidenceAbsorptionResult: if not isinstance(payload, dict) or _payload_is_final_answer(payload): return PMFlowAIEvidenceAbsorptionResult( judgement_state=_build_judgement_state(turn_state=turn_state, new_findings=[]), evidence_quality=turn_state.evidence_state.evidence_quality, ) tool_result = _dict(payload.get("runtime_evidence_tool_result")) findings = _tool_findings_from_payload( tool_result=tool_result, turn_state=turn_state, action=action, ) findings = _merge_findings( [ *findings, *_structured_findings_from_tool_result( tool_result=tool_result, turn_state=turn_state, action=action, ), ], limit=12, ) if not findings: findings = _fallback_findings_from_external_payload( payload=payload, turn_state=turn_state, action=action, ) if not findings: findings = _fallback_findings_from_customer_packet( payload=payload, turn_state=turn_state, action=action, ) sources = _payload_sources(payload) remaining_gaps = _unique( [ *_remaining_gaps(payload=payload, tool_result=tool_result), *_structured_remaining_gaps(tool_result=tool_result), ], limit=10, ) if _has_off_topic_tool_evidence(tool_result=tool_result, action=action, turn_state=turn_state) and not findings: remaining_gaps = _unique([*remaining_gaps, "off_topic_evidence_excluded"], limit=10) merged_state = turn_state.model_copy(deep=True) merged_state.evidence_state.known_facts.extend(findings) merged_state.evidence_state.structured_evidence = _merge_structured_evidence( merged_state.evidence_state.structured_evidence, _structured_evidence_from_tool_result(tool_result=tool_result), ) merged_state.evidence_state.stale_or_missing.extend(remaining_gaps) quality = _evidence_quality(findings=findings, payload=payload) return PMFlowAIEvidenceAbsorptionResult( findings=findings[:10], structured_evidence=_structured_evidence_from_tool_result(tool_result=tool_result), sources=sources, remaining_gaps=remaining_gaps, judgement_state=_build_judgement_state(turn_state=merged_state, new_findings=findings), evidence_quality=quality, summary=str(tool_result.get("summary") or _payload_summary(payload))[:500], ) def _tool_findings_from_payload( *, tool_result: dict[str, Any], turn_state: PMFlowAITurnState, action: PMFlowAIRuntimeAction, ) -> list[PMFlowAIEvidenceFinding]: findings: list[PMFlowAIEvidenceFinding] = [] for item in list(tool_result.get("findings") or [])[:12]: if not isinstance(item, dict): continue fact = _clean(item.get("fact"), limit=700) if not fact: continue if not _tool_finding_is_relevant(fact=fact, action=action, turn_state=turn_state): continue implication = _clean(item.get("implication"), limit=700) or _default_implication(turn_state=turn_state) findings.append( PMFlowAIEvidenceFinding( fact=fact, source=_clean(item.get("source") or tool_result.get("tool_name") or "PMFlow 证据工具", limit=240), implication=implication, relevance_to_goal=_clean(item.get("relevance_to_goal") or turn_state.turn_goal.user_goal, limit=160), confidence=_confidence(item.get("confidence") or tool_result.get("confidence")), ) ) return findings def _structured_findings_from_tool_result( *, tool_result: dict[str, Any], turn_state: PMFlowAITurnState, action: PMFlowAIRuntimeAction, ) -> list[PMFlowAIEvidenceFinding]: if action.action_type != "inspect_project_status": return [] packets = _project_operating_packets(tool_result=tool_result) findings: list[PMFlowAIEvidenceFinding] = [] for packet in packets[:4]: if not _project_packet_matches_primary(packet=packet, turn_state=turn_state): continue title = _clean(packet.get("project_title") or packet.get("project_ref") or "当前项目", limit=160) for key in ( "command_state", "workbench_state", "decision_review_state", "execution_state", "quality_test_state", "release_growth_state", "team_state", "project_memory_state", "risk_blocker_state", "recent_timeline_state", ): facet = _dict(packet.get(key)) summary = _clean(facet.get("summary"), limit=420) if not summary: continue label = _clean(facet.get("display_label") or key, limit=80) coverage = _clean(facet.get("coverage_status"), limit=40) findings.append( PMFlowAIEvidenceFinding( fact=f"{title} / {label}：{summary}", source="PMFlow 项目全流程证据包", implication=_project_operating_implication(label=label, coverage=coverage), relevance_to_goal=turn_state.turn_goal.user_goal, confidence="high" if coverage == "covered" else "medium", ) ) if len(findings) >= 8: break return findings def _structured_remaining_gaps(*, tool_result: dict[str, Any]) -> list[str]: gaps: list[str] = [] for packet in _project_operating_packets(tool_result=tool_result)[:4]: gaps.extend(str(item or "") for item in list(packet.get("remaining_gaps") or [])[:12]) return _unique(gaps, limit=10) def _project_operating_packets(*, tool_result: dict[str, Any]) -> list[dict[str, Any]]: structured = _dict(tool_result.get("structured_evidence")) rows = structured.get("project_operating_evidence_packets") if isinstance(rows, list): return [dict(item) for item in rows if isinstance(item, dict)] single = structured.get("project_operating_evidence_packet") return [dict(single)] if isinstance(single, dict) and single else [] def _structured_evidence_from_tool_result(*, tool_result: dict[str, Any]) -> dict[str, Any]: structured = _dict(tool_result.get("structured_evidence")) packets = _project_operating_packets(tool_result=tool_result) if packets: return { "project_operating_evidence_packets": packets, "project_operating_evidence_packet": packets[0] if len(packets) == 1 else {}, } return structured def _merge_structured_evidence(existing: dict[str, Any], incoming: dict[str, Any]) -> dict[str, Any]: merged = dict(existing or {}) incoming_packets = incoming.get("project_operating_evidence_packets") if isinstance(incoming_packets, list): existing_packets = merged.get("project_operating_evidence_packets") packets = [dict(item) for item in existing_packets if isinstance(item, dict)] if isinstance(existing_packets, list) else [] by_ref = { str(packet.get("project_ref") or packet.get("project_id") or index): packet for index, packet in enumerate(packets) } for packet in incoming_packets: if not isinstance(packet, dict): continue key = str(packet.get("project_ref") or packet.get("project_id") or len(by_ref)) by_ref[key] = dict(packet) merged_packets = list(by_ref.values())[:12] merged["project_operating_evidence_packets"] = merged_packets merged["project_operating_evidence_packet"] = merged_packets[0] if len(merged_packets) == 1 else {} for key, value in incoming.items(): if key not in {"project_operating_evidence_packets", "project_operating_evidence_packet"}: merged[key] = value return merged def _project_packet_matches_primary(*, packet: dict[str, Any], turn_state: PMFlowAITurnState) -> bool: primary = turn_state.object_state.primary_object if primary is None: return False project_ref = _clean(packet.get("project_ref"), limit=160) title = _clean(packet.get("project_title"), limit=160) return bool( (project_ref and project_ref == _clean(primary.object_ref, limit=160)) or (title and title == _clean(primary.title, limit=160)) ) def _project_operating_implication(*, label: str, coverage: str) -> str: if coverage == "missing": return f"{label}证据缺失，最终推进建议必须显式保留该边界。" if coverage == "partial": return f"{label}只拿到部分证据，适合给有限建议，不能假装已完成全量判断。" return f"{label}已形成可用证据，可参与当前项目推进判断。" def _merge_findings(values: list[PMFlowAIEvidenceFinding], *, limit: int) -> list[PMFlowAIEvidenceFinding]: out: list[PMFlowAIEvidenceFinding] = [] seen: set[str] = set() for item in values: key = _clean(item.fact, limit=700) if not key or key in seen: continue seen.add(key) out.append(item) if len(out) >= limit: break return out def _has_off_topic_tool_evidence( *, tool_result: dict[str, Any], action: PMFlowAIRuntimeAction, turn_state: PMFlowAITurnState, ) -> bool: return any( isinstance(item, dict) and _clean(item.get("fact"), limit=700) and not _tool_finding_is_relevant(fact=_clean(item.get("fact"), limit=700), action=action, turn_state=turn_state) for item in list(tool_result.get("findings") or [])[:12] ) def _tool_finding_is_relevant( *, fact: str, action: PMFlowAIRuntimeAction, turn_state: PMFlowAITurnState, ) -> bool: if action.action_type in {"external_search", "external_fetch", "inspect_saved_ideas", "resolve_project_objects"}: return True selected = turn_state.selected_frame operation = selected.operation_kind if selected is not None else "answer_general" if action.action_type == "inspect_surface_context": return operation == "list_collection" or _has_strong_project_primary(turn_state) if action.action_type in {"inspect_project_status", "inspect_internal_projects"}: return _has_strong_project_primary(turn_state) and _fact_mentions_primary(fact=fact, turn_state=turn_state) return True def _has_strong_project_primary(turn_state: PMFlowAITurnState) -> bool: primary = turn_state.object_state.primary_object if primary is None: return False binding = str(primary.binding_strength or "").strip().lower() if binding in {"", "none", "context", "visible_context"}: return False return str(primary.object_type or "").strip().lower() == "project" and bool(str(primary.object_ref or primary.title or "").strip()) def _fact_mentions_primary(*, fact: str, turn_state: PMFlowAITurnState) -> bool: primary = turn_state.object_state.primary_object if primary is None: return False title = _clean(primary.title, limit=160) ref = _clean(primary.object_ref, limit=160) return bool((title and title in fact) or (ref and ref in fact)) def _fallback_findings_from_external_payload( *, payload: dict[str, Any], turn_state: PMFlowAITurnState, action: PMFlowAIRuntimeAction, ) -> list[PMFlowAIEvidenceFinding]: if action.action_type not in {"external_search", "external_fetch"}: return [] findings: list[PMFlowAIEvidenceFinding] = [] for item in _external_evidence_items_from_payload(payload)[:8]: title = _clean(item.get("title") or item.get("source_label") or "外部公开来源", limit=240) snippet = _clean(item.get("snippet"), limit=500) url = _clean(item.get("target_url") or item.get("url"), limit=240) if not title and not snippet: continue fact = f"{title}：{snippet}".strip("：") findings.append( PMFlowAIEvidenceFinding( fact=fact, source=url or title or "外部公开来源", implication=_default_implication(turn_state=turn_state), relevance_to_goal=turn_state.turn_goal.user_goal, confidence="high" if url and snippet else "medium", ) ) return findings def _fallback_findings_from_customer_packet( *, payload: dict[str, Any], turn_state: PMFlowAITurnState, action: PMFlowAIRuntimeAction, ) -> list[PMFlowAIEvidenceFinding]: packet = _dict(payload.get("customer_evidence_packet")) findings: list[PMFlowAIEvidenceFinding] = [] for fact in list(packet.get("facts") or [])[:8]: if not isinstance(fact, dict): continue summary = _clean(fact.get("display_summary") or fact.get("display_label"), limit=700) if not summary: continue findings.append( PMFlowAIEvidenceFinding( fact=summary, source=_source_from_fact(fact=fact, action=action), implication=_customer_fact_implication(fact=fact, turn_state=turn_state), relevance_to_goal=turn_state.turn_goal.user_goal, confidence=_confidence(fact.get("confidence")), ) ) return findings def _build_judgement_state( *, turn_state: PMFlowAITurnState, new_findings: list[PMFlowAIEvidenceFinding], ) -> PMFlowAIJudgementState: findings = list(turn_state.evidence_state.known_facts or []) if not findings: return PMFlowAIJudgementState(confidence="unknown") implications = _unique([item.implication for item in findings if item.implication], limit=8) facts = _unique([item.fact for item in findings if item.fact], limit=8) gaps = _unique([*turn_state.evidence_state.stale_or_missing], limit=6) confidence: EvidenceConfidence = "medium" if any(item.confidence in {"high", "medium"} for item in findings) else "unknown" if new_findings and all(item.confidence == "high" for item in new_findings[:3]): confidence = "high" shape = turn_state.answer_state.answer_shape update: dict[str, Any] = {"confidence": confidence} if shape == "strategic_judgement": update.update( { "opportunity": _join_sentence(implications[:3]) or "已有证据可支撑方向判断，但仍需要把机会收敛到可验证切口。", "can_do": implications[:4] or facts[:4], "should_not_do": _gap_guidance(gaps), "better_angle": _strategic_better_angle(turn_state=turn_state, implications=implications), "mvp": _mvp_from_implications(implications=implications, facts=facts), "risks": _risk_from_gaps(gaps=gaps, findings=findings), "validation_questions": _validation_questions(turn_state=turn_state, gaps=gaps), "next_actions": _next_actions_for_shape(shape=shape), } ) elif shape in {"status_answer", "composite_answer", "fact_list"}: update.update( { "direct_answer": _join_sentence(implications[:3] or facts[:3]), "can_do": implications[:4] or facts[:4], "risks": _risk_from_gaps(gaps=gaps, findings=findings), "validation_questions": _validation_questions(turn_state=turn_state, gaps=gaps), "next_actions": _next_actions_for_shape(shape=shape), } ) elif shape == "execution_guidance": update.update( { "direct_answer": _join_sentence(implications[:3] or facts[:3]), "should_not_do": ["写入、提交、分派或执行类动作必须先经过确认边界。"], "risks": _risk_from_gaps(gaps=gaps, findings=findings), "validation_questions": _validation_questions(turn_state=turn_state, gaps=gaps), "next_actions": _next_actions_for_shape(shape=shape), } ) else: update.update( { "direct_answer": _join_sentence(implications[:3] or facts[:3]), "risks": _risk_from_gaps(gaps=gaps, findings=findings), "next_actions": _next_actions_for_shape(shape=shape), } ) return turn_state.judgement_state.model_copy(update=update) def _payload_sources(payload: dict[str, Any] | None) -> list[dict[str, Any]]: if not isinstance(payload, dict): return [] rows: list[dict[str, Any]] = [] tool_result = _dict(payload.get("runtime_evidence_tool_result")) tool_sources = tool_result.get("sources") if isinstance(tool_sources, list): rows.extend(item for item in tool_sources if isinstance(item, dict)) for key in ("sources_used", "sources", "citations"): value = payload.get(key) if isinstance(value, list): rows.extend(item for item in value if isinstance(item, dict)) return _dedupe_sources(rows)[:12] def _remaining_gaps(*, payload: dict[str, Any], tool_result: dict[str, Any]) -> list[str]: gaps: list[str] = [] for value in list(tool_result.get("remaining_gaps") or []): text = _clean(value, limit=180) if text: gaps.append(text) context_packet = _dict(payload.get("external_context_packet")) for key in ("remaining_gaps", "warnings"): value = context_packet.get(key) if isinstance(value, list): gaps.extend(_clean(item, limit=180) for item in value if _clean(item, limit=180)) return _unique(gaps, limit=10) def _payload_summary(payload: dict[str, Any] | None) -> str: if not isinstance(payload, dict): return "" canvas = _dict(payload.get("answer_canvas")) headline = _clean(canvas.get("headline"), limit=180) if headline: return headline return _clean(payload.get("answer") or payload.get("reply_text"), limit=180) def _payload_is_final_answer(payload: dict[str, Any] | None) -> bool: if not isinstance(payload, dict): return False kind = _clean(payload.get("_runtime_payload_kind"), limit=80) if kind == "evidence": return False if kind == "final_answer": return True return bool(_clean(payload.get("answer") or payload.get("reply_text"), limit=80)) def _external_evidence_items_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]: rows = payload.get("external_evidence_items") if isinstance(rows, list): return [dict(item) for item in rows if isinstance(item, dict)] context_packet = _dict(payload.get("external_context_packet")) verified = context_packet.get("verified_evidence") if isinstance(verified, list): return [dict(item) for item in verified if isinstance(item, dict)] return [] def _source_from_fact(*, fact: dict[str, Any], action: PMFlowAIRuntimeAction) -> str: labels = [_clean(item, limit=180) for item in list(fact.get("source_display_labels") or [])] labels = [item for item in labels if item] if labels: return labels[0] if action.action_type in {"external_search", "external_fetch"}: return "外部公开来源" if action.action_type in {"inspect_surface_context", "inspect_saved_ideas", "resolve_project_objects"}: return "当前空间状态" if action.action_type == "inspect_project_status": return "PMFlow 项目状态证据" return "PMFlow 项目证据" def _customer_fact_implication(*, fact: dict[str, Any], turn_state: PMFlowAITurnState) -> str: values = _dict(fact.get("business_values")) for value in values.values(): text = _clean(value, limit=500) if text: return text return _default_implication(turn_state=turn_state) def _default_implication(*, turn_state: PMFlowAITurnState) -> str: shape = turn_state.answer_state.answer_shape if shape == "strategic_judgement": return "这条证据用于判断机会窗口、定位边界、风险和最小验证切口。" if shape == "status_answer": return "这条证据用于判断当前状态、缺口、风险和下一步。" if shape == "execution_guidance": return "这条证据用于判断可执行动作、风险和确认边界。" if shape == "fact_list": return "这条证据用于对象归位和匹配依据。" return "这条证据用于支撑本轮回答。" def _strategic_better_angle(*, turn_state: PMFlowAITurnState, implications: list[str]) -> str: if implications: return implications[0][:500] subject = _clean(turn_state.problem_frame.subject, limit=120) or "这个方向" return f"{subject}应先收敛到可验证的客户、数据、场景和闭环指标，再扩展为完整方案。" def _mvp_from_implications(*, implications: list[str], facts: list[str]) -> list[str]: candidates = implications[:3] or facts[:3] if candidates: return candidates[:3] return ["先做最小证据闭环：输入、判断、证据、人工确认和结果回流。"] def _risk_from_gaps(*, gaps: list[str], findings: list[PMFlowAIEvidenceFinding]) -> list[str]: risks = list(gaps[:4]) if not risks and any(item.confidence in {"low", "unknown"} for item in findings): risks.append("部分证据置信度不足，不能把方向判断当成已验证结论。") if not risks: risks.append("仍需用真实用户、数据可得性和落地成本验证判断。") return risks[:5] def _gap_guidance(gaps: list[str]) -> list[str]: if gaps: return [f"不要把未补齐的缺口当成已验证结论：{item}" for item in gaps[:4]] return ["不要把外部趋势或内部可见对象直接等同于真实付费需求。"] def _validation_questions(*, turn_state: PMFlowAITurnState, gaps: list[str]) -> list[str]: if gaps: return gaps[:4] if turn_state.problem_frame.ambiguity_points: return turn_state.problem_frame.ambiguity_points[:4] return ["目标客户是谁", "可获得哪些关键数据", "最小可验证场景是什么"] def _next_actions_for_shape(*, shape: str) -> list[str]: if shape == "strategic_judgement": return ["收敛目标客户和高频场景", "补齐近期外部证据和竞品证据", "定义最小验证闭环"] if shape == "status_answer": return ["核对目标对象范围", "补齐缺失证据", "按风险和下一步排序处理"] if shape == "execution_guidance": return ["确认对象", "确认动作边界", "确认写入或执行风险"] return ["补齐关键证据", "给出可验证下一步"] def _evidence_quality(*, findings: list[PMFlowAIEvidenceFinding], payload: dict[str, Any]) -> EvidenceConfidence: if not findings: return "medium" if _payload_sources(payload) else "unknown" if any(item.confidence == "high" for item in findings): return "high" if any(item.confidence == "medium" for item in findings): return "medium" return "unknown" def _confidence(value: Any) -> EvidenceConfidence: normalized = str(value or "").strip().lower() if normalized in {"high", "medium", "low"}: return normalized # type: ignore[return-value] return "unknown" def _dict(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, dict) else {} def _clean(value: Any, *, limit: int) -> str: return " ".join(str(value or "").strip().split())[:limit] def _join_sentence(values: list[str]) -> str: return "；".join(_clean(item, limit=220) for item in values if _clean(item, limit=220))[:500] def _unique(values: list[str], *, limit: int) -> list[str]: out: list[str] = [] for value in values: text = _clean(value, limit=240) if text and text not in out: out.append(text) if len(out) >= limit: break return out def _dedupe_sources(values: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] seen: set[str] = set() for item in values: key = _clean(item.get("target_url") or item.get("url") or item.get("display_label") or item.get("source"), limit=240) if not key or key in seen: continue seen.add(key) out.append(item) return out __all__ = ["PMFlowAIEvidenceAbsorber"]