feat(memory): add namespace/key support + BlueRedChannel coordination class

- SQLite backend: namespace, key, tags, TTL, append support
- PluginMemoryService: full parity with TypeScript/Rust (list, stats, clear, tag, history, export/import)
- BlueRedChannel class for blue-red agent coordination protocol
- 30/30 tests passing

pkg/hanzo-mcp/hanzo_mcp/backends/sqlite_plugin.py

@@ -1,52 +1,154 @@
-"""SQLite backend plugin implementation."""
+"""SQLite backend plugin implementation with namespace, key, tags, and TTL support."""
 
+import json
+import sqlite3
+import uuid
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from hanzo_memory.db.sqlite_client import SQLiteMemoryClient
-from hanzo_memory.models.memory import MemoryCreate
-
 from ..plugin_interface import Capability
 
 
 class SQLiteBackendPlugin:
-    """SQLite backend plugin implementation."""
+    """SQLite backend plugin implementation.
 
-    def __init__(self, db_path: Optional[Path] = None):
-        """Initialize the SQLite backend plugin."""
-        self._db_path = db_path or Path.home() / ".hanzo" / "memory.db"
-        self._client: Optional[SQLiteMemoryClient] = None
+    Supports:
+    - Namespace-scoped memory storage
+    - Key-based exact retrieval and wildcard matching
+    - Tags for filtering and categorization
+    - TTL (time-to-live) with ISO date expiry
+    - Append mode for key-based content concatenation
+    - Version history per key
+    - Export/import
+    """
+
+    # Sentinel for in-memory database (useful for testing)
+    IN_MEMORY = ":memory:"
+
+    def __init__(self, db_path=None):
+        """Initialize the SQLite backend plugin.
+
+        Args:
+            db_path: Path to SQLite DB file, or ":memory:" for in-memory.
+                     Defaults to ~/.hanzo/memory.db.
+        """
+        if db_path == self.IN_MEMORY:
+            self._db_path = self.IN_MEMORY
+        else:
+            self._db_path = db_path or Path.home() / ".hanzo" / "memory.db"
+        self._client = None
+        self._conn: Optional[sqlite3.Connection] = None
         self._initialized = False
 
     @property
     def name(self) -> str:
-        """Unique name of the backend."""
         return "sqlite"
 
     @property
     def capabilities(self) -> List[Capability]:
-        """List of capabilities provided by this backend."""
         return [
             Capability.PERSISTENCE,
             Capability.EMBEDDINGS,
             Capability.MARKDOWN_IMPORT,
             Capability.STRUCTURED_QUERY,
-            # Vector search capability is conditional based on sqlite-vec availability
-            Capability.VECTOR_SEARCH,  # Available if sqlite-vec is installed
+            Capability.VECTOR_SEARCH,
         ]
 
     async def initialize(self) -> None:
         """Initialize the backend."""
         if not self._initialized:
-            self._client = SQLiteMemoryClient(db_path=self._db_path)
-            # The client is initialized in its constructor
+            if self._db_path == self.IN_MEMORY:
+                # Pure in-memory — skip hanzo_memory client, use direct sqlite3
+                self._conn = sqlite3.connect(":memory:", check_same_thread=False)
+                self._conn.row_factory = sqlite3.Row
+            else:
+                # Use hanzo_memory SQLiteMemoryClient if available
+                try:
+                    from hanzo_memory.db.sqlite_client import SQLiteMemoryClient
+
+                    self._client = SQLiteMemoryClient(db_path=self._db_path)
+                    self._conn = self._client.conn
+                except Exception:
+                    db_str = str(self._db_path)
+                    self._conn = sqlite3.connect(db_str, check_same_thread=False)
+                    self._conn.row_factory = sqlite3.Row
+
+            self._ensure_schema()
             self._initialized = True
 
+    def _ensure_schema(self):
+        """Ensure the memories table has all required columns."""
+        conn = self._conn
+        if not conn:
+            return
+
+        # Create the memories table if it doesn't exist (standalone mode)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS memories (
+                id TEXT PRIMARY KEY,
+                memory_id TEXT UNIQUE NOT NULL,
+                project_id TEXT NOT NULL DEFAULT 'default',
+                user_id TEXT NOT NULL DEFAULT 'default',
+                content TEXT NOT NULL,
+                memory_type TEXT DEFAULT 'general',
+                importance REAL DEFAULT 0.5,
+                context TEXT,
+                metadata TEXT,
+                source TEXT,
+                embedding BLOB,
+                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            );
+        """)
+
+        # Add namespace column if missing
+        self._add_column_if_missing("memories", "namespace", "TEXT DEFAULT 'default'")
+        # Add key column if missing
+        self._add_column_if_missing("memories", "key", "TEXT")
+        # Add tags column (JSON array as TEXT)
+        self._add_column_if_missing("memories", "tags", "TEXT")
+        # Add ttl column (ISO date string)
+        self._add_column_if_missing("memories", "ttl", "TEXT")
+
+        # Create indexes for new columns
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_memories_namespace ON memories(namespace);"
+        )
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_memories_key ON memories(key);"
+        )
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_memories_ns_key ON memories(namespace, key);"
+        )
+        conn.commit()
+
+    def _add_column_if_missing(self, table: str, column: str, col_type: str):
+        """Add a column to a table if it doesn't already exist."""
+        conn = self._conn
+        if not conn:
+            return
+        try:
+            conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_type}")
+            conn.commit()
+        except sqlite3.OperationalError:
+            # Column already exists
+            pass
+
     async def shutdown(self) -> None:
         """Shutdown the backend."""
         if self._client:
             self._client.close()
-            self._initialized = False
+            self._client = None
+        elif self._conn:
+            self._conn.close()
+        self._conn = None
+        self._initialized = False
+
+    # ------------------------------------------------------------------ #
+    # Core CRUD (backward-compatible signatures + new params)
+    # ------------------------------------------------------------------ #
 
     async def store_memory(
         self,
@@ -54,38 +156,68 @@ class SQLiteBackendPlugin:
         metadata: Dict[str, Any],
         user_id: str = "default",
         project_id: str = "default",
+        namespace: str = "default",
+        key: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        ttl: Optional[str] = None,
+        append: bool = False,
     ) -> str:
         """Store a memory and return its ID."""
-        if not self._client:
+        if not self._conn:
             raise RuntimeError("Plugin not initialized")
 
-        # For now, we'll store without embeddings to avoid dependency issues
-        # In a real implementation, we'd generate embeddings
-        MemoryCreate(
-            content=content,
-            memory_type=metadata.get("type", "general"),
-            importance=metadata.get("importance", 0.5),
-            context=metadata.get("context", {}),
-            metadata=metadata,
-            source=metadata.get("source", ""),
-        )
+        # Inject namespace/key/tags into metadata for backward compat
+        metadata = dict(metadata)
+        metadata["namespace"] = namespace
+        if key:
+            metadata["key"] = key
+        if tags:
+            metadata["tags"] = tags
+        if ttl:
+            metadata["ttl"] = ttl
 
-        # Add memory directly to the database
-        import uuid
+        # Handle append mode: find existing entry with same key+namespace and concatenate
+        if append and key:
+            existing = self._find_by_key(key, namespace)
+            if existing:
+                new_content = existing["content"] + content
+                self._conn.execute(
+                    "UPDATE memories SET content = ?, metadata = ?, updated_at = CURRENT_TIMESTAMP WHERE memory_id = ?",
+                    (new_content, json.dumps(metadata), existing["memory_id"]),
+                )
+                self._conn.commit()
+                return existing["memory_id"]
 
         memory_id = str(uuid.uuid4())
+        tags_json = json.dumps(tags or [])
+        metadata_json = json.dumps(metadata)
 
-        result = self._client.add_memory(
-            memory_id=memory_id,
-            user_id=user_id,
-            project_id=project_id,
-            content=content,
-            embedding=[],  # Empty embedding for now
-            metadata=metadata,
-            importance=metadata.get("importance", 0.5),
+        self._conn.execute(
+            """
+            INSERT INTO memories
+            (id, memory_id, user_id, project_id, content, importance, context,
+             metadata, source, embedding, namespace, key, tags, ttl)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                str(uuid.uuid4()),
+                memory_id,
+                user_id,
+                project_id,
+                content,
+                metadata.get("importance", 0.5),
+                json.dumps({}),
+                metadata_json,
+                metadata.get("source", ""),
+                None,  # embedding
+                namespace,
+                key,
+                tags_json,
+                ttl,
+            ),
         )
-
-        return result["memory_id"]
+        self._conn.commit()
+        return memory_id
 
     async def retrieve_memory(
         self,
@@ -93,54 +225,51 @@ class SQLiteBackendPlugin:
         user_id: str = "default",
         project_id: str = "default",
         limit: int = 10,
+        namespace: Optional[str] = None,
+        metadata_filter: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
         """Retrieve memories based on query."""
-        if not self._client:
+        if not self._conn:
             raise RuntimeError("Plugin not initialized")
 
-        # For now, return all memories for the user/project since we can't do proper search without embeddings
-        # In a real implementation, we'd search based on the query
-        try:
-            # Try to search if sqlite-vec is available (will fall back to basic search)
-            from hanzo_memory.services.embeddings import get_embedding_service
+        where_parts = ["user_id = ?", "project_id = ?"]
+        params: list = [user_id, project_id]
 
-            embedding_service = get_embedding_service()
-            query_embedding = embedding_service.embed_single(query)
+        if namespace:
+            where_parts.append("namespace = ?")
+            params.append(namespace)
 
-            results = self._client.search_memories(
-                user_id=user_id,
-                query_embedding=query_embedding,
-                project_id=project_id,
-                limit=limit,
-            )
+        if metadata_filter:
+            ns = metadata_filter.get("namespace")
+            if ns:
+                where_parts.append("namespace = ?")
+                params.append(ns)
+            agent = metadata_filter.get("agent")
+            if agent:
+                where_parts.append("json_extract(metadata, '$.agent') = ?")
+                params.append(agent)
+            mtype = metadata_filter.get("type")
+            if mtype:
+                where_parts.append("json_extract(metadata, '$.type') = ?")
+                params.append(mtype)
 
-            return [
-                {
-                    "id": r.get("memory_id", ""),
-                    "content": r.get("content", ""),
-                    "user_id": r.get("user_id", ""),
-                    "project_id": r.get("project_id", ""),
-                    "similarity_score": r.get("similarity_score", 0.0),
-                    "metadata": r.get("metadata", {}),
-                    "timestamp": r.get("created_at", ""),
-                }
-                for r in results
-            ]
-        except Exception:
-            # Fallback to basic retrieval if embedding service not available
-            # This would just return recent memories
-            # For now, we'll simulate by returning a basic result
-            return [
-                {
-                    "id": "fallback-result",
-                    "content": f"No search results for: {query}",
-                    "user_id": user_id,
-                    "project_id": project_id,
-                    "similarity_score": 0.0,
-                    "metadata": {},
-                    "timestamp": "",
-                }
-            ]
+        # Content search via LIKE if query provided
+        if query:
+            where_parts.append("content LIKE ?")
+            params.append(f"%{query}%")
+
+        # Filter out expired TTL entries
+        where_parts.append("(ttl IS NULL OR ttl = '' OR ttl > ?)")
+        params.append(datetime.now(timezone.utc).isoformat())
+
+        where_clause = " AND ".join(where_parts)
+        params.append(limit)
+
+        cursor = self._conn.execute(
+            f"SELECT * FROM memories WHERE {where_clause} ORDER BY created_at DESC LIMIT ?",
+            params,
+        )
+        return [self._row_to_dict(row) for row in cursor.fetchall()]
 
     async def search_memory(
         self,
@@ -148,23 +277,326 @@ class SQLiteBackendPlugin:
         user_id: str = "default",
         project_id: str = "default",
         limit: int = 10,
+        namespace: Optional[str] = None,
+        metadata_filter: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
         """Search memories based on query with scoring."""
-        return await self.retrieve_memory(query, user_id, project_id, limit)
+        return await self.retrieve_memory(
+            query, user_id, project_id, limit, namespace, metadata_filter
+        )
 
     async def delete_memory(
-        self, memory_id: str, user_id: str = "default", project_id: str = "default"
+        self,
+        memory_id: str,
+        user_id: str = "default",
+        project_id: str = "default",
     ) -> bool:
         """Delete a memory by ID."""
-        if not self._client:
+        if not self._conn:
             raise RuntimeError("Plugin not initialized")
 
-        return (
-            self._client.update_memory(
-                memory_id=memory_id,
-                user_id=user_id,
-                project_id=project_id,
-                content="",  # Setting content to empty as a form of deletion
-            )
-            is not None
+        cursor = self._conn.execute(
+            "DELETE FROM memories WHERE memory_id = ?",
+            (memory_id,),
         )
+        self._conn.commit()
+        return cursor.rowcount > 0
+
+    # ------------------------------------------------------------------ #
+    # New operations for TypeScript parity
+    # ------------------------------------------------------------------ #
+
+    async def get_by_key(
+        self,
+        key: str,
+        namespace: str = "default",
+    ):
+        """Get memory by exact key or wildcard within namespace.
+
+        If key contains '*', returns a list of matches.
+        Otherwise returns a single dict or None.
+        """
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        now = datetime.now(timezone.utc).isoformat()
+
+        if "*" in key:
+            # Wildcard: convert glob to SQL LIKE
+            like_pattern = key.replace("*", "%")
+            cursor = self._conn.execute(
+                """SELECT * FROM memories
+                   WHERE key LIKE ? AND namespace = ?
+                   AND (ttl IS NULL OR ttl = '' OR ttl > ?)
+                   ORDER BY created_at DESC""",
+                (like_pattern, namespace, now),
+            )
+            rows = cursor.fetchall()
+            return [self._row_to_dict(r) for r in rows] if rows else []
+        else:
+            row_dict = self._find_by_key(key, namespace)
+            if row_dict and self._is_expired(row_dict):
+                return None
+            return row_dict
+
+    async def list_memories(
+        self,
+        namespace: Optional[str] = None,
+        tag: Optional[str] = None,
+        limit: int = 50,
+    ) -> List[Dict[str, Any]]:
+        """List all memories with optional filters."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        where_parts: list = []
+        params: list = []
+
+        if namespace:
+            where_parts.append("namespace = ?")
+            params.append(namespace)
+
+        if tag:
+            # tags is a JSON array stored as TEXT; use json_each to filter
+            where_parts.append(
+                "EXISTS (SELECT 1 FROM json_each(tags) WHERE json_each.value = ?)"
+            )
+            params.append(tag)
+
+        # Filter expired
+        now = datetime.now(timezone.utc).isoformat()
+        where_parts.append("(ttl IS NULL OR ttl = '' OR ttl > ?)")
+        params.append(now)
+
+        where_clause = " AND ".join(where_parts) if where_parts else "1=1"
+        params.append(limit)
+
+        cursor = self._conn.execute(
+            f"SELECT * FROM memories WHERE {where_clause} ORDER BY created_at DESC LIMIT ?",
+            params,
+        )
+        return [self._row_to_dict(row) for row in cursor.fetchall()]
+
+    async def namespaces(self) -> Dict[str, int]:
+        """Return all namespaces with their memory counts (excludes expired)."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        now = datetime.now(timezone.utc).isoformat()
+        cursor = self._conn.execute(
+            """SELECT namespace, COUNT(*) as cnt FROM memories
+               WHERE (ttl IS NULL OR ttl = '' OR ttl > ?)
+               GROUP BY namespace""",
+            (now,),
+        )
+        return {row[0] or "default": row[1] for row in cursor.fetchall()}
+
+    async def stats(self) -> Dict[str, Any]:
+        """Return count, namespaces, size info."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        ns = await self.namespaces()
+        total = sum(ns.values())
+        return {
+            "count": total,
+            "namespaces": ns,
+        }
+
+    async def clear(self, namespace: Optional[str] = None) -> int:
+        """Clear all or namespace-specific memories. Returns count deleted."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        if namespace:
+            cursor = self._conn.execute(
+                "DELETE FROM memories WHERE namespace = ?", (namespace,)
+            )
+        else:
+            cursor = self._conn.execute("DELETE FROM memories")
+
+        self._conn.commit()
+        return cursor.rowcount
+
+    async def tag_memory(self, memory_id: str, tag: str) -> bool:
+        """Add a tag to a memory."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        cursor = self._conn.execute(
+            "SELECT tags, metadata FROM memories WHERE memory_id = ?", (memory_id,)
+        )
+        row = cursor.fetchone()
+        if not row:
+            return False
+
+        tags = json.loads(row[0]) if row[0] else []
+        if tag not in tags:
+            tags.append(tag)
+
+        # Also update tags in metadata
+        metadata = json.loads(row[1]) if row[1] else {}
+        metadata["tags"] = tags
+
+        self._conn.execute(
+            "UPDATE memories SET tags = ?, metadata = ?, updated_at = CURRENT_TIMESTAMP WHERE memory_id = ?",
+            (json.dumps(tags), json.dumps(metadata), memory_id),
+        )
+        self._conn.commit()
+        return True
+
+    async def untag_memory(self, memory_id: str, tag: str) -> bool:
+        """Remove a tag from a memory."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        cursor = self._conn.execute(
+            "SELECT tags, metadata FROM memories WHERE memory_id = ?", (memory_id,)
+        )
+        row = cursor.fetchone()
+        if not row:
+            return False
+
+        tags = json.loads(row[0]) if row[0] else []
+        if tag in tags:
+            tags.remove(tag)
+        else:
+            return False
+
+        metadata = json.loads(row[1]) if row[1] else {}
+        metadata["tags"] = tags
+
+        self._conn.execute(
+            "UPDATE memories SET tags = ?, metadata = ?, updated_at = CURRENT_TIMESTAMP WHERE memory_id = ?",
+            (json.dumps(tags), json.dumps(metadata), memory_id),
+        )
+        self._conn.commit()
+        return True
+
+    async def history(self, key: str, namespace: str = "default") -> List[Dict[str, Any]]:
+        """Show all versions/entries for a key within a namespace, ordered by creation time."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        cursor = self._conn.execute(
+            "SELECT * FROM memories WHERE key = ? AND namespace = ? ORDER BY created_at ASC",
+            (key, namespace),
+        )
+        return [self._row_to_dict(row) for row in cursor.fetchall()]
+
+    async def export_memories(
+        self, namespace: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Export memories as a list of dicts."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        if namespace:
+            cursor = self._conn.execute(
+                "SELECT * FROM memories WHERE namespace = ? ORDER BY created_at ASC",
+                (namespace,),
+            )
+        else:
+            cursor = self._conn.execute(
+                "SELECT * FROM memories ORDER BY created_at ASC"
+            )
+
+        return [self._row_to_dict(row) for row in cursor.fetchall()]
+
+    async def import_memories(self, data: List[Dict[str, Any]]) -> int:
+        """Import memories from exported data. Returns count imported."""
+        if not self._conn:
+            raise RuntimeError("Plugin not initialized")
+
+        count = 0
+        for entry in data:
+            memory_id = entry.get("memory_id") or entry.get("id") or str(uuid.uuid4())
+            metadata = entry.get("metadata", {})
+            namespace = metadata.get("namespace", entry.get("namespace", "default"))
+            key = metadata.get("key", entry.get("key"))
+            tags = metadata.get("tags", entry.get("tags", []))
+            ttl = metadata.get("ttl", entry.get("ttl"))
+
+            self._conn.execute(
+                """
+                INSERT INTO memories
+                (id, memory_id, user_id, project_id, content, importance, context,
+                 metadata, source, embedding, namespace, key, tags, ttl)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    str(uuid.uuid4()),
+                    memory_id,
+                    entry.get("user_id", "default"),
+                    entry.get("project_id", "default"),
+                    entry.get("content", ""),
+                    entry.get("importance", 0.5),
+                    json.dumps(entry.get("context", {})),
+                    json.dumps(metadata),
+                    entry.get("source", ""),
+                    None,
+                    namespace,
+                    key,
+                    json.dumps(tags if isinstance(tags, list) else []),
+                    ttl,
+                ),
+            )
+            count += 1
+
+        self._conn.commit()
+        return count
+
+    # ------------------------------------------------------------------ #
+    # Internal helpers
+    # ------------------------------------------------------------------ #
+
+    def _find_by_key(self, key: str, namespace: str) -> Optional[Dict[str, Any]]:
+        """Find a single memory by exact key + namespace."""
+        if not self._conn:
+            return None
+        cursor = self._conn.execute(
+            "SELECT * FROM memories WHERE key = ? AND namespace = ? ORDER BY created_at DESC LIMIT 1",
+            (key, namespace),
+        )
+        row = cursor.fetchone()
+        return self._row_to_dict(row) if row else None
+
+    def _is_expired(self, memory: Dict[str, Any]) -> bool:
+        """Check if a memory's TTL has expired."""
+        ttl = memory.get("ttl") or (memory.get("metadata", {}).get("ttl"))
+        if not ttl:
+            return False
+        try:
+            expiry = datetime.fromisoformat(ttl)
+            if expiry.tzinfo is None:
+                expiry = expiry.replace(tzinfo=timezone.utc)
+            return datetime.now(timezone.utc) > expiry
+        except (ValueError, TypeError):
+            return False
+
+    def _row_to_dict(self, row) -> Dict[str, Any]:
+        """Convert a sqlite3.Row to a dict with parsed JSON fields."""
+        if row is None:
+            return {}
+
+        # sqlite3.Row supports both index and key access
+        d = dict(row) if hasattr(row, "keys") else {}
+        if not d:
+            # Fallback for tuple rows
+            return {}
+
+        # Parse JSON fields
+        for field in ("metadata", "context"):
+            if field in d and isinstance(d[field], str):
+                try:
+                    d[field] = json.loads(d[field])
+                except (json.JSONDecodeError, TypeError):
+                    d[field] = {}
+
+        if "tags" in d and isinstance(d["tags"], str):
+            try:
+                d["tags"] = json.loads(d["tags"])
+            except (json.JSONDecodeError, TypeError):
+                d["tags"] = []
+
+        return d

pkg/hanzo-mcp/hanzo_mcp/coordination.py

@@ -0,0 +1,116 @@
+"""Blue-Red agent coordination protocol via shared memory.
+
+The blue-red protocol is a structured review cycle:
+1. Blue agent analyzes code/system and stores a report
+2. Red agent reads Blue's report, finds issues, stores findings
+3. Blue reads Red's findings, implements fixes, stores response
+4. Red re-reviews Blue's fixes, stores verification result
+
+All messages are stored in the "blue-red" namespace with typed keys,
+enabling deterministic retrieval without semantic search.
+"""
+
+import json
+import time
+from typing import Any, Dict, List, Optional
+
+
+class BlueRedChannel:
+    """Shared memory channel for blue-red agent coordination."""
+
+    NAMESPACE = "blue-red"
+
+    def __init__(self, memory_service):
+        """Initialize with a PluginMemoryService instance."""
+        self.memory = memory_service
+
+    async def blue_report(self, report: dict, scope: str = "") -> str:
+        """Blue stores its report for Red to read."""
+        key = f"blue-report-{int(time.time())}"
+        await self.memory.store_memory(
+            content=json.dumps(report),
+            metadata={
+                "agent": "blue",
+                "type": "report",
+                "scope": scope,
+            },
+            namespace=self.NAMESPACE,
+            key=key,
+            tags=["blue", "report"],
+        )
+        return key
+
+    async def red_report(self, findings: dict, scope: str = "") -> str:
+        """Red stores findings for Blue to read."""
+        key = f"red-report-{int(time.time())}"
+        await self.memory.store_memory(
+            content=json.dumps(findings),
+            metadata={
+                "agent": "red",
+                "type": "findings",
+                "scope": scope,
+            },
+            namespace=self.NAMESPACE,
+            key=key,
+            tags=["red", "findings"],
+        )
+        return key
+
+    async def blue_response(self, fixes: dict) -> str:
+        """Blue stores fix response for Red re-review."""
+        key = f"blue-response-{int(time.time())}"
+        await self.memory.store_memory(
+            content=json.dumps(fixes),
+            metadata={
+                "agent": "blue",
+                "type": "response",
+            },
+            namespace=self.NAMESPACE,
+            key=key,
+            tags=["blue", "response", "fixes"],
+        )
+        return key
+
+    async def red_rereview(self, verification: dict) -> str:
+        """Red stores re-review results."""
+        key = f"red-rereview-{int(time.time())}"
+        await self.memory.store_memory(
+            content=json.dumps(verification),
+            metadata={
+                "agent": "red",
+                "type": "rereview",
+            },
+            namespace=self.NAMESPACE,
+            key=key,
+            tags=["red", "rereview"],
+        )
+        return key
+
+    async def get_latest_blue_report(self) -> Optional[Dict[str, Any]]:
+        """Red reads Blue's latest report."""
+        results = await self.memory.search_memory(
+            query="",
+            metadata_filter={
+                "namespace": self.NAMESPACE,
+                "agent": "blue",
+                "type": "report",
+            },
+        )
+        return results[0] if results else None
+
+    async def get_latest_red_findings(self) -> Optional[Dict[str, Any]]:
+        """Blue reads Red's latest findings."""
+        results = await self.memory.search_memory(
+            query="",
+            metadata_filter={
+                "namespace": self.NAMESPACE,
+                "agent": "red",
+                "type": "findings",
+            },
+        )
+        return results[0] if results else None
+
+    async def get_full_cycle(self) -> List[Dict[str, Any]]:
+        """Get all messages in the blue-red cycle, ordered by creation time."""
+        results = await self.memory.list_memories(namespace=self.NAMESPACE)
+        return sorted(results, key=lambda r: r.get("created_at", ""))

pkg/hanzo-mcp/hanzo_mcp/memory_service.py

@@ -1,4 +1,8 @@
-"""Memory service with plugin support."""
+"""Memory service with plugin support.
+
+Provides namespace-scoped, key-addressable memory with tags, TTL,
+and full TypeScript parity for the blue-red agent coordination protocol.
+"""
 
 from pathlib import Path
 from typing import Any, Dict, List, Optional
@@ -8,7 +12,15 @@ from .plugin_interface import Capability, PluginRegistry
 
 
 class PluginMemoryService:
-    """Memory service that uses plugins for backend operations."""
+    """Memory service that uses plugins for backend operations.
+
+    Supports:
+    - namespace: isolate memories into named channels (default: "default")
+    - key: address individual memories by name for exact retrieval
+    - tags: categorize and filter memories
+    - ttl: auto-expire memories after an ISO date
+    - append: concatenate content to an existing keyed memory
+    """
 
     def __init__(self, config_path: Optional[Path] = None):
         """Initialize the service with configuration."""
@@ -21,36 +33,22 @@ class PluginMemoryService:
 
     def _register_available_plugins(self):
         """Register all available backend plugins."""
-        # Register SQLite plugin (always available)
         self.registry.register_plugin(SQLiteBackendPlugin())
 
-        # Conditionally register other plugins if dependencies are available
-        try:
-            # For now, we'll just register the SQLite plugin
-            # In the future, we could register other plugins like:
-            # from .backends.lancedb_plugin import LanceDBBackendPlugin
-            # self.registry.register_plugin(LanceDBBackendPlugin())
-            pass
-        except ImportError:
-            pass  # Skip if dependencies not available
-
     async def initialize(self, enabled_backends: List[str] = None):
         """Initialize the service and load configured plugins."""
         if self._initialized:
             return
 
-        # If no backends specified, enable default SQLite backend
         if enabled_backends is None:
             enabled_backends = ["sqlite"]
 
-        # Enable configured plugins
         for backend_name in enabled_backends:
             if backend_name in self.registry.get_available_plugins():
                 self.registry.enable_plugin(backend_name)
             else:
                 print(f"Warning: Backend '{backend_name}' not available, skipping.")
 
-        # Initialize active plugins
         await self.registry.initialize_all_active()
         self._initialized = True
 
@@ -59,25 +57,62 @@ class PluginMemoryService:
         await self.registry.shutdown_all_active()
         self._initialized = False
 
+    def _get_plugin(self):
+        """Get the first active plugin, initializing if needed."""
+        active = self.registry.get_active_plugins()
+        if not active:
+            raise RuntimeError("No active plugins available")
+        return active[0]
+
+    async def _ensure_init(self):
+        if not self._initialized:
+            await self.initialize()
+
+    # ------------------------------------------------------------------ #
+    # Core CRUD — backward compatible + new params
+    # ------------------------------------------------------------------ #
+
     async def store_memory(
         self,
         content: str,
         metadata: Dict[str, Any],
         user_id: str = "default",
         project_id: str = "default",
+        namespace: str = "default",
+        key: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        ttl: Optional[str] = None,
+        append: bool = False,
     ) -> str:
-        """Store a memory using active plugins."""
-        if not self._initialized:
-            await self.initialize()
+        """Store a memory using active plugins.
 
-        # Use the first active plugin for storage
-        active_plugins = self.registry.get_active_plugins()
-        if not active_plugins:
-            raise RuntimeError("No active plugins available")
+        Args:
+            content: The memory content.
+            metadata: Arbitrary metadata dict.
+            user_id: Owner user ID.
+            project_id: Project scope.
+            namespace: Logical namespace (e.g. "blue-red").
+            key: Optional unique key for retrieval.
+            tags: Optional list of tags.
+            ttl: Optional ISO date string; memory expires after this time.
+            append: If True and key exists, concatenate content.
 
-        # For now, just use the first plugin
-        plugin = active_plugins[0]
-        return await plugin.store_memory(content, metadata, user_id, project_id)
+        Returns:
+            The memory_id of the stored entry.
+        """
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.store_memory(
+            content=content,
+            metadata=metadata,
+            user_id=user_id,
+            project_id=project_id,
+            namespace=namespace,
+            key=key,
+            tags=tags,
+            ttl=ttl,
+            append=append,
+        )
 
     async def retrieve_memory(
         self,
@@ -85,19 +120,20 @@ class PluginMemoryService:
         user_id: str = "default",
         project_id: str = "default",
         limit: int = 10,
+        namespace: Optional[str] = None,
+        metadata_filter: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
         """Retrieve memories using active plugins."""
-        if not self._initialized:
-            await self.initialize()
-
-        # Use the first active plugin for retrieval
-        active_plugins = self.registry.get_active_plugins()
-        if not active_plugins:
-            raise RuntimeError("No active plugins available")
-
-        # For now, just use the first plugin
-        plugin = active_plugins[0]
-        return await plugin.retrieve_memory(query, user_id, project_id, limit)
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.retrieve_memory(
+            query=query,
+            user_id=user_id,
+            project_id=project_id,
+            limit=limit,
+            namespace=namespace,
+            metadata_filter=metadata_filter,
+        )
 
     async def search_memory(
         self,
@@ -105,56 +141,129 @@ class PluginMemoryService:
         user_id: str = "default",
         project_id: str = "default",
         limit: int = 10,
+        namespace: Optional[str] = None,
+        metadata_filter: Optional[Dict[str, Any]] = None,
     ) -> List[Dict[str, Any]]:
         """Search memories using active plugins."""
-        if not self._initialized:
-            await self.initialize()
-
-        # Use the first active plugin for search
-        active_plugins = self.registry.get_active_plugins()
-        if not active_plugins:
-            raise RuntimeError("No active plugins available")
-
-        # For now, just use the first plugin
-        plugin = active_plugins[0]
-        return await plugin.search_memory(query, user_id, project_id, limit)
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.search_memory(
+            query=query,
+            user_id=user_id,
+            project_id=project_id,
+            limit=limit,
+            namespace=namespace,
+            metadata_filter=metadata_filter,
+        )
 
     async def delete_memory(
-        self, memory_id: str, user_id: str = "default", project_id: str = "default"
+        self,
+        memory_id: str,
+        user_id: str = "default",
+        project_id: str = "default",
     ) -> bool:
         """Delete a memory using active plugins."""
-        if not self._initialized:
-            await self.initialize()
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.delete_memory(
+            memory_id=memory_id,
+            user_id=user_id,
+            project_id=project_id,
+        )
 
-        # Use the first active plugin for deletion
-        active_plugins = self.registry.get_active_plugins()
-        if not active_plugins:
-            raise RuntimeError("No active plugins available")
+    # ------------------------------------------------------------------ #
+    # New operations for TypeScript parity
+    # ------------------------------------------------------------------ #
 
-        # For now, just use the first plugin
-        plugin = active_plugins[0]
-        return await plugin.delete_memory(memory_id, user_id, project_id)
+    async def get_by_key(self, key: str, namespace: str = "default"):
+        """Get memory by exact key or wildcard within namespace.
+
+        Exact key returns a single dict or None.
+        Wildcard key (contains '*') returns a list.
+        """
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.get_by_key(key=key, namespace=namespace)
+
+    async def list_memories(
+        self,
+        namespace: Optional[str] = None,
+        tag: Optional[str] = None,
+        limit: int = 50,
+    ) -> List[Dict[str, Any]]:
+        """List all memories with optional namespace/tag filters."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.list_memories(namespace=namespace, tag=tag, limit=limit)
+
+    async def stats(self) -> Dict[str, Any]:
+        """Return count, namespaces, size info."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.stats()
+
+    async def clear(self, namespace: Optional[str] = None) -> int:
+        """Clear all or namespace-specific memories. Returns count deleted."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.clear(namespace=namespace)
+
+    async def tag_memory(self, memory_id: str, tag: str) -> bool:
+        """Add a tag to a memory."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.tag_memory(memory_id=memory_id, tag=tag)
+
+    async def untag_memory(self, memory_id: str, tag: str) -> bool:
+        """Remove a tag from a memory."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.untag_memory(memory_id=memory_id, tag=tag)
+
+    async def namespaces(self) -> Dict[str, int]:
+        """List all namespaces with counts."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.namespaces()
+
+    async def history(self, key: str, namespace: str = "default") -> List[Dict[str, Any]]:
+        """Show all versions of a key."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.history(key=key, namespace=namespace)
+
+    async def export_memories(
+        self, namespace: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Export memories as a list of dicts."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.export_memories(namespace=namespace)
+
+    async def import_memories(self, data: List[Dict[str, Any]]) -> int:
+        """Import memories from exported data. Returns count imported."""
+        await self._ensure_init()
+        plugin = self._get_plugin()
+        return await plugin.import_memories(data=data)
+
+    # ------------------------------------------------------------------ #
+    # Backend management (unchanged)
+    # ------------------------------------------------------------------ #
 
     def get_available_backends(self) -> List[str]:
-        """Get list of available backends."""
         return self.registry.get_available_plugins()
 
     def get_active_backends(self) -> List[str]:
-        """Get list of active backends."""
         return [plugin.name for plugin in self.registry.get_active_plugins()]
 
     def get_backend_capabilities(self, backend_name: str) -> List[Capability]:
-        """Get capabilities of a specific backend."""
         return self.registry.get_plugin_capabilities(backend_name)
 
     def has_capability(self, capability: Capability) -> List[str]:
-        """Get all backends that support a specific capability."""
         return self.registry.has_capability(capability)
 
     def enable_backend(self, backend_name: str) -> bool:
-        """Enable a specific backend."""
         return self.registry.enable_plugin(backend_name)
 
     def disable_backend(self, backend_name: str) -> bool:
-        """Disable a specific backend."""
         return self.registry.disable_plugin(backend_name)

pkg/hanzo-mcp/tests/test_memory_namespace.py

@@ -0,0 +1,390 @@
+"""Tests for memory service namespace, key-based retrieval, tags, ttl, and coordination."""
+
+import asyncio
+import json
+import time
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+
+@pytest.fixture
+async def memory_service():
+    """Create a PluginMemoryService with in-memory SQLite for testing."""
+    from hanzo_mcp.memory_service import PluginMemoryService
+    from hanzo_mcp.backends.sqlite_plugin import SQLiteBackendPlugin
+
+    svc = PluginMemoryService()
+    # Use true in-memory db — each test gets a fresh database
+    plugin = SQLiteBackendPlugin(db_path=SQLiteBackendPlugin.IN_MEMORY)
+    svc.registry._plugins = {"sqlite": plugin}
+    svc.registry._active_plugins = ["sqlite"]
+    await svc.registry.initialize_all_active()
+    svc._initialized = True
+    yield svc
+    await svc.shutdown()
+
+
+class TestNamespaceSupport:
+    """Test namespace parameter on memory operations."""
+
+    @pytest.mark.asyncio
+    async def test_store_with_namespace(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="blue agent report",
+            metadata={"type": "report"},
+            namespace="blue-red",
+        )
+        assert mid
+
+    @pytest.mark.asyncio
+    async def test_store_default_namespace(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="plain memory",
+            metadata={},
+        )
+        assert mid
+
+    @pytest.mark.asyncio
+    async def test_list_memories_filters_by_namespace(self, memory_service):
+        await memory_service.store_memory(content="ns1", metadata={}, namespace="ns1")
+        await memory_service.store_memory(content="ns2", metadata={}, namespace="ns2")
+        await memory_service.store_memory(content="ns1-b", metadata={}, namespace="ns1")
+
+        results = await memory_service.list_memories(namespace="ns1")
+        assert len(results) == 2
+        for r in results:
+            assert r["metadata"].get("namespace") == "ns1"
+
+    @pytest.mark.asyncio
+    async def test_list_memories_no_filter(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, namespace="x")
+        await memory_service.store_memory(content="b", metadata={}, namespace="y")
+        results = await memory_service.list_memories()
+        assert len(results) == 2
+
+    @pytest.mark.asyncio
+    async def test_namespaces_returns_counts(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, namespace="alpha")
+        await memory_service.store_memory(content="b", metadata={}, namespace="alpha")
+        await memory_service.store_memory(content="c", metadata={}, namespace="beta")
+
+        ns = await memory_service.namespaces()
+        assert ns["alpha"] == 2
+        assert ns["beta"] == 1
+
+    @pytest.mark.asyncio
+    async def test_clear_namespace(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, namespace="temp")
+        await memory_service.store_memory(content="b", metadata={}, namespace="keep")
+
+        deleted = await memory_service.clear(namespace="temp")
+        assert deleted >= 1
+
+        results = await memory_service.list_memories()
+        assert len(results) == 1
+        assert results[0]["metadata"].get("namespace") == "keep"
+
+    @pytest.mark.asyncio
+    async def test_clear_all(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, namespace="x")
+        await memory_service.store_memory(content="b", metadata={}, namespace="y")
+
+        deleted = await memory_service.clear()
+        assert deleted >= 2
+
+        results = await memory_service.list_memories()
+        assert len(results) == 0
+
+
+class TestKeyBasedRetrieval:
+    """Test key-based store and get_by_key."""
+
+    @pytest.mark.asyncio
+    async def test_store_with_key(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="keyed content",
+            metadata={"extra": "data"},
+            key="my-key-1",
+            namespace="test",
+        )
+        assert mid
+
+    @pytest.mark.asyncio
+    async def test_get_by_key_exact(self, memory_service):
+        await memory_service.store_memory(
+            content="target", metadata={}, key="exact-key", namespace="ns"
+        )
+        await memory_service.store_memory(
+            content="other", metadata={}, key="other-key", namespace="ns"
+        )
+
+        result = await memory_service.get_by_key(key="exact-key", namespace="ns")
+        assert result is not None
+        assert result["content"] == "target"
+
+    @pytest.mark.asyncio
+    async def test_get_by_key_wildcard(self, memory_service):
+        await memory_service.store_memory(
+            content="report-1", metadata={}, key="blue-report-100", namespace="blue-red"
+        )
+        await memory_service.store_memory(
+            content="report-2", metadata={}, key="blue-report-200", namespace="blue-red"
+        )
+        await memory_service.store_memory(
+            content="other", metadata={}, key="red-report-100", namespace="blue-red"
+        )
+
+        results = await memory_service.get_by_key(
+            key="blue-report-*", namespace="blue-red"
+        )
+        assert isinstance(results, list)
+        assert len(results) == 2
+
+    @pytest.mark.asyncio
+    async def test_get_by_key_not_found(self, memory_service):
+        result = await memory_service.get_by_key(key="nonexistent", namespace="ns")
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_append_mode(self, memory_service):
+        await memory_service.store_memory(
+            content="line1", metadata={}, key="append-key", namespace="test"
+        )
+        await memory_service.store_memory(
+            content="\nline2", metadata={}, key="append-key", namespace="test", append=True
+        )
+
+        result = await memory_service.get_by_key(key="append-key", namespace="test")
+        assert "line1" in result["content"]
+        assert "line2" in result["content"]
+
+
+class TestTagsSupport:
+    """Test tags parameter on memory operations."""
+
+    @pytest.mark.asyncio
+    async def test_store_with_tags(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="tagged memory",
+            metadata={},
+            tags=["blue", "report"],
+        )
+        assert mid
+
+    @pytest.mark.asyncio
+    async def test_list_memories_filter_by_tag(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, tags=["blue"])
+        await memory_service.store_memory(content="b", metadata={}, tags=["red"])
+        await memory_service.store_memory(content="c", metadata={}, tags=["blue", "red"])
+
+        results = await memory_service.list_memories(tag="blue")
+        assert len(results) == 2
+
+    @pytest.mark.asyncio
+    async def test_tag_memory(self, memory_service):
+        mid = await memory_service.store_memory(content="taggable", metadata={})
+        ok = await memory_service.tag_memory(mid, "new-tag")
+        assert ok
+
+        results = await memory_service.list_memories(tag="new-tag")
+        assert len(results) == 1
+
+    @pytest.mark.asyncio
+    async def test_untag_memory(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="untaggable", metadata={}, tags=["remove-me"]
+        )
+        ok = await memory_service.untag_memory(mid, "remove-me")
+        assert ok
+
+        results = await memory_service.list_memories(tag="remove-me")
+        assert len(results) == 0
+
+
+class TestTTLSupport:
+    """Test TTL (time-to-live) on memories."""
+
+    @pytest.mark.asyncio
+    async def test_store_with_ttl(self, memory_service):
+        # TTL in the past = already expired
+        past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
+        mid = await memory_service.store_memory(
+            content="ephemeral", metadata={}, ttl=past
+        )
+        assert mid
+
+        # Should not appear in list (expired)
+        results = await memory_service.list_memories()
+        assert len(results) == 0
+
+    @pytest.mark.asyncio
+    async def test_store_with_future_ttl(self, memory_service):
+        future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+        mid = await memory_service.store_memory(
+            content="long-lived", metadata={}, ttl=future
+        )
+
+        results = await memory_service.list_memories()
+        assert len(results) == 1
+
+
+class TestStats:
+    """Test stats method."""
+
+    @pytest.mark.asyncio
+    async def test_stats_empty(self, memory_service):
+        s = await memory_service.stats()
+        assert s["count"] == 0
+        assert s["namespaces"] == {}
+
+    @pytest.mark.asyncio
+    async def test_stats_with_data(self, memory_service):
+        await memory_service.store_memory(content="a", metadata={}, namespace="ns1")
+        await memory_service.store_memory(content="b", metadata={}, namespace="ns1")
+        await memory_service.store_memory(content="c", metadata={}, namespace="ns2")
+
+        s = await memory_service.stats()
+        assert s["count"] == 3
+        assert s["namespaces"]["ns1"] == 2
+        assert s["namespaces"]["ns2"] == 1
+
+
+class TestHistory:
+    """Test version history for a key."""
+
+    @pytest.mark.asyncio
+    async def test_history_shows_versions(self, memory_service):
+        await memory_service.store_memory(
+            content="v1", metadata={}, key="versioned", namespace="test"
+        )
+        # Store again with same key but NOT append — creates new entry
+        await memory_service.store_memory(
+            content="v2", metadata={}, key="versioned", namespace="test"
+        )
+
+        versions = await memory_service.history(key="versioned", namespace="test")
+        assert len(versions) == 2
+
+
+class TestExportImport:
+    """Test export and import."""
+
+    @pytest.mark.asyncio
+    async def test_export_import_round_trip(self, memory_service):
+        await memory_service.store_memory(
+            content="exportable", metadata={"x": 1}, namespace="exp"
+        )
+
+        data = await memory_service.export_memories(namespace="exp")
+        assert len(data) == 1
+        assert data[0]["content"] == "exportable"
+
+        # Clear and re-import
+        await memory_service.clear()
+        imported = await memory_service.import_memories(data)
+        assert imported == 1
+
+        results = await memory_service.list_memories(namespace="exp")
+        assert len(results) == 1
+
+
+class TestBackwardCompatibility:
+    """Verify existing callers still work."""
+
+    @pytest.mark.asyncio
+    async def test_store_memory_original_signature(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="old style",
+            metadata={"type": "general"},
+            user_id="u1",
+            project_id="p1",
+        )
+        assert mid
+
+    @pytest.mark.asyncio
+    async def test_retrieve_memory_original_signature(self, memory_service):
+        await memory_service.store_memory(
+            content="findme", metadata={}, user_id="u1", project_id="p1"
+        )
+        results = await memory_service.retrieve_memory(
+            query="findme", user_id="u1", project_id="p1"
+        )
+        assert isinstance(results, list)
+
+    @pytest.mark.asyncio
+    async def test_delete_memory_original_signature(self, memory_service):
+        mid = await memory_service.store_memory(
+            content="deleteme", metadata={}, user_id="u1", project_id="p1"
+        )
+        ok = await memory_service.delete_memory(
+            memory_id=mid, user_id="u1", project_id="p1"
+        )
+        assert ok
+
+
+class TestBlueRedChannel:
+    """Test the coordination convenience class."""
+
+    @pytest.mark.asyncio
+    async def test_blue_report(self, memory_service):
+        from hanzo_mcp.coordination import BlueRedChannel
+
+        channel = BlueRedChannel(memory_service)
+        key = await channel.blue_report({"analysis": "code looks good"})
+        assert key.startswith("blue-report-")
+
+    @pytest.mark.asyncio
+    async def test_red_report(self, memory_service):
+        from hanzo_mcp.coordination import BlueRedChannel
+
+        channel = BlueRedChannel(memory_service)
+        key = await channel.red_report({"issues": ["bug in auth"]})
+        assert key.startswith("red-report-")
+
+    @pytest.mark.asyncio
+    async def test_blue_red_cycle(self, memory_service):
+        from hanzo_mcp.coordination import BlueRedChannel
+
+        channel = BlueRedChannel(memory_service)
+
+        # Blue reports
+        await channel.blue_report({"analysis": "reviewed auth module"}, scope="auth")
+
+        # Red reads and reports findings
+        blue_report = await channel.get_latest_blue_report()
+        assert blue_report is not None
+        assert "reviewed auth module" in blue_report["content"]
+
+        await channel.red_report({"issues": ["SQL injection risk"]}, scope="auth")
+
+        # Blue reads red findings
+        red_findings = await channel.get_latest_red_findings()
+        assert red_findings is not None
+        assert "SQL injection risk" in red_findings["content"]
+
+        # Blue responds with fixes
+        await channel.blue_response({"fixes": ["added parameterized queries"]})
+
+        # Red re-reviews
+        await channel.red_rereview({"verdict": "resolved"})
+
+        # Get full cycle
+        cycle = await channel.get_full_cycle()
+        assert len(cycle) == 4
+
+    @pytest.mark.asyncio
+    async def test_get_latest_blue_report_empty(self, memory_service):
+        from hanzo_mcp.coordination import BlueRedChannel
+
+        channel = BlueRedChannel(memory_service)
+        result = await channel.get_latest_blue_report()
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_get_latest_red_findings_empty(self, memory_service):
+        from hanzo_mcp.coordination import BlueRedChannel
+
+        channel = BlueRedChannel(memory_service)
+        result = await channel.get_latest_red_findings()
+        assert result is None

pkg/hanzo-mcp/uv.lock

@@ -1291,7 +1291,7 @@ wheels = [
 
 [[package]]
 name = "hanzo-mcp"
-version = "0.14.0"
+version = "0.15.0"
 source = { editable = "." }
 dependencies = [
     { name = "fastmcp" },
@@ -1328,6 +1328,7 @@ dependencies = [
     { name = "pydantic-settings" },
     { name = "typing-extensions" },
     { name = "uvloop", marker = "sys_platform != 'win32'" },
+    { name = "zap-protocol" },
 ]
 
 [package.optional-dependencies]
@@ -1350,7 +1351,6 @@ docs = [
 ]
 ide = [
     { name = "hanzo-tools-ide" },
-    { name = "websockets" },
 ]
 interactive = [
     { name = "hanzo-tools-browser" },
@@ -1358,7 +1358,6 @@ interactive = [
     { name = "hanzo-tools-repl" },
     { name = "ipykernel" },
     { name = "jupyter-client" },
-    { name = "websockets" },
 ]
 memory = [
     { name = "fastembed" },
@@ -1451,7 +1450,7 @@ requires-dist = [
     { name = "ujson", marker = "extra == 'performance'", specifier = ">=5.7.0" },
     { name = "uvloop", marker = "sys_platform != 'win32'", specifier = ">=0.22.1" },
     { name = "watchdog", marker = "extra == 'dev'", specifier = ">=3.0.0" },
-    { name = "websockets", marker = "extra == 'ide'", specifier = ">=12.0" },
+    { name = "zap-protocol", specifier = ">=0.3.0" },
 ]
 provides-extras = ["memory", "repl", "ide", "interactive", "dev", "docs", "test", "performance", "publish"]
 
@@ -4154,6 +4153,42 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ea/6d/41faa414cde66ec023b0ca8402a8f11cb61731c3dc27c082909cbbd1f929/pybase64-1.4.3-graalpy312-graalpy250_312_native-win_amd64.whl", hash = "sha256:f7537fa22ae56a0bf51e4b0ffc075926ad91c618e1416330939f7ef366b58e3b", size = 36231, upload-time = "2025-12-06T13:26:31.656Z" },
 ]
 
+[[package]]
+name = "pycapnp"
+version = "2.2.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/85/7b/b2f356bc24220068beffc03e94062e8059a1383addb837303794398aec36/pycapnp-2.2.2.tar.gz", hash = "sha256:7f6c23c2283173a3cb6f1a5086dd0114779d508a7cd1b138d25a6357857d02b6", size = 730142, upload-time = "2026-01-21T01:22:13.73Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/8a/76/f8f81d32ddf950e934ec144facbc112e5acbef31a63ba5be0c5f34a00fd5/pycapnp-2.2.2-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:8b86cb8ea5b8011b562c4e022325a826a62f91196ceb5aa33a766c0bea0b8fd3", size = 1605194, upload-time = "2026-01-21T01:20:29.604Z" },
+    { url = "https://files.pythonhosted.org/packages/50/dd/a31be782d56a8648fef899f39aeeab867cf544a6b170871e3f4cbfc58af6/pycapnp-2.2.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:2353531cfa669e3eeb99be9f993573341650276abec46676d687cc12b3e6b6d9", size = 1486613, upload-time = "2026-01-21T01:20:31.415Z" },
+    { url = "https://files.pythonhosted.org/packages/aa/bf/8da830dda94eb7327c6508d6c26fbd964897d742f8c1c0ec48623f0c515b/pycapnp-2.2.2-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:ee27bdc78c7ccd8eaa0fe31e09f0ec4ef31deda3f475fc9373bb4b0de8083053", size = 5186701, upload-time = "2026-01-21T01:20:32.836Z" },
+    { url = "https://files.pythonhosted.org/packages/8d/a1/13d0baa2f337f4f6fe8c2142646ba437a26b9c433f5d7ce016a912bad052/pycapnp-2.2.2-cp312-cp312-manylinux_2_28_i686.whl", hash = "sha256:a8ded808911d1d7a9a2197626c09eea6e269e74dc1276760789538b1efcf6cd5", size = 5239464, upload-time = "2026-01-21T01:20:34.793Z" },
+    { url = "https://files.pythonhosted.org/packages/82/76/0451c64b5f0132e4b75a0afe8cec957c8bf8fa981264a7c0b264cb94663e/pycapnp-2.2.2-cp312-cp312-manylinux_2_28_ppc64le.whl", hash = "sha256:59e92e1db40041d82a95eab0bd8de2676ce50c6b97c1457e2edde4d134b6d046", size = 5542887, upload-time = "2026-01-21T01:20:36.463Z" },
+    { url = "https://files.pythonhosted.org/packages/04/00/d025d68d9a5330d55cbe2d018091cacfef0835c3ad422eb6778c4525041f/pycapnp-2.2.2-cp312-cp312-manylinux_2_28_s390x.whl", hash = "sha256:ee1e9ac2f0b80fa892b922b60e36efc925d072ecf1204ba3e59d8d9ac7c3dc83", size = 5659696, upload-time = "2026-01-21T01:20:38.069Z" },
+    { url = "https://files.pythonhosted.org/packages/58/b7/28f7c539a5f4cbaa12e55ec27d081d11473464230f2e801e4714606d3453/pycapnp-2.2.2-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:53273b385be78ed8ac997ff8697f2a4c760e93c190b509822a937de5531f4861", size = 5413827, upload-time = "2026-01-21T01:20:39.781Z" },
+    { url = "https://files.pythonhosted.org/packages/3c/a7/83bc13d90675f0cee8a38d4ad8401bb2f8662c543b3a6622aeffb7b56b1e/pycapnp-2.2.2-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:812cbdd002bc542b63f969b85c6b9041dfdaf4185613635a6d4feea84c9092fa", size = 6046815, upload-time = "2026-01-21T01:20:42.172Z" },
+    { url = "https://files.pythonhosted.org/packages/0d/8a/80f46baa1684bbcc4754ce22c5a44693a1209a64de6df2b256b85b8b8a97/pycapnp-2.2.2-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:9c330218a44bd649b96f565dbf5326d183fdd20f9887bdedfeabd73f0366c2e1", size = 6367625, upload-time = "2026-01-21T01:20:44.004Z" },
+    { url = "https://files.pythonhosted.org/packages/02/00/60e82eaf6b4e78d887157bf9f18234c852771cc575355e63d1114c4a5d79/pycapnp-2.2.2-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:796aa0ba18bcd4e6b2815471bbed059ad7ee8a815a30e81ac8a9aa030ec7818d", size = 6487265, upload-time = "2026-01-21T01:20:46.137Z" },
+    { url = "https://files.pythonhosted.org/packages/57/6e/2dedd8f95dc22357c50a775ee2b8711b3d711f30344d244141e0e1962c3e/pycapnp-2.2.2-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:251a6abdd64b9b11d2a8e16fc365b922ef6ba6c968959b72a3a3d9d8ec8cc8d7", size = 6576699, upload-time = "2026-01-21T01:20:47.987Z" },
+    { url = "https://files.pythonhosted.org/packages/2f/53/f7f69ed1d11ea30ea4f0f6d8319fbc18bc8781c480c118005e0a394492a7/pycapnp-2.2.2-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6aab811e0fcc27ae8bf5f04dedaa7e0af47e0d4db51d9c85ab0d2dad26a46bd7", size = 6344114, upload-time = "2026-01-21T01:20:50.367Z" },
+    { url = "https://files.pythonhosted.org/packages/ab/78/ab78ee42797ff44c7e1fc0d1aa9396c6742cb05ff01a7cdf9c8f19e0defe/pycapnp-2.2.2-cp312-cp312-win32.whl", hash = "sha256:5061c85dd8f843b2656720ca6976d2a9b418845580c6f6d9602f7119fc2208d5", size = 1047207, upload-time = "2026-01-21T01:20:51.842Z" },
+    { url = "https://files.pythonhosted.org/packages/4c/fb/6edf56d5144c476270fa8b2e6a660ef5a188fb0097193e342618fbcb0210/pycapnp-2.2.2-cp312-cp312-win_amd64.whl", hash = "sha256:700eb8c77405222903af3fb5a371c0d766f86139c3d51f4bff41ccd6403b51f9", size = 1185178, upload-time = "2026-01-21T01:20:53.429Z" },
+    { url = "https://files.pythonhosted.org/packages/ba/70/376c3f1be4ba453584bc96a9e6a7372486ce920cb9c0869c06066e77d626/pycapnp-2.2.2-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:90138fceca1e85ea3eaa0de6656e33c4bef1c8da3c191db0a5b5bacc969f7889", size = 1603886, upload-time = "2026-01-21T01:20:55.383Z" },
+    { url = "https://files.pythonhosted.org/packages/b6/11/2728b563f3f25d826024136cd3aab39f5d1727195de5c90a2ba3d232e897/pycapnp-2.2.2-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:cd0549036bfddb003f8e371c3c4ed3f56ac847953eb57cdd371100bb52afa64e", size = 1485731, upload-time = "2026-01-21T01:20:57.587Z" },
+    { url = "https://files.pythonhosted.org/packages/77/1b/ab9bb376e7314b92dde24843fcf3d6459f10e48d95eb54d25912f973f5d7/pycapnp-2.2.2-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:7c5a8f6d96017a7bb7e202fa8920fcdad119deab0d761f9aca1e6a4755376cef", size = 5209046, upload-time = "2026-01-21T01:20:59.049Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/62/303548df0316740caad513e4b81b18b2db1990785f3f01c36fd19889e7d4/pycapnp-2.2.2-cp313-cp313-manylinux_2_28_i686.whl", hash = "sha256:bba31a8ba8ec32a04c5a14a5e469df7e1f1b85e169f49f7c2edfdfb78ec5075f", size = 5235782, upload-time = "2026-01-21T01:21:00.676Z" },
+    { url = "https://files.pythonhosted.org/packages/5d/ec/df320105d17e118207f01208af1e505d48981856c07fe10b04a8feab39d9/pycapnp-2.2.2-cp313-cp313-manylinux_2_28_ppc64le.whl", hash = "sha256:2decdaeb517120e152f7d9ddede393a830087ac3f1024c5224f2ec58e1735abd", size = 5544647, upload-time = "2026-01-21T01:21:02.812Z" },
+    { url = "https://files.pythonhosted.org/packages/03/47/bc3ea9b0d71cc3e0694993a5907ac56aab2c1ad803697be068fff55a0e1b/pycapnp-2.2.2-cp313-cp313-manylinux_2_28_s390x.whl", hash = "sha256:e879127ac9580005efcd51ec9ae903f1dd98954fb4ea88db9cf534e9a4afb379", size = 5596651, upload-time = "2026-01-21T01:21:04.504Z" },
+    { url = "https://files.pythonhosted.org/packages/84/82/c500220d4eccca845a06f2f9d0747ad8043f8b893ef5278e2019cbc906cc/pycapnp-2.2.2-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:7092c2393191221b8ce1c03ddc1343d1ff26d36129a831017d2371867e9c09bb", size = 5393131, upload-time = "2026-01-21T01:21:06.459Z" },
+    { url = "https://files.pythonhosted.org/packages/ed/61/be35f0b8d81cf107a9d0329d180f1fd8f5d6d75e117fdf7fcad779443f17/pycapnp-2.2.2-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:682812bf9ba4a60309b7150763cd5abed9341a31375398f3d2f60fee26143e13", size = 6071761, upload-time = "2026-01-21T01:21:08.238Z" },
+    { url = "https://files.pythonhosted.org/packages/54/c0/d09da26ebb1bf0a130650a940b8e32f41219241c8d5f9c0891f15343dd9c/pycapnp-2.2.2-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:6f321dbf33767bc2fd77dd66265177284697089f9e45b6dee7fd462b2b5cf918", size = 6369216, upload-time = "2026-01-21T01:21:10.338Z" },
+    { url = "https://files.pythonhosted.org/packages/3d/58/a651de950437d8ef9fa91c3cce84ecd068e05011d421a6cf288f930f331c/pycapnp-2.2.2-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:1d443d38f1cbeaec5b20b420117885fd5de812c5687e3bc8456d2e2c5cba371e", size = 6488273, upload-time = "2026-01-21T01:21:12.127Z" },
+    { url = "https://files.pythonhosted.org/packages/4b/28/254f511272fde9fea1220807d759bcbdb47ecdff64ac860499be9aeddeaa/pycapnp-2.2.2-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:20a31fedb7ef30ec53d5d2d951a7cec8f639c954557093a3ba3d11aba5d174b4", size = 6542078, upload-time = "2026-01-21T01:21:14.581Z" },
+    { url = "https://files.pythonhosted.org/packages/f9/b3/eea39216b7b59cefd21e6783e38a4d78db7e58c4f8f1f45e8351057432b7/pycapnp-2.2.2-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:fbefb388464501899233d0b7d27256edd22c3cebb5a312da5d9f246671e9455e", size = 6333374, upload-time = "2026-01-21T01:21:16.522Z" },
+    { url = "https://files.pythonhosted.org/packages/be/08/42ab52a3e7e5381d8fd7ae227ae75f54454f9e339d214ebe92763bb78115/pycapnp-2.2.2-cp313-cp313-win32.whl", hash = "sha256:a25e3b3ef40d430309acc7c024aa62b1ee6b2e7a85b341a8b7a8a6f8e29d4133", size = 1046173, upload-time = "2026-01-21T01:21:18.074Z" },
+    { url = "https://files.pythonhosted.org/packages/34/d4/fec2023c4709d3631fe24e09dae2badc23d613f1288b4a6b398302945984/pycapnp-2.2.2-cp313-cp313-win_amd64.whl", hash = "sha256:db65c9ec9d69bac6d19092db43b5e1f3cb2c680810418fd1e02316f7ef157fc1", size = 1184414, upload-time = "2026-01-21T01:21:19.334Z" },
+]
+
 [[package]]
 name = "pycparser"
 version = "3.0"
@@ -6289,6 +6324,32 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/73/ae/b48f95715333080afb75a4504487cbe142cae1268afc482d06692d605ae6/yarl-1.22.0-py3-none-any.whl", hash = "sha256:1380560bdba02b6b6c90de54133c81c9f2a453dee9912fe58c1dcced1edb7cff", size = 46814, upload-time = "2025-10-06T14:12:53.872Z" },
 ]
 
+[[package]]
+name = "zap-protocol"
+version = "1.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "zap-schema" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/21/d2/c21faa12aaba9f9e74a07d71a8f87b468deb150f7f0a1601c9fda0aae58d/zap_protocol-1.0.0.tar.gz", hash = "sha256:1f9a30226f15aef5777ff2ab96523e8ece43a6f88bbb230e313b86775d322aea", size = 1513, upload-time = "2026-01-26T19:19:59.934Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/9c/71/bad51bd4ba2fb0ee5d3f6ed504231697d45324ad0abc2633cf75c3894295/zap_protocol-1.0.0-py3-none-any.whl", hash = "sha256:b33bed8c5c42d15f6f42e74ab3a6f7a5f18471b48292a395ea9c3bb2fedb24fe", size = 1816, upload-time = "2026-01-26T19:19:58.599Z" },
+]
+
+[[package]]
+name = "zap-schema"
+version = "1.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio" },
+    { name = "httpx" },
+    { name = "pycapnp" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/7a/f9/54d8d2df784cd7a1b06fdecc5805ea33e06d216df01a352557c2fdf9301b/zap_schema-1.0.0.tar.gz", hash = "sha256:451b1ce05b6ae9c1b016e86fdbc04d62a78a0d720a970a7e61b0579954fb687d", size = 57177, upload-time = "2026-01-26T19:19:36.951Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/bd/4f/db0331ed6c785d37b6965d43eec8fc49e06f9b4033ea8e1a144ad3c5e8de/zap_schema-1.0.0-py3-none-any.whl", hash = "sha256:3102e813fc02d12ef2dc6379bc4c79baf1f752939d146a9bae1c6b125ada62b7", size = 18789, upload-time = "2026-01-26T19:19:35.685Z" },
+]
+
 [[package]]
 name = "zipp"
 version = "3.23.0"