feat(memory): Phase 3 — embedder, zilliz, raptor, worker, __init__

packages/memory/__init__.py

@@ -0,0 +1,37 @@
+"""
+packages/memory/__init__.py
+
+Memory pipeline public API.
+
+Exports:
+    Embedder          — sentence-transformers batch embed, 384-dim
+    ZillizStore       — async upsert/search via AsyncMilvusClient
+    RaptorTree        — RAPTOR tree build + query (Groq summariser, no OpenAI)
+    MemoryWorker      — Redis mem_buffer flush-to-Zilliz background loop
+
+Pre-registered gut-feel bugs (other files that could break this one):
+    W1 [HIGH]   worker.py flush fires while raptor.py build_tree() running on same
+                user_id → partial tree written to Zilliz mid-build → corrupt nodes.
+                Fix: asyncio.Lock per user_id before tree build.
+    W2 [HIGH]   embedder.py loads model at import time → HF Space cold start adds
+                ~8s. If main.py lifespan timeout is tight, Space may return 503
+                before Brain is ready. Fix: lazy-load model inside Embedder.__init__
+                only when first encode() call arrives.
+    W3 [MED]    Zilliz free-tier has 1M vector cap per collection. If mem_buffer
+                flush writes leaf + summary nodes without dedup, cap hit fast.
+                Fix: upsert (not insert) keyed on chunk_id hash.
+    W4 [MED]    RAPTOR summariser calls llm_router → pool key exhausted mid-build
+                → AllKeysExhaustedError raised inside tree build → partial tree
+                never cleaned up. Fix: catch in raptor.py, return partial tree.
+    W5 [LOW]    sentence-transformers model download on first run (~90MB). HF Space
+                build cache doesn't persist between deploys unless /data mount used.
+                Fix: add all-MiniLM-L6-v2 to requirements.txt (triggers cache warm
+                during build, not at runtime).
+"""
+
+from .embedder import Embedder
+from .tier2_zilliz import ZillizStore
+from .raptor import RaptorTree
+from .worker import MemoryWorker
+
+__all__ = ["Embedder", "ZillizStore", "RaptorTree", "MemoryWorker"]

packages/memory/embedder.py

@@ -0,0 +1,102 @@
+"""
+packages/memory/embedder.py
+
+Batch embedding layer using sentence-transformers all-MiniLM-L6-v2.
+384-dim cosine-space vectors. CPU-only (HF Spaces free tier).
+
+Design:
+    - Lazy model load: model downloaded/loaded on first encode() call, not at import.
+    - Async-safe: encode() runs in executor to avoid blocking the event loop.
+    - Batch: encode list[str] → list[list[float]] in one model.encode() call.
+    - Normalise: L2-norm so cosine sim == dot product (Zilliz COSINE metric).
+
+Pre-registered gut-feel bugs (other files that could break this one):
+    E1 [HIGH]   raptor.py calls encode() inside asyncio.gather() across multiple
+                user_ids simultaneously. sentence-transformers model.encode() is
+                NOT thread-safe when batch_size > 1. Fix: asyncio.Lock on encode().
+    E2 [MED]    HF Space free tier has ~2GB RAM. all-MiniLM-L6-v2 uses ~90MB.
+                If browser_agent.py Playwright is also loaded, OOM risk.
+                Fix: track peak RAM at startup, warn via /health endpoint.
+    E3 [LOW]    model.encode() returns numpy arrays. ZillizStore expects list[float].
+                Implicit .tolist() missing → pymilvus TypeError on upsert.
+                Fix: explicit .tolist() in encode() return.
+"""
+
+import asyncio
+import logging
+from functools import lru_cache
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+MODEL_NAME = "all-MiniLM-L6-v2"
+EMBED_DIM = 384
+
+
+@lru_cache(maxsize=1)
+def _load_model():
+    """Load model once, cache in process. Called inside executor thread."""
+    from sentence_transformers import SentenceTransformer  # type: ignore
+    logger.info(f"Loading embedding model: {MODEL_NAME}")
+    model = SentenceTransformer(MODEL_NAME)
+    return model
+
+
+class Embedder:
+    """
+    Async batch embedder.
+
+    Usage:
+        embedder = Embedder()
+        vecs = await embedder.encode(["hello world", "foo bar"])
+        # vecs: List[List[float]], shape (2, 384)
+    """
+
+    def __init__(self):
+        self._lock = asyncio.Lock()  # E1 guard: single encode at a time
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    async def encode(self, texts: List[str]) -> List[List[float]]:
+        """
+        Encode a batch of texts to 384-dim normalised float vectors.
+
+        Args:
+            texts: Non-empty list of strings.
+
+        Returns:
+            List of float lists, one per input text.
+
+        Raises:
+            ValueError: If texts is empty.
+        """
+        if not texts:
+            raise ValueError("encode() called with empty texts list")
+
+        async with self._lock:  # E1: serialise concurrent encode calls
+            loop = asyncio.get_running_loop()
+            # Run CPU-bound model.encode in thread pool — avoids blocking event loop
+            embeddings = await loop.run_in_executor(
+                None, self._sync_encode, texts
+            )
+        return embeddings
+
+    def _sync_encode(self, texts: List[str]) -> List[List[float]]:
+        """Blocking encode — called inside executor thread."""
+        model = _load_model()
+        # normalize_embeddings=True → L2 norm → cosine sim == dot product
+        vecs = model.encode(
+            texts,
+            batch_size=64,
+            normalize_embeddings=True,
+            show_progress_bar=False,
+        )
+        return [v.tolist() for v in vecs]  # E3: explicit .tolist()
+
+    async def encode_one(self, text: str) -> List[float]:
+        """Convenience: encode a single string."""
+        results = await self.encode([text])
+        return results[0]
+
+    @property
+    def dim(self) -> int:
+        return EMBED_DIM

packages/memory/raptor.py

@@ -0,0 +1,278 @@
+"""
+packages/memory/raptor.py
+
+RAPTOR tree — adapted from parthsarthi03/raptor for Ultron V4.
+
+Key divergences from original:
+    - No OpenAI dependency. Summariser = Groq/any provider via llm_router.
+    - No FAISS. Retrieval delegates to ZillizStore (async).
+    - No tiktoken. Token counting via simple split (good enough for Groq 8k ctx).
+    - Clustering: UMAP + GMM from sklearn (same as original).
+    - Tree stored in Zilliz: leaf nodes (layer=0) + summary nodes (layer>0).
+    - build_tree(): embed chunks → cluster → summarise clusters → recurse.
+    - query(): search summary nodes first (collapse_tree mode), fall back raw.
+    - Async throughout. Single asyncio.Lock per user_id during build (W1 guard).
+
+Pre-registered gut-feel bugs (other files that could break this one):
+    R1 [HIGH]   worker.py calls build_tree() after every flush. If flush interval
+                is short (e.g., 10 msgs) and sessions are long, build_tree() runs
+                repeatedly on overlapping chunk sets → duplicate summary nodes.
+                Fix: track last_build_ts per user_id in Redis; skip if < 5min ago.
+    R2 [HIGH]   UMAP requires n_samples > n_neighbors. If chunk batch < 4,
+                global_cluster_embeddings() crashes. Fix: skip RAPTOR if len < 4,
+                write raw leaf nodes only.
+    R3 [MED]    Groq summariser called inside build_tree() → key exhausted →
+                AllKeysExhaustedError → tree build aborts, partial nodes in Zilliz.
+                Fix: catch AllKeysExhaustedError, mark nodes as orphaned, continue.
+    R4 [MED]    RAPTOR max_layers=3 default. For very long sessions (100+ chunks)
+                tree depth may be insufficient → top-level summaries too coarse.
+                Fix: auto-scale layers = max(2, log2(len(chunks)//4)).
+    R5 [LOW]    sklearn GaussianMixture BIC search up to max_clusters=50 is slow
+                on CPU for n_chunks > 200. Fix: cap max_clusters = min(50, n//4).
+"""
+
+import asyncio
+import logging
+import math
+from typing import List, Optional
+
+logger = logging.getLogger(__name__)
+
+MAX_LAYERS = 3
+MIN_CLUSTER_SIZE = 4   # R2: skip RAPTOR if fewer chunks
+MAX_TOKENS_PER_CLUSTER = 3500
+SUMMARY_MAX_TOKENS = 150
+
+# Per-user locks to prevent concurrent tree builds (W1)
+_build_locks: dict[str, asyncio.Lock] = {}
+_build_locks_mutex = asyncio.Lock()
+
+
+async def _get_user_lock(user_id: str) -> asyncio.Lock:
+    async with _build_locks_mutex:
+        if user_id not in _build_locks:
+            _build_locks[user_id] = asyncio.Lock()
+        return _build_locks[user_id]
+
+
+def _count_tokens(text: str) -> int:
+    """Approximate token count via whitespace split. ~4 chars/token is common."""
+    return max(1, len(text) // 4)
+
+
+def _cluster_embeddings(embeddings, threshold: float = 0.1):
+    """
+    RAPTOR clustering: UMAP dim-reduce → GMM soft clustering.
+    Returns list of cluster label arrays per node (node can belong to multiple clusters).
+    Mirrors cluster_utils.perform_clustering from parthsarthi03/raptor.
+    """
+    import numpy as np
+    from sklearn.mixture import GaussianMixture  # type: ignore
+    import umap  # type: ignore
+
+    n = len(embeddings)
+    dim = min(10, n - 2)  # UMAP can't reduce to >= n_samples
+    if dim < 2:
+        # Can't cluster — return all in one cluster
+        return [[0] for _ in embeddings]
+
+    arr = np.array(embeddings)
+    n_neighbors = max(2, int(n ** 0.5))
+
+    # Global reduction
+    reduced = umap.UMAP(
+        n_neighbors=n_neighbors, n_components=dim, metric="cosine"
+    ).fit_transform(arr)
+
+    # GMM — BIC to find optimal clusters
+    max_k = min(50, n // 4)  # R5
+    if max_k < 2:
+        return [[0] for _ in embeddings]
+
+    bics = []
+    for k in range(1, max_k + 1):
+        gm = GaussianMixture(n_components=k, random_state=42)
+        gm.fit(reduced)
+        bics.append(gm.bic(reduced))
+
+    best_k = bics.index(min(bics)) + 1
+    gm = GaussianMixture(n_components=best_k, random_state=42)
+    gm.fit(reduced)
+    probs = gm.predict_proba(reduced)
+    labels = [list(np.where(p > threshold)[0]) or [int(p.argmax())] for p in probs]
+    return labels
+
+
+class RaptorTree:
+    """
+    RAPTOR tree builder and retriever.
+
+    Usage:
+        raptor = RaptorTree(embedder, zilliz_store, llm_fn)
+        await raptor.build_tree(user_id, chunks)
+        context = await raptor.query(user_id, query_text, query_vec)
+    """
+
+    def __init__(self, embedder, zilliz_store, llm_fn):
+        """
+        Args:
+            embedder:     Embedder instance.
+            zilliz_store: ZillizStore instance.
+            llm_fn:       Async callable(messages: list) → str.
+                          Should be make_provider_llm_fn(pool) from llm_router.
+        """
+        self._embedder = embedder
+        self._store = zilliz_store
+        self._llm = llm_fn
+
+    async def build_tree(
+        self,
+        user_id: str,
+        chunks: List[str],
+        max_layers: int = MAX_LAYERS,
+    ) -> None:
+        """
+        Build RAPTOR tree from text chunks and upsert all nodes into Zilliz.
+
+        Steps:
+            1. Embed all chunks.
+            2. Upsert as leaf nodes (layer=0).
+            3. Cluster embeddings → summarise each cluster → upsert summary (layer=1).
+            4. Recurse on summary texts up to max_layers.
+
+        Skips if len(chunks) < MIN_CLUSTER_SIZE (R2).
+        """
+        if not chunks:
+            logger.warning(f"build_tree called with empty chunks for {user_id}")
+            return
+
+        lock = await _get_user_lock(user_id)
+        async with lock:  # W1: serialise per user_id
+            await self._build_layer(user_id, chunks, layer=0, max_layers=max_layers)
+
+    async def _build_layer(
+        self,
+        user_id: str,
+        texts: List[str],
+        layer: int,
+        max_layers: int,
+    ) -> None:
+        """Recursive layer builder."""
+        logger.info(f"RAPTOR build_layer user={user_id} layer={layer} n_chunks={len(texts)}")
+
+        # Embed current layer
+        embeddings = await self._embedder.encode(texts)
+
+        # Upsert current layer to Zilliz
+        node_type = "leaf" if layer == 0 else "summary"
+        await self._store.upsert(user_id, texts, embeddings, node_type=node_type, layer=layer)
+
+        # Base cases: max depth reached or too few chunks to cluster
+        if layer >= max_layers or len(texts) < MIN_CLUSTER_SIZE:  # R2
+            return
+
+        # Cluster
+        try:
+            labels = _cluster_embeddings(embeddings)
+        except Exception as exc:
+            logger.warning(f"RAPTOR clustering failed at layer {layer}: {exc}")
+            return
+
+        # Group texts by cluster
+        clusters: dict[int, List[str]] = {}
+        for i, node_labels in enumerate(labels):
+            for lbl in node_labels:
+                clusters.setdefault(int(lbl), []).append(texts[i])
+
+        # Enforce max token budget per cluster
+        summary_texts: List[str] = []
+        for cluster_texts in clusters.values():
+            # Truncate cluster to MAX_TOKENS_PER_CLUSTER
+            budget = 0
+            selected = []
+            for t in cluster_texts:
+                toks = _count_tokens(t)
+                if budget + toks > MAX_TOKENS_PER_CLUSTER:
+                    break
+                selected.append(t)
+                budget += toks
+            if not selected:
+                selected = cluster_texts[:1]
+
+            summary = await self._summarise_cluster(selected)
+            if summary:
+                summary_texts.append(summary)
+
+        if not summary_texts:
+            return
+
+        # Recurse on summary layer
+        await self._build_layer(user_id, summary_texts, layer=layer + 1, max_layers=max_layers)
+
+    async def _summarise_cluster(self, texts: List[str]) -> Optional[str]:
+        """Call LLM to summarise a cluster of texts. Returns None on failure (R3)."""
+        combined = "\n\n".join(texts)
+        messages = [
+            {
+                "role": "system",
+                "content": "Summarise the following text concisely in 1-3 sentences. Return only the summary.",
+            },
+            {"role": "user", "content": combined[:6000]},  # stay under Groq ctx
+        ]
+        try:
+            from packages.shared.exceptions import AllKeysExhaustedError  # type: ignore
+            summary = await self._llm(messages)
+            return summary.strip() if summary else None
+        except Exception as exc:  # R3: AllKeysExhaustedError + any other
+            logger.warning(f"RAPTOR summarise failed: {exc}")
+            return None
+
+    async def query(
+        self,
+        user_id: str,
+        query_text: str,
+        query_vec: Optional[List[float]] = None,
+        top_k: int = 5,
+    ) -> str:
+        """
+        Retrieve relevant context for a query.
+
+        Strategy (collapse_tree mode from original RAPTOR):
+            1. Search summary nodes first (higher-level context).
+            2. Augment with raw leaf nodes.
+            3. Concatenate unique results, return as context string.
+
+        Args:
+            user_id:    Discord user_id.
+            query_text: Raw query string.
+            query_vec:  Pre-computed embedding (optional; computed if None).
+            top_k:      Number of results per tier.
+
+        Returns:
+            Context string for LLM prompt injection.
+        """
+        if query_vec is None:
+            query_vec = await self._embedder.encode_one(query_text)
+
+        # Search summaries first (tree top-down)
+        summary_hits = await self._store.search(
+            user_id, query_vec, top_k=top_k, node_type_filter="summary"
+        )
+        # Then raw leaves
+        leaf_hits = await self._store.search(
+            user_id, query_vec, top_k=top_k, node_type_filter="leaf"
+        )
+
+        # Deduplicate and combine
+        seen: set[str] = set()
+        parts: List[str] = []
+        for hit in summary_hits + leaf_hits:
+            text = hit["text"]
+            if text not in seen:
+                seen.add(text)
+                parts.append(text)
+
+        if not parts:
+            return ""  # caller falls back to pure LLM
+
+        return "\n---\n".join(parts[:top_k * 2])

packages/memory/tier2_zilliz.py

@@ -0,0 +1,246 @@
+"""
+packages/memory/tier2_zilliz.py
+
+Zilliz (Milvus-as-a-Service) async vector store.
+
+Architecture:
+    - AsyncMilvusClient (pymilvus) — async context manager pattern.
+    - Collection per shard: ultron_mem_{shard_id}, shard = hash(user_id) % 15.
+    - Schema: id (VARCHAR PK, 64 chars), user_id (VARCHAR), chunk_id (VARCHAR),
+      text (VARCHAR, 2048), node_type (VARCHAR, 32), layer (INT64),
+      vector (FLOAT_VECTOR, 384), ts (INT64 unix ms).
+    - Metric: COSINE. Index: AUTOINDEX (Zilliz manages HNSW internally).
+    - Upsert (not insert) keyed on chunk_id — dedup guard (W3).
+    - search() returns top_k results as list[dict].
+    - ensure_collection() is idempotent — safe to call on every startup.
+
+Pre-registered gut-feel bugs (other files that could break this one):
+    Z1 [HIGH]   Zilliz free cluster goes cold after inactivity (~10min). First
+                upsert/search after cold start → gRPC timeout (default 10s).
+                Fix: set timeout=30 on all ops; /health pings each shard daily.
+    Z2 [HIGH]   pymilvus AsyncMilvusClient is EXPERIMENTAL (their own warning).
+                If Zilliz server ≥ 2.5 returns new protobuf field, client may
+                raise decode error. Fix: pin pymilvus==2.4.x in requirements.txt.
+    Z3 [MED]    ensure_collection() called concurrently for same shard on startup
+                → duplicate create_collection race. Fix: asyncio.Lock per shard.
+    Z4 [MED]    FLOAT_VECTOR dim mismatch (embedder returns 384, collection created
+                with wrong dim) → MilvusException on upsert. Fix: assert dim==384
+                in ensure_collection() describe result.
+    Z5 [LOW]    VARCHAR PK max_length=64 but chunk_id is SHA256 hex (64 chars)
+                → exact boundary. If pymilvus adds null terminator, will overflow.
+                Fix: truncate chunk_id to 60 chars.
+"""
+
+import asyncio
+import hashlib
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+NUM_SHARDS = 15
+EMBED_DIM = 384
+COLLECTION_PREFIX = "ultron_mem"
+MAX_TEXT_LEN = 2048
+CHUNK_ID_MAX = 60  # Z5: stay under 64 VARCHAR PK limit
+
+
+def _shard(user_id: str) -> int:
+    """Deterministic shard index for a user_id. Matches locked Zilliz sharding rule."""
+    return int(hashlib.md5(user_id.encode()).hexdigest(), 16) % NUM_SHARDS
+
+
+def _collection_name(user_id: str) -> str:
+    return f"{COLLECTION_PREFIX}_{_shard(user_id)}"
+
+
+def _chunk_id(text: str, user_id: str) -> str:
+    """Stable dedup key: SHA256(user_id + text), truncated to 60 chars."""
+    raw = hashlib.sha256(f"{user_id}:{text}".encode()).hexdigest()
+    return raw[:CHUNK_ID_MAX]
+
+
+class ZillizStore:
+    """
+    Async Zilliz vector store.
+
+    Usage:
+        store = ZillizStore(uri=ZILLIZ_URI, token=ZILLIZ_TOKEN)
+        await store.ensure_collection(user_id)
+        await store.upsert(user_id, chunks, embeddings, node_type="leaf", layer=0)
+        results = await store.search(user_id, query_vec, top_k=5)
+        await store.close()
+    """
+
+    def __init__(self, uri: str, token: str):
+        self._uri = uri
+        self._token = token
+        self._clients: Dict[int, Any] = {}   # shard_id → AsyncMilvusClient
+        self._ensure_locks: Dict[int, asyncio.Lock] = {}  # Z3: per-shard lock
+        self._global_lock = asyncio.Lock()
+
+    async def _get_client(self, shard: int) -> Any:
+        """Lazy init AsyncMilvusClient per shard."""
+        if shard not in self._clients:
+            async with self._global_lock:
+                if shard not in self._clients:  # double-check after lock
+                    from pymilvus import AsyncMilvusClient  # type: ignore
+                    client = AsyncMilvusClient(uri=self._uri, token=self._token)
+                    await client._connect()
+                    self._clients[shard] = client
+                    self._ensure_locks[shard] = asyncio.Lock()
+        return self._clients[shard]
+
+    async def ensure_collection(self, user_id: str) -> None:
+        """
+        Idempotent: create collection + index if not exists.
+        Safe to call every startup. Z3: serialised per shard.
+        """
+        shard = _shard(user_id)
+        client = await self._get_client(shard)
+        col = _collection_name(user_id)
+
+        # Ensure lock exists (may be first call)
+        if shard not in self._ensure_locks:
+            self._ensure_locks[shard] = asyncio.Lock()
+
+        async with self._ensure_locks[shard]:  # Z3
+            exists = await client.has_collection(col, timeout=30)
+            if exists:
+                logger.debug(f"Collection {col} already exists")
+                return
+
+            logger.info(f"Creating collection {col} (shard {shard})")
+            from pymilvus import DataType  # type: ignore
+            from pymilvus.milvus_client import IndexParams  # type: ignore
+
+            schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
+            schema.add_field("id", DataType.VARCHAR, is_primary=True, max_length=64)
+            schema.add_field("user_id", DataType.VARCHAR, max_length=128)
+            schema.add_field("chunk_id", DataType.VARCHAR, max_length=64)
+            schema.add_field("text", DataType.VARCHAR, max_length=MAX_TEXT_LEN)
+            schema.add_field("node_type", DataType.VARCHAR, max_length=32)  # leaf|summary
+            schema.add_field("layer", DataType.INT64)
+            schema.add_field("vector", DataType.FLOAT_VECTOR, dim=EMBED_DIM)  # Z4
+            schema.add_field("ts", DataType.INT64)
+
+            idx = IndexParams()
+            idx.add_index("vector", index_type="AUTOINDEX", metric_type="COSINE")
+
+            await client.create_collection(
+                col, schema=schema, index_params=idx, timeout=30
+            )
+            logger.info(f"Collection {col} created")
+
+    async def upsert(
+        self,
+        user_id: str,
+        texts: List[str],
+        embeddings: List[List[float]],
+        node_type: str = "leaf",
+        layer: int = 0,
+    ) -> int:
+        """
+        Upsert text chunks + embeddings into user's shard.
+
+        Args:
+            user_id:    Discord user_id string.
+            texts:      Raw text chunks (parallel with embeddings).
+            embeddings: 384-dim float vectors.
+            node_type:  'leaf' or 'summary'.
+            layer:      RAPTOR tree layer (0 = leaf).
+
+        Returns:
+            Number of rows upserted.
+        """
+        if len(texts) != len(embeddings):
+            raise ValueError(f"texts/embeddings length mismatch: {len(texts)} vs {len(embeddings)}")
+        if not texts:
+            return 0
+
+        await self.ensure_collection(user_id)
+        shard = _shard(user_id)
+        client = await self._get_client(shard)
+        col = _collection_name(user_id)
+        ts = int(time.time() * 1000)
+
+        rows = []
+        for text, vec in zip(texts, embeddings):
+            cid = _chunk_id(text, user_id)
+            rows.append({
+                "id": cid,           # PK = chunk_id for upsert dedup
+                "user_id": user_id,
+                "chunk_id": cid,
+                "text": text[:MAX_TEXT_LEN],
+                "node_type": node_type,
+                "layer": layer,
+                "vector": vec,
+                "ts": ts,
+            })
+
+        res = await client.upsert(col, rows, timeout=30)
+        count = res.get("upsert_count", len(rows))
+        logger.info(f"Zilliz upsert {count} rows → {col} (layer={layer}, type={node_type})")
+        return count
+
+    async def search(
+        self,
+        user_id: str,
+        query_vec: List[float],
+        top_k: int = 5,
+        node_type_filter: Optional[str] = None,
+    ) -> List[Dict]:
+        """
+        Cosine ANN search in user's shard.
+
+        Args:
+            user_id:           Discord user_id.
+            query_vec:         384-dim query embedding.
+            top_k:             Number of results.
+            node_type_filter:  Optional 'leaf' or 'summary' filter.
+
+        Returns:
+            List of dicts with keys: text, node_type, layer, score, ts.
+        """
+        shard = _shard(user_id)
+        col = _collection_name(user_id)
+        client = await self._get_client(shard)
+
+        filter_expr = ""
+        if node_type_filter:
+            filter_expr = f'node_type == "{node_type_filter}"'
+
+        try:
+            hits = await client.search(
+                collection_name=col,
+                data=[query_vec],
+                anns_field="vector",
+                limit=top_k,
+                output_fields=["text", "node_type", "layer", "ts"],
+                filter=filter_expr or "",
+                timeout=30,
+            )
+        except Exception as exc:
+            logger.warning(f"Zilliz search failed for {user_id}: {exc}")
+            return []
+
+        results = []
+        for hit in hits[0]:  # hits[0] = results for first query vector
+            results.append({
+                "text": hit["entity"].get("text", ""),
+                "node_type": hit["entity"].get("node_type", "leaf"),
+                "layer": hit["entity"].get("layer", 0),
+                "score": hit.get("distance", 0.0),
+                "ts": hit["entity"].get("ts", 0),
+            })
+        return results
+
+    async def close(self) -> None:
+        """Close all open client connections."""
+        for client in self._clients.values():
+            try:
+                await client.close()
+            except Exception:
+                pass
+        self._clients.clear()

packages/memory/worker.py

@@ -0,0 +1,237 @@
+"""
+packages/memory/worker.py
+
+Memory flush worker — drains Redis mem_buffer → embeds → RAPTOR tree build → Zilliz.
+
+Redis key: ultron:mem_buffer:{user_id}  (LIST, right-push on write, left-pop on flush)
+Flush trigger: FLUSH_THRESHOLD items OR FLUSH_INTERVAL_SECS elapsed since last flush.
+Run as background asyncio task started in main.py lifespan.
+
+Flow per user_id:
+    1. LRANGE ultron:mem_buffer:{uid} 0 FLUSH_THRESHOLD-1
+    2. Decode JSON: [{"role": ..., "content": ..., "ts": ...}, ...]
+    3. Chunk content into ~512-token windows (50-token overlap)
+    4. Embedder.encode(chunks)
+    5. RaptorTree.build_tree(uid, chunks)  -- upserts leaf + summary nodes
+    6. LTRIM ultron:mem_buffer:{uid} FLUSH_THRESHOLD -1  (remove flushed items)
+    7. Set ultron:mem_last_flush:{uid} = now (Redis SET EX 86400)
+
+Error handling: Redis unavailable → log + skip (never crash worker loop).
+                Zilliz unavailable → log + skip, retry next interval.
+
+Pre-registered gut-feel bugs (other files that could break this one):
+    MW1 [HIGH]  main.py lifespan cancels the worker task on shutdown. If flush is
+                mid-write (after embed, before upsert), Zilliz gets orphan vectors
+                without tree structure. Fix: asyncio.shield() the upsert call.
+    MW2 [HIGH]  Redis LRANGE returns bytes, not str. JSON decode fails on raw bytes.
+                Fix: decode bytes → str before json.loads().
+    MW3 [MED]   Two worker iterations overlap if flush_interval_secs is shorter than
+                build_tree() runtime. Fix: per-user_id asyncio.Lock in worker loop.
+    MW4 [MED]   chunk_text() splits on whitespace — if a message is one long URL
+                (2000 chars), it becomes a single chunk > 512 tokens. Groq context
+                may handle it but RAPTOR BIC clustering will treat it as noise.
+                Fix: hard-cap chunk size at 512 tokens with forced split.
+    MW5 [LOW]   mem_buffer list unbounded if flush never fires (e.g. Zilliz always
+                fails). Fix: LTRIM to 200 items max as safety in ensure_buffer_size().
+"""
+
+import asyncio
+import json
+import logging
+import time
+from typing import List, Optional
+
+logger = logging.getLogger(__name__)
+
+FLUSH_THRESHOLD = 10          # flush after this many messages buffered
+FLUSH_INTERVAL_SECS = 300     # flush every 5 min regardless of count
+CHUNK_SIZE_CHARS = 2000       # ~500 tokens at 4 chars/token
+CHUNK_OVERLAP_CHARS = 200     # ~50 tokens overlap
+MEM_BUFFER_PREFIX = "ultron:mem_buffer"
+MEM_LAST_FLUSH_PREFIX = "ultron:mem_last_flush"
+MAX_BUFFER_SIZE = 200         # MW5: safety cap
+FLUSH_SLEEP_SECS = 60         # worker poll interval
+
+
+def _chunk_text(text: str) -> List[str]:
+    """
+    Split text into overlapping chunks of ~CHUNK_SIZE_CHARS characters.
+    MW4: hard-cap each chunk to CHUNK_SIZE_CHARS.
+    """
+    if len(text) <= CHUNK_SIZE_CHARS:
+        return [text] if text.strip() else []
+
+    chunks = []
+    start = 0
+    while start < len(text):
+        end = start + CHUNK_SIZE_CHARS
+        chunk = text[start:end]
+        if chunk.strip():
+            chunks.append(chunk)
+        start = end - CHUNK_OVERLAP_CHARS  # overlap
+        if start >= len(text) - CHUNK_OVERLAP_CHARS:
+            break
+    return chunks
+
+
+def _messages_to_chunks(messages: List[dict]) -> List[str]:
+    """Convert role-content message dicts to text chunks."""
+    chunks: List[str] = []
+    for msg in messages:
+        role = msg.get("role", "user")
+        content = msg.get("content", "")
+        if not content or not content.strip():
+            continue
+        text = f"{role}: {content}"
+        chunks.extend(_chunk_text(text))
+    return chunks
+
+
+class MemoryWorker:
+    """
+    Background memory flush worker.
+
+    Usage (in main.py lifespan):
+        worker = MemoryWorker(redis_client, embedder, raptor_tree)
+        task = asyncio.create_task(worker.run())
+        # on shutdown:
+        task.cancel()
+    """
+
+    def __init__(self, redis, embedder, raptor_tree):
+        """
+        Args:
+            redis:       aioredis / redis.asyncio client (already connected).
+            embedder:    Embedder instance.
+            raptor_tree: RaptorTree instance.
+        """
+        self._redis = redis
+        self._embedder = embedder
+        self._raptor = raptor_tree
+        self._user_locks: dict[str, asyncio.Lock] = {}  # MW3
+        self._lock_mutex = asyncio.Lock()
+
+    async def _get_lock(self, user_id: str) -> asyncio.Lock:
+        async with self._lock_mutex:
+            if user_id not in self._user_locks:
+                self._user_locks[user_id] = asyncio.Lock()
+            return self._user_locks[user_id]
+
+    async def run(self) -> None:
+        """Main worker loop. Runs indefinitely until cancelled."""
+        logger.info("MemoryWorker started")
+        while True:
+            try:
+                await self._tick()
+            except asyncio.CancelledError:
+                logger.info("MemoryWorker cancelled")
+                raise
+            except Exception as exc:
+                logger.error(f"MemoryWorker tick error: {exc}")
+            await asyncio.sleep(FLUSH_SLEEP_SECS)
+
+    async def _tick(self) -> None:
+        """Single flush cycle: scan all active buffers."""
+        try:
+            # Find all mem_buffer keys
+            keys = await self._redis.keys(f"{MEM_BUFFER_PREFIX}:*")
+        except Exception as exc:
+            logger.warning(f"Redis keys scan failed: {exc}")
+            return
+
+        if not keys:
+            return
+
+        # MW2: keys may be bytes
+        user_ids = []
+        for k in keys:
+            key_str = k.decode() if isinstance(k, bytes) else k
+            uid = key_str.replace(f"{MEM_BUFFER_PREFIX}:", "")
+            user_ids.append(uid)
+
+        # Flush each user concurrently (bounded to 5 at a time)
+        sem = asyncio.Semaphore(5)
+        tasks = [self._flush_user(uid, sem) for uid in user_ids]
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+    async def _flush_user(self, user_id: str, sem: asyncio.Semaphore) -> None:
+        """Flush one user's mem_buffer if threshold or interval reached."""
+        async with sem:
+            lock = await self._get_lock(user_id)  # MW3
+            async with lock:
+                await self._do_flush(user_id)
+
+    async def _do_flush(self, user_id: str) -> None:
+        """Core flush logic for a single user_id."""
+        buf_key = f"{MEM_BUFFER_PREFIX}:{user_id}"
+        last_key = f"{MEM_LAST_FLUSH_PREFIX}:{user_id}"
+
+        try:
+            buf_len = await self._redis.llen(buf_key)
+        except Exception as exc:
+            logger.warning(f"Redis llen failed for {user_id}: {exc}")
+            return
+
+        if buf_len == 0:
+            return
+
+        # Check flush triggers
+        should_flush = buf_len >= FLUSH_THRESHOLD
+        if not should_flush:
+            try:
+                last_flush = await self._redis.get(last_key)
+                if last_flush:
+                    elapsed = time.time() - float(last_flush)
+                    should_flush = elapsed >= FLUSH_INTERVAL_SECS
+            except Exception:
+                should_flush = False
+
+        if not should_flush:
+            return
+
+        # LRANGE: get up to FLUSH_THRESHOLD items
+        try:
+            raw_items = await self._redis.lrange(buf_key, 0, FLUSH_THRESHOLD - 1)
+        except Exception as exc:
+            logger.warning(f"Redis lrange failed for {user_id}: {exc}")
+            return
+
+        messages = []
+        for item in raw_items:
+            # MW2: decode bytes
+            item_str = item.decode() if isinstance(item, bytes) else item
+            try:
+                messages.append(json.loads(item_str))
+            except json.JSONDecodeError:
+                messages.append({"role": "user", "content": item_str})
+
+        chunks = _messages_to_chunks(messages)
+        if not chunks:
+            logger.debug(f"No chunks extracted for {user_id}")
+            return
+
+        logger.info(f"MemoryWorker flushing {len(chunks)} chunks for {user_id}")
+
+        try:
+            # asyncio.shield: protect upsert from cancellation (MW1)
+            await asyncio.shield(
+                self._raptor.build_tree(user_id, chunks)
+            )
+        except Exception as exc:
+            logger.error(f"RAPTOR build_tree failed for {user_id}: {exc}")
+            return  # don't trim buffer if build failed
+
+        # Trim flushed items from buffer
+        try:
+            await self._redis.ltrim(buf_key, len(raw_items), -1)
+            await self._redis.set(last_key, str(time.time()), ex=86400)
+        except Exception as exc:
+            logger.warning(f"Redis post-flush trim failed for {user_id}: {exc}")
+
+        # MW5: safety cap
+        try:
+            await self._redis.ltrim(buf_key, -MAX_BUFFER_SIZE, -1)
+        except Exception:
+            pass
+
+        logger.info(f"MemoryWorker flush complete for {user_id}")