communication

Tiered Communication System

Status: ✅ Implemented (feature/tiered-communication branch) Tests: 28 unit tests passing Race Conditions: 0 Coverage: Value semantics, reference semantics, auto-promotion, persistence

Overview

The tiered communication system enables efficient agent-to-agent messaging with intelligent routing between value-based and reference-based semantics. This solves the problem of passing large payloads (session state, query results, datasets) between agents without overwhelming the message bus.

Three-Tier Architecture

┌─────────────────────────────────────────────────────┐
│           Tier 1: Always Reference                  │
│  Large, persistent data (session_state, datasets)   │
│  → Always stored in SQLite/Redis with reference ID  │
└─────────────────────────────────────────────────────┘
                        ▼
┌─────────────────────────────────────────────────────┐
│           Tier 2: Auto-Promote                      │
│  Dynamic routing based on payload size threshold    │
│  Small: inline value | Large: reference storage     │
└─────────────────────────────────────────────────────┘
                        ▼
┌─────────────────────────────────────────────────────┐
│           Tier 3: Always Value                      │
│  Small, ephemeral data (control messages, acks)     │
│  → Always passed inline, never stored                │
└─────────────────────────────────────────────────────┘

Components

1. Reference Store Interface

Abstract storage backend for large message payloads:

type ReferenceStore interface {
    Store(ctx context.Context, data []byte, opts StoreOptions) (*loomv1.Reference, error)
    Resolve(ctx context.Context, ref *loomv1.Reference) ([]byte, error)
    Delete(ctx context.Context, refID string) error
    List(ctx context.Context) ([]*loomv1.Reference, error)
    Stats(ctx context.Context) (*StoreStats, error)
    Close() error
}

Implementations:

• ✅ MemoryStore: In-memory storage with TTL-based garbage collection
• ✅ SQLiteStore: Persistent storage with ref_counting and manual GC
• 📋 RedisStore: Planned for distributed deployments

2. Policy Manager

Determines routing strategy (value vs reference) based on message type:

type PolicyManager interface {
    GetPolicy(messageType string) *loomv1.CommunicationPolicy
    ShouldUseReference(messageType string, sizeBytes int64) bool
}

Built-in Policies:

• NewAlwaysReferencePolicy() - Always use reference storage
• NewAlwaysValuePolicy() - Always pass inline value
• NewAutoPromotePolicy(threshold) - Threshold-based routing (default: 10KB)
• NewSessionStatePolicy() - Always reference for session state
• NewWorkflowContextPolicy() - Always reference for workflow context
• NewToolResultPolicy() - Auto-promote for tool results

3. Agent Integration

Agents use Send() and Receive() methods for communication:

// pkg/agent/agent_communication.go

// Send sends a message to another agent using value or reference semantics
func (a *Agent) Send(ctx context.Context, toAgent, messageType string, data interface{}) (*loomv1.CommunicationMessage, error)

// Receive receives and resolves a message from another agent
func (a *Agent) Receive(ctx context.Context, msg *loomv1.CommunicationMessage) (interface{}, error)

Configuration via agent options:

agent := agent.NewAgent(backend, llm,
    agent.WithName("agent1"),
    agent.WithReferenceStore(refStore),
    agent.WithCommunicationPolicy(policyManager))

Usage Examples

Basic Value Message (Tier 3)

Small control messages passed inline:

// Create agents with shared reference store
policyManager := communication.NewPolicyManager()
refStore := communication.NewMemoryStore(5*time.Minute, 10*time.Minute)
defer refStore.Close()

agent1 := agent.NewAgent(backend, llm,
    agent.WithName("agent1"),
    agent.WithReferenceStore(refStore),
    agent.WithCommunicationPolicy(policyManager))

agent2 := agent.NewAgent(backend, llm,
    agent.WithName("agent2"),
    agent.WithReferenceStore(refStore),
    agent.WithCommunicationPolicy(policyManager))

// Agent 1 sends small control message
data := map[string]interface{}{
    "command": "start_processing",
    "priority": "high",
}
msg, err := agent1.Send(ctx, "agent2", "control", data)

// Message uses value semantics (no reference)
if msg.Payload.GetValue() != nil {
    fmt.Println("Using value semantics")
}

// Agent 2 receives inline value
received, err := agent2.Receive(ctx, msg)

Reference Message (Tier 1)

Large payloads stored in reference store:

// Configure policy for session_state
policyManager := communication.NewPolicyManager()
policyManager.RegisterPolicy("session_state", communication.NewSessionStatePolicy())

// Create SQLite store for persistence
refStore, err := communication.NewSQLiteStore("./data/refs.db", 3600)
defer refStore.Close()

agent1 := agent.NewAgent(backend, llm,
    agent.WithReferenceStore(refStore),
    agent.WithCommunicationPolicy(policyManager))

// Agent 1 sends large session state
largeData := map[string]interface{}{
    "session_id":   "session-123",
    "conversation": make([]string, 100),
    "metadata":     map[string]interface{}{...},
}
msg, err := agent1.Send(ctx, "agent2", "session_state", largeData)

// Message uses reference semantics
ref := msg.Payload.GetReference()
fmt.Printf("Stored as reference: %s\n", ref.Id)

// Agent2 resolves reference automatically
received, err := agent2.Receive(ctx, msg)

Auto-Promotion (Tier 2)

Automatic routing based on payload size:

// Default policy uses 10KB threshold
policyManager := communication.NewPolicyManager()
refStore := communication.NewMemoryStore(5*time.Minute, 10*time.Minute)

agent1 := agent.NewAgent(backend, llm,
    agent.WithReferenceStore(refStore),
    agent.WithCommunicationPolicy(policyManager))

// Small message → value semantics
smallData := map[string]interface{}{"status": "ok", "count": 42}
smallMsg, _ := agent1.Send(ctx, "agent2", "tool_result", smallData)
// smallMsg.Payload.GetValue() != nil

// Large message → reference semantics
largeData := map[string]interface{}{
    "data": make([]byte, 15*1024), // 15KB
}
largeMsg, _ := agent1.Send(ctx, "agent2", "tool_result", largeData)
// largeMsg.Payload.GetReference() != nil

Configuration

Server Configuration

The communication system is configured via server config:

// internal/config/config.go
type Config struct {
    Communication CommunicationConfig `yaml:"communication"`
}

type CommunicationConfig struct {
    Backend string            `yaml:"backend"` // "memory" or "sqlite"
    SQLite  SQLiteStoreConfig `yaml:"sqlite"`
}

type SQLiteStoreConfig struct {
    Path       string `yaml:"path"`
    GCInterval int    `yaml:"gc_interval"` // seconds
}

Example YAML:

communication:
  backend: sqlite
  sqlite:
    path: ./data/references.db
    gc_interval: 3600  # 1 hour

Default Configuration:

• Backend: SQLite
• Path: ./data/references.db
• GC Interval: 3600 seconds (1 hour)

Factory Function

Initialize reference store from config:

// pkg/communication/factory.go
refStore, err := communication.NewReferenceStoreFromConfig(cfg.Communication)
if err != nil {
    return fmt.Errorf("failed to create reference store: %w", err)
}
defer refStore.Close()

Reference Store Implementations

MemoryStore

In-memory storage with TTL-based garbage collection:

store := communication.NewMemoryStore(
    5*time.Minute,  // GC interval
    10*time.Minute, // Entry TTL
)
defer store.Close()

// Store data
opts := communication.StoreOptions{
    Type:        loomv1.ReferenceType_REFERENCE_TYPE_SESSION_STATE,
    ContentType: "application/json",
}
ref, err := store.Store(ctx, dataBytes, opts)

// Resolve reference
data, err := store.Resolve(ctx, ref)

Use Cases:

• Development and testing
• Short-lived sessions
• Single-server deployments

Limitations:

• No persistence across restarts
• Not suitable for distributed systems

SQLiteStore

Persistent storage with ref_counting:

store, err := communication.NewSQLiteStore(
    "./data/references.db",
    3600, // GC interval (seconds)
)
defer store.Close()

// Storage persists across sessions
ref, _ := store.Store(ctx, dataBytes, opts)

// Later session can resolve
data, _ := store.Resolve(ctx, ref)

// Check statistics
stats, _ := store.Stats(ctx)
fmt.Printf("Total refs: %d, Total bytes: %d\n",
    stats.TotalRefs, stats.TotalBytes)

Use Cases:

• Production deployments
• Session persistence
• Long-lived references
• Audit trails

Features:

• Reference counting (increment on resolve)
• Garbage collection based on ref_count
• Statistics tracking (total refs, total bytes)
• ACID guarantees
• WAL mode for concurrent access

Testing

Running Tests

# Run all communication tests
go test -race ./pkg/communication -v

# Run specific test
go test -race ./pkg/communication -run TestMemoryStore_StoreAndResolve -v

Test Coverage

Unit Tests (28 tests):

• MemoryStore: creation, store/resolve, deletion, GC, stats, concurrency
• SQLiteStore: creation, store/resolve, persistence, GC, stats, WAL mode
• PolicyManager: default policies, custom policies, policy lookup
• Factory: config parsing, backend selection, error handling

All tests run with -race detector to ensure zero race conditions.

Test Results

✅ All unit tests pass (28 tests)
✅ Zero race conditions detected
✅ Test execution time: ~8s

Architecture Decisions

Why Three Tiers?

1. Tier 1 (Always Reference): Large, persistent data that should never be inline

   • Session state (conversation history, context)
   • Datasets (query results, ML training data)
   • Workflow context (execution state)

2. Tier 2 (Auto-Promote): Dynamic routing based on runtime characteristics

   • Tool results (may be small text or large JSON)
   • API responses (vary widely in size)
   • Document chunks (depends on segmentation)

3. Tier 3 (Always Value): Small, ephemeral data that's cheaper inline

   • Control messages (start/stop/ack)
   • Status updates (progress, health checks)
   • Metadata (timestamps, agent IDs)

Why Reference IDs?

Reference IDs are SHA256 hashes of the serialized data, providing:

• Content-addressable storage (deduplication)
• Cryptographic integrity verification
• Deterministic ID generation
• No coordination required between agents

Why SQLite Default?

SQLite provides:

• Zero configuration (embedded database)
• ACID guarantees (data safety)
• Cross-session persistence (survive restarts)
• Efficient for < 1TB data (most agent workloads)
• No external dependencies (no Redis/Postgres required)

Performance Characteristics

Value Semantics (Tier 3)

• Latency: ~1μs (inline serialization)
• Throughput: 100k+ msgs/sec
• Best for: < 1KB payloads

Reference Semantics (Tier 1)

• Memory Store: ~10μs (hash + memory write)
• SQLite Store: ~1ms (hash + disk write)
• Best for: > 10KB payloads

Auto-Promote (Tier 2)

• Decision overhead: ~5μs (size check + policy lookup)
• Threshold: 10KB default (configurable)
• Best for: Variable-size payloads

Limitations and Future Work

Current Limitations

• ⚠️ No distributed reference store (Redis/etcd planned)
• ⚠️ Manual garbage collection for SQLite (automatic GC planned)
• ⚠️ No reference expiration policies (TTL planned)
• ⚠️ No compression for large payloads (LZ4/Zstd planned)

Planned Features

• 📋 RedisStore for distributed deployments
• 📋 Automatic garbage collection based on access time
• 📋 Compression for payloads > 100KB
• 📋 Reference expiration policies (TTL, LRU)
• 📋 Metrics export to observability (Hawk integration)

API Reference

Agent Methods

// Send sends a message to another agent
func (a *Agent) Send(
    ctx context.Context,
    toAgent string,      // Target agent name
    messageType string,  // Message type for policy lookup
    data interface{},    // Data to send (will be JSON marshaled)
) (*loomv1.CommunicationMessage, error)

// Receive receives and resolves a message
func (a *Agent) Receive(
    ctx context.Context,
    msg *loomv1.CommunicationMessage,
) (interface{}, error)  // Returns unmarshaled data

Agent Options

// WithReferenceStore sets the reference store for this agent
func WithReferenceStore(store communication.ReferenceStore) Option

// WithCommunicationPolicy sets the communication policy manager
func WithCommunicationPolicy(policy communication.PolicyManager) Option

Reference Store Interface

type ReferenceStore interface {
    // Store stores data and returns a reference
    Store(ctx context.Context, data []byte, opts StoreOptions) (*loomv1.Reference, error)

    // Resolve resolves a reference to data
    Resolve(ctx context.Context, ref *loomv1.Reference) ([]byte, error)

    // Delete removes a reference
    Delete(ctx context.Context, refID string) error

    // List lists all references
    List(ctx context.Context) ([]*loomv1.Reference, error)

    // Stats returns store statistics
    Stats(ctx context.Context) (*StoreStats, error)

    // Close closes the store
    Close() error
}

Policy Manager Interface

type PolicyManager interface {
    // GetPolicy returns the policy for a message type
    GetPolicy(messageType string) *loomv1.CommunicationPolicy

    // ShouldUseReference determines if reference should be used
    ShouldUseReference(messageType string, sizeBytes int64) bool

    // RegisterPolicy registers a custom policy
    RegisterPolicy(messageType string, policy *loomv1.CommunicationPolicy)
}

Troubleshooting

"Reference not found" errors

Cause: Reference was garbage collected or never stored

Solutions:

1. Increase GC interval in config
2. Check reference ID is correct
3. Verify reference store is shared between agents

High memory usage with MemoryStore

Cause: Large payloads stored in memory without GC

Solutions:

1. Switch to SQLiteStore for persistence
2. Reduce GC interval and TTL
3. Implement manual cleanup after message processing

Slow SQLite writes

Cause: Disk I/O bottleneck

Solutions:

1. Use WAL mode (enabled by default)
2. Increase SQLite cache size
3. Use tmpfs/RAM disk for database file
4. Consider Redis for distributed deployments

Branch: feature/tiered-communication Status: ✅ Implementation Complete Tests: 28 passing with -race Next: Integration tests and server deployment