"""Context evidence absorption for PMFlow AI. This module turns already-selected thread, memory and topic context into answer-grade obligations. It is deliberately based on source role, recency, scope and generic semantic overlap rather than business-scene keyword routing. """ from __future__ import annotations import hashlib import re from typing import Any from ..schemas.pmflow_ai_agent_runtime import ( PMFlowAIAgentBoundary, PMFlowAIAgentClaim, PMFlowAIAgentWorkspace, clean_text, ) CONTEXT_EVIDENCE_PACKET_VERSION = "pmflow_ai_context_evidence_packet.v1" TOPIC_WORKSPACE_PACKET_VERSION = "pmflow_ai_topic_workspace_packet.v1" def build_context_evidence_packet( *, context_pack: dict[str, Any] | None, user_question: str, ) -> dict[str, Any]: """Build customer-safe context claims and obligations from a context pack.""" pack = context_pack if isinstance(context_pack, dict) else {} question = clean_text(user_question, limit=1200) rows = _context_rows(pack) scored = sorted( rows, key=lambda row: ( -_row_score(row=row, question=question), _source_rank(row.get("source_kind")), -int(row.get("message_index") or 0), ), ) claims: list[dict[str, Any]] = [] for index, row in enumerate(scored[:16], start=1): text = clean_text(row.get("text"), limit=1000) if not text: continue role = _semantic_role(row) claim_key = f"{row.get('source_ref')}:{text[:160]}" claim_id = f"ctxclaim_{hashlib.sha1(claim_key.encode('utf-8')).hexdigest()[:12]}" score = _row_score(row=row, question=question) claims.append( { "claim_id": claim_id, "semantic_role": role, "claim_text": text, "customer_value": _customer_value_for_role(role), "source_kind": str(row.get("source_kind") or "")[:80], "source_role": str(row.get("source_role") or "")[:40], "source_ref": str(row.get("source_ref") or "")[:160], "support_refs": [str(row.get("source_ref") or "")[:160]] if row.get("source_ref") else [], "relevance_score": round(float(score), 4), "required_for_answer": bool(index <= 6 and score > 0), "confidence": row.get("confidence") or ("high" if row.get("source_kind") == "thread_message" else "medium"), } ) if claims and not any(item.get("required_for_answer") for item in claims): claims[0]["required_for_answer"] = True obligations = [ { "obligation_id": f"ctxobl_{item['claim_id'].replace('ctxclaim_', '')}", "claim_id": item["claim_id"], "semantic_role": item["semantic_role"], "requirement": "The final answer must preserve this prior context unless it explicitly marks a narrower missing artifact boundary.", "source_ref": item.get("source_ref"), } for item in claims if item.get("required_for_answer") ] topic = build_topic_workspace_packet( user_question=question, context_claims=claims, context_pack=pack, ) conflicts = _context_conflict_boundaries(pack=pack, context_claims=claims) return { "packet_version": CONTEXT_EVIDENCE_PACKET_VERSION, "current_user_message": question, "claims": claims, "continuation_obligations": obligations[:8], "conflict_boundaries": conflicts[:8], "topic_workspace": topic, "policy": { "context_claims_are_answer_grade_evidence": True, "tool_empty_results_cannot_negate_prior_context": True, "visible_objects_are_candidates_until_explicitly_bound": True, "business_scene_keyword_routing": False, }, } def append_context_evidence_to_workspace( *, workspace: PMFlowAIAgentWorkspace, evidence_packet: dict[str, Any] | None, ) -> int: """Promote context evidence claims into the Agent Workspace claim ledger.""" packet = evidence_packet if isinstance(evidence_packet, dict) else {} rows = list(packet.get("claims") or []) if not rows: return 0 existing_refs = { ref for claim in list(getattr(workspace, "claim_ledger", []) or []) for ref in list(getattr(claim, "support_refs", []) or []) if ref } claims: list[PMFlowAIAgentClaim] = [] for row in rows[:16]: if not isinstance(row, dict): continue text = clean_text(row.get("claim_text"), limit=1000) source_ref = clean_text(row.get("source_ref"), limit=160) if not text or (source_ref and source_ref in existing_refs): continue semantic_role = clean_text(row.get("semantic_role"), limit=80) or "context_evidence" claims.append( PMFlowAIAgentClaim( claim_id=clean_text(row.get("claim_id"), limit=80) or "", claim_type=f"context_{semantic_role}"[:80], claim_text=text, customer_value=clean_text(row.get("customer_value"), limit=500), support_refs=list(row.get("support_refs") or ([source_ref] if source_ref else []))[:12], source_tool="context_evidence_absorber", source_kind=clean_text(row.get("source_kind"), limit=120), relevance_to_user_goal="direct" if bool(row.get("required_for_answer")) else "partial", target_match="direct" if bool(row.get("required_for_answer")) else "partial", time_match="recent" if row.get("source_kind") == "thread_message" else "unknown", authority_level="primary" if row.get("source_kind") == "thread_message" else "credible_secondary", claim_kind="substantive_fact", confidence=_claim_confidence(row), allowed_to_state=True, ) ) boundaries = [] for row in list(packet.get("conflict_boundaries") or [])[:8]: if not isinstance(row, dict): continue reason = clean_text(row.get("customer_safe_reason"), limit=800) if not reason: continue boundaries.append( PMFlowAIAgentBoundary( boundary_type=clean_text(row.get("boundary_type"), limit=80) or "context_boundary", blocked_claim=clean_text(row.get("blocked_claim"), limit=800), customer_safe_reason=reason, allowed_alternative=clean_text(row.get("allowed_alternative"), limit=800), support_refs=list(row.get("support_refs") or [])[:12], ) ) if claims: workspace.claim_ledger = [*list(getattr(workspace, "claim_ledger", []) or []), *claims][-80:] if boundaries: workspace.boundary_ledger = [*list(getattr(workspace, "boundary_ledger", []) or []), *boundaries][-40:] topic = packet.get("topic_workspace") if isinstance(packet.get("topic_workspace"), dict) else {} if topic: workspace.known_facts = [ *list(getattr(workspace, "known_facts", []) or []), { "type": "topic_workspace_context", "summary": clean_text(topic.get("topic_summary"), limit=800), "topic_signature": topic.get("topic_signature"), "current_objective": clean_text(topic.get("current_objective"), limit=500), }, ][-64:] return len(claims) def build_topic_workspace_packet( *, user_question: str, context_claims: list[dict[str, Any]], context_pack: dict[str, Any] | None = None, ) -> dict[str, Any]: pack = context_pack if isinstance(context_pack, dict) else {} topic_pack = _topic_pack_from_context(pack) seed_claims = [item for item in context_claims if isinstance(item, dict) and item.get("claim_text")][:8] seed_texts = [clean_text(user_question, limit=500), *[clean_text(item.get("claim_text"), limit=500) for item in seed_claims[:6]]] summary = clean_text(topic_pack.get("topic_summary") or _summarize_seed_texts(seed_texts), limit=1200) linked_memory_ids = [ str(item.get("source_ref") or "")[:80] for item in seed_claims if str(item.get("source_kind") or "") == "memory" and str(item.get("source_ref") or "").strip() ] linked_thread_refs = [ str(item.get("source_ref") or "")[:80] for item in seed_claims if str(item.get("source_kind") or "") == "thread_message" and str(item.get("source_ref") or "").strip() ] return { "packet_version": TOPIC_WORKSPACE_PACKET_VERSION, "topic_signature": _topic_signature(seed_texts), "topic_summary": summary, "current_objective": clean_text(user_question, limit=800), "known_assumptions": _claims_by_role(seed_claims, {"user_historical_goal", "semantic_memory"}, limit=5), "prior_proposals": _claims_by_role(seed_claims, {"assistant_prior_proposal"}, limit=6), "open_questions": _topic_text_list(topic_pack.get("open_questions"), limit=6), "monetization_hypothesis": clean_text(topic_pack.get("monetization_hypothesis"), limit=600), "next_actionable_package": clean_text(topic_pack.get("next_actionable_package"), limit=800), "linked_thread_refs": _dedupe(linked_thread_refs)[:12], "linked_run_ids": _topic_text_list(topic_pack.get("linked_run_ids"), limit=12), "linked_memory_ids": _dedupe(linked_memory_ids)[:12], "selection_policy": { "semantic_overlap_and_scope_not_title_equality": True, "topic_workspace_is_not_a_project_truth_state": True, }, } def context_evidence_obligation_packet(value: Any) -> dict[str, Any]: packet = value if isinstance(value, dict) else {} claims = [item for item in list(packet.get("claims") or []) if isinstance(item, dict)] obligations = [item for item in list(packet.get("continuation_obligations") or []) if isinstance(item, dict)] return { "packet_version": "pmflow_ai_context_obligation_summary.v1", "claim_count": len(claims), "required_claim_ids": [str(item.get("claim_id") or "")[:80] for item in claims if item.get("required_for_answer")][:12], "obligation_count": len(obligations), "topic_signature": str((packet.get("topic_workspace") or {}).get("topic_signature") or "")[:80] if isinstance(packet.get("topic_workspace"), dict) else "", } def _context_rows(pack: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] thread_context = pack.get("thread_context") if isinstance(pack.get("thread_context"), dict) else {} memory_context = pack.get("memory_context") if isinstance(pack.get("memory_context"), dict) else {} topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {} seen: set[str] = set() for item in [*list(thread_context.get("relevant_messages") or []), *list(thread_context.get("recent_messages") or [])]: if not isinstance(item, dict): continue text = clean_text(item.get("content") or item.get("summary"), limit=1200) source_ref = str(item.get("message_id") or f"thread:{item.get('message_index') or ''}")[:160] key = f"thread:{source_ref}:{text[:120]}" if not text or key in seen: continue seen.add(key) rows.append( { "source_kind": "thread_message", "source_role": str(item.get("role") or "")[:40], "source_ref": source_ref, "message_index": item.get("message_index") or 0, "text": text, "confidence": "high", } ) for item in list(memory_context.get("memories") or []): if not isinstance(item, dict): continue text = clean_text(item.get("summary"), limit=1200) source_ref = str(item.get("memory_id") or item.get("scope_ref") or "")[:160] key = f"memory:{source_ref}:{text[:120]}" if not text or key in seen: continue seen.add(key) rows.append( { "source_kind": "memory", "source_role": str(item.get("memory_scope") or "")[:40], "source_ref": source_ref, "message_index": 0, "text": text, "confidence": _confidence_from_score(item.get("confidence_score")), } ) for item in list(topic_context.get("topic_workspaces") or []): if not isinstance(item, dict): continue text = clean_text( "；".join( str(value) for value in ( item.get("topic_summary"), item.get("current_objective"), item.get("next_actionable_package"), ) if value ), limit=1200, ) source_ref = str(item.get("topic_id") or item.get("topic_signature") or "")[:160] key = f"topic:{source_ref}:{text[:120]}" if not text or key in seen: continue seen.add(key) rows.append( { "source_kind": "topic_workspace", "source_role": "topic", "source_ref": source_ref, "message_index": 0, "text": text, "confidence": _confidence_from_score(item.get("confidence_score")), } ) return rows def _semantic_role(row: dict[str, Any]) -> str: source_kind = str(row.get("source_kind") or "") if source_kind == "memory": return "semantic_memory" if source_kind == "topic_workspace": return "topic_workspace" role = str(row.get("source_role") or "").strip().lower() if role == "assistant": return "assistant_prior_proposal" if role == "user": return "user_historical_goal" return "thread_context" def _row_score(*, row: dict[str, Any], question: str) -> float: question_terms = _text_terms(question) text_terms = _text_terms(row.get("text")) overlap = len(question_terms & text_terms) union = len(question_terms | text_terms) or 1 score = overlap / union if row.get("source_kind") == "thread_message": score += 0.18 if row.get("source_kind") == "topic_workspace": score += 0.15 if row.get("source_kind") == "memory": score += 0.12 if str(row.get("source_role") or "").lower() == "assistant": score += 0.05 try: index = int(row.get("message_index") or 0) except Exception: index = 0 if index: score += min(0.08, index / 1000) return score def _context_conflict_boundaries(*, pack: dict[str, Any], context_claims: list[dict[str, Any]]) -> list[dict[str, Any]]: topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {} rows = list(topic_context.get("topic_workspaces") or []) if not context_claims or not rows: return [] return [] def _topic_pack_from_context(pack: dict[str, Any]) -> dict[str, Any]: topic_context = pack.get("topic_context") if isinstance(pack.get("topic_context"), dict) else {} rows = [item for item in list(topic_context.get("topic_workspaces") or []) if isinstance(item, dict)] return rows[0] if rows else {} def _claims_by_role(claims: list[dict[str, Any]], roles: set[str], *, limit: int) -> list[str]: return _dedupe( [ clean_text(item.get("claim_text"), limit=360) for item in claims if str(item.get("semantic_role") or "") in roles and clean_text(item.get("claim_text"), limit=360) ] )[:limit] def _topic_text_list(value: Any, *, limit: int) -> list[str]: rows = value if isinstance(value, list) else [] return _dedupe([clean_text(item, limit=360) for item in rows if clean_text(item, limit=360)])[:limit] def _summarize_seed_texts(values: list[str]) -> str: rows = _dedupe([clean_text(item, limit=360) for item in values if clean_text(item, limit=360)]) return "；".join(rows[:4])[:1200] def _topic_signature(values: list[str]) -> str: terms: list[str] = [] for value in values: terms.extend(sorted(_text_terms(value))) key_terms = _dedupe(terms)[:24] raw = "|".join(key_terms) or "|".join(clean_text(item, limit=80) for item in values[:3]) return f"topic_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}" def _text_terms(value: Any) -> set[str]: text = str(value or "").lower() alnum = {item for item in re.findall(r"[a-z0-9_]{2,}", text)} cjk = re.findall(r"[\u4e00-\u9fff]", text) cjk_terms = {"".join(cjk[index : index + 2]) for index in range(max(0, len(cjk) - 1))} return {item for item in [*alnum, *cjk_terms] if item} def _source_rank(value: Any) -> int: source = str(value or "") return {"thread_message": 0, "topic_workspace": 1, "memory": 2}.get(source, 9) def _customer_value_for_role(role: str) -> str: return { "assistant_prior_proposal": "保留上一轮已经给过用户的方案和判断，支持连续追问。", "user_historical_goal": "保留用户此前表达过的目标、约束和关注点。", "semantic_memory": "保留长期语义记忆，支持跨轮和跨线程连续性。", "topic_workspace": "保留持续主题工作区，避免新线程丢失同一主题进展。", }.get(role, "保留当前上下文中与本轮目标相关的可复用材料。") def _confidence_from_score(value: Any) -> str: try: score = float(value) except Exception: return "medium" if score >= 0.8: return "high" if score >= 0.45: return "medium" return "low" def _claim_confidence(row: dict[str, Any]) -> str: confidence = str(row.get("confidence") or "medium").strip() return confidence if confidence in {"high", "medium", "low", "unknown"} else "medium" def _dedupe(values: list[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for value in values: text = clean_text(value, limit=600) key = re.sub(r"\s+", "", text).lower() if not text or key in seen: continue seen.add(key) out.append(text) return out __all__ = [ "CONTEXT_EVIDENCE_PACKET_VERSION", "TOPIC_WORKSPACE_PACKET_VERSION", "append_context_evidence_to_workspace", "build_context_evidence_packet", "build_topic_workspace_packet", "context_evidence_obligation_packet", ]