"""Layered semantic memory and compaction for Space PMFlow AI. The service keeps long-context behavior on a product/runtime contract: thread, user, workspace and project memories are selected by explicit scope anchors and generic relevance, not by business-scene keyword branches. """ from __future__ import annotations import json import re import uuid from datetime import datetime from typing import Any from sqlalchemy import and_, desc, select from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR from ..models import PMFlowAIContextMemory, PMFlowAIThread from .ai_function_runtime_profile_service import AIFunctionRuntimeProfileService from .llm_service import llm_service from .prompt_runtime_service import PromptRuntimeRenderResult, PromptRuntimeService from .runtime_audit_context import bind_agent_audit_context PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR = "PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR" FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR = "pmflow_ai_context_memory_compactor" SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME = "space_pmflow_ai_agent_runtime" MEMORY_KEY_CURRENT = "semantic_compaction.current" MEMORY_PACKET_VERSION = "pmflow_ai_layered_memory_pack.v1" _MEMORY_SCOPES = ("thread", "user", "workspace", "project") _ACTIVE_MEMORY_STATUSES = ("active", "candidate") class PMFlowAIContextMemoryService: """Build and update layered semantic memories for PMFlow AI.""" def __init__(self, db: AsyncSession, *, tenant_id: int, generator: Any | None = None) -> None: self.db = db self.tenant_id = int(tenant_id or 1) self.generator = generator or llm_service async def build_memory_context_pack( self, *, thread: PMFlowAIThread | None, current_user: Any | None = None, conversation_context: dict[str, Any] | None = None, tenant_project_context: dict[str, Any] | None = None, user_question: str = "", limit: int = 12, ) -> dict[str, Any]: """Load relevant long-context memory rows for the current turn.""" scope_refs = _scope_refs( thread=thread, current_user=current_user, conversation_context=conversation_context, tenant_project_context=tenant_project_context, ) if not scope_refs or not callable(getattr(self.db, "execute", None)): return {} rows: list[PMFlowAIContextMemory] = [] for scope, refs in scope_refs.items(): if not refs: continue result = await self.db.execute( select(PMFlowAIContextMemory) .where( and_( PMFlowAIContextMemory.tenant_id == self.tenant_id, PMFlowAIContextMemory.memory_scope == scope, PMFlowAIContextMemory.scope_ref.in_(refs[:6]), PMFlowAIContextMemory.status.in_(_ACTIVE_MEMORY_STATUSES), ) ) .order_by( desc(PMFlowAIContextMemory.confidence_score), desc(PMFlowAIContextMemory.last_validated_at), desc(PMFlowAIContextMemory.last_compacted_at), desc(PMFlowAIContextMemory.updated_at), ) .limit(max(1, limit)) ) rows.extend(list(result.scalars().all())) ranked = _rank_memory_rows(question=user_question, rows=rows, limit=limit) now = datetime.utcnow() for row in ranked: row.last_accessed_at = now return _memory_pack_from_rows(ranked, scope_refs=scope_refs) async def compact_and_persist_after_turn( self, *, thread: PMFlowAIThread | None, current_user: Any | None, user_message: str, assistant_answer: str, context_pack: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, run_public_id: str | None = None, user_message_index: int | None = None, assistant_message_index: int | None = None, ) -> dict[str, Any]: """Compact the latest turn into layered long-context memories.""" if thread is None or not getattr(thread, "id", None) or not callable(getattr(self.db, "execute", None)): return {} scope_refs = _scope_refs( thread=thread, current_user=current_user, conversation_context={}, tenant_project_context=tenant_project_context, ) if not scope_refs: return {} previous_pack = await self.build_memory_context_pack( thread=thread, current_user=current_user, conversation_context={}, tenant_project_context=tenant_project_context, user_question=user_message, limit=16, ) compaction = await self._compact_turn( user_message=user_message, assistant_answer=assistant_answer, previous_memory_pack=previous_pack, context_pack=context_pack or {}, user_message_index=user_message_index, assistant_message_index=assistant_message_index, ) memories = _normalize_compaction_memories(compaction) if not memories: memories = _deterministic_compaction_memories( user_message=user_message, assistant_answer=assistant_answer, user_message_index=user_message_index, assistant_message_index=assistant_message_index, ) persisted = [] for scope in _MEMORY_SCOPES: refs = scope_refs.get(scope) or [] if not refs: continue memory = dict(memories.get(scope) or memories.get("thread") or {}) if not _memory_has_content(memory): continue row = await self._upsert_memory( memory_scope=scope, scope_ref=refs[0], memory=memory, thread=thread, run_public_id=run_public_id, user_message_index=user_message_index, assistant_message_index=assistant_message_index, compaction=compaction, ) if row is None: continue persisted.append( { "memory_scope": row.memory_scope, "scope_ref": row.scope_ref, "status": row.status, "confidence_score": float(row.confidence_score or 0.0), "summary": row.summary[:500], "token_estimate": int(row.token_estimate or 0), } ) return { "packet_version": "pmflow_ai_memory_compaction_result.v1", "persisted": persisted, "compaction_source": str(compaction.get("compaction_source") or ""), "scope_count": len(persisted), } async def _compact_turn( self, *, user_message: str, assistant_answer: str, previous_memory_pack: dict[str, Any], context_pack: dict[str, Any], user_message_index: int | None, assistant_message_index: int | None, ) -> dict[str, Any]: generate = getattr(self.generator, "generate", None) if not callable(generate) or not callable(getattr(self.db, "execute", None)): return { "compaction_source": "deterministic_fallback", "memories": _deterministic_compaction_memories( user_message=user_message, assistant_answer=assistant_answer, user_message_index=user_message_index, assistant_message_index=assistant_message_index, ), } try: prompt_result = await self._render_prompt( user_message=user_message, assistant_answer=assistant_answer, previous_memory_pack=previous_memory_pack, context_pack=context_pack, user_message_index=user_message_index, assistant_message_index=assistant_message_index, ) runtime_profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile( function_name=FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR, scene_key=SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME, prompt_key=PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR, ) with bind_agent_audit_context(agent_step_key="context_memory_compaction"): raw = await generate( prompt_result.prompt, db=self.db, tenant_id=self.tenant_id, function_name=FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR, use_prompt_profile=False, temperature=float(runtime_profile.temperature), response_mode=runtime_profile.response_mode, max_tokens=int(runtime_profile.max_tokens), routing_meta={ "scene_key": SCENE_KEY_PMFLOW_AI_AGENT_RUNTIME, "runtime": "pmflow_ai_context_memory", "scene_contract": "semantic_memory_compaction", **runtime_profile.routing_meta(), **prompt_result.routing_meta(), "allow_low_to_high_escalation": False, "disable_thinking": True, }, ) parsed = _extract_json_object(raw) parsed["compaction_source"] = "llm_semantic_compaction" parsed["prompt_meta"] = prompt_result.routing_meta() return parsed except Exception as exc: return { "compaction_source": "deterministic_fallback", "compaction_error": exc.__class__.__name__, "memories": _deterministic_compaction_memories( user_message=user_message, assistant_answer=assistant_answer, user_message_index=user_message_index, assistant_message_index=assistant_message_index, ), } async def _render_prompt( self, *, user_message: str, assistant_answer: str, previous_memory_pack: dict[str, Any], context_pack: dict[str, Any], user_message_index: int | None, assistant_message_index: int | None, ) -> PromptRuntimeRenderResult: return await PromptRuntimeService(self.db).render( tenant_id=self.tenant_id, prompt_key=PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR, fallback_template=PROMPT_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR, latest_turn_json=_dump_json( { "user_message_index": user_message_index, "assistant_message_index": assistant_message_index, "user_message": _clean_text(user_message, limit=1600), "assistant_answer": _clean_text(assistant_answer, limit=2400), }, limit=6000, ), previous_memory_pack_json=_dump_json(previous_memory_pack, limit=12000), context_pack_json=_dump_json(context_pack, limit=16000), ) async def _upsert_memory( self, *, memory_scope: str, scope_ref: str, memory: dict[str, Any], thread: PMFlowAIThread, run_public_id: str | None, user_message_index: int | None, assistant_message_index: int | None, compaction: dict[str, Any], ) -> PMFlowAIContextMemory | None: result = await self.db.execute( select(PMFlowAIContextMemory).where( and_( PMFlowAIContextMemory.tenant_id == self.tenant_id, PMFlowAIContextMemory.memory_scope == memory_scope, PMFlowAIContextMemory.scope_ref == scope_ref, PMFlowAIContextMemory.memory_key == MEMORY_KEY_CURRENT, ) ) ) row = result.scalar_one_or_none() now = datetime.utcnow() gate = _memory_write_gate(memory=memory, existing=row) if gate["action"] == "skip": if row is not None: row.last_validated_at = now row.compaction_payload = { **(dict(row.compaction_payload or {}) if isinstance(row.compaction_payload, dict) else {}), "last_gate": gate, } return None if row is None: row = PMFlowAIContextMemory( tenant_id=self.tenant_id, public_memory_id=f"ctxmem_{uuid.uuid4().hex[:16]}", memory_scope=memory_scope, scope_ref=scope_ref, memory_key=MEMORY_KEY_CURRENT, ) self.db.add(row) merged_memory = _merge_memory(existing=row if gate["action"] != "create" else None, incoming=memory) row.source_thread_id = int(thread.id) row.source_run_public_id = str(run_public_id or "")[:80] or None row.status = str(gate["status"])[:32] row.confidence_score = float(gate["confidence_score"]) row.supersedes_memory_id = int(gate["supersedes_memory_id"]) if gate.get("supersedes_memory_id") else None row.invalidation_reason = str(gate.get("invalidation_reason") or "")[:1000] or None row.summary = _clean_text(merged_memory.get("summary") or "", limit=4000) row.durable_facts = _safe_list(merged_memory.get("durable_facts"), limit=24, item_limit=900) row.open_loops = _safe_list(merged_memory.get("open_loops"), limit=16, item_limit=700) row.decisions = _safe_list(merged_memory.get("decisions"), limit=16, item_limit=700) row.preferences = _safe_list(merged_memory.get("preferences"), limit=16, item_limit=700) row.object_refs = _safe_list(merged_memory.get("object_refs"), limit=16, item_limit=500) row.source_message_range = { "start": _safe_int(user_message_index), "end": _safe_int(assistant_message_index), } row.confidence = str(memory.get("confidence") or "medium")[:32] row.token_estimate = _estimated_tokens( { "summary": row.summary, "durable_facts": row.durable_facts, "open_loops": row.open_loops, "decisions": row.decisions, "preferences": row.preferences, } ) row.compaction_payload = { "packet_version": "pmflow_ai_context_memory_compaction.v1", "source": str(compaction.get("compaction_source") or "")[:80], "prompt_meta": compaction.get("prompt_meta") if isinstance(compaction.get("prompt_meta"), dict) else {}, "write_gate": gate, "conflict_resolution": gate.get("conflict_resolution") if isinstance(gate.get("conflict_resolution"), dict) else {}, "raw_memory_keys": sorted([str(key) for key in dict(memory or {}).keys()])[:32], } row.last_compacted_at = now row.last_validated_at = now row.last_accessed_at = now return row def _scope_refs( *, thread: PMFlowAIThread | None, current_user: Any | None, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, ) -> dict[str, list[str]]: context = conversation_context if isinstance(conversation_context, dict) else {} tenant = tenant_project_context if isinstance(tenant_project_context, dict) else {} refs: dict[str, list[str]] = {scope: [] for scope in _MEMORY_SCOPES} if thread is not None: public_thread_id = str(getattr(thread, "public_thread_id", "") or "").strip() thread_id = str(getattr(thread, "id", "") or "").strip() if public_thread_id: refs["thread"].append(f"thread:{public_thread_id}") elif thread_id: refs["thread"].append(f"thread_id:{thread_id}") user_id = _safe_int(getattr(current_user, "id", None)) or _safe_int(tenant.get("user_id")) if user_id: refs["user"].append(f"user:{user_id}") workspace_scope = str( context.get("conversation_scope") or context.get("workspace_scope") or tenant.get("workspace_scope") or context.get("surface") or "space" ).strip() if workspace_scope: refs["workspace"].append(f"workspace:{workspace_scope[:120]}") project_id = _safe_int(context.get("current_project_id")) or _safe_int(tenant.get("current_project_id")) if project_id: refs["project"].append(f"project:{project_id}") else: for item in _project_refs_from_context(context, tenant): if item not in refs["project"]: refs["project"].append(item) return {scope: values for scope, values in refs.items() if values} def _project_refs_from_context(context: dict[str, Any], tenant: dict[str, Any]) -> list[str]: refs: list[str] = [] roots = [context, tenant.get("surface_context") if isinstance(tenant.get("surface_context"), dict) else {}] for root in roots: for key in ("current_focus_object", "object_ledger"): value = root.get(key) if isinstance(root, dict) else None if isinstance(value, dict): project_id = _safe_int(value.get("project_id") or value.get("id") or value.get("object_id")) object_type = str(value.get("object_type") or value.get("type") or "").strip() if project_id and (not object_type or object_type == "project"): refs.append(f"project:{project_id}") for item in list(root.get("visible_objects") or [])[:12] if isinstance(root, dict) else []: if not isinstance(item, dict): continue project_id = _safe_int(item.get("project_id") or item.get("id") or item.get("object_id")) object_type = str(item.get("object_type") or item.get("type") or "").strip() if project_id and (not object_type or object_type == "project"): refs.append(f"project:{project_id}") out: list[str] = [] for ref in refs: if ref not in out: out.append(ref) return out[:4] def _memory_pack_from_rows(rows: list[PMFlowAIContextMemory], *, scope_refs: dict[str, list[str]]) -> dict[str, Any]: if not rows: return {} return { "packet_version": MEMORY_PACKET_VERSION, "scope_refs": scope_refs, "memories": [_memory_row_packet(row) for row in rows], "context_policy": { "long_term_memory_is_semantic_compaction_not_raw_transcript": True, "same_thread_recent_messages_override_memory_when_conflicting": True, "memory_scope_binding_is_anchor_based_not_scene_keyword_based": True, }, } def _memory_row_packet(row: PMFlowAIContextMemory) -> dict[str, Any]: return { "memory_id": str(row.public_memory_id or "")[:80], "memory_scope": str(row.memory_scope or "")[:40], "scope_ref": str(row.scope_ref or "")[:160], "memory_key": str(row.memory_key or "")[:120], "status": str(getattr(row, "status", "") or "active")[:32], "summary": _clean_text(row.summary, limit=1800), "durable_facts": _safe_list(row.durable_facts, limit=10, item_limit=700), "open_loops": _safe_list(row.open_loops, limit=8, item_limit=500), "decisions": _safe_list(row.decisions, limit=8, item_limit=500), "preferences": _safe_list(row.preferences, limit=8, item_limit=500), "object_refs": _safe_list(row.object_refs, limit=8, item_limit=300), "confidence": str(row.confidence or "")[:32], "confidence_score": float(getattr(row, "confidence_score", 0.0) or 0.0), "token_estimate": int(row.token_estimate or 0), "last_compacted_at": _iso(row.last_compacted_at), } def _rank_memory_rows(*, question: str, rows: list[PMFlowAIContextMemory], limit: int) -> list[PMFlowAIContextMemory]: if not rows: return [] question_terms = _text_terms(question) def score(row: PMFlowAIContextMemory) -> tuple[int, str]: text = " ".join( [ str(row.summary or ""), _dump_json(row.durable_facts, limit=3000), _dump_json(row.open_loops, limit=2000), _dump_json(row.decisions, limit=2000), ] ) terms = _text_terms(text) overlap = len(question_terms & terms) scope_weight = {"thread": 8, "project": 6, "workspace": 4, "user": 3}.get(str(row.memory_scope or ""), 1) confidence_weight = int(float(getattr(row, "confidence_score", 0.0) or 0.0) * 4) status_weight = 2 if str(getattr(row, "status", "") or "active") == "active" else 0 return (overlap + scope_weight + confidence_weight + status_weight, str(row.last_compacted_at or row.updated_at or "")) ranked = sorted(rows, key=score, reverse=True) out: list[PMFlowAIContextMemory] = [] seen: set[str] = set() for row in ranked: key = f"{row.memory_scope}:{row.scope_ref}:{row.memory_key}" if key in seen: continue seen.add(key) out.append(row) if len(out) >= max(1, limit): break return out def _normalize_compaction_memories(value: dict[str, Any]) -> dict[str, dict[str, Any]]: raw = value.get("memories") if isinstance(value, dict) else {} if not isinstance(raw, dict): return {} out: dict[str, dict[str, Any]] = {} for scope in _MEMORY_SCOPES: item = raw.get(scope) if isinstance(item, dict): out[scope] = _sanitize_memory(item) return out def _memory_write_gate(*, memory: dict[str, Any], existing: PMFlowAIContextMemory | None) -> dict[str, Any]: """Decide whether a semantic memory should be written or merged. The gate only uses lifecycle signals, generic text novelty and structured field deltas. It deliberately avoids business-domain keyword rules. """ sanitized = _sanitize_memory(memory) status = _normalize_memory_status(sanitized.get("status")) confidence_score = _confidence_score(sanitized.get("confidence")) incoming_text = _memory_text(sanitized) if not incoming_text: return _gate_result( "skip", "empty_memory", status=status, confidence_score=confidence_score, conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=existing), ) if status in {"stale", "invalidated", "rejected"}: if existing is None: return _gate_result( "skip", "cannot_invalidate_missing_memory", status=status, confidence_score=confidence_score, conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=None), ) return _gate_result( "update", "lifecycle_status_update", status=status, confidence_score=max(confidence_score, float(getattr(existing, "confidence_score", 0.0) or 0.0)), supersedes_memory_id=int(existing.id) if getattr(existing, "id", None) is not None else None, invalidation_reason=sanitized.get("invalidation_reason") or sanitized.get("summary") or "", conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=existing), ) if existing is None: create_status = "candidate" if confidence_score < 0.5 else "active" return _gate_result( "create", "new_scope_memory", status=create_status, confidence_score=confidence_score, conflict_resolution=_resolve_memory_conflict(memory=sanitized, existing=None), ) existing_text = _memory_text(_row_memory_dict(existing)) incoming_terms = _text_terms(incoming_text) existing_terms = _text_terms(existing_text) new_terms = incoming_terms - existing_terms novelty_ratio = len(new_terms) / max(1, len(incoming_terms)) structured_delta = _structured_delta_count(sanitized, _row_memory_dict(existing)) existing_score = float(getattr(existing, "confidence_score", 0.0) or 0.0) supersedes = bool(memory.get("supersedes_current") or memory.get("supersedes_existing")) conflict_resolution = _resolve_memory_conflict( memory={**sanitized, "supersedes_current": supersedes}, existing=existing, novelty_ratio=novelty_ratio, structured_delta=structured_delta, ) if not supersedes and novelty_ratio < 0.08 and structured_delta <= 0 and confidence_score <= existing_score + 0.05: return _gate_result( "skip", "duplicate_or_lower_confidence", status=str(getattr(existing, "status", "") or "active"), confidence_score=existing_score, conflict_resolution=conflict_resolution, ) return _gate_result( "update", "merge_new_semantic_delta", status="active", confidence_score=max(confidence_score, existing_score), supersedes_memory_id=int(existing.id) if supersedes and getattr(existing, "id", None) is not None else None, invalidation_reason=sanitized.get("invalidation_reason") if supersedes else "", novelty_ratio=round(novelty_ratio, 4), structured_delta=structured_delta, conflict_resolution=conflict_resolution, ) def _resolve_memory_conflict( *, memory: dict[str, Any], existing: PMFlowAIContextMemory | None, novelty_ratio: float | None = None, structured_delta: int | None = None, ) -> dict[str, Any]: sanitized = _sanitize_memory(memory) status = _normalize_memory_status(sanitized.get("status")) incoming_score = _confidence_score(sanitized.get("confidence")) if existing is None: return { "packet_version": "pmflow_ai_memory_conflict_resolution.v1", "resolution": "create", "reason": "no_existing_memory", "confidence_movement": {"incoming": incoming_score, "existing": None}, } existing_memory = _row_memory_dict(existing) existing_score = float(getattr(existing, "confidence_score", 0.0) or 0.0) if status in {"stale", "invalidated", "rejected"}: return { "packet_version": "pmflow_ai_memory_conflict_resolution.v1", "resolution": "invalidate", "reason": "incoming_lifecycle_status_invalidates_existing", "old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80], "confidence_movement": {"incoming": incoming_score, "existing": existing_score}, } if bool(memory.get("supersedes_current") or memory.get("supersedes_existing")): return { "packet_version": "pmflow_ai_memory_conflict_resolution.v1", "resolution": "supersede", "reason": "incoming_memory_declares_supersession", "old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80], "confidence_movement": {"incoming": incoming_score, "existing": existing_score}, } delta = int(structured_delta if structured_delta is not None else _structured_delta_count(sanitized, existing_memory)) novelty = float(novelty_ratio if novelty_ratio is not None else 0.0) if delta > 0 and novelty >= 0.35: resolution = "keep_both" reason = "incoming_adds_distinct_structured_context" elif delta > 0 or novelty >= 0.08: resolution = "merge" reason = "incoming_extends_existing_memory" else: resolution = "no_change" reason = "incoming_memory_duplicates_existing" return { "packet_version": "pmflow_ai_memory_conflict_resolution.v1", "resolution": resolution, "reason": reason, "old_memory_id": str(getattr(existing, "public_memory_id", "") or "")[:80], "novelty_ratio": round(novelty, 4), "structured_delta": delta, "confidence_movement": {"incoming": incoming_score, "existing": existing_score}, } def _gate_result( action: str, reason: str, *, status: str, confidence_score: float, supersedes_memory_id: int | None = None, invalidation_reason: str = "", **extra: Any, ) -> dict[str, Any]: return { "packet_version": "pmflow_ai_memory_write_gate.v1", "action": action, "reason": reason, "status": _normalize_memory_status(status), "confidence_score": max(0.0, min(float(confidence_score or 0.0), 1.0)), "supersedes_memory_id": supersedes_memory_id, "invalidation_reason": _clean_text(invalidation_reason, limit=1000), **extra, } def _merge_memory(*, existing: PMFlowAIContextMemory | None, incoming: dict[str, Any]) -> dict[str, Any]: new_memory = _sanitize_memory(incoming) if existing is None: return new_memory old_memory = _row_memory_dict(existing) old_summary = _clean_text(old_memory.get("summary") or "", limit=3000) new_summary = _clean_text(new_memory.get("summary") or "", limit=3000) if not old_summary: summary = new_summary elif not new_summary or _text_terms(new_summary).issubset(_text_terms(old_summary)): summary = old_summary else: summary = _clean_text(f"{old_summary}\n{new_summary}", limit=4000) return { "summary": summary, "durable_facts": _merge_list_rows(old_memory.get("durable_facts"), new_memory.get("durable_facts"), limit=24), "open_loops": _merge_list_rows(old_memory.get("open_loops"), new_memory.get("open_loops"), limit=16), "decisions": _merge_list_rows(old_memory.get("decisions"), new_memory.get("decisions"), limit=16), "preferences": _merge_list_rows(old_memory.get("preferences"), new_memory.get("preferences"), limit=16), "object_refs": _merge_list_rows(old_memory.get("object_refs"), new_memory.get("object_refs"), limit=16), "confidence": _higher_confidence(old_memory.get("confidence"), new_memory.get("confidence")), "status": new_memory.get("status") or old_memory.get("status") or "active", "invalidation_reason": new_memory.get("invalidation_reason") or old_memory.get("invalidation_reason") or "", } def _sanitize_memory(value: dict[str, Any]) -> dict[str, Any]: summary = _clean_text(value.get("summary") or "", limit=4000) return { "summary": summary, "durable_facts": _safe_list(value.get("durable_facts"), limit=24, item_limit=900), "open_loops": _safe_list(value.get("open_loops"), limit=16, item_limit=700), "decisions": _safe_list(value.get("decisions"), limit=16, item_limit=700), "preferences": _safe_list(value.get("preferences"), limit=16, item_limit=700), "object_refs": _safe_list(value.get("object_refs"), limit=16, item_limit=500), "confidence": str(value.get("confidence") or "medium")[:32], "status": _normalize_memory_status(value.get("status")), "invalidation_reason": _clean_text(value.get("invalidation_reason") or "", limit=1000), } def _deterministic_compaction_memories( *, user_message: str, assistant_answer: str, user_message_index: int | None, assistant_message_index: int | None, ) -> dict[str, dict[str, Any]]: user_text = _clean_text(user_message, limit=900) answer_text = _clean_text(assistant_answer, limit=1400) summary = "；".join(item for item in [f"用户提出：{user_text}" if user_text else "", f"PMFlow AI 回答：{answer_text}" if answer_text else ""] if item) base = { "summary": summary[:3600], "durable_facts": [ { "kind": "turn_summary", "summary": summary[:900], "source_message_indexes": [item for item in [user_message_index, assistant_message_index] if item is not None], } ] if summary else [], "open_loops": [], "decisions": [], "preferences": [], "object_refs": [], "confidence": "medium", } return {scope: dict(base) for scope in _MEMORY_SCOPES} def _row_memory_dict(row: PMFlowAIContextMemory) -> dict[str, Any]: return { "summary": str(getattr(row, "summary", "") or ""), "durable_facts": _safe_list(getattr(row, "durable_facts", None), limit=24, item_limit=900), "open_loops": _safe_list(getattr(row, "open_loops", None), limit=16, item_limit=700), "decisions": _safe_list(getattr(row, "decisions", None), limit=16, item_limit=700), "preferences": _safe_list(getattr(row, "preferences", None), limit=16, item_limit=700), "object_refs": _safe_list(getattr(row, "object_refs", None), limit=16, item_limit=500), "confidence": str(getattr(row, "confidence", "") or "medium")[:32], "status": str(getattr(row, "status", "") or "active")[:32], "invalidation_reason": str(getattr(row, "invalidation_reason", "") or "")[:1000], } def _memory_text(memory: dict[str, Any]) -> str: return " ".join( item for item in [ str(memory.get("summary") or ""), _dump_json(memory.get("durable_facts") or [], limit=5000), _dump_json(memory.get("open_loops") or [], limit=3000), _dump_json(memory.get("decisions") or [], limit=3000), _dump_json(memory.get("preferences") or [], limit=3000), ] if item ) def _merge_list_rows(left: Any, right: Any, *, limit: int) -> list[Any]: out: list[Any] = [] seen: set[str] = set() for item in [*list(left or []), *list(right or [])]: key = _dedupe_key(item) if not key or key in seen: continue seen.add(key) out.append(item) if len(out) >= limit: break return out def _structured_delta_count(left: dict[str, Any], right: dict[str, Any]) -> int: delta = 0 for key in ("durable_facts", "open_loops", "decisions", "preferences", "object_refs"): right_keys = {_dedupe_key(item) for item in list(right.get(key) or [])} for item in list(left.get(key) or []): item_key = _dedupe_key(item) if item_key and item_key not in right_keys: delta += 1 return delta def _dedupe_key(value: Any) -> str: if isinstance(value, dict): text = str(value.get("summary") or value.get("display_name") or value.get("object_id") or value) else: text = str(value or "") return re.sub(r"\s+", "", text.strip().lower())[:240] def _normalize_memory_status(value: Any) -> str: status = str(value or "").strip().lower() if status in {"active", "candidate", "stale", "invalidated", "rejected"}: return status return "active" def _confidence_score(value: Any) -> float: if isinstance(value, (int, float)): return max(0.0, min(float(value), 1.0)) label = str(value or "").strip().lower() if label == "high": return 0.85 if label == "low": return 0.35 return 0.6 def _higher_confidence(left: Any, right: Any) -> str: return str(left or "medium") if _confidence_score(left) >= _confidence_score(right) else str(right or "medium") def _memory_has_content(memory: dict[str, Any]) -> bool: return bool(str(memory.get("summary") or "").strip() or memory.get("durable_facts") or memory.get("open_loops")) def _extract_json_object(raw: Any) -> dict[str, Any]: if isinstance(raw, dict): return raw text = str(raw or "").strip() if not text: return {} try: parsed = json.loads(text) return parsed if isinstance(parsed, dict) else {} except Exception: pass match = re.search(r"\{.*\}", text, flags=re.S) if not match: return {} try: parsed = json.loads(match.group(0)) return parsed if isinstance(parsed, dict) else {} except Exception: return {} 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 _safe_list(value: Any, *, limit: int, item_limit: int) -> list[Any]: rows = value if isinstance(value, list) else [] out: list[Any] = [] for item in rows[: max(0, limit)]: if isinstance(item, dict): cleaned = { str(key)[:80]: _clean_text(val, limit=item_limit) if not isinstance(val, (dict, list)) else val for key, val in item.items() if val not in (None, "", [], {}) } if cleaned: out.append(cleaned) continue text = _clean_text(item, limit=item_limit) if text: out.append(text) return out def _dump_json(value: Any, *, limit: int) -> str: try: text = json.dumps(value, ensure_ascii=False, default=str) except Exception: text = str(value or "") return text[: max(0, limit)] 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 | None: try: if value is None or value == "": return None return int(value) except Exception: return None def _estimated_tokens(value: Any) -> int: text = _dump_json(value, limit=60000) return max(1, int(len(text) / 3.5)) def _iso(value: Any) -> str: if value is None: return "" try: return value.isoformat() except Exception: return str(value or "") __all__ = [ "FUNCTION_NAME_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR", "MEMORY_PACKET_VERSION", "PMFlowAIContextMemoryService", "PROMPT_KEY_PMFLOW_AI_CONTEXT_MEMORY_COMPACTOR", "_resolve_memory_conflict", ]