Manual Memory Management Guide¶

add_memory(
    project_id="my-project",
    memory_type="decision",
    title="Use JWT for authentication",
    content="Decided to use JWT tokens instead of session-based authentication",
    reason="Need stateless authentication for mobile clients and microservices",
    importance=0.9,
    tags=["auth", "security", "architecture"]
)

from src.codebase_rag.services.memory import memory_store

result = await memory_store.add_memory(
    project_id="my-project",
    memory_type="decision",
    title="Use JWT for authentication",
    content="Decided to use JWT tokens",
    reason="Need stateless auth for mobile clients",
    importance=0.9,
    tags=["auth", "security"]
)

memory_id = result['memory_id']

add_memory(
    project_id="web-app",
    memory_type="decision",
    title="Adopt microservices architecture",
    content="Transitioning from monolith to microservices for user service, order service, and payment service",
    reason="Need independent scaling and deployment. User service has different load patterns than order service",
    importance=0.95,
    tags=["architecture", "microservices", "scaling"],
    related_refs=[
        "ref://file/docs/architecture/microservices.md",
        "ref://file/services/user-service/README.md"
    ]
)

add_memory(
    project_id="web-app",
    memory_type="preference",
    title="Use functional components in React",
    content="Team prefers functional components with hooks over class components",
    reason="Hooks provide better code reuse and easier testing. Team is more familiar with functional approach",
    importance=0.6,
    tags=["react", "frontend", "coding-style"]
)

add_memory(
    project_id="web-app",
    memory_type="experience",
    title="Redis connection timeout in Docker Compose",
    content="Redis connections were timing out when using 'localhost:6379' in Docker environment",
    reason="Docker Compose networking requires using service name 'redis:6379' instead of 'localhost'. Docker creates an internal network where services communicate by name",
    importance=0.7,
    tags=["docker", "redis", "networking", "deployment"],
    related_refs=["ref://file/docker-compose.yml#L15"]
)

add_memory(
    project_id="web-app",
    memory_type="convention",
    title="API endpoints use kebab-case",
    content="All REST API endpoints must use kebab-case naming convention. Example: /api/user-profiles instead of /api/userProfiles",
    reason="Consistency across API, better readability in URLs",
    importance=0.5,
    tags=["api", "naming", "conventions"]
)

add_memory(
    project_id="web-app",
    memory_type="plan",
    title="Add rate limiting to public API",
    content="Plan to implement rate limiting on all public API endpoints. Use Redis-based rate limiter with sliding window algorithm. Limit: 100 requests per minute per IP",
    reason="Prevent API abuse and ensure fair usage",
    importance=0.6,
    tags=["api", "security", "todo", "performance"]
)

add_memory(
    project_id="web-app",
    memory_type="note",
    title="Production database backup location",
    content="Production PostgreSQL backups are stored in S3 bucket 'prod-db-backups' with 30-day retention. Daily backups run at 2 AM UTC",
    reason="Critical operational information for disaster recovery",
    importance=0.7,
    tags=["operations", "backup", "production"]
)

add_memory(
    project_id="api-service",
    memory_type="decision",
    title="Use dependency injection for services",
    content="Implemented dependency injection pattern for all service classes",
    reason="Improves testability and reduces coupling",
    importance=0.8,
    tags=["architecture", "patterns"],
    related_refs=[
        "ref://file/src/core/container.py",
        "ref://file/src/services/user_service.py#L25",
        "ref://symbol/UserService",
        "ref://symbol/inject_dependencies"
    ]
)

add_memory(
    project_id="web-app",
    memory_type="decision",
    title="Switch to PostgreSQL 15",
    content="Upgraded from PostgreSQL 13 to 15",
    reason="Better performance and new JSON features",
    importance=0.8,
    tags=["database", "upgrade"],
    metadata={
        "migration_date": "2024-11-01",
        "downtime": "5 minutes",
        "rollback_plan": "restore from backup",
        "approved_by": "tech-lead"
    }
)

get_memory(memory_id="550e8400-e29b-41d4-a716-446655440000")

result = await memory_store.get_memory("550e8400-e29b-41d4-a716-446655440000")

if result['success']:
    memory = result['memory']
    print(f"Title: {memory['title']}")
    print(f"Content: {memory['content']}")
    print(f"Related refs: {memory['related_refs']}")

update_memory(
    memory_id="550e8400-e29b-41d4-a716-446655440000",
    importance=0.95,
    tags=["auth", "security", "critical", "production"]
)

result = await memory_store.update_memory(
    memory_id="550e8400-e29b-41d4-a716-446655440000",
    title="Use JWT with refresh token rotation",
    content="Updated implementation to include refresh token rotation",
    importance=0.95,
    tags=["auth", "security", "critical"]
)

# Security vulnerability discovered
update_memory(
    memory_id=auth_memory_id,
    importance=1.0,
    tags=["auth", "security", "critical", "vulnerability"]
)

# Found additional information
update_memory(
    memory_id=redis_memory_id,
    content=original_content + "\n\nUpdate: Also affects Redis Sentinel configuration. Must use sentinel service names.",
    reason=original_reason + " Additionally, Sentinel failover requires proper service name configuration."
)

# Initially thought important, but turned out routine
update_memory(
    memory_id=config_memory_id,
    importance=0.4  # Downgrade from 0.7
)

delete_memory(memory_id="550e8400-e29b-41d4-a716-446655440000")

result = await memory_store.delete_memory("550e8400-e29b-41d4-a716-446655440000")

if result['success']:
    print("Memory deleted")
else:
    print(f"Error: {result['error']}")

supersede_memory(
    old_memory_id="abc-123-def-456",
    new_memory_type="decision",
    new_title="Migrate from MySQL to PostgreSQL",
    new_content="Migrated from MySQL to PostgreSQL for production database",
    new_reason="Need advanced JSON support, full-text search, and better geospatial features",
    new_tags=["database", "postgresql", "migration"],
    new_importance=0.9
)

result = await memory_store.supersede_memory(
    old_memory_id="abc-123-def-456",
    new_memory_data={
        "memory_type": "decision",
        "title": "Migrate from MySQL to PostgreSQL",
        "content": "Migrated from MySQL to PostgreSQL",
        "reason": "Need advanced features",
        "tags": ["database", "postgresql"],
        "importance": 0.9
    }
)

new_memory_id = result['new_memory_id']
old_memory_id = result['old_memory_id']

mysql_memory = add_memory(
    project_id="web-app",
    memory_type="decision",
    title="Use MySQL as primary database",
    content="Selected MySQL for application database",
    reason="Team familiarity and existing infrastructure",
    importance=0.7,
    tags=["database", "mysql"]
)
# memory_id: "original-123"

postgres_v1 = supersede_memory(
    old_memory_id="original-123",
    new_memory_type="decision",
    new_title="Migrate to PostgreSQL",
    new_content="Migrating from MySQL to PostgreSQL",
    new_reason="Need better JSON support and full-text search",
    new_tags=["database", "postgresql", "migration"],
    new_importance=0.8
)
# new_memory_id: "update-456"
# "original-123" now marked as superseded

postgres_v2 = supersede_memory(
    old_memory_id="update-456",
    new_memory_type="decision",
    new_title="PostgreSQL 15 in production",
    new_content="Completed migration to PostgreSQL 15. All services migrated",
    new_reason="Migration successful. Using advanced features: JSONB, GiST indexes, full-text search",
    new_tags=["database", "postgresql", "production"],
    new_importance=0.9
)
# new_memory_id: "final-789"
# "update-456" now marked as superseded

get_project_summary(project_id="web-app")

result = await memory_store.get_project_summary("web-app")

summary = result['summary']
print(f"Total memories: {summary['total_memories']}")

for memory_type, data in summary['by_type'].items():
    print(f"{memory_type}: {data['count']}")
    for mem in data['top_memories'][:3]:
        print(f"  - {mem['title']} (importance: {mem['importance']})")

# Start of session: Get context
async def start_session(project_id: str, task_area: str):
    # 1. Get project overview
    summary = await memory_store.get_project_summary(project_id)
    print(f"Project has {summary['summary']['total_memories']} memories")

    # 2. Search for relevant context
    context = await memory_store.search_memories(
        project_id=project_id,
        query=task_area,
        min_importance=0.6
    )

    # 3. Review top decisions and experiences
    for memory in context['memories'][:5]:
        print(f"Relevant: {memory['title']}")

    return context

# During work: Check conventions
async def check_conventions(project_id: str, area: str):
    conventions = await memory_store.search_memories(
        project_id=project_id,
        memory_type="convention",
        tags=[area]
    )
    return conventions

# End of session: Save learnings
async def save_learnings(project_id: str, new_knowledge: dict):
    result = await memory_store.add_memory(
        project_id=project_id,
        **new_knowledge
    )
    return result['memory_id']

async def cleanup_low_value_memories(project_id: str):
    """Remove low-importance notes older than 6 months"""

    # Search for low-importance notes
    old_notes = await memory_store.search_memories(
        project_id=project_id,
        memory_type="note",
        min_importance=0.0,
        limit=100
    )

    from datetime import datetime, timedelta
    cutoff_date = datetime.utcnow() - timedelta(days=180)

    deleted_count = 0
    for memory in old_notes['memories']:
        created_at = datetime.fromisoformat(memory['created_at'])

        if memory['importance'] < 0.3 and created_at < cutoff_date:
            await memory_store.delete_memory(memory['id'])
            deleted_count += 1
            print(f"Deleted old note: {memory['title']}")

    return deleted_count

async def migrate_memory(memory_id: str, from_project: str, to_project: str):
    """Copy memory from one project to another"""

    # Get original memory
    result = await memory_store.get_memory(memory_id)
    memory = result['memory']

    # Create copy in new project
    new_result = await memory_store.add_memory(
        project_id=to_project,
        memory_type=memory['type'],
        title=f"[Migrated] {memory['title']}",
        content=memory['content'],
        reason=memory.get('reason'),
        tags=memory.get('tags', []) + ['migrated'],
        importance=memory.get('importance', 0.5),
        metadata={
            **memory.get('metadata', {}),
            'migrated_from': from_project,
            'original_memory_id': memory_id
        }
    )

    return new_result['memory_id']

async def organize_by_tags(project_id: str):
    """Get memories organized by tag"""

    # Get all memories
    all_memories = await memory_store.search_memories(
        project_id=project_id,
        limit=100
    )

    # Organize by tag
    by_tag = {}
    for memory in all_memories['memories']:
        for tag in memory.get('tags', []):
            if tag not in by_tag:
                by_tag[tag] = []
            by_tag[tag].append({
                'id': memory['id'],
                'title': memory['title'],
                'importance': memory['importance']
            })

    # Sort tags by memory count
    sorted_tags = sorted(
        by_tag.items(),
        key=lambda x: len(x[1]),
        reverse=True
    )

    print("Top tags:")
    for tag, memories in sorted_tags[:10]:
        print(f"{tag}: {len(memories)} memories")

    return by_tag

# Critical architecture decision
importance=0.95

# Important feature decision
importance=0.8

# Team preference
importance=0.6

# Minor convention
importance=0.4

# Future plan
importance=0.3

# Initially seemed important
add_memory(..., importance=0.7)

# Later found to be critical
update_memory(memory_id, importance=0.95)

tags = [
    "auth",           # Domain
    "security",       # Category
    "jwt",            # Technology
    "production",     # Environment
    "critical"        # Status
]

# ❌ Too vague
tags = ["backend", "code"]

# ✅ Specific and useful
tags = ["auth", "jwt", "refresh-token", "security"]

# ❌ Too brief
content = "Using Redis"

# ✅ Comprehensive
content = """
Implementing Redis as caching layer for user sessions and API responses.

Configuration:
- Redis 7.0 in cluster mode
- 3 master nodes, 3 replicas
- Maxmemory policy: allkeys-lru
- Persistence: RDB + AOF

Cache strategy:
- User sessions: TTL 24h
- API responses: TTL 5min
- Invalidation on data updates
"""

# ❌ No rationale
reason = "Better performance"

# ✅ Clear rationale
reason = """
Need to reduce database load by 70%. Current response times averaging 500ms,
target is 100ms. Redis provides:
1. Sub-millisecond latency
2. Horizontal scaling
3. Built-in data structures
4. Proven at scale by similar companies
"""

related_refs = [
    "ref://file/src/cache/redis_client.py",
    "ref://file/config/redis.yml",
    "ref://file/docs/cache-strategy.md"
]

# ✅ Point to exact implementation
related_refs = [
    "ref://file/src/auth/jwt.py#L45",  # JWT generation
    "ref://symbol/verify_token"         # Verification function
]

# Monthly review
async def monthly_review(project_id: str):
    # Check for outdated plans
    plans = await memory_store.search_memories(
        project_id=project_id,
        memory_type="plan",
        min_importance=0.0
    )

    # Review and update or delete completed plans
    for plan in plans['memories']:
        print(f"Review: {plan['title']}")
        # Manual decision: delete, update, or keep

# When decision evolves, supersede instead of update
# Preserves history
await memory_store.supersede_memory(old_id, new_data)

async def feature_workflow(project_id: str, feature_name: str):
    # 1. Check existing decisions
    decisions = await memory_store.search_memories(
        project_id=project_id,
        query=feature_name,
        memory_type="decision"
    )

    # 2. Check conventions
    conventions = await memory_store.search_memories(
        project_id=project_id,
        memory_type="convention"
    )

    # 3. Check past experiences
    experiences = await memory_store.search_memories(
        project_id=project_id,
        query=feature_name,
        memory_type="experience"
    )

    # 4. Implement feature...

    # 5. Save new knowledge
    await memory_store.add_memory(
        project_id=project_id,
        memory_type="decision",
        title=f"Implemented {feature_name}",
        content="...",
        reason="...",
        importance=0.7
    )

async def document_bug_fix(project_id: str, bug_description: str, solution: str):
    # Save the experience
    result = await memory_store.add_memory(
        project_id=project_id,
        memory_type="experience",
        title=f"Bug: {bug_description}",
        content=f"Problem: {bug_description}\n\nSolution: {solution}",
        reason="Prevent recurrence of this issue",
        importance=0.7,
        tags=["bug", "debugging"]
    )

    return result['memory_id']

result = await memory_store.get_memory(memory_id)

if not result['success']:
    if "not found" in result['error'].lower():
        print("Memory doesn't exist or was deleted")
    else:
        print(f"Error: {result['error']}")

# Make sure at least one field is provided
result = await memory_store.update_memory(
    memory_id=memory_id,
    importance=0.9  # At least one field required
)

if not result['success']:
    if "No updates" in result['error']:
        print("Must provide at least one field to update")

# Try broader search
result = await memory_store.search_memories(
    project_id=project_id,
    query="auth",  # Remove filters
    limit=100      # Increase limit
)

if result['total_count'] == 0:
    print("No memories found for this project")