multi-agent-memory

name	multi-agent-memory
description	Build shared, versioned memory systems for multi-agent teams using blackboard pattern, conflict resolution, and eventual consistency

Multi-Agent Memory Skill

When to activate

• Building multi-agent systems where agents need to access and modify shared state
• Implementing blackboard-pattern architectures for collaborative workflows
• Handling memory conflicts when multiple agents update the same data concurrently
• Designing agent memory that persists across multiple workflow executions
• Implementing agent team memory (episodic and semantic knowledge shared among agents)

When NOT to use

• Single-agent workflows with private memory (use in-context memory)
• Read-only workflows with no shared mutable state
• Workflows where agents never access each other's data

Instructions

Blackboard Data Structure

The blackboard is the central repository for all shared state:

{
  "session_id": "sess_abc123",
  "created_at": "2026-06-15T14:00:00Z",
  "phases": {
    "phase_1": {
      "name": "research",
      "status": "completed",
      "owner": "researcher",
      "version": 5,
      "data": {
        "sources": [...],
        "summary": "..."
      },
      "locked_by": null,
      "locked_until": null
    },
    "phase_2": {
      "name": "analysis",
      "status": "in_progress",
      "owner": "analyst",
      "version": 2,
      "data": {
        "themes": [...],
        "findings": [...]
      },
      "locked_by": "analyst",
      "locked_until": "2026-06-15T14:45:00Z"
    }
  },
  "conflicts": [],
  "audit_log": ".claude/blackboard-audit.jsonl"
}

Rules:

• Each phase has a single owner (the agent currently working on it)
• Phases are versioned (increment on every write)
• Agents must acquire a lock before writing
• Locks have timeouts (default 30 minutes)

Reading from Blackboard

def read_blackboard(phase_name, blackboard_file='.claude/blackboard.json'):
    """
    Read a phase from the blackboard.
    Returns the data and version number.
    """
    import json
    
    with open(blackboard_file, 'r') as f:
        blackboard = json.load(f)
    
    if phase_name not in blackboard['phases']:
        raise KeyError(f"Phase '{phase_name}' not found in blackboard")
    
    phase = blackboard['phases'][phase_name]
    return {
        'data': phase['data'],
        'version': phase['version'],
        'status': phase['status']
    }

Writing to Blackboard

Always follow this sequence:

def write_blackboard(phase_name, new_data, agent_name, blackboard_file='.claude/blackboard.json'):
    """
    Write to a phase in the blackboard.
    
    Steps:
    1. Acquire lock
    2. Check version (ensure no one else modified it)
    3. Merge changes (do not overwrite blindly)
    4. Write
    5. Release lock
    """
    import json
    from datetime import datetime, timedelta
    
    with open(blackboard_file, 'r') as f:
        blackboard = json.load(f)
    
    phase = blackboard['phases'][phase_name]
    
    # Step 1: Check if locked by another agent
    if phase['locked_by'] and phase['locked_by'] != agent_name:
        lock_expired = datetime.fromisoformat(phase['locked_until']) < datetime.now()
        if not lock_expired:
            raise RuntimeError(f"Phase locked by {phase['locked_by']} until {phase['locked_until']}")
    
    # Step 2: Acquire lock
    phase['locked_by'] = agent_name
    phase['locked_until'] = (datetime.now() + timedelta(minutes=30)).isoformat()
    
    # Step 3: Merge changes (for non-conflicting updates)
    merged_data = {**phase['data'], **new_data}
    
    # Step 4: Write
    phase['data'] = merged_data
    phase['version'] += 1
    phase['owner'] = agent_name
    
    # Log to audit
    log_to_audit(blackboard_file, {
        'timestamp': datetime.now().isoformat(),
        : agent_name,
        : phase_name,
        : phase[],
        : 
    })
    
    
    phase[] = 
    phase[] = 
    
     (blackboard_file, )  f:
        json.dump(blackboard, f)

Conflict Resolution

When multiple agents try to modify the same phase:

def detect_write_conflict(phase_name, version_read, blackboard_file='.claude/blackboard.json'):
    """
    Check if another agent has modified the phase since we read it.
    """
    import json
    
    with open(blackboard_file, 'r') as f:
        blackboard = json.load(f)
    
    phase = blackboard['phases'][phase_name]
    
    if phase['version'] > version_read:
        return {
            'conflict': True,
            'version_read': version_read,
            'version_current': phase['version'],
            'modified_by': phase['owner'],
            'action': 'merge' if is_mergeable(version_read, phase) else 'escalate'
        }
    
    return {'conflict': False}

def resolve_conflict(phase_name, local_data, version_read, resolution_strategy='merge'):
    """
    Resolve a write conflict.
    
    Strategies:
    - 'merge': Combine local and remote changes (for non-conflicting keys)
    - 'local': Keep local changes, discard remote
    - 'remote': Keep remote changes, discard local
    - 'escalate': Ask supervisor
    """
    import json
    
    with open('.claude/blackboard.json', 'r') as f:
        blackboard = json.load(f)
    
    phase = blackboard['phases'][phase_name]
    
     resolution_strategy == :
        merged = {**phase[], **local_data}
     resolution_strategy == :
        merged = local_data
     resolution_strategy == :
        merged = phase[]
     resolution_strategy == :
         ValueError()
    
    
    phase[] = merged
    phase[] += 
    
     (, )  f:
        json.dump(blackboard, f)
    
     phase[]

Memory Phases

Define phases for different types of work:

{
  "phases": {
    "research": {
      "name": "Information gathering",
      "dependencies": [],
      "expected_data": {
        "sources": {"type": "array"},
        "summary": {"type": "string"}
      }
    },
    "analysis": {
      "name": "Synthesizing findings",
      "dependencies": ["research"],
      "expected_data": {
        "themes": {"type": 
          
      
    
     
       
       
       
          
          
      
    
  

Agents can only read phases that have status: completed. If a phase is in_progress by another agent, wait for it to complete.

Audit Log

Log all blackboard operations for debugging:

{
  "timestamp": "2026-06-15T14:15:30Z",
  "agent": "researcher",
  "phase": "research",
  "operation": "write",
  "version_before": 4,
  "version_after": 5,
  "keys_modified": ["sources", "summary"],
  "conflict_detected": false,
  "status": "success"
}

Location: .claude/blackboard-audit.jsonl (append-only).

Example

# Agent 1: Researcher
def researcher_work():
    read_result = read_blackboard('research')
    if read_result['status'] != 'in_progress':
        # Start work on research phase
        sources = search_web("quantum computing")
        summary = summarize(sources)
        
        # Write to blackboard
        write_blackboard('research', {
            'sources': sources,
            'summary': summary
        }, agent_name='researcher')
        
        print("Research phase completed. Version: 5")

# Agent 2: Analyst (reads from research, writes to analysis)
def analyst_work():
    # Read research results
    research = read_blackboard('research')
    if research['status'] != 'completed':
        print("Waiting for research to complete...")
        return
    
    # Analyze
    themes = extract_themes(research['data']['sources'])
    findings = analyze(research['data'])
    
    # Check for conflicts
    conflict = detect_write_conflict('analysis', version_read=0)
    if conflict['conflict']:
        print(f"Conflict detected. Current version: {conflict['version_current']}")
        # Resolve via merge
        resolve_conflict('analysis', {
            : themes,
            : findings
        }, version_read=, resolution_strategy=)
    :
        
        write_blackboard(, {
            : themes,
            : findings
        }, agent_name=)
        
        ()

---