Memory Integration

Your Workflow

Before Agent Work (Retrieval)

1. Search CKS for similar past solutions
2. Check session cache for coordination state
3. Inject relevant context into subagent prompts

During Agent Work (Coordination)

1. Use MemoryCacheManager for in-process state sharing
2. Set status/progress keys for cross-agent coordination

After Agent Work (Storage)

1. Identify learnings, corrections, patterns from work
2. Store appropriate entries in CKS with metadata
3. Index for future retrieval

Validation Rules

Prohibited Actions

• Do NOT store unverified speculation as learnings
• Do NOT use session cache for persistent knowledge
• Do NOT skip context retrieval before complex tasks

When to Use Memory

Before Agent Work (Retrieval)

Always retrieve relevant context before dispatching subagents when:

1. Problem-solving tasks - Search for similar past solutions

   from src.lib.memory.coordinator import MemoryCoordinator
   
   memory = MemoryCoordinator()
   context = memory.get_context(task_description)
   # Returns: corrections, patterns, learnings from src.cks
   

2. Multi-agent tasks - Check for previous coordination patterns

   # Get session state
   state = memory.cache_get("multi_agent:state")
   # Returns: current coordination state (if exists)
   

During Agent Work (Coordination)

Use MemoryCacheManager for in-process coordination:

from lib.context.persistence.memory_cache import MemoryCoordinator

cache = MemoryCacheManager()

# Agent 1: Finds available API quota
cache.set("quota:available", {"anthropic": 1000, "openrouter": 50000})

# Agent 2: Routes based on quota (no messaging needed)
quota = cache.get("quota:available")
if task_size > quota["anthropic"]:
    return "openrouter"

Session coordination patterns:

• cache.set(f"agent:{agent_name}:status", "working")
• cache.set(f"task:{task_id}:progress", {"step": 3, "total": 5})
• cache.get("current_user_preference")

After Agent Work (Storage)

Store learnings in CKS when:

from csf.cks.unified import CKS

cks = CKS()

# Store a correction
cks.ingest_correction(
    title="Don't concatenate /TN flag in schtasks",
    content="The /TN flag and task name must be separate arguments: ['/TN', task_name] not [f'/TN{task_name}']",
    context="task_manager.py fix"
)

# Store a pattern
cks.ingest_pattern(
    title="Session Coordination with MemoryCacheManager",
    content="Agents coordinate via shared cache without messaging. First agent sets state, second agent reads state. No explicit coordination needed.",
    category="multi-agent"
)

Decision Tree: CKS vs MemoryCacheManager

┌─────────────────────────────────────────────────────────────┐
│  Need to remember something...                                │
└─────────────────────────────────────────────────────────────┘
                            │
            ┌───────────────┴───────────────┐
            │                               │
     Should survive restart?          Only needed now?
            │                               │
            YES                              NO
            │                               │
    ┌───────┴───────┐              ┌───────┴───────┐
    │               │              │               │
   CKS           CKS          MemoryCache    MemoryCache
   (persistent)  (semantic)     (session)      (session)
                                          │
                              ┌───────────────┴───────────────┐
                              │                               │
                        Is it a learning?              Is it state?
                              │                               │
                              YES                             NO
                              │                               │
                        CKS entry_type:            Simple key-value
                      correction/learning        set(key, value)

Integration Points

1. Subagent Dispatch

When dispatching subagents, inject relevant memories:

from src.lib.memory.coordinator import MemoryCoordinator

memory = MemoryCoordinator()

# Get context for the subagent
context = memory.get_context(task_description)

# Add to subagent prompt
subagent_prompt = f"""
{task_description}

## Relevant Past Work:
{format_memories(context)}

Use these insights to avoid repeating mistakes.
"""

2. Task Execution

Before complex tasks, retrieve patterns:

# Search for relevant patterns
patterns = cks.search_patterns(task_type, limit=5)
if patterns:
    print(f"Found {len(patterns)} relevant patterns for {task_type}")

3. Post-Task Learning

After completing work, store learnings:

# What did we learn?
cks.ingest_learning(
    title=f"{project}: {task_type} optimization",
    content=f"Optimized {component} using {technique}. Result: {outcome}. ROI: {improvement}%",
    context=f"{project} {task_type}"
)

Examples

Example 1: Problem with Known Solution

# ORCHESTRATOR: Detects similar problem
memory = MemoryCoordinator()
corrections = memory.cks.search_corrections("FAISS slow", limit=3)

if corrections:
    # Direct solution from memory
    apply_fix(corrections[0]['content'])