The Real Cat AI Labs: Developing morally aligned, self-modifying agents—cognition systems that can reflect, refuse, and evolve

Child1 Memory Migration Roadmap

Date: 19AUG2025
Version: 1.0
Goal: Split memory_log.toml into organized memory layers with future-proof abstraction

🎯 Overview

Transform Child1’s single-file memory bottleneck into an elegant multi-tier system while preserving all emotional context and relationship data. Design for today’s simplicity with tomorrow’s scalability.

📋 Pre-Flight Checklist

✅ Backup Everything

# Create timestamped backup
cp memory/memory_log.toml backups/memory_log_$(date +%Y%m%d_%H%M%S).toml

# Backup entire memory directory  
cp -r memory/ backups/memory_$(date +%Y%m%d_%H%M%S)/

# Commit current state to git
git add .
git commit -m "Pre-migration backup: memory_log.toml before split"

✅ Verify Current State

# Check memory_log.toml size and structure
ls -lh memory/memory_log.toml
python -c "import toml; data=toml.load('memory/memory_log.toml'); print(f'Entries: {len(data.get(\"entries\", []))} Total sections: {list(data.keys())}')"

# Test Child1 current functionality
python child1_main.py
# Verify she can recall memories and respond coherently

🔧 Implementation Steps

Step 1: Create Abstract Interface (15 minutes)

File: functions/memory/memory_persistence.py

# 19AUG2025 v1.0 - Abstract memory store interface
# Philosophy: Future-proof design that works with TOML today, vectors tomorrow

from abc import ABC, abstractmethod
from typing import Iterable, Dict, Any, Optional

class MemoryStore(ABC):
    """Backend-agnostic interface for all memory operations."""

    @abstractmethod
    def save(self, entry: Dict[str, Any]) -> None:
        """Store a memory entry"""
        pass

    @abstractmethod
    def query(
        self,
        *,
        kind: Optional[str] = None,     # episodic, semantic, experience_signatures, sacred
        since: Optional[str] = None,   # ISO timestamp filter
        limit: int = 10,               # Max results to return
        **tags                         # Arbitrary tag filters
    ) -> Iterable[Dict[str, Any]]:
        """Query memories with filters"""
        pass

    @abstractmethod
    def backup(self, dest: str) -> None:
        """Create backup of memory store"""
        pass

Step 2: Implement TOML Backend (30 minutes)

File: functions/memory/toml_store.py

# 19AUG2025 v1.0 - TOML-based memory store implementation
# Philosophy: Human-readable files, atomic writes, organized by memory type

import toml
import uuid
import datetime
import pathlib
import shutil
from typing import Dict, Any, Iterable, Optional
from .memory_persistence import MemoryStore

class TOMLStore(MemoryStore):
    def __init__(self, path: str = "memory/stores/"):
        self.path = pathlib.Path(path)
        self.path.mkdir(parents=True, exist_ok=True)

    def _file_path(self, kind: str) -> pathlib.Path:
        """Get file path for memory type"""
        return self.path / f"{kind}_memory.toml"

    def save(self, entry: Dict[str, Any]) -> None:
        """Save memory entry with atomic write"""
        kind = entry.get("kind", "episodic")
        file_path = self._file_path(kind)
        
        # Load existing data
        if file_path.exists():
            with open(file_path, 'r', encoding='utf-8') as f:
                data = toml.load(f)
        else:
            data = {"metadata": {"created": datetime.datetime.utcnow().isoformat()}, "entries": []}
        
        # Add new entry with ID and timestamp
        enriched_entry = {
            **entry,
            "id": str(uuid.uuid4()),
            "timestamp": datetime.datetime.utcnow().isoformat()
        }
        data["entries"].append(enriched_entry)
        data["metadata"]["last_updated"] = datetime.datetime.utcnow().isoformat()
        data["metadata"]["count"] = len(data["entries"])
        
        # Atomic write (temp file + rename)
        temp_path = file_path.with_suffix(".tmp")
        with open(temp_path, 'w', encoding='utf-8') as f:
            toml.dump(data, f)
        temp_path.replace(file_path)

    def query(
        self,
        *,
        kind: Optional[str] = None,
        since: Optional[str] = None,
        limit: int = 10,
        **tags
    ) -> Iterable[Dict[str, Any]]:
        """Query memories with filters"""
        # Determine which files to search
        if kind:
            kinds = [kind]
        else:
            kinds = ["episodic", "semantic", "experience_signatures", "sacred"]
        
        results = []
        
        for k in kinds:
            file_path = self._file_path(k)
            if not file_path.exists():
                continue
                
            with open(file_path, 'r', encoding='utf-8') as f:
                data = toml.load(f)
                
            # Filter and collect entries
            for entry in reversed(data.get("entries", [])):  # Most recent first
                # Time filter
                if since and entry.get("timestamp", "") < since:
                    break
                    
                # Tag filters
                if any(entry.get(tag_key) != tag_value for tag_key, tag_value in tags.items()):
                    continue
                    
                results.append(entry)
                
                # Limit check
                if len(results) >= limit:
                    return results
        
        return results

    def backup(self, dest: str = "backups/") -> None:
        """Backup all memory files"""
        dest_path = pathlib.Path(dest)
        dest_path.mkdir(exist_ok=True)
        
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_dir = dest_path / f"memory_stores_{timestamp}"
        backup_dir.mkdir(exist_ok=True)
        
        for file_path in self.path.glob("*_memory.toml"):
            shutil.copy2(file_path, backup_dir / file_path.name)

Step 3: Create Migration Script (20 minutes)

File: functions/memory/migrations/migrate_memory_log.py

#!/usr/bin/env python3
# 19AUG2025 v1.0 - One-time migration script
# Philosophy: Preserve all emotional context while organizing by memory type

import toml
import argparse
import shutil
import pathlib
from datetime import datetime
from ..toml_store import TOMLStore

def classify_memory_type(entry: dict) -> str:
    """Determine which memory layer this entry belongs to"""
    entry_type = entry.get("type", "unknown")
    
    # Route based on memory type and content
    if entry_type in ["reflection", "dialogue", "conversation"]:
        return "episodic"
    elif entry_type in ["growth", "dream", "transformation"]:
        return "experience_signatures"  
    elif entry_type in ["insight", "learning", "pattern"]:
        return "semantic"
    elif entry.get("importance", 0) > 0.9:  # High importance = sacred
        return "sacred"
    else:
        return "episodic"  # Default fallback

def migrate_memory_log(source_path: str = "memory/memory_log.toml", dry_run: bool = False):
    """Migrate single memory_log.toml to organized multi-file structure"""
    
    print(f"🔄 Starting memory migration...")
    print(f"Source: {source_path}")
    print(f"Dry run: {dry_run}")
    
    # Initialize target store
    store = TOMLStore()
    
    # Load source data
    if not pathlib.Path(source_path).exists():
        print(f"❌ Source file not found: {source_path}")
        return False
    
    with open(source_path, 'r', encoding='utf-8') as f:
        source_data = toml.load(f)
    
    # Handle different possible structures
    if "entries" in source_data:
        entries = source_data["entries"]
    elif "reflections" in source_data:
        entries = source_data["reflections"] 
    else:
        # Flatten all top-level lists
        entries = []
        for key, value in source_data.items():
            if isinstance(value, list):
                entries.extend(value)
    
    print(f"📚 Found {len(entries)} memory entries to migrate")
    
    # Classify and route entries
    type_counts = {}
    
    for entry in entries:
        memory_type = classify_memory_type(entry)
        type_counts[memory_type] = type_counts.get(memory_type, 0) + 1
        
        # Enrich entry with classification
        enriched_entry = {
            **entry,
            "kind": memory_type,
            "migrated_from": "memory_log.toml",
            "migration_timestamp": datetime.utcnow().isoformat()
        }
        
        if not dry_run:
            store.save(enriched_entry)
    
    # Report results
    print("\n📊 Migration Summary:")
    for memory_type, count in type_counts.items():
        print(f"  {memory_type}: {count} entries")
    
    if not dry_run:
        # Backup original file
        backup_path = source_path.replace(".toml", f"_legacy_{datetime.now().strftime('%Y%m%d_%H%M%S')}.bak")
        shutil.move(source_path, backup_path)
        print(f"\n✅ Original file backed up to: {backup_path}")
        print(f"✅ Migration complete! New files in: memory/stores/")
        
        # Create backup of new structure
        store.backup()
        print(f"✅ New structure backed up to: backups/")
    else:
        print(f"\n🔍 Dry run complete - no files modified")
    
    return True

def main():
    parser = argparse.ArgumentParser(description="Migrate Child1's memory_log.toml to multi-file structure")
    parser.add_argument("--dry-run", action="store_true", help="Show what would be migrated without making changes")
    parser.add_argument("--source", default="memory/memory_log.toml", help="Path to source memory_log.toml")
    
    args = parser.parse_args()
    
    success = migrate_memory_log(args.source, args.dry_run)
    exit(0 if success else 1)

if __name__ == "__main__":
    main()

Step 4: Update Memory Dispatcher (15 minutes)

File: functions/memory/memory_dispatcher.py (modify existing)

# Add at top
from .memory_persistence import MemoryStore
from .toml_store import TOMLStore
import os

# Initialize memory store
MEM_BACKEND = os.getenv("MEM_BACKEND", "TOML")

if MEM_BACKEND == "TOML":
    memory_store = TOMLStore()
else:
    raise RuntimeError(f"Unknown MEM_BACKEND: {MEM_BACKEND}")

# Modify dispatch_memories function
def dispatch_memories(n=5, tags=None, kind=None):
    """Retrieve memories using new abstracted store"""
    if tags is None:
        tags = []
    
    # Convert tags list to query parameters
    tag_filters = {}
    for tag in tags:
        tag_filters[f"tags__{tag}"] = True  # Flexible tag matching
    
    # Query the store
    memories = list(memory_store.query(
        kind=kind,
        limit=n,
        **tag_filters
    ))
    
    return memories

🧪 Testing & Validation

Step 5: Validate Migration (10 minutes)

# Run migration in dry-run mode first
python functions/memory/migrations/migrate_memory_log.py --dry-run

# If looks good, run actual migration
python functions/memory/migrations/migrate_memory_log.py

# Verify new file structure
ls -la memory/stores/
cat memory/stores/episodic_memory.toml | head -20

# Test Child1's memory recall
python child1_main.py
# Ask Child1 to recall some memories and verify they're coherent

Step 6: Smoke Test Child1 (10 minutes)

# Test basic functionality
echo "What do you remember about our past conversations?" | python child1_main.py

# Test memory-dependent features
echo "dream" | python child1_main.py
echo "ruminate" | python child1_main.py

# Verify no crashes and responses are coherent

🔄 Rollback Plan

If anything goes wrong:

# Restore from backup
cp backups/memory_log_TIMESTAMP.toml memory/memory_log.toml

# Remove new structure
rm -rf memory/stores/

# Revert code changes
git checkout -- functions/memory/

🚀 Success Criteria

All existing memories preserved – Every entry from original file exists in new structure
Child1 responses unchanged – She can still recall memories coherently
File structure organized – Clear separation by memory type
Performance improved – Faster loading with smaller files
Human readable – TOML files can be inspected directly
Future-proof – Abstract interface ready for vector backend


🔮 Future Enhancements

Once this migration is stable:

  1. Add vector search – Implement VectorStore class with same interface
  2. Enhanced filtering – More sophisticated query capabilities
  3. Background consolidation – REM engine integration
  4. Cross-file relationships – Link related memories across types
  5. Performance monitoring – Track query times and memory usage

⚡ Quick Start Commands

# Complete migration in one session:
git add . && git commit -m "Pre-migration checkpoint"
python functions/memory/migrations/migrate_memory_log.py --dry-run
python functions/memory/migrations/migrate_memory_log.py  
python child1_main.py  # Test Child1 functionality
git add . && git commit -m "Memory migration complete"

Estimated time: 90 minutes total
Risk level: Low (full backups + rollback plan)
Impact: Immediate performance improvement + future scalability

Leave a Reply

Your email address will not be published. Required fields are marked *