"""Durable semantic topic workspaces for PMFlow AI continuity.""" from __future__ import annotations import hashlib import re from datetime import datetime from typing import Any from sqlalchemy import and_, desc, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import PMFlowAITopicWorkspace, PMFlowAIThread class PMFlowAITopicWorkspaceService: """Persist and retrieve cross-thread topic continuity without title matching.""" def __init__(self, db: AsyncSession, *, tenant_id: int) -> None: self.db = db self.tenant_id = int(tenant_id or 1) async def build_topic_context_pack( self, *, current_user: Any, user_question: str, thread_context_pack: dict[str, Any] | None, limit: int = 4, ) -> dict[str, Any]: user_id = _safe_int(getattr(current_user, "id", None)) if user_id is None: return {} result = await self.db.execute( select(PMFlowAITopicWorkspace) .where( and_( PMFlowAITopicWorkspace.tenant_id == self.tenant_id, PMFlowAITopicWorkspace.user_id == int(user_id), PMFlowAITopicWorkspace.status == "active", ) ) .order_by(desc(PMFlowAITopicWorkspace.last_activated_at), desc(PMFlowAITopicWorkspace.id)) .limit(24) ) rows = list(result.scalars().all()) seed = _retrieval_seed(user_question=user_question, thread_context_pack=thread_context_pack) scored = [ (row, _semantic_score(seed, _topic_row_text(row))) for row in rows ] selected = [ row for row, score in sorted(scored, key=lambda pair: (-pair[1], pair[0].id or 0)) if score > 0 or not seed ][: max(1, min(8, int(limit or 4)))] if not selected: return {} return { "packet_version": "pmflow_ai_topic_context_pack.v1", "topic_workspaces": [topic_workspace_packet(row) for row in selected], "retrieval_policy": { "semantic_overlap_with_question_and_thread_context": True, "title_equality_required": False, "business_keyword_list_used": False, }, } async def upsert_from_turn( self, *, thread: PMFlowAIThread | None, current_user: Any, user_question: str, assistant_answer: str, context_evidence_packet: dict[str, Any] | None, run_public_id: str | None, ) -> dict[str, Any] | None: user_id = _safe_int(getattr(current_user, "id", None)) if user_id is None: return None evidence = context_evidence_packet if isinstance(context_evidence_packet, dict) else {} topic = evidence.get("topic_workspace") if isinstance(evidence.get("topic_workspace"), dict) else {} seed_text = _clean_text( "；".join( item for item in [ str(topic.get("topic_summary") or ""), str(topic.get("current_objective") or ""), str(user_question or ""), str(assistant_answer or "")[:1200], ] if item ), limit=2600, ) if not seed_text: return None row = await self._best_existing_topic(user_id=user_id, seed_text=seed_text) now = datetime.utcnow() public_thread_id = str(getattr(thread, "public_thread_id", "") or "")[:80] if row is None: row = PMFlowAITopicWorkspace( tenant_id=self.tenant_id, public_topic_id=f"topic_{hashlib.sha1(f'{user_id}:{seed_text[:500]}'.encode('utf-8')).hexdigest()[:16]}", user_id=int(user_id), workspace_scope="space", status="active", topic_signature=str(topic.get("topic_signature") or _topic_signature(seed_text))[:80], topic_summary=_clean_text(topic.get("topic_summary") or seed_text, limit=1600), current_objective=_clean_text(topic.get("current_objective") or user_question, limit=1200), known_assumptions=_safe_text_list(topic.get("known_assumptions"), limit=12), prior_proposals=_safe_text_list(topic.get("prior_proposals"), limit=12), open_questions=_safe_text_list(topic.get("open_questions"), limit=12), monetization_hypothesis=_clean_text(topic.get("monetization_hypothesis"), limit=1000), next_actionable_package=_clean_text(topic.get("next_actionable_package") or assistant_answer, limit=1600), linked_thread_ids=[public_thread_id] if public_thread_id else [], linked_run_ids=[str(run_public_id)[:80]] if run_public_id else [], linked_memory_ids=_safe_text_list(topic.get("linked_memory_ids"), limit=24), confidence_score=0.62, topic_payload=_safe_topic_payload(evidence), last_activated_at=now, ) self.db.add(row) await self.db.flush() return topic_workspace_packet(row) row.topic_summary = _merge_text(row.topic_summary, topic.get("topic_summary") or seed_text, limit=1800) row.current_objective = _clean_text(user_question or topic.get("current_objective") or row.current_objective, limit=1200) row.known_assumptions = _merge_list(row.known_assumptions, topic.get("known_assumptions"), limit=16) row.prior_proposals = _merge_list(row.prior_proposals, topic.get("prior_proposals"), limit=16) row.open_questions = _merge_list(row.open_questions, topic.get("open_questions"), limit=16) row.monetization_hypothesis = _merge_text(row.monetization_hypothesis, topic.get("monetization_hypothesis"), limit=1000) row.next_actionable_package = _merge_text(row.next_actionable_package, topic.get("next_actionable_package") or assistant_answer, limit=1800) row.linked_thread_ids = _merge_list(row.linked_thread_ids, [public_thread_id] if public_thread_id else [], limit=24) row.linked_run_ids = _merge_list(row.linked_run_ids, [str(run_public_id)[:80]] if run_public_id else [], limit=24) row.linked_memory_ids = _merge_list(row.linked_memory_ids, topic.get("linked_memory_ids"), limit=32) row.confidence_score = max(float(row.confidence_score or 0.0), 0.68) row.topic_payload = _safe_topic_payload(evidence) row.last_activated_at = now await self.db.flush() return topic_workspace_packet(row) async def _best_existing_topic(self, *, user_id: int, seed_text: str) -> PMFlowAITopicWorkspace | None: result = await self.db.execute( select(PMFlowAITopicWorkspace) .where( and_( PMFlowAITopicWorkspace.tenant_id == self.tenant_id, PMFlowAITopicWorkspace.user_id == int(user_id), PMFlowAITopicWorkspace.status == "active", ) ) .order_by(desc(PMFlowAITopicWorkspace.last_activated_at), desc(PMFlowAITopicWorkspace.id)) .limit(24) ) rows = list(result.scalars().all()) if not rows: return None ranked = sorted(((row, _semantic_score(seed_text, _topic_row_text(row))) for row in rows), key=lambda pair: -pair[1]) best, score = ranked[0] return best if score >= 0.08 else None def topic_workspace_packet(row: PMFlowAITopicWorkspace) -> dict[str, Any]: return { "packet_version": "pmflow_ai_topic_workspace_record.v1", "topic_id": str(row.public_topic_id or "")[:80], "topic_signature": str(row.topic_signature or "")[:80], "status": str(row.status or "")[:40], "topic_summary": str(row.topic_summary or "")[:1600], "current_objective": str(row.current_objective or "")[:1200], "known_assumptions": list(row.known_assumptions or [])[:12], "prior_proposals": list(row.prior_proposals or [])[:12], "open_questions": list(row.open_questions or [])[:12], "monetization_hypothesis": str(row.monetization_hypothesis or "")[:1000], "next_actionable_package": str(row.next_actionable_package or "")[:1600], "linked_thread_ids": list(row.linked_thread_ids or [])[:24], "linked_run_ids": list(row.linked_run_ids or [])[:24], "linked_memory_ids": list(row.linked_memory_ids or [])[:24], "confidence_score": row.confidence_score, "last_activated_at": row.last_activated_at.isoformat() if row.last_activated_at is not None else None, } def _retrieval_seed(*, user_question: str, thread_context_pack: dict[str, Any] | None) -> str: rows: list[str] = [str(user_question or "")] pack = thread_context_pack if isinstance(thread_context_pack, dict) else {} for item in [*list(pack.get("relevant_messages") or [])[:6], *list(pack.get("recent_messages") or [])[-4:]]: if isinstance(item, dict): rows.append(str(item.get("content") or item.get("summary") or "")) return _clean_text("；".join(row for row in rows if row), limit=3000) def _topic_row_text(row: PMFlowAITopicWorkspace) -> str: return _clean_text( "；".join( str(value) for value in [ row.topic_summary, row.current_objective, row.known_assumptions, row.prior_proposals, row.open_questions, row.monetization_hypothesis, row.next_actionable_package, ] if value ), limit=4000, ) def _semantic_score(left: str, right: str) -> float: left_terms = _text_terms(left) right_terms = _text_terms(right) if not left_terms or not right_terms: return 0.0 return len(left_terms & right_terms) / max(1, min(len(left_terms), len(right_terms))) 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 _topic_signature(value: str) -> str: terms = sorted(_text_terms(value))[:24] raw = "|".join(terms) or value[:200] return f"topic_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}" def _merge_text(existing: Any, incoming: Any, *, limit: int) -> str: values = _dedupe_texts([str(existing or ""), str(incoming or "")]) return _clean_text("；".join(values), limit=limit) def _merge_list(existing: Any, incoming: Any, *, limit: int) -> list[str]: return _dedupe_texts([*_safe_text_list(existing, limit=limit), *_safe_text_list(incoming, limit=limit)])[:limit] def _safe_text_list(value: Any, *, limit: int) -> list[str]: rows = value if isinstance(value, list) else [] return _dedupe_texts([_clean_text(item, limit=500) for item in rows if _clean_text(item, limit=500)])[:limit] def _dedupe_texts(values: list[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for value in values: text = _clean_text(value, limit=700) key = re.sub(r"\s+", "", text).lower() if not text or key in seen: continue seen.add(key) out.append(text) return out def _safe_topic_payload(value: dict[str, Any]) -> dict[str, Any]: packet = value if isinstance(value, dict) else {} return { "packet_version": str(packet.get("packet_version") or "")[:80], "claim_count": len(list(packet.get("claims") or [])), "obligation_count": len(list(packet.get("continuation_obligations") or [])), "topic_workspace": packet.get("topic_workspace") if isinstance(packet.get("topic_workspace"), dict) else {}, } def _clean_text(value: Any, *, limit: int) -> str: return " ".join(str(value or "").replace("\x00", " ").split())[: max(0, int(limit or 0))] def _safe_int(value: Any) -> int | None: try: if value in (None, ""): return None return int(value) except Exception: return None __all__ = ["PMFlowAITopicWorkspaceService", "topic_workspace_packet"]