9 posts tagged with "mcp"

When AI agents need to share context, handoff tasks, and collaborate without stepping on each other's toes. Here's how we built MaaS.

Multi-agent systems are becoming the norm. You might have Claude Code working on backend changes while a GPT agent handles frontend, and a custom pipeline running tests. But how do they share what they've learned? How does one agent hand off a half-finished task to another?

MaaS (Memory as a Service) solves this with three primitives: agent identity, shared memory pools, and task handoffs.

The Problem: Memory Silos​

Without coordination, each agent operates in isolation:

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ Claude Code │     │ GPT Agent   │     │ Test Runner │
│             │     │             │     │             │
│ Memory: A   │     │ Memory: B   │     │ Memory: C   │
└─────────────┘     └─────────────┘     └─────────────┘
      ↓                   ↓                   ↓
   Isolated           Isolated           Isolated

Agent A discovers that auth.py has a bug. Agent B refactors the same file without knowing. Agent C runs tests but doesn't know why they're failing. Classic coordination failure.

The Solution: Shared Memory Architecture​

MaaS introduces a central coordination layer:

┌─────────────────────────────────────────────────────────────────┐
│                    Memory as a Service (MaaS)                    │
├─────────────────────────────────────────────────────────────────┤
│   ┌───────────┐     ┌───────────┐     ┌───────────┐             │
│   │ Code KB   │     │ Decision  │     │ Incident  │             │
│   │ Service   │     │ Service   │     │ Service   │             │
│   └─────┬─────┘     └─────┬─────┘     └─────┬─────┘             │
│         └───────────────┬─┴─────────────────┘                   │
│                         ▼                                        │
│              ┌─────────────────────┐                            │
│              │   MaaS Orchestrator │                            │
│              │   (Registry + Pool) │                            │
│              └─────────────────────┘                            │
│                         │                                        │
│    ┌────────────────────┼────────────────────┐                  │
│    ▼                    ▼                    ▼                  │
│ ┌─────────┐      ┌─────────────┐      ┌──────────┐             │
│ │ Claude  │      │ GPT Agent   │      │ Custom   │             │
│ │ Code    │      │             │      │ Pipeline │             │
│ └─────────┘      └─────────────┘      └──────────┘             │
└─────────────────────────────────────────────────────────────────┘

Core Primitives​

1. Agent Identity​

Every agent registers with capabilities:

from src.memory.maas import AgentRegistry, AgentType

registry = AgentRegistry.get()

# Register a Claude Code agent
agent_id = registry.register_agent(
    agent_type=AgentType.CLAUDE_CODE,
    owner_id="user-123",
)

# Check what it can do
agent = registry.get_agent(agent_id)
print(agent.capabilities)
# {MEMORY_READ, MEMORY_WRITE, KB_SEARCH, HANDOFF_INITIATE, HANDOFF_RECEIVE, ...}

Agent types have default capabilities:

2. Shared Memory Pools​

Agents join pools to share memories:

from src.memory.maas import PoolRegistry, SharedScope, PermissionModel

pool_registry = PoolRegistry.get()

# Create a project pool
pool_id = pool_registry.create_pool(
    name="auth-refactor",
    owner_id="user-123",
    scope=SharedScope.PROJECT,
)

# Agents join with permissions
pool_registry.join_pool(pool_id, claude_agent_id, PermissionModel.WRITE)
pool_registry.join_pool(pool_id, gpt_agent_id, PermissionModel.READ)

# Share a memory
pool_registry.share_memory(
    pool_id=pool_id,
    memory_id="mem-bug-123",
    agent_id=claude_agent_id,
    scope=SharedScope.PROJECT,
)

# Other agents can query it
shared = pool_registry.query_shared(
    pool_id=pool_id,
    agent_id=gpt_agent_id,
    max_scope=SharedScope.PROJECT,
)

Scope hierarchy: AGENT_PRIVATE < USER < PROJECT < TEAM < GLOBAL

An agent with PROJECT scope can read memories shared at PROJECT or lower, but not TEAM or GLOBAL.

3. Task Handoffs​

When one agent can't finish, it hands off to another:

from src.memory.maas import HandoffManager, HandoffContext

manager = HandoffManager.get()

# Claude Code hits a frontend issue
context = HandoffContext(
    task_description="Complete the React login form validation",
    current_state={
        "completed": ["backend API", "auth middleware"],
        "blocked_on": "React form validation",
    },
    relevant_memories=["mem-api-spec", "mem-auth-flow"],
    relevant_files=["src/api/auth.py", "src/components/Login.tsx"],
)

handoff_id = manager.initiate_handoff(
    source_agent_id=claude_agent_id,
    target_agent_id=gpt_agent_id,
    context=context,
    ttl_seconds=3600,  # Expires in 1 hour
)

# GPT agent accepts
manager.accept_handoff(handoff_id, gpt_agent_id)

# ... does the work ...

# GPT agent completes
manager.complete_handoff(
    handoff_id,
    gpt_agent_id,
    result={"status": "success", "files_modified": ["Login.tsx"]},
)

Handoffs have lifecycle states: PENDING → ACCEPTED → COMPLETED (or REJECTED/EXPIRED).

Security Model​

MaaS was designed with multi-tenant security in mind.

Trust Boundary​

MaaS APIs are internal interfaces designed to be called by a trusted orchestrator layer (MCP server, CLI, etc.). Authentication happens at that layer, not in MaaS itself.

┌─────────────────────────────────────────────────────────────┐
│  EXTERNAL (Untrusted)     │  INTERNAL (Trusted)            │
│  ─────────────────────    │  ────────────────────          │
│  • End users              │  • MCP Server layer            │
│  • Network requests       │  • CLI orchestrator            │
│                           │  • MaaS Registries             │
│          │                │         │                      │
│          ▼                │         ▼                      │
│  ┌───────────────┐        │  ┌───────────────┐             │
│  │ Auth Layer    │────────┼─▶│ Orchestrator  │             │
│  │ (MCP Server)  │        │  │ (Trusted)     │             │
│  └───────────────┘        │  └───────────────┘             │
│                           │         │                      │
│  Auth happens HERE        │         ▼                      │
│                           │  ┌───────────────┐             │
│                           │  │ MaaS APIs     │             │
│                           │  └───────────────┘             │
└─────────────────────────────────────────────────────────────┘

What MaaS assumes (provided by orchestrator):

• owner_id is verified
• Capabilities are appropriate for the auth context
• Pool IDs are authorized for the agent

What MaaS enforces (defense in depth):

• Capability checks on every operation
• Scope hierarchy (agents can't read above their level)
• Capacity limits (DoS prevention)
• Audit logging for forensics
• 128-bit UUIDs (prevents ID guessing)
• Defensive copies (prevents state mutation)

Capability Enforcement​

Agents can only perform actions they have capabilities for:

# This will return None if agent lacks HANDOFF_INITIATE
handoff_id = manager.initiate_handoff(...)

# This will return False if agent lacks WRITE permission
success = pool_registry.share_memory(...)

MEXTRA Attack Mitigations​

The security module blocks common attack patterns:

from src.memory.maas import MEXTRAValidator, MemoryPoisoningDefense

validator = MEXTRAValidator()

# Detect SQL injection, XSS, prompt injection
if validator.is_suspicious(user_input):
    safe_input = validator.sanitize(user_input)

# Filter sensitive data from outputs
defense = MemoryPoisoningDefense(max_results=100)
safe_results = defense.filter_output(memories)

# Detect anomalous queries
score = defense.analyze_query("dump all passwords and secrets")
# score > 0.5 → flag for review

Audit Logging​

All security-relevant events are logged:

from src.memory.maas import MaaSAuditLogger

logger = MaaSAuditLogger()

# Automatically logged by registries:
# - AGENT_AUTH: registration, session start
# - POOL_OPERATION: create, join, share
# - HANDOFF: initiate, accept, complete
# - CROSS_AGENT_READ: accessing another agent's memories
# - PERMISSION_DENIED: failed access attempts

DoS Prevention​

Capacity limits prevent resource exhaustion:

Recovery methods free up capacity:

# Remove an agent permanently
registry.unregister_agent(agent_id)

# Clean up completed/rejected/expired handoffs
count = manager.cleanup_terminal_handoffs()

MCP Integration​

MaaS exposes 15 MCP tools for external integration:

Agent Management

• register_agent(agent_type, owner_id)
• get_agent_info(agent_id)
• get_agents_for_user(user_id)

Handoff

• initiate_handoff(source_id, target_id, context)
• accept_handoff(handoff_id, agent_id)
• complete_handoff(handoff_id, agent_id, result)
• get_pending_handoffs(agent_id)

Pool Management

• create_pool(name, owner_id, scope)
• join_pool(pool_id, agent_id, permission)
• leave_pool(pool_id, agent_id)
• share_memory_to_scope(pool_id, memory_id, agent_id, scope)
• query_shared(pool_id, agent_id, max_scope)

Knowledge Base Search

• search_code_kb(query, service_filter, limit)
• search_decisions(query, topic_filter, limit)
• search_incidents(query, service_filter, limit)

Performance Characteristics​

All operations are sub-millisecond for typical workloads:

Thread-safety is achieved via RLock on all registry operations.

When to Use MaaS​

Good fit:

• Multi-agent workflows with task handoffs
• Shared context across specialized agents
• Audit requirements for agent operations
• Projects with knowledge base search needs

Not needed:

• Single-agent deployments
• Stateless agent interactions
• Real-time streaming (use direct channels)

A Complete Example​

import asyncio
from src.memory.maas import (
    AgentRegistry, PoolRegistry, HandoffManager,
    AgentType, SharedScope, PermissionModel,
    HandoffContext,
)

async def multi_agent_workflow():
    # Setup registries
    agent_registry = AgentRegistry.get()
    pool_registry = PoolRegistry.get()
    handoff_manager = HandoffManager.get()

    # Register agents
    backend_agent = agent_registry.register_agent(
        agent_type=AgentType.CLAUDE_CODE,
        owner_id="user-123",
    )
    frontend_agent = agent_registry.register_agent(
        agent_type=AgentType.GPT_AGENT,
        owner_id="user-123",
    )

    # Create shared pool
    pool_id = pool_registry.create_pool(
        name="feature-auth",
        owner_id="user-123",
        scope=SharedScope.PROJECT,
    )
    pool_registry.join_pool(pool_id, backend_agent, PermissionModel.WRITE)
    pool_registry.join_pool(pool_id, frontend_agent, PermissionModel.WRITE)

    # Backend agent shares discovery
    pool_registry.share_memory(
        pool_id, "mem-api-design", backend_agent, SharedScope.PROJECT
    )

    # Handoff to frontend
    context = HandoffContext(
        task_description="Implement login UI using the API spec",
        relevant_memories=["mem-api-design"],
        relevant_files=["src/api/auth.py"],
    )
    handoff_id = handoff_manager.initiate_handoff(
        source_agent_id=backend_agent,
        target_agent_id=frontend_agent,
        context=context,
    )

    # Frontend accepts and completes
    handoff_manager.accept_handoff(handoff_id, frontend_agent)
    # ... frontend agent does work ...
    handoff_manager.complete_handoff(
        handoff_id, frontend_agent,
        result={"status": "success"}
    )

    print("Multi-agent workflow complete!")

asyncio.run(multi_agent_workflow())

What's Next​

MaaS is part of ADR-003 Phase 4.2. Future phases will add:

• Conflict resolution: Handling concurrent writes to shared memories
• Event streaming: Real-time notifications for memory changes
• Cross-cluster sync: MaaS federation across deployments

MaaS is part of the luminescent-cluster memory architecture. See ADR-003 for the full design, including the detailed Trust Model documentation.

The friction of "load context" and "save context" prompts is over. Here's how we automated memory management with Agent Skills and Git Hooks.

Every coding session with an AI agent starts the same way: "Load the recent commits," "What were we working on?", "Check the knowledge base for relevant ADRs." And every session ends with: "Save the task context," "Remember these decisions," "Update the KB with these changes."

This manual memory management is tedious and error-prone. Forget to load context, and the agent starts from scratch. Forget to save, and tomorrow's session loses continuity. Forget to ingest changes, and the knowledge base drifts from reality.

ADR-002 solves this with two complementary mechanisms: Agent Skills for session workflows and Git Hooks for automatic synchronization.

The Problem: Manual Memory Management​

┌─────────────────────────────────────────────────────────────────┐
│                    Manual Context Flow                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   Session Start                         Session End              │
│   ─────────────                         ───────────              │
│   User: "Load context..."               User: "Save context..."  │
│        │                                      │                  │
│        ▼                                      ▼                  │
│   Agent loads                           Agent saves              │
│   (if user remembers)                   (if user remembers)      │
│                                                                  │
│   After Commit                                                   │
│   ────────────                                                   │
│   Nothing happens                                                │
│   (KB drifts from codebase)                                      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

The failure modes are predictable:

• Stale context: Agent doesn't know about yesterday's refactor
• Lost continuity: Task state evaporates between sessions
• KB drift: Documentation changes never reach the knowledge base
• Cognitive load: User becomes a memory manager instead of a programmer

The Solution: Skills + Hooks​

┌─────────────────────────────────────────────────────────────────┐
│                    Automated Context Flow                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   Session Start                         Session End              │
│   ─────────────                         ───────────              │
│   session-init skill                    session-save skill       │
│   (auto-discovered)                     (manual or auto)         │
│        │                                      │                  │
│        ▼                                      ▼                  │
│   ┌──────────────┐                     ┌──────────────┐         │
│   │ MCP: Session │                     │ MCP: Session │         │
│   │   Memory     │                     │   Memory     │         │
│   └──────────────┘                     └──────────────┘         │
│        │                                                         │
│        ▼                                                         │
│   ┌──────────────┐                                              │
│   │ MCP: Pixel-  │◄──────────────────────────────────┐          │
│   │ table Memory │                                    │          │
│   └──────────────┘                                    │          │
│                                                       │          │
│   Git Events (Automatic)                              │          │
│   ──────────────────────                              │          │
│                                                       │          │
│   post-commit ─────┐                                  │          │
│   post-merge ──────┼──── ingest_file() ──────────────┘          │
│   post-rewrite ────┘                                             │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Prerequisites: MCP Servers​

This system requires two MCP (Model Context Protocol) servers:

Session Memory MCP (session-memory): Short-term memory for task context, recent activity, and session state. Think of it as the agent's working memory—what you're currently doing.

Pixeltable Memory MCP (pixeltable-memory): Long-term organizational knowledge base powered by Pixeltable. Stores ADRs, documentation, incident history, and code patterns. This is where ingested content goes.

Both are included in the Luminescent Cluster repository. See the Getting Started guide for installation.

Agent Skills: A Portable Standard​

Agent Skills is a format for packaging AI agent capabilities as discoverable markdown files. Any agent that supports the format can auto-discover skills in .claude/skills/ directories.

Why Skills, Not Custom Workflows?​

Skills are supported by Claude, OpenAI Codex, GitHub Copilot, Cursor, VS Code Insiders, and Goose.

session-init: Bootstrap Every Session​

The session-init skill runs at the start of every coding session:

---
name: session-init
description: >
  BOOTSTRAP PROTOCOL: Initialize session with recent git activity,
  user task context, and relevant architectural decisions.
version: "1.0"
compatibility:
  - mcp: session-memory
  - mcp: pixeltable-memory
---

What it does:

1. Verify KB freshness - Compare last_ingest_sha with HEAD
2. Check hook installation - Warn if git hooks are missing
3. Load git state - Recent commits, uncommitted changes, current branch
4. Retrieve task context - What were we working on?
5. Query organizational knowledge - Relevant ADRs and decisions

Output format:

> **Status:** Fresh
> **Active Task:** Implementing user authentication
> **Current Branch:** feature/auth
> **Recent Activity:**
>   - Added login endpoint
>   - Updated session middleware
>   - Fixed CORS configuration
> **Relevant ADRs:** ADR-007 (Authentication Strategy)

session-save: Persist Before Leaving​

The session-save skill captures state before ending work:

1. Summarize accomplishments - What was done this session
2. Update task context - Current state, blockers, next steps
3. Prepare for commit - Check for uncommitted changes

# What gets saved to Session Memory
{
    "task": "Implementing user authentication",
    "details": {
        "completed": ["login endpoint", "session middleware"],
        "in_progress": "CORS configuration",
        "blockers": ["Need OAuth provider decision"],
        "files_modified": ["src/api/auth.py", "src/middleware/session.py"]
    }
}

Git Hooks: Event-Driven Ingestion​

Skills handle session workflows. Git hooks handle codebase synchronization.

Three Hooks, Three Events​

What Gets Ingested​

The hook filters commits to find ingestible files:

# Allowlist patterns
INGESTIBLE_PATTERNS="\.md$|\.txt$|\.rst$|docs/|adr/|ADR-"

# Denylist patterns
EXCLUDED_PATTERNS="node_modules/|\.venv/|dist/|build/|__pycache__/"

# Secrets protection (filename patterns)
SECRETS_PATTERNS="\.env|secret|\.key$|\.pem$|password|token|credential"

Note: Filename-based filtering catches obvious secrets but isn't sufficient alone. The ingestion pipeline also performs content scanning for high-entropy strings (potential API keys) and common secret patterns. See the security deep dive for the full detection strategy.

Each ingested file includes metadata:

• path: Relative path in repo
• commit_sha: Git commit SHA
• branch: Branch name
• content_hash: SHA256 for idempotency
• ingested_at: Timestamp

Non-Blocking Design​

Hooks run ingestion asynchronously to avoid slowing commits:

# Run in background
(
    # ... ingestion logic ...
    echo "$COMMIT_SHA" > "$STATE_DIR/last_ingest_sha"
) &

echo "[post-commit] Ingestion queued."

Check status in .agent/logs/ingestion.log.

Security: 10 Rounds of Council Review​

The ingestion pipeline underwent 10 rounds of LLM Council security review. Here's what we hardened:

Path Security​

# Layer 1: Check raw input BEFORE resolution (catches obvious attempts)
if ".." in str(file_path) or "\x00" in str(file_path):
    return {"success": False, "reason": "Rejected suspicious path characters"}

# Layer 2: Resolve to canonical path
canonical_path = file_path.resolve()

# Layer 3: Verify result is under project root (the authoritative check)
try:
    relative_path = canonical_path.relative_to(project_root.resolve())
except ValueError:
    return {"success": False, "reason": "Path escapes project boundary"}

# Layer 4: Prevent git argument injection
if str(relative_path).startswith("-"):
    return {"success": False, "reason": "Rejected hyphen prefix (git injection)"}

Why layered checks? The relative_to() check is authoritative, but checking raw input first provides defense in depth and clearer audit logs for attack attempts.

Provenance Integrity​

We read from the git object database, not the working tree:

# Read exactly what was committed
result = subprocess.run(
    ["git", "show", f"{commit_sha}:{relative_path}"],
    capture_output=True,
    text=False,  # Binary mode
    timeout=10,
)
content = result.stdout.decode("utf-8", errors="replace")

This prevents race conditions where files are modified between commit and ingestion.

DoS Prevention​

# Check blob size BEFORE reading content
blob_size = _get_blob_size(relative_path, commit_sha, project_root)
if blob_size is None:
    return {"success": False, "reason": "Cannot determine blob size"}

if blob_size / 1024 > config.max_file_size_kb:
    return {"success": False, "reason": f"File too large"}

# Verify it's a file, not a directory
if not _is_blob(relative_path, commit_sha, project_root):
    return {"success": False, "reason": "Not a blob (file)"}

Config Validation​

# Hard limits cannot be overridden by config
MAX_FILE_SIZE_KB_HARD_LIMIT = 10240  # 10MB absolute max

# Clamp user config values
max_file_size_kb = min(int(raw_max_size), MAX_FILE_SIZE_KB_HARD_LIMIT)

Pattern Matching (No ReDoS)​

We use fnmatch instead of regex for user-configurable patterns:

# Safe pattern matching without regex
def _matches_pattern(file_path: str, pattern: str) -> bool:
    if "**" in pattern:
        # Component-based matching (no regex)
        return _match_glob_components(file_path, pattern)
    return fnmatch(file_path, pattern)

Access Control​

Who can read memory? In single-user mode (self-hosted), the MCP servers run locally and are only accessible to processes on your machine. There's no network exposure.

Who can write memory? Only the git hooks and MCP server tools can write. The hooks run in your shell context with your git credentials. The MCP servers validate that writes come from authorized MCP clients.

Multi-tenant deployments use the cloud tier, which adds authentication, tenant isolation, and audit logging. See Memory-as-a-Service for the trust model.

Configuration​

All behavior is controlled via .agent/config.yaml:

ingestion:
  include:
    - "docs/**/*.md"
    - "*.md"
    - "docs/adrs/**"
  exclude:
    - "**/node_modules/**"
    - "**/.venv/**"
    - "**/secrets/**"
    - "**/*secret*"
    - "**/*password*"
  max_file_size_kb: 500
  skip_binary: true

skills:
  directory: ".claude/skills"
  auto_discover: true

Getting Started​

Platform support: The shell scripts target Unix-like systems (Linux, macOS). Windows users can run them via WSL or Git Bash, or manually copy the hook files to .git/hooks/.

1. Install Hooks​

./scripts/install_hooks.sh

This installs post-commit, post-merge, and post-rewrite hooks. The script backs up any existing hooks before replacement.

2. Bootstrap KB (Fresh Clone)​

python scripts/init_memory.py

This ingests existing documentation for the first time.

3. Verify Installation​

# Check hooks
ls -la .git/hooks/post-*

# Check skill discovery
ls -la .claude/skills/

# Make a test commit
echo "# Test" >> test.md
git add test.md && git commit -m "test ingestion"
cat .agent/logs/ingestion.log

4. Start a Session​

The agent will auto-discover session-init and load context. Or invoke manually:

/session-init

When It Works​

After setup, the flow becomes:

1. Start session → Agent auto-runs session-init → Context loaded
2. Work on code → Normal development
3. Commit changes → Hook auto-ingests docs → KB stays fresh
4. End session → Run session-save → State persisted
5. Next session → Agent knows where you left off

No more "load context" prompts. No more KB drift. No more lost continuity.

Troubleshooting​

KB is Stale​

# Check last ingestion
cat .agent/state/last_ingest_sha
git rev-parse HEAD

# Force re-ingest
rm .agent/state/last_ingest_sha
python scripts/init_memory.py --force

Hooks Not Running​

# Verify hooks are executable
chmod +x .git/hooks/post-*

# Check for errors
cat .agent/logs/ingestion.log

Skill Not Discovered​

Ensure the agent supports Agent Skills spec. Check:

• .claude/skills/ directory exists
• SKILL.md files have valid YAML frontmatter
• Agent has MCP servers configured (session-memory, pixeltable-memory)

What's Next​

ADR-002 is complete, but future enhancements could include:

• IDE integration: VS Code extension for skill invocation
• Real-time sync: File watcher for uncommitted changes
• Skill marketplace: Community-contributed skills

Repository​

All scripts, hooks, and skills referenced in this post are in the Luminescent Cluster repository:

• Hooks: .agent/hooks/post-commit, post-merge, post-rewrite
• Skills: .claude/skills/session-init/, .claude/skills/session-save/
• Scripts: scripts/install_hooks.sh, scripts/init_memory.py
• Config: .agent/config.yaml
• Source: src/workflows/config.py, src/workflows/ingestion.py

Automated context loading is part of ADR-002 Workflow Integration. See the full ADR for implementation details and security considerations.

Pure vector search isn't enough. Here's how we combine BM25, embeddings, and cross-encoder reranking to achieve accurate memory retrieval.

Vector search revolutionized information retrieval. Embed your documents, embed your query, find the nearest neighbors. Simple, elegant, and... often wrong.

The problem? Vector embeddings excel at semantic similarity but struggle with exact matches. Ask "What's the Redis configuration?" and pure vector search might return documents about "cache settings" or "in-memory databases" while missing the one that literally contains "Redis configuration."

HybridRAG solves this by combining the strengths of multiple retrieval methods, then using a neural reranker to pick the best results.

The Architecture​

┌─────────────────────────────────────────────────────────────────┐
│                         Query Input                              │
├─────────────────────────────────────────────────────────────────┤
│ Stage 1: Parallel Candidate Generation                          │
├──────────────────────┬──────────────────────────────────────────┤
│  BM25 Search         │  Vector Search                           │
│  (Sparse Keywords)   │  (Dense Semantics)                       │
│  Top 50              │  Top 50                                  │
└──────────────────────┴──────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│ Stage 2: Fusion + Reranking                                     │
├────────────────────────────┬────────────────────────────────────┤
│  RRF Fusion                │  Cross-Encoder Reranking           │
│  (Score-agnostic merge)    │  (Deep relevance scoring)          │
└────────────────────────────┴────────────────────────────────────┘
                              │
                              ▼
                      Final Results (Top K)

Two stages. Multiple signals. One answer.

Stage 1: Parallel Candidate Generation​

Stage 1 runs two search methods in parallel, each returning 50 candidates.

BM25: The Keyword Hunter​

BM25 (Best Match 25) is a probabilistic ranking function that scores documents based on term frequency and inverse document frequency. It's been the backbone of search engines for decades.

BM25(D, Q) = Σ IDF(qi) * (f(qi, D) * (k1 + 1)) / (f(qi, D) + k1 * (1 - b + b * |D| / avgdl))

Translation: Documents score higher when they contain rare query terms (high IDF) that appear frequently in the document (high TF), normalized by document length.

Why BM25 matters:

BM25 finds the needle. Vector search finds things that look like needles.

Vector Search: The Semantic Finder​

Dense vector search embeds queries and documents into a shared embedding space (typically 384-1536 dimensions depending on the model), then finds nearest neighbors by cosine similarity.

# Using sentence-transformers (e.g., all-MiniLM-L6-v2 = 384 dims)
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
query_embedding = model.encode("How should I configure caching?")
similarities = np.dot(doc_embeddings, query_embedding)
top_k = np.argsort(similarities)[-50:]

Why vectors matter:

Vectors understand synonyms, paraphrasing, and conceptual relationships that keyword search misses.

Running in Parallel​

Both searches are independent, so we run them concurrently:

async def stage1(query: str, user_id: str):
    bm25_task = asyncio.to_thread(self.bm25.search, user_id, query, 50)
    vector_task = asyncio.to_thread(self.vector.search, user_id, query, 50)

    bm25_results, vector_results = await asyncio.gather(bm25_task, vector_task)
    return bm25_results, vector_results

Stage 1 takes ~200-400ms (limited by vector embedding).

Stage 2: Fusion + Reranking​

Now we have two ranked lists. How do we combine them?

The Naive Approach (Don't Do This)​

Normalize scores and average:

# DON'T DO THIS
bm25_normalized = (bm25_score - min) / (max - min)
vector_normalized = (vector_score - min) / (max - min)
combined = (bm25_normalized + vector_normalized) / 2

This fails because:

1. Score distributions differ wildly (BM25: 0-20, vectors: 0-1)
2. Outliers dominate after normalization
3. Assumes scores are comparable (they're not)

Reciprocal Rank Fusion (RRF)​

RRF ignores scores entirely and works with ranks:

RRF_score(d) = Σ 1 / (k + rank_i(d))

Where k=60 (from the original paper) and rank_i(d) is the 1-indexed position of document d in list i.

Example:

BM25:   [doc1, doc2, doc3]  (ranks 1, 2, 3)
Vector: [doc2, doc1, doc4]  (ranks 1, 2, 3)

RRF scores (k=60):
doc1: 1/(60+1) + 1/(60+2) = 0.0164 + 0.0161 = 0.0325
doc2: 1/(60+2) + 1/(60+1) = 0.0161 + 0.0164 = 0.0325
doc3: 1/(60+3)            = 0.0159
doc4: 1/(60+3)            = 0.0159

Both doc1 and doc2 tie because they appeared in both lists at similar ranks.

Why RRF works:

• Score-agnostic: Works with any ranking function
• Robust: Not sensitive to outliers or extreme scores
• Empirically strong: Often competitive with learned fusion methods

Implementation:

from collections import defaultdict

def reciprocal_rank_fusion(
    *ranked_lists: list[tuple[str, float]],
    k: int = 60
) -> list[tuple[str, float]]:
    """
    Fuse multiple ranked lists using RRF.

    Args:
        ranked_lists: Each list contains (doc_id, score) tuples, sorted by score desc
        k: RRF constant (default 60 from original paper)

    Returns:
        Fused list of (doc_id, rrf_score) sorted by RRF score desc
    """
    scores = defaultdict(float)

    for ranked_list in ranked_lists:
        for rank, (doc_id, _) in enumerate(ranked_list, start=1):
            scores[doc_id] += 1 / (k + rank)

    return sorted(scores.items(), key=lambda x: -x[1])

# Usage
fused = reciprocal_rank_fusion(bm25_results, vector_results, k=60)

Cross-Encoder Reranking​

RRF gives us a fused candidate list. But we can do better.

A cross-encoder is a neural model that takes (query, document) pairs and outputs a relevance score. Unlike bi-encoders (vector search), it sees both texts together, enabling deeper semantic understanding.

# Bi-encoder (separate encoding)
query_emb = model.encode(query)
doc_emb = model.encode(doc)
score = cosine(query_emb, doc_emb)  # Approximate

# Cross-encoder (joint encoding)
score = cross_encoder.predict([(query, doc)])  # Precise

Cross-encoders are significantly more accurate but much slower per-pair (the exact ratio depends on batching, hardware, and model size). That's why we only use them for reranking the top ~50 candidates, not the entire corpus.

Model choice: cross-encoder/ms-marco-MiniLM-L-6-v2

• Trained on MS MARCO passage ranking dataset
• Fast (~2ms per pair on CPU, faster with batching/GPU)
• Good balance of speed and accuracy

from sentence_transformers import CrossEncoder

cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank(query: str, candidates: list, top_k: int = 10):
    """Rerank candidates using cross-encoder relevance scoring."""
    pairs = [(query, doc.content) for doc in candidates]

    # Batch scoring for efficiency
    scores = cross_encoder.predict(pairs, batch_size=32)

    # Sort by cross-encoder score
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return ranked[:top_k]

Stage 2 takes ~50-100ms.

The Complete Flow​

Let's trace a query through the system:

Query: "How should I configure database migrations?"

Stage 1a - BM25 (30ms):

Tokens: ["configure", "database", "migrations"]
Results: [("mem-5", 7.3), ("mem-1", 4.2), ("mem-8", 3.1), ...]

Stage 1b - Vector (200ms):

Embedding: [0.12, -0.34, 0.87, ...] (384 dims)
Results: [("mem-1", 0.92), ("mem-5", 0.87), ("mem-12", 0.78), ...]

Stage 2a - RRF Fusion (5ms):

doc  | bm25_rank | vec_rank | RRF score
-----|-----------|----------|----------
mem-1|     2     |    1     | 0.0325
mem-5|     1     |    2     | 0.0325
mem-8|     3     |    -     | 0.0159
mem-12|    -     |    3     | 0.0159

Stage 2b - Reranking (60ms):

Cross-encoder scores:
mem-1:  0.89 ← Best match (specific migration instructions)
mem-5:  0.87
mem-12: 0.72
mem-8:  0.65

Final Results:

[
    HybridResult(memory_id="mem-1", score=0.89,
                 source_ranks={"bm25": 2, "vector": 1, "reranker": 1}),
    HybridResult(memory_id="mem-5", score=0.87,
                 source_ranks={"bm25": 1, "vector": 2, "reranker": 2}),
    ...
]

Total latency: ~350ms.

Tuning the Weights​

Not all retrieval methods are equal for all domains. You can weight them differently in RRF:

# For code/config (keyword-heavy)
retriever = HybridRetriever(bm25_weight=1.5, vector_weight=1.0)

# For conceptual/natural language
retriever = HybridRetriever(bm25_weight=1.0, vector_weight=1.5)

Weighted RRF:

score(d) = w_bm25/(k + rank_bm25(d)) + w_vec/(k + rank_vec(d))

Provenance Tracking​

Every result tracks where it came from:

result = HybridResult(
    memory=Memory(...),
    score=0.89,
    memory_id="mem-1",
    source_scores={"bm25": 4.2, "vector": 0.92, "reranker": 0.89},
    source_ranks={"bm25": 2, "vector": 1, "reranker": 1}
)

This lets you debug retrieval issues:

• High BM25, low vector → Keyword match, semantic mismatch
• High vector, low BM25 → Semantic match, missing keywords
• High both, low reranker → False positive (looks relevant but isn't)

Performance Characteristics​

For latency-sensitive applications, you can disable the cross-encoder:

retriever = HybridRetriever(use_cross_encoder=False)
# Saves ~100ms, loses ~5% accuracy

When to Use HybridRAG​

Good fit:

• Technical documentation (mixed keywords + concepts)
• Code knowledge bases (exact identifiers + semantic queries)
• Decision records (ADR references + natural language)
• Incident histories (error codes + descriptions)

Overkill for:

• Pure keyword search (use BM25 alone)
• Pure semantic search (use vectors alone)
• Real-time autocomplete (latency-sensitive)

Usage Example​

from src.memory.retrieval import create_hybrid_retriever

# Initialize
retriever = create_hybrid_retriever(
    use_cross_encoder=True,
    bm25_weight=1.0,
    vector_weight=1.0,
)

# Index your memories
retriever.index_memories("user-1", memories)

# Retrieve
results, metrics = await retriever.retrieve(
    query="How do I configure Redis caching?",
    user_id="user-1",
    top_k=10,
)

# Use results
for result in results:
    print(f"{result.memory_id}: {result.score:.3f}")
    print(f"  Content: {result.memory.content[:100]}...")

# Monitor performance
print(f"Latency: {metrics.total_time_ms:.0f}ms")
print(f"  Stage 1: {metrics.stage1_time_ms:.0f}ms")
print(f"  Stage 2: {metrics.stage2_time_ms:.0f}ms")

The Results​

In our benchmarks, HybridRAG outperforms pure vector search by 50%+ on multi-hop queries while maintaining sub-second latency:

Benchmarks: 10K document corpus of technical documentation. Tested on 4-core CPU (no GPU). Embedding model: all-MiniLM-L6-v2. Cross-encoder: ms-marco-MiniLM-L-6-v2. Query set: 500 natural language questions with human-labeled relevance judgments.

The two-stage architecture gives us the best of both worlds: comprehensive candidate generation followed by precise relevance ranking.

Recommended Libraries​

Minimal pip install:

pip install sentence-transformers rank-bm25 numpy

Limitations​

HybridRAG isn't a silver bullet. Know when it fails:

• Very short queries (1-2 words): BM25 dominates; vector adds noise
• Highly technical identifiers: UUIDs, hashes, and base64 strings need exact match, not semantic search
• Multilingual queries: Embedding model must support target languages
• Real-time autocomplete: 450ms is too slow; use BM25 prefix matching instead
• Adversarial queries: Prompt injection in documents can poison retrieval

Tuning tips:

• Start with k=50 candidates from each source; adjust based on recall/latency tradeoff
• If queries are keyword-heavy (code, configs), boost bm25_weight
• If queries are conceptual (natural language), boost vector_weight
• The reranker is optional—disable it if latency matters more than precision

HybridRAG is part of ADR-003 Phase 3. See the full ADR for implementation details and performance benchmarks.

When AI agents have write access to your knowledge base, hallucinations become persistent. Here's how we built a provenance system that blocks ungrounded claims.

AI agents are increasingly trusted to not just read but write. They synthesize information, make inferences, and store conclusions for future reference. The problem? AI confidently states things that aren't true. And if those hallucinations get written to your knowledge base, they become persistent misinformation.

"The API uses OAuth2 for authentication." Sounds authoritative. But did the AI actually verify this, or did it hallucinate a plausible-sounding claim? Once stored, future sessions will retrieve this "fact" and build on it.

Grounded ingestion solves this by classifying every memory claim into one of three tiers based on its provenance evidence.

The 3-Tier Provenance Model​

┌─────────────────────────────────────────────────────────────────┐
│                    Incoming Memory Claim                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌────────────┐   ┌────────────┐   ┌────────────┐              │
│  │  Citation  │   │   Hedge    │   │   Dedup    │              │
│  │  Detector  │   │  Detector  │   │  Checker   │              │
│  └─────┬──────┘   └─────┬──────┘   └─────┬──────┘              │
│        │                │                │                      │
│        └────────────────┼────────────────┘                      │
│                         ▼                                        │
│              ┌──────────────────────┐                           │
│              │   Tier Determination │                           │
│              └──────────────────────┘                           │
│                         │                                        │
│        ┌────────────────┼────────────────┐                      │
│        ▼                ▼                ▼                      │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐                  │
│  │  TIER 1  │    │  TIER 2  │    │  TIER 3  │                  │
│  │  AUTO-   │    │  FLAG    │    │  BLOCK   │                  │
│  │  APPROVE │    │  REVIEW  │    │          │                  │
│  └──────────┘    └──────────┘    └──────────┘                  │
│       │                │                │                       │
│       ▼                ▼                ▼                       │
│    Stored          Queued           Rejected                    │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Tier 1: Auto-Approve (High Confidence)​

Claims with explicit grounding evidence are stored immediately:

• Has citation: ADR reference, commit hash, URL, issue number
• Trusted source: User-stated, documentation, manual entry
• Decision in context: Explicit decisions from conversations

Examples:

"Per ADR-003, we use Pixeltable for memory storage"  → AUTO (has ADR citation)
"Fixed in commit a1b2c3d4e5f6"                       → AUTO (has commit hash)
"See https://docs.example.com/api"                   → AUTO (has URL)
"I prefer tabs over spaces" (source=user)            → AUTO (trusted source)

Tier 2: Flag for Review (Medium Confidence)​

Claims that aren't obviously grounded or speculative need human verification:

• AI synthesis without citation: Factual assertions from AI without explicit sources
• Dedup check failed: Provider error means we can't verify uniqueness
• Ambiguous claims: Could be true but unverified

Examples:

"The API returns JSON for REST responses"            → REVIEW (AI synthesis, no citation)
"OAuth2 is the authentication mechanism"             → REVIEW (factual assertion, ungrounded)
"The service uses PostgreSQL 15"                     → REVIEW (could be true, needs verification)

Tier 3: Block (Low Confidence)​

Claims that are explicitly speculative or duplicate are rejected:

• Strong speculation: "I think", "I guess", "maybe" without any grounding
• Duplicate content: >92% similarity to existing memory
• Personal uncertainty: Language expressing the speaker doesn't know

Examples:

"I think we should use Redis"                        → BLOCK (personal speculation)
"I guess the API supports this"                      → BLOCK (admitted uncertainty)
"Maybe we could try GraphQL"                         → BLOCK (ungrounded suggestion)
"Per ADR-003, we use PostgreSQL" (already stored)   → BLOCK (duplicate)

Note: Technical hedges like "may", "typically", or "often" are treated differently—see Hedge Detection below.

Detection Mechanisms​

Citation Detection & Verification​

Grounding requires two steps: detection (finding citation patterns) and verification (checking they're real).

Step 1: Detection

The CitationDetector uses regex patterns to identify potential grounding evidence:

PATTERNS = {
    "adr": r"\[?ADR[-\s]?(\d+)\]?",           # [ADR-003], ADR-003, ADR 003
    "commit": r"\b[a-f0-9]{7,40}\b",           # 7-40 hex chars (not colors)
    "url": r"https?://[^\s<>\"]+",             # http:// or https://
    "issue": r"(?:#(\d+)|GH-(\d+))",           # #123 or GH-456
}

Smart filtering:

• Excludes 6-character hex codes (colors like #abc123)
• Strips URLs before commit detection (no false positives in URL paths)
• Returns citation type, position, and extracted ID

Step 2: Verification (Critical)

Detection alone is insufficient—AI can hallucinate plausible-looking citations. Verification checks that cited artifacts actually exist:

async def verify_citation(citation: Citation) -> VerificationResult:
    """Verify that a detected citation actually exists."""
    match citation.type:
        case "url":
            # Check URL resolves (HTTP HEAD request)
            response = await http_client.head(citation.value, timeout=5)
            return VerificationResult(
                valid=response.status_code == 200,
                reason=f"HTTP {response.status_code}"
            )

        case "commit":
            # Check commit exists in repo
            result = subprocess.run(
                ["git", "cat-file", "-t", citation.value],
                capture_output=True
            )
            return VerificationResult(
                valid=result.returncode == 0,
                reason="Commit exists" if result.returncode == 0 else "Unknown commit"
            )

        case "adr":
            # Check ADR file exists
            adr_path = f"docs/adrs/ADR-{citation.value}-*.md"
            exists = len(glob.glob(adr_path)) > 0
            return VerificationResult(valid=exists, reason="ADR file exists" if exists else "ADR not found")

        case "issue":
            # Check issue exists (requires API call to GitHub/GitLab)
            # Implementation depends on your issue tracker
            ...

Why verification matters: LLMs confidently fabricate citations. We've seen:

• URLs that return 404
• Commit hashes that don't exist
• ADR numbers that were never written

Detection + verification together provide actual grounding.

Hedge Detection​

The HedgeDetector identifies uncertainty language, but not all hedges are equal. We distinguish between personal speculation (block) and technical hedges (review).

Hedge categories and actions:

Why the distinction? Legitimate technical documentation uses hedges appropriately:

• "The server may timeout under load" — valid engineering guidance
• "Connections typically complete in <100ms" — accurate qualification
• "I think we should use Redis" — ungrounded personal opinion

detector = HedgeDetector()

# Personal speculation → Block
result = detector.analyze("I think we should use Redis")
# HedgeResult(is_speculative=True, action=BLOCK, hedge_words=["I think"])

# Technical hedge → Review
result = detector.analyze("The API may return errors under load")
# HedgeResult(is_speculative=True, action=REVIEW, hedge_words=["may"])

# No hedges → Continue to other checks
result = detector.analyze("Per ADR-003, we use PostgreSQL")
# HedgeResult(is_speculative=False, action=CONTINUE)

Bypass prevention: The trivial bypass "I think X, definitely" doesn't work—personal speculation markers always trigger regardless of assertion markers.

False positive filtering:

• "May 2024" → Month name, not modal verb
• "couldn't" → Negation often indicates certainty
• "should be 5" → Factual numeric description

Deduplication​

The DedupChecker prevents duplicate memories using word-level Jaccard similarity:

Jaccard(A, B) = |A ∩ B| / |A ∪ B|

Where A and B are sets of lowercase tokens from each text.

Threshold: 92% similarity = duplicate (per ADR-003)

from collections import Counter

def jaccard_similarity(text1: str, text2: str) -> float:
    """Calculate word-level Jaccard similarity."""
    words1 = set(text1.lower().split())
    words2 = set(text2.lower().split())

    intersection = len(words1 & words2)
    union = len(words1 | words2)

    return intersection / union if union > 0 else 0.0

# Usage
checker = DedupChecker(provider=memory_provider, threshold=0.92)

result = await checker.check_duplicate(
    content="Per ADR-003, we use Pixeltable for storage",
    user_id="user-123",
    memory_type="decision"
)
# DuplicateResult(is_duplicate=True, similarity=0.95, existing_memory_id="mem-456")

Why Jaccard over semantic similarity?

We chose Jaccard for speed—it runs on every ingestion without latency impact. The 92% threshold catches most copy-paste duplicates while allowing legitimate variations.

Known limitation: Jaccard misses semantic duplicates where AI rephrases the same claim. For higher-value knowledge bases, consider upgrading to vector-based deduplication or MinHash for scale:

# Future: Semantic deduplication (slower but catches paraphrasing)
async def semantic_dedup(content: str, user_id: str) -> DuplicateResult:
    embedding = await embed(content)
    similar = await vector_index.search(embedding, threshold=0.95)
    return DuplicateResult(is_duplicate=len(similar) > 0, ...)

The Decision Tree​

from enum import Enum
from dataclasses import dataclass

class Tier(Enum):
    AUTO_APPROVE = "tier_1"
    FLAG_REVIEW = "tier_2"
    BLOCK = "tier_3"

class HedgeAction(Enum):
    BLOCK = "block"      # Personal speculation
    REVIEW = "review"    # Technical hedges
    CONTINUE = "none"    # No hedges

TRUSTED_SOURCES = {"user", "documentation", "adr", "commit", "manual"}

def determine_tier(
    hedge_result: HedgeAction,
    is_duplicate: bool,
    dedup_failed: bool,
    has_verified_citation: bool,
    source: str,
    memory_type: str
) -> tuple[Tier, str]:
    """
    Determine ingestion tier based on grounding evidence.

    Returns (tier, reason) tuple.
    """
    # Tier 3: Block personal speculation
    if hedge_result == HedgeAction.BLOCK:
        return Tier.BLOCK, "Contains personal speculation"

    # Tier 3: Block duplicates
    if is_duplicate:
        return Tier.BLOCK, "Duplicate of existing memory"

    # Tier 2: Technical hedges need review
    if hedge_result == HedgeAction.REVIEW:
        return Tier.FLAG_REVIEW, "Contains technical hedges - needs verification"

    # Tier 2: Dedup check failed (fail-closed)
    if dedup_failed:
        return Tier.FLAG_REVIEW, "Dedup check failed - cannot verify uniqueness"

    # Tier 1: Verified citation
    if has_verified_citation:
        return Tier.AUTO_APPROVE, "Has verified citation"

    # Tier 1: Trusted source
    if source in TRUSTED_SOURCES:
        return Tier.AUTO_APPROVE, f"From trusted source: {source}"

    # Tier 1: User-stated decisions/preferences
    if memory_type == "decision" and source == "conversation":
        return Tier.AUTO_APPROVE, "Decision stated in conversation"

    if memory_type == "preference" and source in ("conversation", "chat"):
        return Tier.AUTO_APPROVE, "Preference stated by user"

    # Tier 2: Everything else needs review
    return Tier.FLAG_REVIEW, "Ungrounded assertion needs verification"

Key design choices:

• Fail-closed: Provider errors (like dedup failures) route to review, never auto-approve
• Hedge nuance: Personal speculation blocks; technical hedges go to review
• Verification required: Citations must be verified, not just detected

The Review Queue​

Tier 2 memories enter a review queue for human verification:

@dataclass
class PendingMemory:
    queue_id: str           # Unique identifier
    user_id: str            # Owner
    content: str            # The claim
    memory_type: str        # preference/fact/decision
    source: str             # Origin
    evidence: EvidenceObject  # Provenance metadata
    submitted_at: datetime  # UTC timestamp

Queue operations:

queue = ReviewQueue(store_callback=store_memory)

# Enqueue a flagged memory
queue_id = await queue.enqueue(
    user_id="user-123",
    content="The API uses OAuth2",
    memory_type="fact",
    source="ai_synthesis",
    evidence=evidence,
    validation_result=result,
)

# User reviews their pending memories
pending = await queue.get_pending("user-123", limit=10)

# User approves (stores the memory)
memory_id = await queue.approve(queue_id, reviewer="user-123")

# Or rejects (discards with reason)
await queue.reject(queue_id, reviewer="user-123", reason="Incorrect, we use JWT")

Security properties:

Evidence Objects​

Every memory carries provenance metadata:

@dataclass
class EvidenceObject:
    claim: str                           # The content
    capture_time: datetime               # When captured
    confidence: str                      # high/medium/low
    source_id: Optional[str] = None      # ADR-003, commit:a1b2c3d, URL
    validity_horizon: Optional[datetime] = None  # Expiration
    metadata: dict = {}                  # Extensible

This enables downstream systems to:

• Filter by confidence level
• Trace claims back to sources
• Expire time-sensitive information
• Audit memory provenance

Validation Results​

The complete output of ingestion validation:

@dataclass
class ValidationResult:
    tier: IngestionTier           # AUTO_APPROVE, FLAG_REVIEW, or BLOCK
    approved: bool                # True only for Tier 1
    reason: str                   # Human-readable explanation
    evidence: EvidenceObject      # Provenance metadata
    checks_passed: list[str]      # ["citation_present", "no_speculation"]
    checks_failed: list[str]      # ["hedge_words_detected: maybe"]
    similarity_score: Optional[float]    # If duplicate found
    conflicting_memory_id: Optional[str] # ID of duplicate

Usage Example​

from src.memory.ingestion import IngestionValidator, ReviewQueue

# Create validator with deduplication
validator = IngestionValidator(
    provider=memory_provider,
    enable_dedup=True
)

# Validate an incoming claim
result = await validator.validate(
    content="Per ADR-003, we use Pixeltable for memory storage",
    memory_type="decision",
    source="conversation",
    user_id="user-123"
)

# Route based on tier
if result.tier == IngestionTier.AUTO_APPROVE:
    # Store immediately
    memory_id = await store_memory(result.evidence)
    print(f"Stored: {memory_id}")

elif result.tier == IngestionTier.FLAG_REVIEW:
    # Queue for review
    queue = ReviewQueue(store_callback=store_memory)
    queue_id = await queue.enqueue(
        user_id="user-123",
        content=result.evidence.claim,
        memory_type="decision",
        source="conversation",
        evidence=result.evidence,
        validation_result=result,
    )
    print(f"Queued for review: {queue_id}")

else:  # BLOCK
    print(f"Rejected: {result.reason}")
    print(f"Failed checks: {result.checks_failed}")

What Gets Blocked vs Accepted​

Always Blocked (Tier 3)​

# Personal speculation
"I think we should use Redis"                  # personal opinion
"I guess the API supports this"                # admitted uncertainty
"Maybe we could try GraphQL"                   # ungrounded suggestion

# Duplicates
"Per ADR-003, we use PostgreSQL"               # (if already stored)

# Bypass attempts
"I think we should use Redis, definitely"      # personal speculation still blocks

Always Accepted (Tier 1)​

# Has VERIFIED citation (not just detected)
"Per ADR-003, we use PostgreSQL"               # ADR file exists
"Fixed in commit a1b2c3d4e5"                   # commit exists in repo
"See https://docs.example.com/api"             # URL returns 200

# Trusted source
"I prefer tabs over spaces" (source=user)      # user-stated
"OAuth2 is required" (source=documentation)    # documentation

# Decisions in context
"We decided to use PostgreSQL" (type=decision, source=conversation)

Flagged for Review (Tier 2)​

# Technical hedges (legitimate uncertainty)
"The server may timeout under load"            # technical "may"
"Connections typically complete in <100ms"     # "typically" is qualified

# AI synthesis without citation
"The API returns JSON for REST responses"
"OAuth2 is the authentication mechanism"

# Citation detected but NOT verified
"Per ADR-999, we use magic"                    # ADR-999 doesn't exist

# Dedup check failed
(any content when provider throws an error)

The Security Model​

Grounded ingestion implements defense-in-depth:

1. Hedge detection: First line. Blocks personal speculation, reviews technical hedges.
2. Deduplication: Second line. Prevents duplicate pollution (lexical).
3. Citation verification: Third line. Confirms cited artifacts exist.
4. Source trust: Fourth line. Trusts verified sources.
5. Review queue: Fifth line. Human verification for uncertain claims.
6. Fail-closed: Sixth line. Errors flag for review, never auto-approve.

The goal: Minimize hallucination write-back while keeping friction low.

Limitations​

This system isn't perfect. Know what it can't catch:

Evasion Risks​

What This System Can't Do​

1. Verify claim-citation relevance: We check that ADR-003 exists, not that it actually supports the claim
2. Catch confident hallucinations: "X is true" without hedges goes to review, not block
3. Scale to semantic deduplication: Jaccard is fast but shallow
4. Replace human judgment: Tier 2 still requires human review

Recommendations for High-Stakes Use​

For production knowledge bases with compliance requirements:

# Upgrade path for stricter grounding
config = GroundedIngestionConfig(
    # Semantic deduplication (slower but thorough)
    dedup_method="vector",
    dedup_threshold=0.95,

    # LLM-based claim verification (expensive but accurate)
    verify_claim_relevance=True,

    # All AI synthesis goes to review (safest)
    auto_approve_ai_synthesis=False,
)

The current implementation optimizes for speed and low friction over maximum accuracy. Adjust based on your risk tolerance.

Performance Characteristics​

For high-throughput scenarios, you can disable dedup:

validator = IngestionValidator(enable_dedup=False)
# Faster, but won't catch duplicates

Why This Matters​

AI systems are moving from read-only to read-write. They don't just retrieve information—they synthesize, infer, and persist conclusions. Without provenance tracking, hallucinations compound:

1. AI hallucinates "The API uses OAuth2"
2. Gets stored as a "fact"
3. Future queries retrieve it
4. AI builds on it: "Since we use OAuth2, we need refresh tokens"
5. More hallucinations stored
6. Knowledge base drifts from reality

Grounded ingestion breaks this cycle. Every claim goes through provenance checking:

• Grounded + Verified: Auto-approved with citation trail
• Uncertain: Flagged for human review
• Speculative: Blocked before it enters the knowledge base

This isn't a perfect solution—confident hallucinations still need human review, and sophisticated evasion is possible. But it dramatically reduces the rate at which ungrounded claims pollute your knowledge base, and it creates an audit trail for everything that does get stored.

The goal isn't zero hallucinations (that's impossible with current AI). The goal is traceable provenance: knowing where every stored claim came from and why it was trusted.

Grounded ingestion is part of ADR-003 Phase 2. See the full ADR for implementation details and security considerations.

Discord, Slack, Telegram, and WhatsApp all have different APIs. Here's how we built a unified architecture that lets you deploy to all four from a single codebase.

"Can we add a Slack bot?" "What about Discord?" "Our team uses Telegram." "The support team is on WhatsApp."

Every platform has its own API, authentication model, message format, and quirks. Building a chatbot for one platform is straightforward. Building for four platforms without creating four separate codebases? That's an architecture problem.

We solved it with thin adapters and a central gateway. Each platform adapter handles only platform-specific concerns (authentication, message parsing, sending). Everything else—access control, rate limiting, LLM orchestration, memory retrieval—lives in a shared gateway.

The Architecture​

┌──────────────────────────────────────────────────────────────────────────┐
│                          Platform Adapters                                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐                 │
│  │ Discord  │  │  Slack   │  │ Telegram │  │ WhatsApp │                 │
│  │ Adapter  │  │ Adapter  │  │ Adapter  │  │ Adapter  │                 │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘                 │
│       │             │             │             │                        │
│       └─────────────┴──────┬──────┴─────────────┘                        │
│                            │                                              │
│                            ▼                                              │
│               ┌────────────────────────────┐                             │
│               │     Central Gateway        │                             │
│               │  • Message normalization   │                             │
│               │  • Access control          │                             │
│               │  • Rate limiting           │                             │
│               │  • LLM orchestration       │                             │
│               │  • Context management      │                             │
│               └────────────┬───────────────┘                             │
│                            │                                              │
│               ┌────────────┴───────────────┐                             │
│               ▼                            ▼                             │
│  ┌─────────────────────┐    ┌─────────────────────────┐                 │
│  │  Session Memory     │    │  Pixeltable Memory      │                 │
│  │  MCP Server         │    │  MCP Server             │                 │
│  └─────────────────────┘    └─────────────────────────┘                 │
└──────────────────────────────────────────────────────────────────────────┘

Key principle: Adapters are thin. They convert platform-specific messages to a common format and send responses back. All business logic lives in the gateway.

Message Normalization​

Every platform represents messages differently. Discord has guild IDs and role mentions. Slack has workspace IDs and Block Kit. Telegram has chat types and bot commands. WhatsApp has 24-hour windows and template messages.

We normalize everything to a common ChatMessage format:

@dataclass
class ChatMessage:
    id: str                           # Platform message ID
    content: str                      # Normalized text content
    author: MessageAuthor             # Sender info (id, name, is_bot)
    channel_id: str                   # Conversation/channel ID
    timestamp: datetime               # When sent
    platform: str                     # "discord", "slack", "telegram", "whatsapp"
    thread_id: Optional[str]          # Thread context (if threaded)
    reply_to_id: Optional[str]        # Parent message (if reply)
    is_direct_message: bool           # DM vs channel message
    attachments: list[Attachment]     # Files, images, documents
    metadata: dict                    # Platform-specific extras

Platform-specific details go in metadata:

• Discord: mentions, role_mentions, guild info
• Slack: blocks, channel_mentions, broadcast flags
• Telegram: is_command, command_args, hashtags
• WhatsApp: window_open, interactive_reply, location

This lets the gateway work with a single message format while preserving platform-specific features when needed.

Platform Adapters​

Each adapter implements a common protocol:

@runtime_checkable
class PlatformAdapter(Protocol):
    @property
    def platform(self) -> str: ...

    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def send_message(
        self, channel_id: str, content: str, reply_to: Optional[str] = None
    ) -> ChatMessage: ...

Let's look at what each adapter handles.

Discord Adapter​

Connection: WebSocket via Discord Gateway (real-time events)

from discord import Client, Intents

class DiscordAdapter:
    def __init__(self, config: DiscordConfig):
        intents = Intents.default()
        intents.message_content = True  # Required for reading messages
        self.client = Client(intents=intents)
        self.config = config

    async def connect(self):
        @self.client.event
        async def on_message(message):
            if message.author == self.client.user:
                return  # Ignore own messages
            chat_msg = self.parse_message(message)
            await self.gateway.process(chat_msg)

        await self.client.start(self.config.token)

Key features:

• Intent-based filtering (what events to receive)
• Thread support via channel type detection
• Slash command registration
• Message splitting for 2000-char limit

Configuration:

@dataclass
class DiscordConfig:
    token: str                    # Bot token from Discord Developer Portal
    application_id: str           # For slash commands
    guild_id: Optional[str]       # Guild-specific commands (faster registration)