"""
Memory formatting utilities for Coppermind × Hermes.
Converts raw API responses into context strings suitable for
system prompt injection via Hermes's pre_llm_call hook.
"""
import logging
from typing import Any
logger = logging.getLogger("coppermind")
# Maximum characters of memory context to inject per turn.
# Prevents flooding the system prompt with too much context.
MAX_CONTEXT_CHARS = 1000
# Maximum characters per individual memory fragment.
MAX_FRAGMENT_CHARS = 500
MAX_MEMORY_SUMMARY_CHARS = 220
def format_memories(api_response: dict[str, Any]) -> str:
"""Format an API response into an XML context block for prompt injection.
The XML format matches the canonical Coppermind context format used
across all integrations (daemon, gateway, SDK).
Args:
api_response: Response from search_memory() or get_memory().
Returns:
Formatted context string, or empty string if no memories.
"""
# Handle the "context" key (returned by /v1/memory)
context = api_response.get("context")
if isinstance(context, str) and context.strip():
normalized = context.strip()
if normalized.startswith("") and normalized.endswith(""):
return normalized
return _wrap_context(normalized)
# Handle the "fragments" key (returned by /v1/memory/search)
fragments = api_response.get("fragments", [])
if not fragments:
return ""
lines: list[str] = []
total_chars = 0
max_fragment_chars = MAX_FRAGMENT_CHARS
max_context_chars = MAX_CONTEXT_CHARS
for fragment in fragments:
content = _extract_content(fragment)
if not content:
continue
content_len = len(content)
# Truncate individual fragments
if content_len > max_fragment_chars:
content = content[:max_fragment_chars].rstrip() + "…"
content_len = max_fragment_chars
# Check total budget
new_total = total_chars + content_len
if new_total > max_context_chars:
break
lines.append(f" - {content}")
total_chars = new_total
if not lines:
return ""
return _wrap_context("\n".join(lines))
def format_conversation_for_ingest(
user_message: str,
assistant_response: str,
session_id: str | None = None,
) -> str:
"""Format a conversation turn for memory ingestion.
Creates a structured text representation that preserves the
conversational context for later retrieval and extraction.
Args:
user_message: The user's message.
assistant_response: The assistant's response.
session_id: Optional session identifier for grouping.
Returns:
Formatted string suitable for ingest().
"""
parts: list[str] = []
if session_id:
parts.append(f"[Session: {session_id}]")
parts.append(f"User: {user_message}")
# Truncate very long responses to avoid storing excessive data
if len(assistant_response) > 2000:
assistant_response = assistant_response[:2000].rstrip() + "… [truncated]"
parts.append(f"Assistant: {assistant_response}")
return "\n".join(parts)
def distill_turn_for_ingest(
user_message: str,
assistant_response: str,
session_id: str | None = None,
) -> dict[str, Any] | None:
"""Reduce a Hermes turn to a durable memory candidate, or skip it.
We intentionally avoid storing raw turn transcripts as first-class durable
memories. Only stable preference/policy/goal signals or explicit release
outcomes should survive.
"""
user_text = _normalize(user_message)
assistant_text = _normalize(assistant_response)
if not user_text:
return None
preference = _matches(
user_text,
[
r"\bi prefer\b",
r"\bi like\b",
r"\bi dislike\b",
r"\bi use\b",
r"\bmy machine\b",
r"\bmy setup\b",
r"\bmy default\b",
r"\bdefault to\b",
r"\balways use\b",
r"\bnever use\b",
],
)
policy = _matches(
user_text,
[r"\bshould\b", r"\bmust\b", r"\balways\b", r"\bnever\b", r"\bensure\b", r"\bkeep\b"],
) and _mentions_signal_keyword(user_text)
goal = _matches(
user_text,
[r"\bnext we need\b", r"\bwe need to\b", r"\bneed to\b", r"\bwant to\b", r"\bplan to\b", r"\btrying to\b", r"\blet'?s\b"],
) and _matches(
user_text,
[r"\btest\b", r"\brelease\b", r"\bpublish\b", r"\bmerge\b", r"\bship\b", r"\bversion\b", r"\bdoctor\b", r"\bsetup\b", r"\bmemory\b", r"\bauth\b", r"\bsearch\b", r"\bstats\b", r"\blocal ai\b", r"\bsurreal\b"],
)
release_outcome = _matches(
assistant_text,
[r"\bpublished\b", r"\breleased\b", r"\btagged\b", r"\bmerged\b", r"\bpushed\b", r"\bshipped\b", r"\bdeployed\b", r"\bv?\d+\.\d+\.\d+\b", r"\brelease gate\b"],
)
if preference:
return {
"content": _summarize(user_text),
"tags": ["hermes", "distilled", "preference"],
"metadata": {
"source_client": "hermes",
"capture_kind": "distilled_signal",
"durable_signal": True,
"memoryType": "preference",
"importance": 0.85,
"retrievalPriority": 4,
"session_id": session_id,
},
}
if policy:
return {
"content": _summarize(user_text),
"tags": ["hermes", "distilled", "workflow"],
"metadata": {
"source_client": "hermes",
"capture_kind": "distilled_signal",
"durable_signal": True,
"memoryType": "workflow",
"importance": 0.8,
"retrievalPriority": 3,
"session_id": session_id,
},
}
if goal:
return {
"content": _summarize(user_text),
"tags": ["hermes", "distilled", "goal"],
"metadata": {
"source_client": "hermes",
"capture_kind": "distilled_signal",
"durable_signal": True,
"memoryType": "goal",
"importance": 0.7,
"retrievalPriority": 2,
"session_id": session_id,
},
}
if release_outcome:
return {
"content": _summarize(assistant_text),
"tags": ["hermes", "distilled", "project_state"],
"metadata": {
"source_client": "hermes",
"capture_kind": "release_outcome",
"durable_signal": True,
"memoryType": "project_state",
"importance": 0.72,
"retrievalPriority": 2,
"session_id": session_id,
},
}
return None
def summarize_conversation(conversation_history: list[dict]) -> str:
"""Create a summary from a conversation history for session-end ingestion.
This produces a human-readable summary of the key topics covered
in the conversation, suitable for high-level memory recall.
Args:
conversation_history: List of message dicts with "role" and "content".
Returns:
Summary string.
"""
if not conversation_history:
return ""
# Extract user messages only (they contain the topics and questions)
user_messages: list[str] = []
for msg in conversation_history:
if msg.get("role") == "user":
content = msg.get("content", "")
if isinstance(content, str) and content.strip():
user_messages.append(content.strip())
if not user_messages:
return ""
# Build a topic summary
parts = [f"Session summary ({len(user_messages)} user messages):"]
for i, msg in enumerate(user_messages[:10], 1): # Cap at 10 messages
# Truncate long messages
preview = msg[:200].rstrip() + ("…" if len(msg) > 200 else "")
parts.append(f" {i}. {preview}")
if len(user_messages) > 10:
parts.append(f" ... and {len(user_messages) - 10} more messages")
return "\n".join(parts)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _wrap_context(content: str) -> str:
"""Wrap memory content in the canonical Coppermind XML block."""
return f"\n{content}\n"
def _normalize(text: str) -> str:
return " ".join(text.split()).strip()
def _summarize(text: str) -> str:
normalized = _normalize(text)
if not normalized:
return ""
sentences = [part.strip() for part in normalized.replace("\n", " ").split(". ") if part.strip()]
summary = sentences[0] if sentences else normalized
if len(summary) > MAX_MEMORY_SUMMARY_CHARS:
return summary[:MAX_MEMORY_SUMMARY_CHARS].rstrip() + "…"
return summary
def _matches(text: str, patterns: list[str]) -> bool:
import re
return any(re.search(pattern, text, flags=re.IGNORECASE) for pattern in patterns)
def _mentions_signal_keyword(text: str) -> bool:
lowered = text.lower()
return any(
keyword in lowered
for keyword in [
"coppermind",
"doctor",
"setup",
"uninstall",
"daemon",
"memory",
"search",
"inspect",
"stats",
"local ai",
"surreal",
"surrealdb",
"release",
"publish",
"merge",
"version",
]
)
def _extract_content(fragment: dict) -> str:
"""Extract text content from a memory fragment.
Optimized: uses next() with generator to avoid multiple dict lookups
and short-circuit on first match.
"""
# Try common field names in priority order
for key in ("content", "contentSummary", "text", "summary"):
value = fragment.get(key)
if value:
return value.strip() if isinstance(value, str) else ""
return ""