"""Thread-level context pack builder for Space PMFlow AI. The agent runtime should not infer multi-turn continuity from product objects alone. This service builds a customer-safe context pack from the persisted thread transcript and uses generic text relevance to surface prior turns. """ from __future__ import annotations import re from datetime import datetime from typing import Any from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import PMFlowAIMessage, PMFlowAIThread class PMFlowAIThreadContextService: """Build durable, model-ready context from PMFlow AI thread messages.""" def __init__(self, db: AsyncSession, *, tenant_id: int) -> None: self.db = db self.tenant_id = int(tenant_id) async def build_context_pack( self, *, thread: PMFlowAIThread | None, user_question: str, recent_limit: int = 12, search_limit: int = 8, scan_limit: int = 80, ) -> dict[str, Any]: if thread is None or not getattr(thread, "id", None): return {} messages = await self._load_messages(thread=thread, limit=scan_limit) if not messages: return {} recent_messages = messages[-max(1, recent_limit):] relevant_messages = _rank_relevant_messages( question=user_question, messages=messages, limit=search_limit, ) return { "packet_version": "pmflow_ai_thread_context_pack.v1", "thread_id": str(thread.public_thread_id or ""), "message_count": len(messages), "context_policy": { "same_thread_history_is_primary_for_multi_turn_recall": True, "recent_messages_are_customer_safe_transcript": True, "relevant_messages_are_generic_text_retrieval_not_scene_keywords": True, "project_state_must_not_replace_thread_history": True, }, "recent_messages": [_message_for_context(item) for item in recent_messages], "relevant_messages": [_message_for_context(item) for item in relevant_messages], "rolling_summary": _rolling_summary_from_thread(thread), "thread_outline": _thread_outline(messages), } async def _load_messages(self, *, thread: PMFlowAIThread, limit: int) -> list[PMFlowAIMessage]: stmt = ( select(PMFlowAIMessage) .where( and_( PMFlowAIMessage.tenant_id == self.tenant_id, PMFlowAIMessage.thread_id == int(thread.id), ) ) .order_by(PMFlowAIMessage.message_index.desc(), PMFlowAIMessage.id.desc()) .limit(max(1, limit)) ) rows = list((await self.db.execute(stmt)).scalars().all()) return list(reversed(rows)) def slim_thread_context_pack(value: Any, *, recent_limit: int = 10, relevant_limit: int = 8) -> dict[str, Any]: """Return a prompt-safe projection of a thread context pack.""" if not isinstance(value, dict): return {} recent = [_slim_message(item) for item in list(value.get("recent_messages") or [])[:recent_limit]] relevant = [_slim_message(item) for item in list(value.get("relevant_messages") or [])[:relevant_limit]] outline = value.get("thread_outline") if isinstance(value.get("thread_outline"), dict) else {} result = { "packet_version": value.get("packet_version") or "pmflow_ai_thread_context_pack.v1", "thread_id": str(value.get("thread_id") or "")[:80], "message_count": _safe_int(value.get("message_count")), "context_policy": { "same_thread_history_is_primary_for_multi_turn_recall": True, "project_state_must_not_replace_thread_history": True, }, "recent_messages": [item for item in recent if item], "relevant_messages": [item for item in relevant if item], "thread_outline": { "first_user_message": str(outline.get("first_user_message") or "")[:500], "last_user_message": str(outline.get("last_user_message") or "")[:500], "last_assistant_message": str(outline.get("last_assistant_message") or "")[:700], }, } rolling_summary = value.get("rolling_summary") if isinstance(value.get("rolling_summary"), dict) else {} if rolling_summary: result["rolling_summary"] = { "summary_version": str(rolling_summary.get("summary_version") or "pmflow_ai_thread_summary.v1")[:80], "covered_message_count": _safe_int(rolling_summary.get("covered_message_count")), "durable_points": [ { "kind": str(item.get("kind") or "")[:40], "summary": str(item.get("summary") or "")[:700], "source_message_indexes": list(item.get("source_message_indexes") or [])[:6], } for item in list(rolling_summary.get("durable_points") or [])[:16] if isinstance(item, dict) and str(item.get("summary") or "").strip() ], } return {key: item for key, item in result.items() if item not in ("", [], {}, None)} def update_thread_context_summary_metadata( *, thread_metadata: Any, user_message: str, assistant_answer: str, user_message_index: int, assistant_message_index: int, ) -> dict[str, Any]: """Append a compact durable turn summary into thread metadata.""" metadata = dict(thread_metadata or {}) if isinstance(thread_metadata, dict) else {} existing = metadata.get("thread_context_summary") if isinstance(metadata.get("thread_context_summary"), dict) else {} points = [dict(item) for item in list(existing.get("durable_points") or []) if isinstance(item, dict)] user_text = _clean_text(user_message, limit=500) answer_text = _clean_text(assistant_answer, limit=900) if user_text: points.append( { "kind": "user_turn", "summary": user_text, "source_message_indexes": [int(user_message_index)], } ) if answer_text: points.append( { "kind": "assistant_turn", "summary": answer_text, "source_message_indexes": [int(assistant_message_index)], } ) summary = { "summary_version": "pmflow_ai_thread_summary.v1", "covered_message_count": max( _safe_int(existing.get("covered_message_count")), int(assistant_message_index or 0), ), "durable_points": _dedupe_summary_points(points)[-24:], "updated_at": datetime.utcnow().isoformat(), } metadata["thread_context_summary"] = summary return metadata def thread_context_claim_rows(value: Any, *, limit: int = 8) -> list[dict[str, str]]: """Project relevant thread messages as generic claim rows for answer runtime.""" pack = slim_thread_context_pack(value, recent_limit=6, relevant_limit=limit) rows: list[dict[str, str]] = [] seen: set[str] = set() for item in [ *list(pack.get("relevant_messages") or []), *list(pack.get("recent_messages") or [])[-4:], ]: if not isinstance(item, dict): continue content = str(item.get("content") or "").strip() if not content: continue key = f"{item.get('message_index')}:{item.get('role')}:{content[:80]}" if key in seen: continue seen.add(key) rows.append( { "message_id": str(item.get("message_id") or "")[:80], "role": str(item.get("role") or "")[:24], "message_index": str(item.get("message_index") or "")[:20], "content": content[:900], } ) if len(rows) >= limit: break return rows def _message_for_context(row: PMFlowAIMessage) -> dict[str, Any]: content = _clean_text(getattr(row, "content", "") or "", limit=1800) return { "message_id": str(getattr(row, "public_message_id", "") or "")[:80], "role": str(getattr(row, "role", "") or "")[:24], "message_index": _safe_int(getattr(row, "message_index", None)), "run_id": str(getattr(row, "run_public_id", "") or "")[:80], "sent_at": _iso(getattr(row, "sent_at", None)), "content": content, } def _slim_message(value: Any) -> dict[str, Any]: if not isinstance(value, dict): return {} content = _clean_text(value.get("content") or "", limit=1200) if not content: return {} return { "message_id": str(value.get("message_id") or "")[:80], "role": str(value.get("role") or "")[:24], "message_index": _safe_int(value.get("message_index")), "content": content, } def _rank_relevant_messages(*, question: str, messages: list[PMFlowAIMessage], limit: int) -> list[PMFlowAIMessage]: query_tokens = _semantic_tokens(question) if not query_tokens: return [] scored: list[tuple[float, int, PMFlowAIMessage]] = [] total = max(1, len(messages)) for index, row in enumerate(messages): content = str(getattr(row, "content", "") or "") tokens = _semantic_tokens(content) if not tokens: continue overlap = len(query_tokens.intersection(tokens)) if overlap <= 0: continue density = overlap / max(8, len(query_tokens)) recency = (index + 1) / total role_boost = 0.12 if str(getattr(row, "role", "") or "") == "assistant" else 0.0 scored.append((density + recency * 0.08 + role_boost, index, row)) scored.sort(key=lambda item: (item[0], item[1]), reverse=True) return [row for _, _, row in scored[: max(1, limit)]] def _semantic_tokens(value: str) -> set[str]: text = _clean_text(value, limit=4000).lower() if not text: return set() tokens = set(re.findall(r"[a-z0-9][a-z0-9_+-]{1,}|[\u4e00-\u9fff]{2,}", text)) for chunk in re.findall(r"[\u4e00-\u9fff]{3,}", text): tokens.update(chunk[index : index + 2] for index in range(0, max(0, len(chunk) - 1))) tokens.update(chunk[index : index + 3] for index in range(0, max(0, len(chunk) - 2))) return {token for token in tokens if len(token) >= 2} def _thread_outline(messages: list[PMFlowAIMessage]) -> dict[str, Any]: first_user = next((row for row in messages if str(getattr(row, "role", "") or "") == "user"), None) user_rows = [row for row in messages if str(getattr(row, "role", "") or "") == "user"] assistant_rows = [row for row in messages if str(getattr(row, "role", "") or "") == "assistant"] return { "first_user_message": _clean_text(getattr(first_user, "content", "") if first_user else "", limit=500), "last_user_message": _clean_text(getattr(user_rows[-1], "content", "") if user_rows else "", limit=500), "last_assistant_message": _clean_text(getattr(assistant_rows[-1], "content", "") if assistant_rows else "", limit=700), } def _rolling_summary_from_thread(thread: PMFlowAIThread) -> dict[str, Any]: metadata = getattr(thread, "thread_metadata", None) if not isinstance(metadata, dict): return {} summary = metadata.get("thread_context_summary") return dict(summary) if isinstance(summary, dict) else {} def _dedupe_summary_points(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] seen: set[str] = set() for item in rows: summary = _clean_text(item.get("summary") if isinstance(item, dict) else "", limit=900) if not summary: continue key = f"{item.get('kind')}:{summary[:120].lower()}" if key in seen: continue seen.add(key) out.append( { "kind": str(item.get("kind") or "turn")[:40], "summary": summary, "source_message_indexes": [ _safe_int(value) for value in list(item.get("source_message_indexes") or [])[:6] if _safe_int(value) > 0 ], } ) return out def _clean_text(value: Any, *, limit: int) -> str: text = " ".join(str(value or "").replace("\x00", " ").split()) return text[: max(0, limit)] def _safe_int(value: Any) -> int: try: return int(value) except Exception: return 0 def _iso(value: Any) -> str | None: if isinstance(value, datetime): return value.isoformat() return None __all__ = [ "PMFlowAIThreadContextService", "slim_thread_context_pack", "thread_context_claim_rows", "update_thread_context_summary_metadata", ]