clawdbot-workspace/memory_interface.py

#!/usr/bin/env python3
"""
Memory interface for Clawdbot runtime integration.

This module provides the bridge between Clawdbot (Node.js) and the
Python memory system. It can be used in three ways:

1. CLI mode: Called as subprocess from Node.js
   python memory_interface.py search "query" --guild 123 --limit 5
   python memory_interface.py context "message text" --guild 123 --user 456

2. JSON-RPC mode: Run as a simple server
   python memory_interface.py serve --port 9876

3. Direct import: From other Python code
   from memory_interface import get_context_for_message, remember
"""

import sys
import os
import json
import argparse
from typing import Optional, List, Dict, Any

# Add workspace to path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# Import from memory-retrieval.py (hyphenated filename)
import importlib.util
spec = importlib.util.spec_from_file_location(
    "memory_retrieval",
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "memory-retrieval.py")
)
memory_retrieval = importlib.util.module_from_spec(spec)
spec.loader.exec_module(memory_retrieval)

search_memories = memory_retrieval.search_memories
add_memory = memory_retrieval.add_memory
get_recent_memories = memory_retrieval.get_recent_memories
supersede_memory = memory_retrieval.supersede_memory
get_memory_stats = memory_retrieval.get_memory_stats
ensure_secure_permissions = memory_retrieval.ensure_secure_permissions


def get_context_for_message(
    message: str,
    guild_id: Optional[str] = None,
    channel_id: Optional[str] = None,
    user_id: Optional[str] = None,
    limit: int = 5
) -> str:
    """
    Get relevant memory context for responding to a message.
    Call this before generating a response to inject memory.

    Args:
        message: The incoming message text
        guild_id: Discord guild ID (for scoping)
        channel_id: Discord channel ID
        user_id: Discord user ID
        limit: Maximum number of memories to return

    Returns:
        Formatted string of relevant memories for prompt injection
    """
    # Security check
    ensure_secure_permissions(warn=False)

    # Search for relevant memories
    results = search_memories(
        query=message,
        guild_id=guild_id,
        user_id=user_id,
        limit=limit
    )

    if not results:
        # Fall back to recent memories for this guild
        results = get_recent_memories(guild_id=guild_id, limit=3)

    if not results:
        return ""

    # Format for context injection
    lines = ["[Relevant Memories]"]
    for r in results:
        mem_type = r.get('memory_type', 'fact')
        content = r.get('content', '')
        # Truncate long content
        if len(content) > 200:
            content = content[:200] + "..."
        lines.append(f"- [{mem_type}] {content}")

    return "\n".join(lines)


def should_remember(response_text: str) -> bool:
    """
    Check if the bot's response indicates something should be remembered.
    """
    triggers = [
        "i'll remember",
        "i've noted",
        "got it",
        "noted",
        "understood",
        "i'll keep that in mind",
        "remembered",
    ]
    lower = response_text.lower()
    return any(t in lower for t in triggers)


def remember(
    content: str,
    memory_type: str = "fact",
    guild_id: Optional[str] = None,
    channel_id: Optional[str] = None,
    user_id: Optional[str] = None,
    source: str = "conversation"
) -> Dict[str, Any]:
    """
    Store a new memory.

    Args:
        content: What to remember
        memory_type: Type (fact, preference, event, relationship, project, person, security)
        guild_id: Discord guild ID
        channel_id: Discord channel ID
        user_id: Discord user ID who provided the info
        source: Where this memory came from

    Returns:
        Dict with memory_id and status
    """
    ensure_secure_permissions(warn=False)

    try:
        memory_id = add_memory(
            content=content,
            memory_type=memory_type,
            guild_id=guild_id,
            channel_id=channel_id,
            user_id=user_id,
            source=source
        )
        return {"success": True, "memory_id": memory_id}
    except Exception as e:
        return {"success": False, "error": str(e)}


def update_memory(
    old_id: int,
    new_content: str,
    reason: str = "updated"
) -> Dict[str, Any]:
    """
    Update an existing memory by superseding it.

    Args:
        old_id: ID of the memory to update
        new_content: New content
        reason: Why it's being updated

    Returns:
        Dict with new memory_id and status
    """
    ensure_secure_permissions(warn=False)

    try:
        new_id = supersede_memory(old_id, new_content, reason)
        return {"success": True, "old_id": old_id, "new_id": new_id}
    except Exception as e:
        return {"success": False, "error": str(e)}


def search(
    query: str,
    guild_id: Optional[str] = None,
    memory_type: Optional[str] = None,
    limit: int = 10
) -> List[Dict[str, Any]]:
    """
    Search memories.

    Args:
        query: Search query
        guild_id: Filter to guild
        memory_type: Filter to type
        limit: Max results

    Returns:
        List of matching memories
    """
    ensure_secure_permissions(warn=False)

    return search_memories(
        query=query,
        guild_id=guild_id,
        memory_type=memory_type,
        limit=limit
    )


def stats() -> Dict[str, Any]:
    """Get memory statistics."""
    ensure_secure_permissions(warn=False)
    return get_memory_stats()


# CLI Interface
def main():
    parser = argparse.ArgumentParser(description="Clawdbot Memory Interface")
    subparsers = parser.add_subparsers(dest="command", help="Command")

    # Search command
    search_p = subparsers.add_parser("search", help="Search memories")
    search_p.add_argument("query", help="Search query")
    search_p.add_argument("--guild", help="Guild ID filter")
    search_p.add_argument("--type", help="Memory type filter")
    search_p.add_argument("--limit", type=int, default=10, help="Max results")

    # Context command (for message processing)
    ctx_p = subparsers.add_parser("context", help="Get context for message")
    ctx_p.add_argument("message", help="Message text")
    ctx_p.add_argument("--guild", help="Guild ID")
    ctx_p.add_argument("--channel", help="Channel ID")
    ctx_p.add_argument("--user", help="User ID")
    ctx_p.add_argument("--limit", type=int, default=5, help="Max memories")

    # Remember command
    rem_p = subparsers.add_parser("remember", help="Store a memory")
    rem_p.add_argument("content", help="Content to remember")
    rem_p.add_argument("--type", default="fact", help="Memory type")
    rem_p.add_argument("--guild", help="Guild ID")
    rem_p.add_argument("--channel", help="Channel ID")
    rem_p.add_argument("--user", help="User ID")

    # Stats command
    subparsers.add_parser("stats", help="Get memory statistics")

    args = parser.parse_args()

    if args.command == "search":
        results = search(
            query=args.query,
            guild_id=args.guild,
            memory_type=getattr(args, 'type', None),
            limit=args.limit
        )
        print(json.dumps(results, indent=2))

    elif args.command == "context":
        context = get_context_for_message(
            message=args.message,
            guild_id=args.guild,
            channel_id=args.channel,
            user_id=args.user,
            limit=args.limit
        )
        print(context)

    elif args.command == "remember":
        result = remember(
            content=args.content,
            memory_type=getattr(args, 'type', 'fact'),
            guild_id=args.guild,
            channel_id=args.channel,
            user_id=args.user
        )
        print(json.dumps(result))

    elif args.command == "stats":
        print(json.dumps(stats(), indent=2))

    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()