memory

Memory System

Package: memory Architecture: Event Sourcing + CQRS Test Coverage: 84.5% (event package)

Overview

The memory system provides persistent, intelligent memory capabilities for the AI assistant using Event Sourcing and CQRS patterns. It transforms conversations into a structured knowledge base that evolves over time with full auditability and correction support.

Architecture

The system implements Event Sourcing with clear separation between command and query responsibilities:

• Commands: All state changes are captured as immutable events
• Queries: Optimized read models built from event projections
• Eventual Consistency: Decoupled write and read paths for resilience

Event Sourcing Flow

User Input → Extraction → Decision → Event Creation → Event Store → Projection → Read Model
                                            ↓
                                      Event Bus → Async Projections

Why Event Sourcing?

1. Complete Audit Trail: Every change is recorded with who, what, when, why
2. Time Travel: Can reconstruct state at any point in time
3. Debugging: Can replay events to understand how state evolved
4. Correction Tracking: Natural fit for tracking memory corrections

Key Features

Event-Driven Processing

• All memory changes are recorded as immutable events (Created, Updated, Archived, Merged, Corrected)
• Complete audit trail with source tracking (user_direct, user_confirm, user_correction, assistant_infer)
• Support for event replay and temporal queries
• Version-controlled event store with optimistic locking

Memory Correction Tracking

• Explicit correction events with corrected_by relationships
• Source-based confidence adjustment (corrections get highest confidence)
• Correction chains support (memory A corrects B which corrected C)
• Filtered queries automatically exclude corrected memories

Intelligent Memory Management

• LLM-based fact extraction from conversations
• Semantic similarity detection using pgvector HNSW index (768-dimensional embeddings)
• Automatic conflict resolution and deduplication
• Relationship graph between related memories

Asynchronous Pipeline

• Non-blocking memory processing via job queue
• Retry logic with exponential backoff
• Dead letter queue for failed operations
• EventBus for decoupled projection updates

Knowledge Curation

• Background consolidation of similar memories
• Tiered deduplication strategy:
  • Identical (>0.98): Mark as duplicate and deactivate
  • Highly Similar (0.90-0.98): Merge memories
  • Related (0.80-0.90): Create relationship links
• Automatic archival of outdated information
• Conflict detection and resolution

Usage

// Create memory system
system, err := memory.NewSystem(db, aiClient, logger)
if err != nil {
    return err
}

// Start background processing
system.Start(ctx)
defer system.Stop()

// Process messages asynchronously
messages := []ai.Message{
    {Role: ai.User, Content: "I love coffee"},
    {Role: ai.Assistant, Content: "Great! I'll remember that"},
}
err = system.ProcessAsync(ctx, messages)

// Search memories
memories, err := system.Search(ctx, "coffee preferences", 10)

// Get specific memory
memory, err := system.GetByID(ctx, memoryID)

Components

Core System

• system.go - Public API (System) with Start/Stop lifecycle
• orchestrator.go - Coordinates processing flow and event projection
• processor.go - Pure business logic (Facts → Events) with correction detection

Event Sourcing (event/ subpackage)

• event/event.go - Event types (Created, Updated, Archived, Merged, Corrected)
• event/store.go - Event store with version control and optimistic locking
• event/replay.go - Event replay functionality for rebuilding state
• event/history.go - Event history and audit trail

Projection & Read Model

• projector.go - Projects events to read model with duplicate detection
• reader.go - Query interface with correction filtering and confidence adjustment
• query.go - Advanced query capabilities with vector search

Memory Processing

• extractor.go - LLM-based fact extraction with correction pattern detection
• decider.go - Decision logic (ADD/UPDATE/DELETE/NONE) for memory actions
• processor.go - Processes facts into events with deduplication

Session Management

• session.go - Manages conversation context and temporary state
• context.go - Provides context building for LLM interactions

Infrastructure

• task.go - Async job processing with retry logic
• errors.go - Typed error definitions
• constants.go - System constants and thresholds

Testing

The memory system has comprehensive test coverage (84.5%) including:

Unit Tests

• *_test.go - Standard unit tests for each component
• processor_fuzz_test.go - Fuzzing tests for input validation
• Mock implementations for AI clients and embedders

Integration Tests

• Database integration with PostgreSQL + pgvector
• Event store with version control
• End-to-end memory processing pipeline

Test Files

• event/event_test.go - Event construction and marshaling (100% coverage)
• event/store_test.go - Event store operations (LoadFrom, GetCurrentVersion, etc.)
• event/replay_test.go - Event replay and state reconstruction
• event/history_test.go - Event history and audit trail
• reader_test.go - Memory queries with correction filtering
• projector_test.go - Event projection to read model
• processor_test.go - Fact processing and correction detection
• extractor_test.go - Fact extraction from messages
• decider_test.go - Decision logic for memory actions

Run tests with:

TEST_DATABASE_URL="postgres://postgres:postgres@localhost:5432/assistant_test?sslmode=disable" \
  go test -tags integration -v ./internal/memory/...

Generate coverage report:

go test -tags integration -coverprofile=coverage.out ./internal/memory/...
go tool cover -html=coverage.out -o coverage.html

Configuration

The system uses sensible defaults but can be configured:

// Custom configuration example
config := memory.Config{
    MaxWorkers:     8,        // Concurrent processors
    BatchSize:      10,       // Events per batch
    RetryAttempts:  3,        // Retry failed operations
    ProjectionMode: "async",  // async or sync
}

Database Schema

Requires PostgreSQL 14+ with pgvector extension:

Tables

• memory_events - Event store (append-only) with optimistic locking

  • Stores all events with version control
  • Immutable audit trail

• memories - Read model (CQRS projections)

  • Current state of memories
  • HNSW index on embedding column for fast similarity search
  • Tracks corrected_by relationships
  • Source tracking (user_direct, user_correction, etc.)

Key Columns

• semantic_id - Stable identifier for memory updates
• embedding vector(768) - Gemini embeddings for similarity search
• supersedes uuid[] - Memories this one replaces
• corrected_by uuid - Reference to correcting memory
• source memory_source - How memory was created

See migrations in internal/storage/database/migrations/:

• 001_initial_schema.up.sql - Base tables
• 002_memory_event_sourcing.up.sql - Event store
• 003_memory_source_tracking.up.sql - Correction tracking

Performance Considerations

• Vector embeddings are generated asynchronously
• Batch processing for improved throughput
• Connection pooling for database efficiency
• In-memory EventBus for low latency

Implementation Details

Event Store with Optimistic Locking

The event store ensures consistency using version-based optimistic locking:

CREATE TABLE memory_events (
    id UUID PRIMARY KEY,
    stream_id UUID NOT NULL,       -- Memory ID
    type VARCHAR(50) NOT NULL,     -- Event type
    version INT NOT NULL,          -- Version for optimistic locking
    timestamp TIMESTAMPTZ NOT NULL,
    data JSONB NOT NULL,           -- Event payload
    
    CONSTRAINT unique_stream_version UNIQUE (stream_id, version)
);

Memory Attributes System

Strongly-typed attributes replacing map[string]interface{}:

type MemoryAttributes struct {
    Source            MemorySource `json:"source"`
    IsCorrection      bool        `json:"is_correction"`
    CorrectionPattern string      `json:"correction_pattern"`
    Category          string      `json:"category"`
    Tags              []string    `json:"tags"`
    Importance        float32     `json:"importance"`
    // ... tracking and relationships
}

Trust Levels:

• user_correction (Trust: 4) - User explicitly corrected
• user_confirm (Trust: 3) - User confirmed
• user_direct (Trust: 2) - User directly stated
• assistant_infer (Trust: 1) - Assistant inferred

Correction Detection Patterns

The system detects corrections through pattern matching:

// Chinese: "不是...是", "錯了", "正確是", "其實是"
// English: "no, it's actually", "wrong, it's"

When corrections are detected, memories get the highest trust level and supersede previous incorrect memories.

Vector Search with HNSW Index

Optimized for fast similarity search:

CREATE INDEX idx_memories_embedding 
ON memories 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

Hybrid scoring combines:

• Vector similarity (70%)
• Recency boost (up to 15%)
• Frequency boost (up to 10%)
• Source trust boost (up to 5%)

Parallel Processing

The system uses parallel processing for performance:

g, ctx := errgroup.WithContext(ctx)
semaphore := make(chan struct{}, maxConcurrency)

for _, fact := range facts {
    g.Go(func() error {
        // Acquire semaphore for concurrency control
        select {
        case semaphore <- struct{}{}:
            defer func() { <-semaphore }()
        case <-ctx.Done():
            return ctx.Err()
        }
        // Process fact...
    })
}