fix(chat): stop code-block button flicker during streaming (#3023)

Render streamed markdown incrementally (freeze finalized blocks,
re-render only the growing tail) instead of re-rendering the whole
message every token, which recreated every <pre> and dropped CSS :hover.

routes/embedding_routes.py

@@ -316,6 +316,16 @@ def setup_embedding_routes():
             reset_http_embed_state()
         except Exception:
             pass
+        try:
+            from src.embedding_lanes import reset_embedding_lane_state
+            reset_embedding_lane_state()
+        except Exception:
+            pass
+        try:
+            from src.tool_index import reset_tool_index
+            reset_tool_index()
+        except Exception:
+            pass
 
         # Reset ChromaDB client (collections will be recreated with new embeddings)
         try:
@@ -347,6 +357,16 @@ def setup_embedding_routes():
             reset_http_embed_state()
         except Exception:
             pass
+        try:
+            from src.embedding_lanes import reset_embedding_lane_state
+            reset_embedding_lane_state()
+        except Exception:
+            pass
+        try:
+            from src.tool_index import reset_tool_index
+            reset_tool_index()
+        except Exception:
+            pass
 
         # Reset ChromaDB client
         try:

src/agent_loop.py

@@ -67,6 +67,7 @@ The block executes automatically and you see the output."""
 _AGENT_RULES = """\
 ## Rules
 - Only use tools when needed. Don't search for things you already know.
+- For web lookup/search/latest/current requests, use `web_search` or `web_fetch`. Do NOT use `bash`, `python`, `curl`, `requests`, or scraping code for web lookup unless web tools are disabled or already failed.
 - These exact tags execute automatically. For showing code examples, use ```shell, ```sh, ```py, etc. instead.
 - Multiple tool blocks per response OK. 60s timeout per tool, 10K char output limit.
 - Code/content >15 lines → ```create_document (NOT in chat). Short snippets OK in chat.
@@ -113,6 +114,7 @@ _API_AGENT_RULES = """\
 - Prefer native tool/function calling when tools are needed.
 - Only call tools when they materially help answer the request.
 - You MUST use tools to take action — do not describe what you would do. Act, don't narrate.
+- For web lookup/search/latest/current requests, call `web_search` or `web_fetch`. Do NOT use shell, Python, curl, requests, or scraping code for web lookup unless web tools are unavailable or already failed.
 - Keep answers concise unless the user asks for depth.
 - For long code or content, use document tools instead of pasting large blocks into chat.
 - Editing an existing document: ALWAYS use `edit_document` with find/replace. Only use `update_document` for genuine full rewrites (>50% changed) — do NOT echo the entire file back for small edits.
@@ -176,7 +178,8 @@ TOOL_SECTIONS = {
 ```bash
 <shell command>
 ```
-Run any shell command. Output is returned to you. Use for: installing packages, checking files, git, curl, system info, etc.
+Run any shell command. Output is returned to you. Use for: installing packages, checking files, git, system info, process management, etc.
+Do NOT use bash/curl for web lookup/search/latest/current requests when `web_search` or `web_fetch` is available.
 NEVER use bash to create or change files — no `>`/`>>` redirects, no heredocs (`cat > f << 'EOF'`), no `tee`, `sed -i`, `awk -i`, no `python -c` that writes. To CREATE or fully rewrite a file use `write_file`; to change part of an existing file use `edit_file`. Those show a diff and are the ONLY allowed way to write files. (bash is for read-only inspection: `ls`, `cat` to READ, `grep`, `git status`/`git diff`, builds, installs.)
 For LONG-running commands (package installs, pip/npm, ffmpeg, model downloads, training, builds — anything that may take more than ~20s), make the FIRST line `#!bg` to run it in the BACKGROUND. You get a job id back immediately and are automatically re-invoked with the full output when it finishes — so you never block the chat waiting. Example:
 ```bash
@@ -190,7 +193,8 @@ NEVER pipe multi-line Python through `python -c "..."` — shell quoting eats re
 ```python
 <python code>
 ```
-Execute Python code. Use for computation, data processing, scripting. NOT for writing code for the user (use create_document for that). Same sandbox limits as bash — no TTY, no GUI, no `input()`; for anything the user should interact with, generate a single HTML file with inline JS instead.""",
+Execute Python code. Use for computation, data processing, scripting. NOT for writing code for the user (use create_document for that). Same sandbox limits as bash — no TTY, no GUI, no `input()`; for anything the user should interact with, generate a single HTML file with inline JS instead.
+Do NOT use Python/requests for web lookup/search/latest/current requests when `web_search` or `web_fetch` is available.""",
 
     "web_search": """\
 ```web_search
@@ -200,7 +204,8 @@ Or with JSON for fresh news:
 ```web_search
 {"query": "<your query>", "time_filter": "day"}
 ```
-Search the web for a SINGLE quick fact/lookup mid-task. For news / "today" / "latest" queries, pass `time_filter` ("day", "week", "month", or "year"). NOT for "research X" / "do research on X" / "look into X" requests — those mean a multi-source DEEP RESEARCH job: use `trigger_research` instead (it runs in the Deep Research sidebar and produces a full report). web_search = one quick query; trigger_research = a researched report.""",
+Search the web for a SINGLE quick fact/lookup mid-task. For news / "today" / "latest" queries, pass `time_filter` ("day", "week", "month", or "year"). NOT for "research X" / "do research on X" / "look into X" requests — those mean a multi-source DEEP RESEARCH job: use `trigger_research` instead (it runs in the Deep Research sidebar and produces a full report). web_search = one quick query; trigger_research = a researched report.
+Use this instead of `bash`, `curl`, `python`, `requests`, or scraping code for web lookup/search/latest/current requests.""",
 
     "web_fetch": """\
 ```web_fetch
@@ -323,6 +328,7 @@ Bulk delete/archive/mark emails. Use this for "delete all those" after listing e
 {"action": "create_event", "summary": "<event title>", "dtstart": "<natural language or ISO datetime>"}
 ```
 Calendar event management (CalDAV). Actions: `list_events`, `create_event`, `update_event`, `delete_event`, `list_calendars`. \
+For `list_events`: {start?, end?, calendar?}; prefer `start`/`end` for the range, though start_date/end_date and from/to aliases are accepted. \
 For `create_event`: {summary, dtstart, dtend?, duration?, calendar?, location?, description?, reminder_minutes?, rrule?}. \
 `dtstart` accepts natural language ("tomorrow at 1pm", "in 2 hours", "next monday 9am") or ISO ("2026-05-12T13:00:00"). \
 If `dtend` omitted, defaults to dtstart+1h (or +1d when `all_day: true`). \

src/deep_research.py

@@ -16,7 +16,8 @@ from typing import Callable, Dict, List, Optional, Set
 
 from src.research_utils import strip_thinking, is_low_quality
 
-from src.goal_based_extractor import EXTRACTOR_PROMPT
+from src.goal_based_extractor import EXTRACTOR_SYSTEM
+from src.prompt_security import untrusted_context_message
 
 logger = logging.getLogger(__name__)
 
@@ -625,11 +626,12 @@ class DeepResearcher:
             else:
                 content = truncated
 
-        prompt = EXTRACTOR_PROMPT.format(webpage_content=content, goal=question)
-
         try:
             response = await self._llm(
-                [{"role": "user", "content": prompt}],
+                [
+                    {"role": "user", "content": EXTRACTOR_SYSTEM.format(goal=question)},
+                    untrusted_context_message("webpage", content),
+                ],
                 temperature=0.2,
                 max_tokens=2048,
                 timeout=self.extraction_timeout,

src/embedding_lanes.py

@@ -0,0 +1,380 @@
+"""
+embedding_lanes.py
+
+Helpers for keeping FastEmbed fallback vectors separate from user-configured
+embedding vectors. ChromaDB fixes a collection's dimension on first insert, so
+different embedding models must never share one collection.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import logging
+import os
+from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence
+
+logger = logging.getLogger(__name__)
+
+LANE_FASTEMBED = "fastembed"
+LANE_CUSTOM = "custom"
+
+
+@dataclass
+class EmbeddingLane:
+    name: str
+    client: Any
+    collection: Any
+    collection_name: str
+    model: str
+    url: str
+    dimension: int
+    fingerprint: str
+
+    @property
+    def healthy(self) -> bool:
+        return self.collection is not None and self.client is not None
+
+    def encode(self, texts: Sequence[str]) -> List[List[float]]:
+        vecs = self.client.encode(list(texts), normalize_embeddings=True)
+        return vecs.tolist() if hasattr(vecs, "tolist") else [list(v) for v in vecs]
+
+    def count(self) -> int:
+        try:
+            return int(self.collection.count())
+        except Exception:
+            return 0
+
+    def stats(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "collection": self.collection_name,
+            "model": self.model,
+            "url": self.url,
+            "dimension": self.dimension,
+            "fingerprint": self.fingerprint,
+            "count": self.count(),
+            "healthy": self.healthy,
+        }
+
+
+def reset_embedding_lane_state() -> None:
+    """Reset process-local embedding lane state after endpoint config changes."""
+    try:
+        from src.embeddings import reset_http_embed_state
+        reset_http_embed_state()
+    except Exception:
+        pass
+
+
+def collection_name(base_name: str, lane_name: str) -> str:
+    return f"{base_name}_{lane_name}"
+
+
+def _fingerprint(lane_name: str, url: str, model: str, dimension: int) -> str:
+    raw = f"{lane_name}\n{url}\n{model}\n{dimension}"
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
+
+
+def _metadata(lane_name: str, url: str, model: str, dimension: int, fingerprint: str) -> Dict[str, Any]:
+    return {
+        "hnsw:space": "cosine",
+        "embedding_lane": lane_name,
+        "embedding_url": url,
+        "embedding_model": model,
+        "embedding_dimension": dimension,
+        "embedding_fingerprint": fingerprint,
+    }
+
+
+def _load_custom_endpoint() -> Dict[str, str]:
+    try:
+        from src.embeddings import _load_persisted_endpoint
+        persisted = _load_persisted_endpoint()
+    except Exception:
+        persisted = {}
+
+    url = persisted.get("url") or os.environ.get("EMBEDDING_URL", "")
+    if not url:
+        return {}
+
+    model = persisted.get("model") or os.environ.get("EMBEDDING_MODEL", "")
+    api_key = persisted.get("api_key") or os.environ.get("EMBEDDING_API_KEY", "")
+    if persisted.get("api_key"):
+        try:
+            from src.secret_storage import decrypt
+            api_key = decrypt(api_key)
+        except Exception:
+            logger.warning("Could not decrypt saved embedding endpoint API key")
+            api_key = ""
+
+    return {"url": url, "model": model, "api_key": api_key}
+
+
+def _build_fastembed_client():
+    from src.embeddings import FastEmbedClient
+
+    client = FastEmbedClient()
+    client.get_sentence_embedding_dimension()
+    return client
+
+
+def _build_custom_client():
+    from src.embeddings import EmbeddingClient, get_embedding_client
+
+    client = get_embedding_client()
+    if isinstance(client, EmbeddingClient):
+        return client
+    raise RuntimeError("HTTP embedding lane unavailable")
+
+
+def _encode_with_client(client: Any, texts: Sequence[str]) -> List[List[float]]:
+    vecs = client.encode(list(texts), normalize_embeddings=True)
+    return vecs.tolist() if hasattr(vecs, "tolist") else [list(v) for v in vecs]
+
+
+def _get_or_reset_collection(chroma_client, name: str, metadata: Dict[str, Any], client: Any):
+    try:
+        collection = chroma_client.get_collection(name)
+    except Exception:
+        return chroma_client.get_or_create_collection(name=name, metadata=metadata)
+
+    current = collection.metadata or {}
+    if not (
+        current.get("embedding_fingerprint") not in (None, metadata["embedding_fingerprint"])
+        or current.get("embedding_dimension") not in (None, metadata["embedding_dimension"])
+        or current.get("embedding_lane") not in (None, metadata["embedding_lane"])
+    ):
+        return collection
+
+    logger.info(
+        "Recreating Chroma collection %s for embedding lane change (%s -> %s)",
+        name,
+        current.get("embedding_fingerprint"),
+        metadata["embedding_fingerprint"],
+    )
+    preserved = {"ids": [], "documents": [], "metadatas": [], "embeddings": []}
+    try:
+        preserved = collection.get(include=["documents", "metadatas", "embeddings"]) or preserved
+    except Exception as e:
+        raise RuntimeError(f"Could not preserve documents before resetting {name}: {e}") from e
+
+    ids = preserved.get("ids") or []
+    docs = preserved.get("documents") or []
+    metas = preserved.get("metadatas") or []
+    prepared_batches = []
+    if ids and docs:
+        try:
+            for start in range(0, len(ids), 100):
+                batch_ids = ids[start:start + 100]
+                batch_docs = docs[start:start + 100]
+                batch_metas = metas[start:start + 100]
+                if len(batch_metas) < len(batch_ids):
+                    batch_metas += [{}] * (len(batch_ids) - len(batch_metas))
+                prepared_batches.append((
+                    batch_ids,
+                    batch_docs,
+                    batch_metas,
+                    _encode_with_client(client, batch_docs),
+                ))
+        except Exception as e:
+            raise RuntimeError(f"Could not re-embed preserved rows for {name}: {e}") from e
+
+    chroma_client.delete_collection(name)
+    collection = chroma_client.get_or_create_collection(name=name, metadata=metadata)
+
+    try:
+        for batch_ids, batch_docs, batch_metas, embeddings in prepared_batches:
+            collection.add(
+                ids=batch_ids,
+                documents=batch_docs,
+                metadatas=batch_metas,
+                embeddings=embeddings,
+            )
+    except Exception as e:
+        logger.warning("Could not write reset collection %s; restoring previous rows: %s", name, e)
+        try:
+            chroma_client.delete_collection(name)
+            restored = chroma_client.get_or_create_collection(name=name, metadata=current)
+            old_embeddings = preserved.get("embeddings") or []
+            if ids and docs and old_embeddings:
+                for start in range(0, len(ids), 100):
+                    batch_ids = ids[start:start + 100]
+                    batch_docs = docs[start:start + 100]
+                    batch_metas = metas[start:start + 100]
+                    batch_embeddings = old_embeddings[start:start + 100]
+                    if len(batch_metas) < len(batch_ids):
+                        batch_metas += [{}] * (len(batch_ids) - len(batch_metas))
+                    restored.add(
+                        ids=batch_ids,
+                        documents=batch_docs,
+                        metadatas=batch_metas,
+                        embeddings=batch_embeddings,
+                    )
+        except Exception as restore_error:
+            logger.warning("Could not restore previous collection %s: %s", name, restore_error)
+        raise RuntimeError(f"Could not write reset collection {name}: {e}") from e
+    if prepared_batches:
+        logger.info("Re-embedded %s rows after resetting %s", len(ids), name)
+
+    return collection
+
+
+def _create_lane(chroma_client, base_name: str, lane_name: str, client: Any) -> EmbeddingLane:
+    dimension = int(client.get_sentence_embedding_dimension())
+    model = getattr(client, "model", "")
+    url = getattr(client, "url", "")
+    fp = _fingerprint(lane_name, url, model, dimension)
+    name = collection_name(base_name, lane_name)
+    metadata = _metadata(lane_name, url, model, dimension, fp)
+    collection = _get_or_reset_collection(chroma_client, name, metadata, client)
+    return EmbeddingLane(
+        name=lane_name,
+        client=client,
+        collection=collection,
+        collection_name=name,
+        model=model,
+        url=url,
+        dimension=dimension,
+        fingerprint=fp,
+    )
+
+
+def build_embedding_lanes(base_name: str) -> List[EmbeddingLane]:
+    """Return healthy lanes in retrieval preference order: custom, fastembed."""
+    from src.chroma_client import get_chroma_client
+
+    chroma_client = get_chroma_client()
+    lanes: List[EmbeddingLane] = []
+
+    try:
+        custom = _build_custom_client()
+        if custom is not None:
+            lanes.append(_create_lane(chroma_client, base_name, LANE_CUSTOM, custom))
+    except Exception as e:
+        logger.warning("Custom embedding lane unavailable for %s: %s", base_name, e)
+
+    try:
+        fastembed = _build_fastembed_client()
+        lanes.append(_create_lane(chroma_client, base_name, LANE_FASTEMBED, fastembed))
+    except Exception as e:
+        logger.warning("FastEmbed lane unavailable for %s: %s", base_name, e)
+
+    return lanes
+
+
+def migrate_legacy_collection(base_name: str, lanes: Sequence[EmbeddingLane]) -> None:
+    """Backfill empty lanes from a legacy unsuffixed collection, if present."""
+    if not lanes:
+        return
+
+    try:
+        from src.chroma_client import get_chroma_client
+
+        chroma_client = get_chroma_client()
+        legacy = chroma_client.get_collection(base_name)
+        data = legacy.get(include=["documents", "metadatas"])
+    except Exception:
+        return
+
+    ids = data.get("ids") or []
+    docs = data.get("documents") or []
+    metas = data.get("metadatas") or []
+    if not ids or not docs:
+        return
+
+    for lane in lanes:
+        try:
+            existing = lane.collection.get(ids=ids)
+            existing_ids = set(existing.get("ids") or [])
+        except Exception:
+            existing_ids = set()
+        all_metas = list(metas or [])
+        if len(all_metas) < len(ids):
+            all_metas += [{}] * (len(ids) - len(all_metas))
+        missing = [
+            (row_id, doc, meta)
+            for row_id, doc, meta in zip(ids, docs, all_metas)
+            if row_id not in existing_ids
+        ]
+        if not missing:
+            continue
+
+        for start in range(0, len(missing), 100):
+            batch = missing[start:start + 100]
+            batch_ids = [row_id for row_id, _doc, _meta in batch]
+            batch_docs = [doc for _row_id, doc, _meta in batch]
+            batch_metas = [meta or {} for _row_id, _doc, meta in batch]
+            if len(batch_metas) < len(batch_ids):
+                batch_metas += [{}] * (len(batch_ids) - len(batch_metas))
+            try:
+                embeddings = lane.encode(batch_docs)
+                lane.collection.add(
+                    ids=batch_ids,
+                    documents=batch_docs,
+                    metadatas=batch_metas,
+                    embeddings=embeddings,
+                )
+            except Exception as e:
+                logger.warning(
+                    "Could not backfill %s lane from legacy collection %s: %s",
+                    lane.name,
+                    base_name,
+                    e,
+                )
+                break
+        else:
+            logger.info("Backfilled %s %s lane rows from legacy collection %s", len(missing), lane.name, base_name)
+
+
+def lane_count(lanes: Sequence[EmbeddingLane]) -> int:
+    return max((lane.count() for lane in lanes), default=0)
+
+
+def dedupe_results(results: Iterable[Dict[str, Any]], id_key: str = "id", limit: Optional[int] = None) -> List[Dict[str, Any]]:
+    seen = set()
+    out: List[Dict[str, Any]] = []
+    for row in results:
+        row_id = row.get(id_key)
+        if not row_id or row_id in seen:
+            continue
+        seen.add(row_id)
+        out.append(row)
+        if limit is not None and len(out) >= limit:
+            break
+    return out
+
+
+def query_lanes(
+    lanes: Sequence[EmbeddingLane],
+    query: str,
+    n_results: Callable[[EmbeddingLane], int],
+    include: Sequence[str],
+    where: Optional[Dict[str, Any]] = None,
+    raise_if_all_failed: bool = False,
+) -> List[tuple[EmbeddingLane, Dict[str, Any]]]:
+    out: List[tuple[EmbeddingLane, Dict[str, Any]]] = []
+    attempted = 0
+    failures: List[str] = []
+    for lane in lanes:
+        try:
+            count = lane.count()
+            if count == 0:
+                continue
+            attempted += 1
+            n = min(n_results(lane), count)
+            if n <= 0:
+                continue
+            results = lane.collection.query(
+                query_embeddings=lane.encode([query]),
+                n_results=n,
+                where=where,
+                include=list(include),
+            )
+            out.append((lane, results))
+        except Exception as e:
+            failures.append(f"{lane.name}: {e}")
+            logger.warning("%s lane query failed for %s: %s", lane.name, lane.collection_name, e)
+    if raise_if_all_failed and attempted and not out and failures:
+        raise RuntimeError("; ".join(failures))
+    return out

src/goal_based_extractor.py

@@ -3,22 +3,18 @@
 Goal-based content extraction prompt inspired by Alibaba Tongyi DeepResearch.
 """
 
-EXTRACTOR_PROMPT = """Please process the following webpage content and user goal to extract relevant information:
+EXTRACTOR_SYSTEM = """Extract relevant information from a webpage for a given research goal.
 
-## **Webpage Content**
-{webpage_content}
+Goal: {goal}
 
-## **User Goal**
-{goal}
+Task guidelines:
+1. Locate the specific sections directly related to the goal within the provided webpage content.
+2. Identify and extract the most relevant information; output full original context where possible, up to three or more paragraphs.
+3. Organize into a concise paragraph with logical flow, judging each piece of information's contribution to the goal.
 
-## **Task Guidelines**
-1. **Content Scanning for Rational**: Locate the **specific sections/data** directly related to the user's goal within the webpage content
-2. **Key Extraction for Evidence**: Identify and extract the **most relevant information** from the content, you never miss any important information, output the **full original context** of the content as far as possible, it can be more than three paragraphs.
-3. **Summary Output for Summary**: Organize into a concise paragraph with logical flow, prioritizing clarity and judge the contribution of the information to the goal.
+Respond in JSON with exactly these fields: "rational", "evidence", "summary".
 
-**Final Output Format using JSON format has "rational", "evidence", "summary" fields**
-
-Example output:
+Example:
 {{
     "rational": "This section discusses X which directly relates to the goal of understanding Y",
     "evidence": "Full quotes and context from the page...",

src/memory_vector.py

@@ -9,6 +9,16 @@ Stores pre-computed embeddings (ChromaDB does not manage embedding).
 import logging
 from typing import List, Dict, Optional
 
+from src.embedding_lanes import (
+    LANE_CUSTOM,
+    LANE_FASTEMBED,
+    build_embedding_lanes,
+    collection_name,
+    dedupe_results,
+    lane_count,
+    migrate_legacy_collection,
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -20,30 +30,28 @@ class MemoryVectorStore:
     def __init__(self, data_dir: str, embedding_model=None):
         self._model = embedding_model
         self._collection = None
+        self._lanes = []
         self._healthy = False
 
         self._initialize()
 
     def _initialize(self):
         try:
-            from src.chroma_client import get_chroma_client
-
-            if self._model is None:
-                from src.embeddings import get_embedding_client
-                self._model = get_embedding_client()
-                if self._model is None:
-                    raise RuntimeError("No embedding backend available")
-                logger.info(f"MemoryVectorStore using embeddings: {self._model.url}")
-
-            client = get_chroma_client()
-            self._collection = client.get_or_create_collection(
-                name=self.COLLECTION_NAME,
-                metadata={"hnsw:space": "cosine"},
-            )
+            self._lanes = build_embedding_lanes(self.COLLECTION_NAME)
+            if not self._lanes:
+                raise RuntimeError("No embedding lanes available")
 
             self._healthy = True
-            count = self._collection.count()
-            logger.info(f"MemoryVectorStore ready (entries={count})")
+            self._collection = next(
+                (lane.collection for lane in self._lanes if lane.name == LANE_FASTEMBED),
+                self._lanes[0].collection,
+            )
+            migrate_legacy_collection(self.COLLECTION_NAME, self._lanes)
+            logger.info(
+                "MemoryVectorStore ready (lanes=%s entries=%s)",
+                [lane.name for lane in self._lanes],
+                self.count(),
+            )
 
         except Exception as e:
             logger.error(f"MemoryVectorStore init failed: {e}")
@@ -53,39 +61,73 @@ class MemoryVectorStore:
         return self._healthy
 
     def _embed(self, texts: List[str]) -> List[List[float]]:
-        vecs = self._model.encode(texts, normalize_embeddings=True)
-        return vecs.tolist()
+        if not self._lanes:
+            return []
+        return self._lanes[0].encode(texts)
 
     def count(self) -> int:
         """Return the number of stored vectors."""
         if not self._healthy:
             return 0
-        return self._collection.count()
+        return lane_count(self._lanes)
+
+    def _collections_for_delete(self):
+        collections = []
+        seen = set()
+
+        def add(collection) -> None:
+            if collection is None:
+                return
+            key = getattr(collection, "name", None) or id(collection)
+            if key in seen:
+                return
+            seen.add(key)
+            collections.append(collection)
+
+        for lane in self._lanes:
+            add(lane.collection)
+
+        try:
+            from src.chroma_client import get_chroma_client
+
+            client = get_chroma_client()
+            for lane_name in (LANE_CUSTOM, LANE_FASTEMBED):
+                try:
+                    add(client.get_collection(collection_name(self.COLLECTION_NAME, lane_name)))
+                except Exception:
+                    pass
+        except Exception:
+            pass
+
+        return collections
 
     def add(self, memory_id: str, text: str):
         """Add a single memory entry to the vector index."""
         if not self._healthy:
             return
-        # Skip if already exists
-        existing = self._collection.get(ids=[memory_id])
-        if existing["ids"]:
-            return
-        embeddings = self._embed([text])
-        self._collection.add(
-            ids=[memory_id],
-            embeddings=embeddings,
-            documents=[text],
-            metadatas=[{"source": "memory"}],
-        )
+        for lane in self._lanes:
+            try:
+                existing = lane.collection.get(ids=[memory_id])
+                if existing["ids"]:
+                    continue
+                lane.collection.add(
+                    ids=[memory_id],
+                    embeddings=lane.encode([text]),
+                    documents=[text],
+                    metadatas=[{"source": "memory"}],
+                )
+            except Exception as e:
+                logger.warning("memory add failed in %s lane for %s: %s", lane.name, memory_id, e)
 
     def remove(self, memory_id: str):
         """Remove a memory entry. O(1) — no rebuild needed."""
         if not self._healthy:
             return
-        try:
-            self._collection.delete(ids=[memory_id])
-        except Exception as e:
-            logger.warning(f"memory remove {memory_id}: {e}")
+        for collection in self._collections_for_delete():
+            try:
+                collection.delete(ids=[memory_id])
+            except Exception as e:
+                logger.warning(f"memory remove {memory_id}: {e}")
 
     def search(self, query: str, k: int = 8) -> List[Dict]:
         """Search for the most relevant memory IDs by semantic similarity.
@@ -94,41 +136,53 @@ class MemoryVectorStore:
         ChromaDB cosine distance = 1 - cosine_similarity.
         We convert back: similarity = 1.0 - distance.
         """
-        if not self._healthy or self._collection.count() == 0:
+        if not self._healthy or self.count() == 0:
             return []
 
-        embeddings = self._embed([query])
-        actual_k = min(k, self._collection.count())
-        results = self._collection.query(
-            query_embeddings=embeddings,
-            n_results=actual_k,
-        )
-
         out = []
-        for idx, mid in enumerate(results["ids"][0]):
-            distance = results["distances"][0][idx]
-            out.append({
-                "memory_id": mid,
-                "score": round(1.0 - distance, 4),
-            })
-        return out
+        lane_priority = {LANE_CUSTOM: 0, LANE_FASTEMBED: 1}
+        for lane in self._lanes:
+            try:
+                if lane.count() == 0:
+                    continue
+                results = lane.collection.query(
+                    query_embeddings=lane.encode([query]),
+                    n_results=min(k, lane.count()),
+                    include=["distances"],
+                )
+                for idx, mid in enumerate(results["ids"][0]):
+                    distance = results["distances"][0][idx]
+                    out.append({
+                        "memory_id": mid,
+                        "score": round(1.0 - distance, 4),
+                        "embedding_lane": lane.name,
+                    })
+            except Exception as e:
+                logger.warning("memory search failed in %s lane: %s", lane.name, e)
+        out.sort(key=lambda row: (-row["score"], lane_priority.get(row["embedding_lane"], 99)))
+        return dedupe_results(out, id_key="memory_id", limit=k)
 
     def find_similar(self, text: str, threshold: float = 0.92) -> Optional[str]:
         """Check if a near-duplicate exists. Returns memory_id if found, else None."""
-        if not self._healthy or self._collection.count() == 0:
+        if not self._healthy or self.count() == 0:
             return None
 
-        embeddings = self._embed([text])
-        results = self._collection.query(
-            query_embeddings=embeddings,
-            n_results=1,
-        )
-
-        if results["ids"][0]:
-            distance = results["distances"][0][0]
-            similarity = 1.0 - distance
-            if similarity >= threshold:
-                return results["ids"][0][0]
+        for lane in self._lanes:
+            try:
+                if lane.count() == 0:
+                    continue
+                results = lane.collection.query(
+                    query_embeddings=lane.encode([text]),
+                    n_results=1,
+                    include=["distances"],
+                )
+                if results["ids"][0]:
+                    distance = results["distances"][0][0]
+                    similarity = 1.0 - distance
+                    if similarity >= threshold:
+                        return results["ids"][0][0]
+            except Exception as e:
+                logger.warning("memory similarity search failed in %s lane: %s", lane.name, e)
         return None
 
     def rebuild(self, memories: List[Dict]):
@@ -139,15 +193,23 @@ class MemoryVectorStore:
 
         from src.chroma_client import get_chroma_client
 
-        # Delete and recreate collection for a clean rebuild
         client = get_chroma_client()
-        try:
-            client.delete_collection(self.COLLECTION_NAME)
-        except Exception:
-            pass
-        self._collection = client.get_or_create_collection(
-            name=self.COLLECTION_NAME,
-            metadata={"hnsw:space": "cosine"},
+        lane_names = [
+            self.COLLECTION_NAME,
+            collection_name(self.COLLECTION_NAME, LANE_CUSTOM),
+            collection_name(self.COLLECTION_NAME, LANE_FASTEMBED),
+        ]
+        for name in lane_names:
+            try:
+                client.delete_collection(name)
+            except Exception:
+                pass
+        # Explicit rebuilds must start from the supplied memory list, so clear
+        # legacy unsuffixed collections too.
+        self._lanes = build_embedding_lanes(self.COLLECTION_NAME)
+        self._collection = next(
+            (lane.collection for lane in self._lanes if lane.name == LANE_FASTEMBED),
+            self._lanes[0].collection if self._lanes else None,
         )
 
         texts = []
@@ -161,15 +223,29 @@ class MemoryVectorStore:
 
         if texts:
             # Batch in chunks of 100 to avoid oversized requests
+            failed_lanes = set()
             for i in range(0, len(texts), 100):
                 batch_texts = texts[i:i + 100]
                 batch_ids = ids[i:i + 100]
-                embeddings = self._embed(batch_texts)
-                self._collection.add(
-                    ids=batch_ids,
-                    embeddings=embeddings,
-                    documents=batch_texts,
-                    metadatas=[{"source": "memory"}] * len(batch_ids),
-                )
+                for lane in self._lanes:
+                    if lane.name in failed_lanes:
+                        continue
+                    try:
+                        lane.collection.add(
+                            ids=batch_ids,
+                            embeddings=lane.encode(batch_texts),
+                            documents=batch_texts,
+                            metadatas=[{"source": "memory"}] * len(batch_ids),
+                        )
+                    except Exception as e:
+                        failed_lanes.add(lane.name)
+                        logger.warning("memory rebuild failed in %s lane: %s", lane.name, e)
 
-        logger.info(f"MemoryVectorStore rebuilt with {len(ids)} entries")
+        logger.info(f"MemoryVectorStore rebuilt with {len(ids)} entries across {len(self._lanes)} lanes")
+
+    def get_stats(self) -> Dict:
+        return {
+            "healthy": self.healthy,
+            "count": self.count(),
+            "lanes": [lane.stats() for lane in self._lanes],
+        }

src/rag_vector.py

@@ -14,6 +14,17 @@ import numpy as np
 from typing import List, Dict, Any, Optional, Set
 from pathlib import Path
 
+from src.embedding_lanes import (
+    LANE_CUSTOM,
+    LANE_FASTEMBED,
+    build_embedding_lanes,
+    collection_name,
+    dedupe_results,
+    lane_count,
+    migrate_legacy_collection,
+    query_lanes,
+)
+
 logger = logging.getLogger(__name__)
 
 DEFAULT_FILE_EXTENSIONS: Set[str] = {
@@ -44,6 +55,7 @@ class VectorRAG:
         self.persist_directory = persist_directory
         self._collection = None
         self._model = None
+        self._lanes = []
         self._healthy = False
 
         Path(self.persist_directory).mkdir(parents=True, exist_ok=True)
@@ -55,22 +67,20 @@ class VectorRAG:
 
     def _initialize_system(self) -> bool:
         try:
-            from src.chroma_client import get_chroma_client
-            from src.embeddings import get_embedding_client
-
-            self._model = get_embedding_client()
-            if self._model is None:
-                raise RuntimeError("No embedding backend available")
-            logger.info(f"Embedding: {self._model.url} model={self._model.model}")
-
-            client = get_chroma_client()
-            self._collection = client.get_or_create_collection(
-                name=COLLECTION_NAME,
-                metadata={"hnsw:space": "cosine"},
+            self._lanes = build_embedding_lanes(COLLECTION_NAME)
+            if not self._lanes:
+                raise RuntimeError("No embedding lanes available")
+            self._collection = next(
+                (lane.collection for lane in self._lanes if lane.name == LANE_FASTEMBED),
+                self._lanes[0].collection,
+            )
+            self._model = self._lanes[0].client
+            migrate_legacy_collection(COLLECTION_NAME, self._lanes)
+            logger.info(
+                "VectorRAG ready (lanes=%s docs=%s)",
+                [lane.name for lane in self._lanes],
+                lane_count(self._lanes),
             )
-
-            count = self._collection.count()
-            logger.info(f"VectorRAG ready ({count} docs)")
             self._healthy = True
             return True
 
@@ -80,8 +90,9 @@ class VectorRAG:
             return False
 
     def _embed(self, texts: List[str]) -> List[List[float]]:
-        vecs = self._model.encode(texts, normalize_embeddings=True)
-        return np.array(vecs, dtype=np.float32).tolist()
+        if not self._lanes:
+            return []
+        return np.array(self._lanes[0].encode(texts), dtype=np.float32).tolist()
 
     # ------------------------------------------------------------------
     # Properties
@@ -89,13 +100,57 @@ class VectorRAG:
 
     @property
     def healthy(self) -> bool:
-        return self._healthy and self._collection is not None
+        if getattr(self, "_lanes", None):
+            return self._healthy and bool(self._lanes)
+        return self._healthy and getattr(self, "_collection", None) is not None
 
     @property
     def collection(self):
         """Expose the ChromaDB collection for direct access by personal_routes etc."""
         return self._collection
 
+    def _active_collections(self):
+        lanes = getattr(self, "_lanes", None)
+        if lanes:
+            return [(lane.name, lane.collection) for lane in lanes]
+        collection = getattr(self, "_collection", None)
+        return [("legacy", collection)] if collection is not None else []
+
+    def _collections_for_delete(self):
+        collections = []
+        seen = set()
+
+        def add(lane_name: str, collection) -> None:
+            if collection is None:
+                return
+            key = getattr(collection, "name", None) or id(collection)
+            if key in seen:
+                return
+            seen.add(key)
+            collections.append((lane_name, collection))
+
+        for lane_name, collection in self._active_collections():
+            add(lane_name, collection)
+
+        if getattr(self, "_lanes", None):
+            try:
+                from src.chroma_client import get_chroma_client
+
+                client = get_chroma_client()
+                try:
+                    add("legacy", client.get_collection(COLLECTION_NAME))
+                except Exception:
+                    pass
+                for lane_name in (LANE_CUSTOM, LANE_FASTEMBED):
+                    try:
+                        add(lane_name, client.get_collection(collection_name(COLLECTION_NAME, lane_name)))
+                    except Exception:
+                        pass
+            except Exception:
+                pass
+
+        return collections
+
     # ------------------------------------------------------------------
     # Document operations
     # ------------------------------------------------------------------
@@ -109,23 +164,24 @@ class VectorRAG:
         if not metadata or not isinstance(metadata, dict):
             return False
 
-        try:
-            doc_id = _generate_doc_id(text, metadata.get("owner") or "")
-            # Check if already exists
-            existing = self._collection.get(ids=[doc_id])
-            if existing["ids"]:
-                return True  # already exists
-            embeddings = self._embed([text])
-            self._collection.add(
-                ids=[doc_id],
-                embeddings=embeddings,
-                documents=[text],
-                metadatas=[metadata],
-            )
-            return True
-        except Exception as e:
-            logger.error(f"add_document failed: {e}")
-            return False
+        doc_id = _generate_doc_id(text, metadata.get("owner") or "")
+        wrote = False
+        for lane in self._lanes:
+            try:
+                existing = lane.collection.get(ids=[doc_id])
+                if existing["ids"]:
+                    wrote = True
+                    continue
+                lane.collection.add(
+                    ids=[doc_id],
+                    embeddings=lane.encode([text]),
+                    documents=[text],
+                    metadatas=[metadata],
+                )
+                wrote = True
+            except Exception as e:
+                logger.warning("add_document failed in %s lane: %s", lane.name, e)
+        return wrote
 
     def add_documents_batch(self, docs: List[tuple]) -> Dict[str, Any]:
         if not self.healthy:
@@ -140,42 +196,57 @@ class VectorRAG:
         if not valid:
             return {"success": False, "message": "No valid documents"}
 
-        try:
-            # Get existing IDs to avoid duplicates
+        added_ids = set()
+        attempted_new = False
+        write_failed = False
+        for lane in self._lanes:
+            all_ids = [_generate_doc_id(t, m.get("owner") or "") for t, m in valid]
+            try:
+                existing = lane.collection.get(ids=all_ids)
+                existing_ids = set(existing.get("ids") or [])
+            except Exception:
+                existing_ids = set()
+
             new_texts = []
             new_metas = []
             new_ids = []
-            for t, m in valid:
-                doc_id = _generate_doc_id(t, m.get("owner") or "")
-                existing = self._collection.get(ids=[doc_id])
-                if not existing["ids"]:
-                    new_texts.append(t)
-                    new_metas.append(m)
+            for (text, meta), doc_id in zip(valid, all_ids):
+                if doc_id not in existing_ids:
+                    new_texts.append(text)
+                    new_metas.append(meta)
                     new_ids.append(doc_id)
 
             if new_texts:
-                # Batch in chunks of 100
+                attempted_new = True
+                lane_failed = False
                 for i in range(0, len(new_texts), 100):
                     batch_texts = new_texts[i:i + 100]
                     batch_ids = new_ids[i:i + 100]
                     batch_metas = new_metas[i:i + 100]
-                    embeddings = self._embed(batch_texts)
-                    self._collection.add(
-                        ids=batch_ids,
-                        embeddings=embeddings,
-                        documents=batch_texts,
-                        metadatas=batch_metas,
-                    )
+                    try:
+                        lane.collection.add(
+                            ids=batch_ids,
+                            embeddings=lane.encode(batch_texts),
+                            documents=batch_texts,
+                            metadatas=batch_metas,
+                        )
+                    except Exception as e:
+                        lane_failed = True
+                        write_failed = True
+                        logger.warning("add_documents_batch failed in %s lane: %s", lane.name, e)
+                        break
+                if not lane_failed:
+                    added_ids.update(new_ids)
 
-            return {
-                "success": True,
-                "added_count": len(new_texts),
-                "total_count": len(docs),
-                "failed_count": len(docs) - len(valid),
-            }
-        except Exception as e:
-            logger.error(f"add_documents_batch failed: {e}")
-            return {"success": False, "message": str(e)}
+        if attempted_new and write_failed and not added_ids:
+            return {"success": False, "message": "No embedding lane accepted the batch"}
+
+        return {
+            "success": True,
+            "added_count": len(added_ids),
+            "total_count": len(docs),
+            "failed_count": len(docs) - len(valid),
+        }
 
     # ------------------------------------------------------------------
     # Search — hybrid: vector similarity + keyword overlap
@@ -186,58 +257,51 @@ class VectorRAG:
             return []
         if not query or not isinstance(query, str):
             return []
-        if self._collection.count() == 0:
+        if lane_count(self._lanes) == 0:
             return []
 
         try:
-            # Fetch extra candidates when owner-filtering
-            fetch_k = min(k * 3, max(k, 20), self._collection.count())
-            if owner:
-                fetch_k = min(fetch_k * 2, self._collection.count())
-
-            query_embeddings = self._embed([query])
-
-            # Use ChromaDB where filter for owner if specified
             where_filter = {"owner": owner} if owner else None
-
-            results = self._collection.query(
-                query_embeddings=query_embeddings,
-                n_results=fetch_k,
-                where=where_filter,
-                include=["documents", "metadatas", "distances"],
-            )
-
             query_words = set(query.lower().split())
             candidates = []
 
-            for idx in range(len(results["ids"][0])):
-                doc_id = results["ids"][0][idx]
-                distance = results["distances"][0][idx]
-                doc_text = results["documents"][0][idx]
-                meta = results["metadatas"][0][idx]
+            for lane, results in query_lanes(
+                self._lanes,
+                query,
+                n_results=lambda lane: min(
+                    (k * 6 if owner else k * 3),
+                    max(k, 20),
+                    lane.count(),
+                ),
+                where=where_filter,
+                include=["documents", "metadatas", "distances"],
+                raise_if_all_failed=True,
+            ):
+                for idx in range(len(results["ids"][0])):
+                    doc_id = results["ids"][0][idx]
+                    distance = results["distances"][0][idx]
+                    doc_text = results["documents"][0][idx]
+                    meta = results["metadatas"][0][idx]
 
-                # ChromaDB cosine distance = 1 - cosine_similarity
-                vector_sim = 1.0 - distance
+                    vector_sim = 1.0 - distance
+                    doc_words = set(doc_text.lower().split())
+                    overlap = len(query_words & doc_words)
+                    keyword_score = overlap / len(query_words) if query_words else 0.0
+                    hybrid_score = (VECTOR_WEIGHT * vector_sim) + (KEYWORD_WEIGHT * keyword_score)
 
-                # Keyword overlap score
-                doc_words = set(doc_text.lower().split())
-                overlap = len(query_words & doc_words)
-                keyword_score = overlap / len(query_words) if query_words else 0.0
-
-                hybrid_score = (VECTOR_WEIGHT * vector_sim) + (KEYWORD_WEIGHT * keyword_score)
-
-                candidates.append({
-                    "id": doc_id,
-                    "document": doc_text,
-                    "metadata": meta,
-                    "distance": round(distance, 4),
-                    "similarity": round(hybrid_score, 4),
-                    "vector_similarity": round(vector_sim, 4),
-                    "keyword_score": round(keyword_score, 4),
-                })
+                    candidates.append({
+                        "id": doc_id,
+                        "document": doc_text,
+                        "metadata": meta,
+                        "distance": round(distance, 4),
+                        "similarity": round(hybrid_score, 4),
+                        "vector_similarity": round(vector_sim, 4),
+                        "keyword_score": round(keyword_score, 4),
+                        "embedding_lane": lane.name,
+                    })
 
             candidates.sort(key=lambda c: c["similarity"], reverse=True)
-            top = candidates[:k]
+            top = dedupe_results(candidates, limit=k)
             logger.info(f"Hybrid search for '{query[:60]}': {len(top)} results")
             return top
 
@@ -247,39 +311,36 @@ class VectorRAG:
 
     def _keyword_search_fallback(self, query: str, k: int = 5, owner: Optional[str] = None) -> List[Dict[str, Any]]:
         try:
-            if self._collection.count() == 0:
-                return []
-
-            # Fetch all documents for keyword search fallback
-            all_docs = self._collection.get(include=["documents", "metadatas"])
-            if not all_docs["ids"]:
+            if not self._active_collections():
                 return []
 
             query_words = query.lower().split()
             scored = []
-            for i, doc in enumerate(all_docs["documents"]):
-                meta = all_docs["metadatas"][i]
-                if owner:
-                    # Match the primary path's strict where={"owner": owner}
-                    # filter. The old `if doc_owner and doc_owner != owner`
-                    # let docs with a missing/empty owner fall through, leaking
-                    # owner-less documents into another user's results.
-                    if meta.get("owner") != owner:
+            for lane_name, collection in self._active_collections():
+                if collection.count() == 0:
+                    continue
+                all_docs = collection.get(include=["documents", "metadatas"])
+                if not all_docs["ids"]:
+                    continue
+                for i, doc in enumerate(all_docs["documents"]):
+                    meta = all_docs["metadatas"][i]
+                    if owner and meta.get("owner") != owner:
                         continue
-                doc_lower = doc.lower()
-                score = sum(1 for w in query_words if w in doc_lower)
-                if score > 0:
-                    scored.append({
-                        "id": all_docs["ids"][i],
-                        "document": doc,
-                        "metadata": meta,
-                        "distance": 0,
-                        "similarity": score,
-                        "search_type": "keyword_fallback",
-                    })
+                    doc_lower = doc.lower()
+                    score = sum(1 for w in query_words if w in doc_lower)
+                    if score > 0:
+                        scored.append({
+                            "id": all_docs["ids"][i],
+                            "document": doc,
+                            "metadata": meta,
+                            "distance": 0,
+                            "similarity": score,
+                            "search_type": "keyword_fallback",
+                            "embedding_lane": lane_name,
+                        })
 
             scored.sort(key=lambda x: x["similarity"], reverse=True)
-            return scored[:k]
+            return dedupe_results(scored, limit=k)
         except Exception as e:
             logger.error(f"keyword fallback failed: {e}")
             return []
@@ -296,9 +357,20 @@ class VectorRAG:
                 client.delete_collection(COLLECTION_NAME)
             except Exception:
                 pass
-            self._collection = client.get_or_create_collection(
-                name=COLLECTION_NAME,
-                metadata={"hnsw:space": "cosine"},
+            for name in (
+                collection_name(COLLECTION_NAME, LANE_CUSTOM),
+                collection_name(COLLECTION_NAME, LANE_FASTEMBED),
+            ):
+                try:
+                    client.delete_collection(name)
+                except Exception:
+                    pass
+            # Rebuild means empty current lanes. Clear the legacy unsuffixed
+            # collection too so startup migration cannot resurrect stale docs.
+            self._lanes = build_embedding_lanes(COLLECTION_NAME)
+            self._collection = next(
+                (lane.collection for lane in self._lanes if lane.name == LANE_FASTEMBED),
+                self._lanes[0].collection if self._lanes else None,
             )
             self._healthy = True
             return True
@@ -312,10 +384,11 @@ class VectorRAG:
             return {"error": "Collection not initialized"}
         try:
             return {
-                "document_count": self._collection.count(),
-                "embedding_model": f"{self._model.model} @ {self._model.url}" if self._model else "N/A",
+                "document_count": lane_count(self._lanes),
+                "embedding_model": f"{self._lanes[0].model} @ {self._lanes[0].url}" if self._lanes else "N/A",
                 "persist_directory": self.persist_directory,
                 "collection_name": COLLECTION_NAME,
+                "embedding_lanes": [lane.stats() for lane in self._lanes],
                 "healthy": True,
             }
         except Exception as e:
@@ -400,19 +473,23 @@ class VectorRAG:
             return {"success": False, "message": "Collection not initialized"}
         directory = os.path.abspath(directory)
         try:
-            results = self._collection.get(include=["metadatas"])
-            ids = [
-                results["ids"][i]
-                for i, m in enumerate(results["metadatas"])
-                if isinstance(m, dict)
-                and isinstance(m.get("source"), str)
-                and (m["source"] == directory or m["source"].startswith(directory + os.sep))
-            ]
-            if not ids:
+            removed_ids = set()
+            for _lane_name, collection in self._collections_for_delete():
+                results = collection.get(include=["metadatas"])
+                ids = [
+                    results["ids"][i]
+                    for i, m in enumerate(results["metadatas"])
+                    if isinstance(m, dict)
+                    and isinstance(m.get("source"), str)
+                    and (m["source"] == directory or m["source"].startswith(directory + os.sep))
+                ]
+                if ids:
+                    collection.delete(ids=ids)
+                    removed_ids.update(ids)
+            if not removed_ids:
                 return {"success": True, "removed_count": 0, "message": "No docs found"}
 
-            self._collection.delete(ids=ids)
-            n = len(ids)
+            n = len(removed_ids)
             logger.info(f"Removed {n} chunks from {directory}")
             return {"success": True, "removed_count": n, "message": f"Removed {n} chunks"}
         except Exception as e:
@@ -504,16 +581,18 @@ class VectorRAG:
         if not self.healthy:
             return 0
         try:
-            results = self._collection.get(
-                where={"source": source},
-                include=[],
-            )
-            ids = results.get("ids", [])
-            if not ids:
-                return 0
-            self._collection.delete(ids=ids)
-            logger.info(f"Deleted {len(ids)} chunks for source={source}")
-            return len(ids)
+            removed_ids = set()
+            for _lane_name, collection in self._collections_for_delete():
+                results = collection.get(
+                    where={"source": source},
+                    include=[],
+                )
+                ids = results.get("ids", [])
+                if ids:
+                    collection.delete(ids=ids)
+                    removed_ids.update(ids)
+            logger.info(f"Deleted {len(removed_ids)} chunks for source={source}")
+            return len(removed_ids)
         except Exception as e:
             logger.error(f"delete_by_source failed: {e}")
             return 0

src/tool_implementations.py

@@ -2111,6 +2111,13 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
         """Parse agent event datetimes in the user's timezone when available."""
         return _parse_dt_pair(parse_due_for_user(raw))
 
+    def _first_nonempty_arg(*names: str):
+        for name in names:
+            value = args.get(name)
+            if value not in (None, ""):
+                return value
+        return None
+
     def _create_calendar_reminder(summary: str, location: str, dtstart: datetime,
                                   all_day: bool, minutes_before: int,
                                   is_utc: bool = False) -> tuple[Optional[str], Optional[str]]:
@@ -2168,12 +2175,18 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
 
         elif action == "list_events":
             try:
-                if args.get("start"):
-                    start_dt = _parse_dt(args["start"])
+                start_raw = _first_nonempty_arg(
+                    "start", "start_date", "range_start", "from", "dtstart", "since"
+                )
+                end_raw = _first_nonempty_arg(
+                    "end", "end_date", "range_end", "to", "dtend", "until"
+                )
+                if start_raw:
+                    start_dt = _parse_dt(start_raw)
                 else:
                     start_dt = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
-                if args.get("end"):
-                    end_dt = _parse_dt(args["end"])
+                if end_raw:
+                    end_dt = _parse_dt(end_raw)
                 else:
                     end_dt = start_dt + timedelta(days=14)
             except ValueError as e:

src/tool_index.py

@@ -12,6 +12,14 @@ import re
 import time
 from typing import Dict, List, Optional, Set
 
+from src.embedding_lanes import (
+    LANE_CUSTOM,
+    LANE_FASTEMBED,
+    build_embedding_lanes,
+    dedupe_results,
+    migrate_legacy_collection,
+)
+
 try:
     import numpy as np
 except ImportError:
@@ -82,9 +90,9 @@ COLLECTION_NAME = "odysseus_tool_index"
 # Each tool gets a searchable description that helps retrieval.
 # These are richer than the system prompt one-liners — they're for embedding.
 BUILTIN_TOOL_DESCRIPTIONS: Dict[str, str] = {
-    "bash": "Run shell commands on the server. Install packages, check files, git operations, curl, system info, process management, networking.",
-    "python": "Execute Python code for computation, data processing, math, scripting, parsing, API calls. Not for writing code for the user.",
-    "web_search": "Quick single web lookup for a fact, current event, or doc mid-task. NOT for 'research X' / 'do research on X' requests — those are deep-research jobs (use trigger_research). web_search = one query; trigger_research = a full researched report in the sidebar.",
+    "bash": "Run shell commands on the server. Install packages, check files, git operations, system info, and process management. Do not use for web lookup/search; use web_search or web_fetch when web tools are available.",
+    "python": "Execute Python code for computation, data processing, math, scripting, and parsing. Not for writing code for the user. Do not use for web lookup/search; use web_search or web_fetch when web tools are available.",
+    "web_search": "Quick single web lookup for a fact, current event, latest/current information, or doc mid-task. Use this instead of bash/curl/python/requests for web searches. NOT for 'research X' / 'do research on X' requests — those are deep-research jobs (use trigger_research). web_search = one query; trigger_research = a full researched report in the sidebar.",
     "web_fetch": "Fetch and read the text content of a specific URL/website the user names (e.g. 'check example.com', 'open this link'). Use when you have a concrete URL; for open-ended lookups use web_search instead.",
     "read_file": "Read a file from disk and return its contents. View source code, config files, logs. Supports an optional line range (offset/limit) for large files.",
     "grep": "Search file CONTENTS for a regex across a directory tree (ripgrep-backed, honours .gitignore). Returns file:line:match. Use to find where code/symbols/strings live — prefer over bash grep.",
@@ -155,32 +163,30 @@ class ToolIndex:
     """ChromaDB-backed tool index for RAG-based tool selection."""
 
     def __init__(self):
-        from src.chroma_client import get_chroma_client
-        from src.embeddings import get_embedding_client
-
-        self._embedder = get_embedding_client()
-        if not self._embedder:
-            raise RuntimeError("No embedding client available")
-
-        client = get_chroma_client()
-        self._collection = client.get_or_create_collection(
-            name=COLLECTION_NAME,
-            metadata={"hnsw:space": "cosine"},
+        self._lanes = build_embedding_lanes(COLLECTION_NAME)
+        if not self._lanes:
+            raise RuntimeError("No embedding lanes available")
+        self._embedder = self._lanes[0].client
+        self._collection = next(
+            (lane.collection for lane in self._lanes if lane.name == LANE_FASTEMBED),
+            self._lanes[0].collection,
         )
+        migrate_legacy_collection(COLLECTION_NAME, self._lanes)
         self._fingerprint = ""
         self._mcp_generation = -1
         self._healthy = True
-        logger.info("ToolIndex initialized")
+        logger.info("ToolIndex initialized (lanes=%s)", [lane.name for lane in self._lanes])
 
     @property
     def healthy(self):
         return self._healthy
 
     def _embed(self, texts: List[str]) -> List[List[float]]:
-        vecs = self._embedder.encode(texts, normalize_embeddings=True)
+        if not self._lanes:
+            return []
+        vecs = self._lanes[0].encode(texts)
         if np is not None:
             return np.array(vecs, dtype=np.float32).tolist()
-        # Fallback without numpy
         return [list(v) for v in vecs]
 
     def index_builtin_tools(self):
@@ -201,23 +207,31 @@ class ToolIndex:
         # registry (e.g. removed tools like the old vault_* set).
         # Without this, upsert leaves them in place and RAG keeps
         # surfacing tools that no longer exist.
-        try:
-            existing = self._collection.get(where={"tool_type": "builtin"})
-            existing_ids = (existing or {}).get("ids") or []
-            stale = [i for i in existing_ids if i not in set(ids)]
-            if stale:
-                self._collection.delete(ids=stale)
-                logger.info(f"Pruned {len(stale)} stale builtin tool entries from index")
-        except Exception as e:
-            logger.debug(f"Stale-pruning skipped: {e}")
+        indexed = False
+        for lane in self._lanes:
+            try:
+                existing = lane.collection.get(where={"tool_type": "builtin"})
+                existing_ids = (existing or {}).get("ids") or []
+                stale = [i for i in existing_ids if i not in set(ids)]
+                if stale:
+                    lane.collection.delete(ids=stale)
+                    logger.info(f"Pruned {len(stale)} stale builtin tool entries from {lane.name} index")
+            except Exception as e:
+                logger.debug(f"Stale-pruning skipped for {lane.name}: {e}")
 
-        embeddings = self._embed(docs)
-        self._collection.upsert(
-            ids=ids,
-            documents=docs,
-            embeddings=embeddings,
-            metadatas=metadatas,
-        )
+            try:
+                lane.collection.upsert(
+                    ids=ids,
+                    documents=docs,
+                    embeddings=lane.encode(docs),
+                    metadatas=metadatas,
+                )
+                indexed = True
+            except Exception as e:
+                logger.warning("Builtin tool indexing failed in %s lane: %s", lane.name, e)
+        if not indexed:
+            self._healthy = False
+            raise RuntimeError("Builtin tool indexing failed in all embedding lanes")
         self._fingerprint = hashlib.sha256(
             ",".join(sorted(BUILTIN_TOOL_DESCRIPTIONS.keys())).encode()
         ).hexdigest()
@@ -232,15 +246,15 @@ class ToolIndex:
         gen = getattr(mcp_mgr, '_generation', 0)
         if gen == self._mcp_generation:
             return
-        self._mcp_generation = gen
 
         # Remove old MCP entries
-        try:
-            existing = self._collection.get(where={"tool_type": "mcp"})
-            if existing and existing["ids"]:
-                self._collection.delete(ids=existing["ids"])
-        except Exception:
-            pass
+        for lane in self._lanes:
+            try:
+                existing = lane.collection.get(where={"tool_type": "mcp"})
+                if existing and existing["ids"]:
+                    lane.collection.delete(ids=existing["ids"])
+            except Exception:
+                pass
 
         # Get current MCP tools
         try:
@@ -249,6 +263,7 @@ class ToolIndex:
             all_tools = ""
 
         if not all_tools:
+            self._mcp_generation = gen
             return
 
         # Parse MCP tool descriptions from the prompt text
@@ -276,39 +291,59 @@ class ToolIndex:
                     metadatas.append({"tool_name": name, "tool_type": "mcp"})
 
         if not docs:
+            self._mcp_generation = gen
             return
 
-        embeddings = self._embed(docs)
-        self._collection.upsert(
-            ids=ids,
-            documents=docs,
-            embeddings=embeddings,
-            metadatas=metadatas,
-        )
+        indexed = False
+        for lane in self._lanes:
+            try:
+                lane.collection.upsert(
+                    ids=ids,
+                    documents=docs,
+                    embeddings=lane.encode(docs),
+                    metadatas=metadatas,
+                )
+                indexed = True
+            except Exception as e:
+                logger.warning("MCP tool indexing failed in %s lane: %s", lane.name, e)
+        if not indexed:
+            logger.warning("MCP tool indexing failed in all embedding lanes")
+            return
+        self._mcp_generation = gen
         logger.info(f"Indexed {len(docs)} MCP tools")
 
     def retrieve(self, query: str, k: int = 8) -> List[str]:
         """Retrieve the top-K most relevant tool names for a query."""
-        try:
-            query_embedding = self._embed([query])
-            results = self._collection.query(
-                query_embeddings=query_embedding,
-                n_results=min(k, self._collection.count() or k),
-                include=["metadatas", "distances"],
-            )
-            if not results or not results.get("metadatas"):
-                return []
-
-            tool_names = []
-            for meta_list in results["metadatas"]:
-                for meta in meta_list:
-                    name = meta.get("tool_name", "")
-                    if name and name not in tool_names:
-                        tool_names.append(name)
-            return tool_names
-        except Exception as e:
-            logger.warning(f"Tool retrieval failed: {e}")
-            return []
+        rows = []
+        lane_priority = {LANE_CUSTOM: 0, LANE_FASTEMBED: 1}
+        for lane in self._lanes:
+            try:
+                count = lane.count()
+                if count == 0:
+                    continue
+                results = lane.collection.query(
+                    query_embeddings=lane.encode([query]),
+                    n_results=min(k, count),
+                    include=["metadatas", "distances"],
+                )
+                if not results or not results.get("metadatas"):
+                    continue
+                distances = results.get("distances") or []
+                for list_idx, meta_list in enumerate(results["metadatas"]):
+                    distance_list = distances[list_idx] if list_idx < len(distances) else []
+                    for idx, meta in enumerate(meta_list):
+                        name = meta.get("tool_name", "")
+                        if name:
+                            distance = distance_list[idx] if idx < len(distance_list) else 1.0
+                            rows.append({
+                                "tool_name": name,
+                                "score": round(1.0 - distance, 4),
+                                "embedding_lane": lane.name,
+                            })
+            except Exception as e:
+                logger.warning("Tool retrieval failed in %s lane: %s", lane.name, e)
+        rows.sort(key=lambda row: (-row["score"], lane_priority.get(row["embedding_lane"], 99)))
+        return [row["tool_name"] for row in dedupe_results(rows, id_key="tool_name", limit=k)]
 
     # Structural recurring-schedule intent. Typo-resilient (matches "every dya"
     # via "every <word>"), and catches bare clock times ("at 7:30 am", "7am").
@@ -511,3 +546,10 @@ def get_tool_index() -> Optional[ToolIndex]:
         logger.warning(f"ToolIndex init failed (will retry in {_RETRY_INTERVAL}s): {e}")
         _tool_index = None
         return None
+
+
+def reset_tool_index() -> None:
+    """Clear the singleton so embedding endpoint changes rebuild tool lanes."""
+    global _tool_index, _last_attempt
+    _tool_index = None
+    _last_attempt = 0.0

src/tool_parsing.py

@@ -5,9 +5,10 @@ Regex-based parsing of tool invocations from LLM response text.
 Supports fenced code blocks, [TOOL_CALL] blocks, and XML-style <invoke> blocks.
 """
 
-import re
+import ast
 import json
 import logging
+import re
 from typing import List, Optional
 
 from src.agent_tools import ToolBlock, TOOL_TAGS
@@ -176,11 +177,108 @@ _TOOL_NAME_MAP = {
     "todos": "manage_notes",
 }
 
+_MISFENCED_WEB_TOOL_NAMES = {
+    "web_search": "web_search",
+    "websearch": "web_search",
+    "google_search": "web_search",
+    "google_search_retrieval": "web_search",
+    "google_search_grounding": "web_search",
+    "web_fetch": "web_fetch",
+    "webfetch": "web_fetch",
+    "fetch_url": "web_fetch",
+}
+
 
 # ---------------------------------------------------------------------------
 # Parsing functions
 # ---------------------------------------------------------------------------
 
+def _literal_string(value) -> Optional[str]:
+    """Return a string from a small literal AST node, or None."""
+    try:
+        parsed = ast.literal_eval(value)
+    except (ValueError, SyntaxError, TypeError):
+        return None
+    if isinstance(parsed, str):
+        return parsed.strip()
+    if isinstance(parsed, list):
+        for item in parsed:
+            if isinstance(item, str) and item.strip():
+                return item.strip()
+    return None
+
+
+def _parse_misfenced_web_lookup(content: str) -> Optional[ToolBlock]:
+    """Recover simple web_search/web_fetch calls wrapped in python/bash fences.
+
+    Some local fenced-tool models write:
+
+        ```python
+        web_search("latest python release")
+        ```
+
+    That is an intended tool call, not Python code. Keep this intentionally
+    narrow: only a single bare function call to a known web tool alias converts.
+    """
+    try:
+        module = ast.parse(content.strip(), mode="exec")
+    except SyntaxError:
+        return None
+    if len(module.body) != 1 or not isinstance(module.body[0], ast.Expr):
+        return None
+    call = module.body[0].value
+    if not isinstance(call, ast.Call) or not isinstance(call.func, ast.Name):
+        return None
+
+    mapped = _MISFENCED_WEB_TOOL_NAMES.get(call.func.id.lower())
+    if mapped not in ("web_search", "web_fetch"):
+        return None
+    if len(call.args) > 1:
+        return None
+
+    args = {}
+    if call.args:
+        key = "url" if mapped == "web_fetch" else "query"
+        value = _literal_string(call.args[0])
+        if not value:
+            return None
+        args[key] = value
+
+    allowed = {"query", "queries", "url", "time_filter", "freshness", "max_pages"}
+    for keyword in call.keywords:
+        if keyword.arg not in allowed:
+            return None
+        key = "query" if keyword.arg == "queries" else keyword.arg
+        value = _literal_string(keyword.value)
+        if value is not None:
+            args[key] = value
+            continue
+        try:
+            parsed = ast.literal_eval(keyword.value)
+        except (ValueError, SyntaxError, TypeError):
+            return None
+        if key == "max_pages" and isinstance(parsed, int):
+            args[key] = parsed
+            continue
+        return None
+
+    if mapped == "web_search":
+        query = args.get("query")
+        if not query:
+            return None
+        payload = {"query": query}
+        for key in ("time_filter", "freshness", "max_pages"):
+            if key in args:
+                payload[key] = args[key]
+        if len(payload) == 1:
+            return ToolBlock("web_search", query)
+        return ToolBlock("web_search", json.dumps(payload))
+
+    url = args.get("url")
+    if not url:
+        return None
+    return ToolBlock("web_fetch", url)
+
 def _parse_tool_call_block(raw: str) -> Optional[ToolBlock]:
     """Parse a [TOOL_CALL] block into a ToolBlock.
 
@@ -363,6 +461,11 @@ def parse_tool_blocks(text: str) -> List[ToolBlock]:
             # a python/bash block — e.g. a hyphenated/namespaced tool name that
             # _XML_INVOKE_RE's \w+ can't match would otherwise be executed as code.
             continue
+        if tag in ("python", "bash"):
+            block = _parse_misfenced_web_lookup(content)
+            if block:
+                blocks.append(block)
+                continue
         blocks.append(ToolBlock(tag, content))
 
     # Pattern 2: [TOOL_CALL] blocks (only if no fenced blocks found)

src/tool_schemas.py

@@ -545,8 +545,8 @@ FUNCTION_TOOL_SCHEMAS = [
                     "uid": {"type": "string", "description": "Event UID (for update/delete)"},
                     "calendar_href": {"type": "string", "description": "Specific calendar URL (optional; defaults to first calendar)"},
                     "calendar": {"type": "string", "description": "Filter list_events by calendar name or href"},
-                    "start": {"type": "string", "description": "list_events range start (ISO datetime); defaults to today"},
-                    "end": {"type": "string", "description": "list_events range end (ISO datetime); defaults to +14 days"},
+                    "start": {"type": "string", "description": "list_events range start (ISO datetime); defaults to today. Prefer start; backend also accepts start_date, range_start, from, dtstart, since."},
+                    "end": {"type": "string", "description": "list_events range end (ISO datetime); defaults to +14 days. Prefer end; backend also accepts end_date, range_end, to, dtend, until."},
                     "event_type": {"type": "string", "description": "Tag / category for the event. Common values: work, personal, health, travel, meal, social, admin, other. Aliases accepted: tag, category, type."},
                     "importance": {"type": "string", "enum": ["low", "normal", "high", "critical"], "description": "Priority level (defaults to 'normal')"},
                     "reminder_minutes": {"type": "integer", "description": "For create_event: create an Odysseus reminder this many minutes before the event, e.g. 5 for 'reminder 5 min before'."},

static/js/chat.js

@@ -23,6 +23,7 @@ import * as emailInbox from './emailInbox.js';
 import codeRunnerModule from './codeRunner.js';
 import slashCommands, { initSlashCommands, isCommand, handleSlashCommand, handleSetupInput, handleSetupWizard, typewriterInto } from './slashCommands.js';
 import createResearchSynapse from './researchSynapse.js';
+import { createStreamRenderer } from './streamingRenderer.js';
   const RESEARCH_TIMEOUT_MS = 360000;
   const DEFAULT_TIMEOUT_MS = 120000;
   const RESEARCH_SVG = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><circle cx="11" cy="11" r="8"/><path d="M21 21l-4.35-4.35"/></svg>';
@@ -1167,9 +1168,6 @@ import createResearchSynapse from './researchSynapse.js';
       let _liveThinkToggle = null;
       let _liveThinkDomId = null;
 
-      // Offscreen measurement div — reused across renders
-      let _measureDiv = null;
-
       function _replyAfterClosedThinking(text) {
         const closeRe = /<\/(?:think(?:ing)?|thought)>|<channel\|>/gi;
         let match = null;
@@ -1224,19 +1222,18 @@ import createResearchSynapse from './researchSynapse.js';
             }
           }
           if (replyTrimmed) {
-            const replyHtml = markdownModule.mdToHtml(markdownModule.squashOutsideCode(replyTrimmed));
-            const prevLen = liveReply._prevTextLen || 0;
-            liveReply.innerHTML = replyHtml;
-            _fadeNewTokens(liveReply, prevLen);
-            liveReply._prevTextLen = liveReply.textContent.length;
-            if (window.hljs) liveReply.querySelectorAll('pre code').forEach((b) => window.hljs.highlightElement(b));
+            const r = liveReply._streamRenderer ||
+              (liveReply._streamRenderer = createStreamRenderer(liveReply, {
+                render: (t) => markdownModule.mdToHtml(markdownModule.squashOutsideCode(t)),
+                hljs: window.hljs,
+              }));
+            r.update(replyTrimmed);
           }
           // Reply empty or not — preserve thinking bar, don't fall through to full re-render
           uiModule.scrollHistory();
           return;
         }
 
-        const prevLen = contentEl._prevTextLen || 0;
         // If thinking is still streaming (unclosed <think>), show indicator instead of raw text
         if (markdownModule.hasUnclosedThinkTag && markdownModule.hasUnclosedThinkTag(dt)) {
           const thinkStart = dt.search(/<(?:think(?:ing)?|thought)(?:\s+[^>]*)?>|<\|channel>thought/i);
@@ -1250,66 +1247,26 @@ import createResearchSynapse from './researchSynapse.js';
           contentEl.innerHTML =
             '<div class="thinking-section"><div class="thinking-header"><div class="thinking-header-left">Thinking' +
             (lines > 1 ? ` (${lines} lines)` : '') + '</div></div></div>';
-          contentEl._prevTextLen = 0;
+          // The stream renderer self-heals when it next sees this overwritten
+          // container (streamingRenderer.js), so no explicit reset is needed here.
           uiModule.scrollHistory();
           return;
         }
-        const html = markdownModule.processWithThinking(markdownModule.squashOutsideCode(dt));
 
-        // Smooth expand only for regular chat text (not thinking/agent blocks)
-        const _hasThinking = html.includes('thinking-section');
-        const _isAgentRound = roundHolder !== holder;
-        if (!_hasThinking && !_isAgentRound) {
-          // Render into offscreen clone to measure new height before swapping
-          if (!_measureDiv) {
-            _measureDiv = document.createElement('div');
-            _measureDiv.style.cssText = 'position:absolute;visibility:hidden;pointer-events:none;z-index:-1;';
-          }
-          _measureDiv.style.width = contentEl.offsetWidth + 'px';
-          _measureDiv.className = contentEl.className;
-          _measureDiv.innerHTML = html;
-          contentEl.parentNode.appendChild(_measureDiv);
-          const measuredH = _measureDiv.offsetHeight;
-          _measureDiv.remove();
-          const curMin = parseFloat(contentEl.style.minHeight) || 0;
-          contentEl.style.minHeight = Math.max(curMin, measuredH) + 'px';
-        } else {
-          contentEl.style.minHeight = '';
-        }
-
-        contentEl.innerHTML = html;
-        _fadeNewTokens(contentEl, prevLen);
-        contentEl._prevTextLen = contentEl.textContent.length;
-        if (window.hljs) contentEl.querySelectorAll('pre code').forEach((b) => window.hljs.highlightElement(b));
+        // Incremental streaming render: freeze finalized blocks, re-render only the
+        // growing tail, and highlight each code block once on completion. This is
+        // what keeps code-block hover buttons from flickering and avoids the O(N^2)
+        // re-parse/re-highlight of the whole message on every token.
+        // See streamingRenderer.js / streamingSegmenter.js.
+        const renderer = contentEl._streamRenderer ||
+          (contentEl._streamRenderer = createStreamRenderer(contentEl, {
+            render: (t) => markdownModule.processWithThinking(markdownModule.squashOutsideCode(t)),
+            hljs: window.hljs,
+          }));
+        renderer.update(dt);
         uiModule.scrollHistory();
       };
 
-      // Walk text nodes, skip past `prevLen` characters of old text,
-      // wrap everything after that in <span class="token-new"> for fade-in
-      function _fadeNewTokens(container, prevLen) {
-        if (!prevLen) return; // First chunk — skip, whole msg already has entrance anim
-        const walker = document.createTreeWalker(container, NodeFilter.SHOW_TEXT);
-        let charCount = 0;
-        const toWrap = [];
-        while (walker.nextNode()) {
-          const node = walker.currentNode;
-          const len = node.textContent.length;
-          if (charCount + len <= prevLen) { charCount += len; continue; }
-          const splitAt = charCount < prevLen ? prevLen - charCount : 0;
-          toWrap.push({ node, splitAt });
-          charCount += len;
-        }
-        for (const { node, splitAt } of toWrap) {
-          const parent = node.parentNode;
-          if (!parent || parent.closest('pre, .think-content')) continue;
-          const target = splitAt > 0 ? node.splitText(splitAt) : node;
-          const span = document.createElement('span');
-          span.className = 'token-new';
-          parent.replaceChild(span, target);
-          span.appendChild(target);
-        }
-      }
-
       let _nextIsError = false;
       let _streamSawDone = false;
 

static/js/sessions.js

@@ -2184,6 +2184,10 @@ async function _checkServerStream(sessionId) {
     box.appendChild(holder);
     uiModule.scrollHistory();
 
+    // sessions.js executes before chat.js in module order, so window.chatModule
+    // may not be set yet when _checkServerStream first runs. Retry resumeStream
+    // on the first poll tick where it becomes available.
+    let _resumeRetried = false;
     const pollId = setInterval(async () => {
       if (getCurrentSessionId() !== sessionId) {
         clearInterval(pollId);
@@ -2191,6 +2195,16 @@ async function _checkServerStream(sessionId) {
         if (holder.parentNode) holder.remove();
         return;
       }
+      if (!_resumeRetried && window.chatModule && window.chatModule.resumeStream) {
+        _resumeRetried = true;
+        const attached = await window.chatModule.resumeStream(sessionId);
+        if (attached) {
+          clearInterval(pollId);
+          spinner.destroy();
+          if (holder.parentNode) holder.remove();
+          return;
+        }
+      }
       try {
         const r = await fetch(`${API_BASE}/api/chat/stream_status/${sessionId}`);
         if (!r.ok || (await r.json()).status !== 'streaming') {

static/js/streamingRenderer.js

@@ -0,0 +1,206 @@
+// streamingRenderer.js
+//
+// The DOM shell for incremental streaming markdown rendering. One instance owns
+// the DOM of one streaming assistant message and is the only thing that writes to
+// it while it streams.
+//
+// It keeps the message as two regions, separated by an invisible comment marker so
+// the rendered blocks are direct children of the container (no wrapper elements to
+// disturb CSS):
+//
+//     [ finalized block, frozen ][ finalized block, frozen ] <!--tail--> [ live tail ]
+//
+//   - Finalized blocks are rendered once and never touched again — so code-block
+//     hover buttons can't flicker and code is highlighted exactly once.
+//   - The live tail (the still-growing trailing block) is re-rendered each token,
+//     except an open code fence, which streams in append-mode (text appended to a
+//     stable <pre>, highlighted once when it closes).
+//
+// All the "is this safe to freeze?" logic lives in the pure segmenter; this file
+// is deliberately mechanical. If anything throws, it latches into a full-re-render
+// fallback so a bug can never produce broken output — only today's behavior.
+
+import { splitFinalized, describeOpenFence } from './streamingSegmenter.js';
+
+// Compile-time escape hatch: set to false to force the plain full-re-render path.
+// (The per-instance try/catch `degraded` fallback below is the runtime safety net.)
+const ENABLED = true;
+
+export function createStreamRenderer(contentEl, { render, hljs } = {}) {
+  let started = false;
+  let tailMarker = null; // finalized nodes precede it; live-tail nodes follow it
+  let committedLen = 0; // chars of source already frozen
+  let lastText = ''; // most recent full text (for finalize)
+  let tailShownLen = 0; // rendered-text length of the live tail (drives token fade)
+  let appendMode = null; // { codeText: Text, appendedLen } while an open fence streams
+  let degraded = !ENABLED; // true once we fall back to full re-render
+
+  function start() {
+    contentEl.textContent = '';
+    tailMarker = document.createComment('tail');
+    contentEl.appendChild(tailMarker);
+    started = true;
+  }
+
+  function highlight(root) {
+    if (hljs) root.querySelectorAll('pre code').forEach((b) => hljs.highlightElement(b));
+  }
+
+  function clearTail() {
+    while (tailMarker.nextSibling) tailMarker.nextSibling.remove();
+  }
+
+  // Render `src` and freeze the nodes before the tail marker. Highlighting happens
+  // here, once, on the detached fragment before the nodes are ever shown.
+  function freeze(src) {
+    const holder = document.createElement('div');
+    holder.innerHTML = render(src);
+    highlight(holder);
+    while (holder.firstChild) contentEl.insertBefore(holder.firstChild, tailMarker);
+  }
+
+  // Re-render the live tail. An open trailing fence streams in append-mode.
+  function renderTail(tailText) {
+    const fence = tailText ? describeOpenFence(tailText) : null;
+    if (fence) {
+      appendOpenFence(tailText, fence);
+      return;
+    }
+    appendMode = null;
+    clearTail();
+    if (!tailText) {
+      tailShownLen = 0;
+      return;
+    }
+    const holder = document.createElement('div');
+    holder.innerHTML = render(tailText);
+    fadeNewText(holder, tailShownLen);
+    tailShownLen = holder.textContent.length;
+    while (holder.firstChild) contentEl.appendChild(holder.firstChild);
+  }
+
+  // Stream the body of an unterminated code fence by appending only the new
+  // characters to a stable <pre><code> text node — no re-parse, no re-highlight.
+  function appendOpenFence(tailText, fence) {
+    if (!appendMode) {
+      clearTail();
+      const pre = document.createElement('pre');
+      const code = document.createElement('code');
+      if (fence.lang) code.className = `language-${fence.lang}`;
+      const textNode = document.createTextNode('');
+      code.appendChild(textNode);
+      pre.appendChild(code);
+      contentEl.appendChild(pre);
+      appendMode = { codeText: textNode, appendedLen: 0 };
+      tailShownLen = 0; // code is never faded; prose after the fence fades fresh
+    }
+    const code = tailText.slice(fence.contentStart);
+    if (code.length > appendMode.appendedLen) {
+      appendMode.codeText.appendData(code.slice(appendMode.appendedLen));
+      appendMode.appendedLen = code.length;
+    }
+  }
+
+  // Wrap tail text past `prevLen` characters in <span class="token-new"> for the
+  // streaming fade-in. Skips code (<pre>) and thinking blocks (.thinking-content).
+  // Note: the original chat.js helper checked `.think-content`, a class that exists
+  // nowhere in the app, so thinking text used to fade; matching the real
+  // `.thinking-content` corrects that. Operates on the detached fragment before insertion.
+  function fadeNewText(container, prevLen) {
+    if (!prevLen) return;
+    const walker = document.createTreeWalker(container, NodeFilter.SHOW_TEXT);
+    let count = 0;
+    const toWrap = [];
+    while (walker.nextNode()) {
+      const node = walker.currentNode;
+      const len = node.textContent.length;
+      if (count + len <= prevLen) {
+        count += len;
+        continue;
+      }
+      toWrap.push({ node, splitAt: count < prevLen ? prevLen - count : 0 });
+      count += len;
+    }
+    for (const { node, splitAt } of toWrap) {
+      const parent = node.parentNode;
+      if (!parent || parent.closest('pre, .thinking-content')) continue;
+      const target = splitAt > 0 ? node.splitText(splitAt) : node;
+      const span = document.createElement('span');
+      span.className = 'token-new';
+      parent.replaceChild(span, target);
+      span.appendChild(target);
+    }
+  }
+
+  function fullRender(fullText) {
+    contentEl.innerHTML = render(fullText);
+    highlight(contentEl);
+  }
+
+  // Render the latest full source text.
+  //
+  // PRECONDITION: callers must pass append-only text — each call's `fullText` must
+  // extend the previous one with the already-seen prefix UNCHANGED. Finalized
+  // blocks are frozen and never re-rendered, so a feed that rewrites earlier text
+  // would leave stale frozen blocks (corrected only by the next full re-render).
+  // chat.js satisfies this: its stripToolBlocks output only strips not-yet-finalized
+  // trailing tool syntax, never text that has already been frozen.
+  function update(fullText) {
+    lastText = fullText;
+    if (degraded) {
+      fullRender(fullText);
+      return;
+    }
+    try {
+      // Self-heal: if our DOM was replaced out from under us — chat.js writes
+      // contentEl.innerHTML directly for thinking indicators and tool blocks, and
+      // finalize() removes the marker — our tail marker is no longer a child of the
+      // container. Rebuild from scratch so we never append onto foreign content or
+      // touch a detached marker.
+      if (started && (!tailMarker || tailMarker.parentNode !== contentEl)) {
+        started = false;
+        committedLen = 0;
+        tailShownLen = 0;
+        appendMode = null;
+      }
+      if (!started) start();
+      const next = splitFinalized(fullText, render, committedLen);
+      if (next > committedLen) {
+        freeze(fullText.slice(committedLen, next));
+        committedLen = next;
+        appendMode = null; // whatever was streaming is now frozen
+        tailShownLen = 0;
+      }
+      renderTail(fullText.slice(committedLen));
+    } catch (err) {
+      degraded = true;
+      console.error('streamingRenderer: falling back to full render', err);
+      fullRender(fullText);
+    }
+  }
+
+  // Stream finished: freeze whatever is left canonically and flatten away the
+  // marker so the container holds exactly what a single full render would produce.
+  // chat.js currently re-renders the finished message from source for its own
+  // reasons and so doesn't call this, but it completes the renderer's lifecycle and
+  // is exercised by the tests.
+  function finalize() {
+    if (degraded) return;
+    try {
+      if (!started) start();
+      clearTail();
+      appendMode = null;
+      const rest = lastText.slice(committedLen);
+      if (rest.trim()) freeze(rest);
+      tailMarker.remove();
+      tailMarker = null;
+      committedLen = lastText.length;
+    } catch (err) {
+      degraded = true;
+      console.error('streamingRenderer: falling back to full render', err);
+      fullRender(lastText);
+    }
+  }
+
+  return { update, finalize };
+}

static/js/streamingSegmenter.js

@@ -0,0 +1,190 @@
+// streamingSegmenter.js
+//
+// Pure logic for incremental ("block-at-a-time") streaming markdown rendering.
+//
+// While an assistant message streams in, re-rendering the whole accumulated
+// markdown on every token is wasteful (O(N^2)) and recreates DOM nodes, which
+// makes code-block hover buttons flicker. The fix is to FREEZE the leading part
+// of the message that can no longer change, and only re-render the growing tail.
+//
+// This module answers the one hard question that makes freezing safe:
+//
+//     Given the full markdown received so far, how many leading characters can
+//     be finalized without changing the rendered output?
+//
+// The contract callers rely on (`render` is the canonical markdown renderer):
+//
+//     const n = splitFinalized(text, render);
+//     render(text.slice(0, n)) + render(text.slice(n))  ===  render(text)
+//
+// The module is intentionally DOM-free and renderer-agnostic so it can be unit
+// tested in isolation and reused for any markdown renderer with no long-range
+// cross-block dependencies (no reference-style links / footnotes).
+//
+// Known limitations (both bounded by the same mitigation):
+//   - cutIsRenderSafe proves only PRESENT-tense equivalence. If the renderer pairs
+//     an inline delimiter across a blank line (e.g. markdown.js will turn
+//     `*a\n\nb*` into emphasis spanning two paragraphs), a block frozen before the
+//     closing delimiter arrives can disagree with the final full render.
+//   - afterClosedFence boundaries are trusted without the equivalence check, so a
+//     fence the real renderer parses differently (e.g. a stray 4-backtick line) can
+//     be mis-detected as a close.
+//   Both only occur for input the renderer itself handles oddly, and both are
+//   transient: chat.js re-renders the finished message from source, so the settled
+//   output is always canonical.
+
+// A fenced-code delimiter line: up to 3 leading spaces, then >=3 backticks or
+// tildes, then an optional info string.
+const FENCE_RE = /^ {0,3}(`{3,}|~{3,})(.*)$/;
+
+/**
+ * Scan `text` starting at `fromOffset` — which MUST be at top level (callers only
+ * ever advance to a finalized boundary, never into a fence) — and collect the
+ * candidate cut points.
+ *
+ * @returns {{ boundaries: Array<{offset:number, afterClosedFence:boolean}>, inFence:boolean }}
+ *   - A blank-line run at top level yields a boundary at the start of the next
+ *     non-blank line (`afterClosedFence: false`).
+ *   - A fence close yields a boundary just past the closing fence line
+ *     (`afterClosedFence: true`) — such a cut is unconditionally safe, since
+ *     nothing can ever merge into a completed code block.
+ */
+function findBoundaries(text, fromOffset) {
+  const boundaries = [];
+  const n = text.length;
+  let inFence = false;
+  let fenceMarker = '';
+  let i = fromOffset;
+
+  while (i < n) {
+    const nl = text.indexOf('\n', i);
+    const lineEnd = nl === -1 ? n : nl;
+    const afterNl = nl === -1 ? n : nl + 1;
+    const line = text.slice(i, lineEnd);
+    const fence = line.match(FENCE_RE);
+
+    if (fence) {
+      const marker = fence[1];
+      if (!inFence) {
+        inFence = true;
+        fenceMarker = marker;
+      } else if (
+        marker[0] === fenceMarker[0] &&
+        marker.length >= fenceMarker.length &&
+        fence[2].trim() === '' // a closing fence carries no info string
+      ) {
+        inFence = false;
+        fenceMarker = '';
+        boundaries.push({ offset: afterNl, afterClosedFence: true });
+      }
+      i = afterNl;
+    } else if (!inFence && line.trim() === '') {
+      // Consume the entire run of blank lines; the boundary is the start of the
+      // next non-blank line so the finalized side owns the separator and the tail
+      // starts clean.
+      let j = afterNl;
+      while (j < n) {
+        const nl2 = text.indexOf('\n', j);
+        const lineEnd2 = nl2 === -1 ? n : nl2;
+        if (text.slice(j, lineEnd2).trim() !== '') break;
+        if (nl2 === -1) {
+          j = n;
+          break;
+        }
+        j = nl2 + 1;
+      }
+      boundaries.push({ offset: j, afterClosedFence: false });
+      i = j;
+    } else {
+      i = afterNl;
+    }
+  }
+
+  return { boundaries, inFence };
+}
+
+/**
+ * Does cutting between `before` and `after` leave the rendered output unchanged?
+ * This is the self-verifying safety check: it directly compares rendering the two
+ * sides separately against rendering them joined, so constructs that span the cut
+ * (loose lists, setext headings, lazy blockquote continuations, tables) are caught
+ * with no hand-coded grammar rules.
+ *
+ * Renderer non-determinism (e.g. mermaid ids seeded with Date.now()) can only make
+ * this return a false negative, never a false positive — so the bias is always
+ * toward under-finalizing, which is the safe direction.
+ */
+function cutIsRenderSafe(before, after, render) {
+  return render(before) + render(after) === render(before + after);
+}
+
+/**
+ * Return how many leading characters of `text` can be safely finalized, scanning
+ * forward from `committedLen` (the amount already finalized).
+ *
+ * Guarantees `render(text.slice(0, n)) + render(text.slice(n)) === render(text)`,
+ * and `committedLen <= n <= text.length`.
+ *
+ * @param {string} text       Full markdown accumulated so far.
+ * @param {(src:string)=>string} render  Canonical markdown renderer.
+ * @param {number} [committedLen=0]  Characters already finalized (always a prior boundary).
+ * @returns {number}
+ */
+export function splitFinalized(text, render, committedLen = 0) {
+  const { boundaries } = findBoundaries(text, committedLen);
+
+  let best = committedLen;
+  let segStart = committedLen;
+
+  for (let k = 0; k < boundaries.length; k++) {
+    const { offset, afterClosedFence } = boundaries[k];
+
+    if (afterClosedFence) {
+      // A completed code block — always safe to freeze through here.
+      best = offset;
+    } else {
+      // A prose/list/table boundary. We need a following block to compare
+      // against (the last block must stay live, it can still grow), and the cut
+      // must be render-equivalent locally.
+      const nextOffset = k + 1 < boundaries.length ? boundaries[k + 1].offset : text.length;
+      const before = text.slice(segStart, offset);
+      const after = text.slice(offset, nextOffset);
+      if (after.trim() !== '' && cutIsRenderSafe(before, after, render)) {
+        best = offset;
+      }
+    }
+    segStart = offset;
+  }
+
+  return best;
+}
+
+/**
+ * If `text` begins with a fenced-code opener whose fence never closes, describe it
+ * so the renderer can stream the code in append-mode instead of re-rendering it.
+ * Returns `{ lang, contentStart }` (contentStart = offset of the first code char),
+ * or null when `text` does not start with a still-open fence.
+ *
+ * The opener line must be complete (terminated by a newline) so the info string /
+ * language is known before append-mode begins.
+ */
+export function describeOpenFence(text) {
+  const open = text.match(/^( {0,3})(`{3,}|~{3,})([^\n]*)\n/);
+  if (!open) return null;
+  const marker = open[2];
+  const contentStart = open[0].length;
+
+  for (let i = contentStart; i < text.length; ) {
+    const nl = text.indexOf('\n', i);
+    const line = text.slice(i, nl === -1 ? text.length : nl);
+    const close = line.match(/^ {0,3}(`{3,}|~{3,})\s*$/);
+    if (close && close[1][0] === marker[0] && close[1].length >= marker.length) {
+      return null; // the fence closes — let the normal finalize path handle it
+    }
+    if (nl === -1) break;
+    i = nl + 1;
+  }
+
+  const lang = (open[3] || '').trim().split(/\s+/)[0] || '';
+  return { lang, contentStart };
+}

static/style.css

@@ -842,7 +842,7 @@ body.bg-pattern-sparkles {
       display: flex; gap: 6px; flex-wrap: wrap;
       max-width: calc(100vw - 24px);
       padding: 4px;
-      z-index: 10020;
+      z-index: 100;
       pointer-events: none;
     }
     .minimized-dock-chip {

tests/streaming/corpus.mjs

@@ -0,0 +1,27 @@
+// A spread of markdown samples exercising the constructs the renderer supports.
+// Used by the streaming-invariant fuzz test (fed token-by-token) and the renderer
+// integration test. Keep samples small but structurally varied — the fuzz test
+// runs every prefix of every sample, so length is quadratic on cost.
+export const CORPUS = [
+  ['plain paragraph', 'Just a single sentence of text.'],
+  ['two paragraphs', 'First paragraph here.\n\nSecond paragraph here.'],
+  ['three paragraphs', 'Alpha block.\n\nBravo block.\n\nCharlie block.'],
+  ['atx headings', '# Title\n\nIntro line.\n\n## Section\n\nBody text.'],
+  ['setext heading', 'The Title\n=========\n\nA paragraph under it.'],
+  ['inline formatting', 'Some **bold**, *italic*, `code`, and a [link](https://x.com).'],
+  ['tight unordered list', '- one\n- two\n- three\n\ndone'],
+  ['ordered list then text', 'Before\n\n1. first\n2. second\n3. third\n\nAfter'],
+  ['loose list then paragraph', '- a\n\n- b\n\n- c\n\nClosing paragraph.'],
+  ['nested list', '- top\n  - nested one\n  - nested two\n- back to top\n\nend'],
+  ['blockquote', '> quoted line one\n> quoted line two\n\nplain after'],
+  ['thematic break', 'above the line\n\n---\n\nbelow the line'],
+  ['python code fence', 'Run this:\n\n```python\nprint("hi")\nfor i in range(3):\n    print(i)\n```\n\nThat prints numbers.'],
+  ['fence with blank lines inside', '```js\nconst a = 1;\n\nconst b = 2;\n```\n\nafter the code'],
+  ['two consecutive fences', '```\nfirst block\n```\n\n```\nsecond block\n```\n\ntail'],
+  ['mermaid diagram', 'Diagram:\n\n```mermaid\ngraph TD\nA-->B\n```\n\nafter diagram'],
+  ['gfm table', 'Data:\n\n| A | B |\n|---|---|\n| 1 | 2 |\n| 3 | 4 |\n\nafter table'],
+  [
+    'mixed document',
+    '# Report\n\nIntro paragraph with a `symbol`.\n\n```python\nx = 1\n```\n\n- bullet one\n- bullet two\n\n> a quote\n\nFinal words.',
+  ],
+];

tests/streaming/invariant.test.mjs

@@ -0,0 +1,107 @@
+// The centerpiece correctness test: stream every corpus sample in token-by-token,
+// driving the segmenter exactly as the renderer will, and assert the freeze/tail
+// split stays render-equivalent to a single full render at EVERY step.
+//
+//   finalized-html (accumulated from committed deltas) + render(live tail)  ===  render(prefix)
+//
+// This is run with no DOM and no safety net, so any segmenter bug fails here
+// rather than reaching the browser.
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { loadMarkdown, normalizeRender } from './markdownHarness.mjs';
+import { splitFinalized } from '../../static/js/streamingSegmenter.js';
+import { CORPUS } from './corpus.mjs';
+
+const md = await loadMarkdown();
+const render = (t) => md.mdToHtml(t);
+
+// The two render pipelines chat.js actually feeds streamed text through. BOTH wrap
+// the source in squashOutsideCode; the main path additionally runs
+// processWithThinking (which floats <think> blocks to the top — a non-local
+// transform). Fuzzing the corpus through these — not just bare mdToHtml — closes
+// the gap where a squashOutsideCode whitespace/fence edge could break the split.
+const renderLiveReply = (t) => md.mdToHtml(md.squashOutsideCode(t)); // chat.js live-reply path
+const renderMain = (t) => md.processWithThinking(md.squashOutsideCode(t)); // chat.js main path
+
+// Reproduce the renderer's exact use of the segmenter over a sequence of prefixes.
+function simulate(text, prefixLengths, renderFn = render) {
+  let committed = 0;
+  let finalizedHtml = '';
+  for (const len of prefixLengths) {
+    const prefix = text.slice(0, len);
+    const next = splitFinalized(prefix, renderFn, committed);
+
+    assert.ok(
+      next >= committed && next <= prefix.length,
+      `committed must stay monotonic and in range (${committed} -> ${next} at length ${len})`,
+    );
+    if (next > committed) {
+      // The renderer renders each finalized delta once and never touches it again.
+      finalizedHtml += renderFn(prefix.slice(committed, next));
+      committed = next;
+    }
+
+    const got = normalizeRender(finalizedHtml + renderFn(prefix.slice(committed)));
+    const want = normalizeRender(renderFn(prefix));
+    assert.equal(got, want, `invariant broke at prefix length ${len} of ${JSON.stringify(text)}`);
+  }
+}
+
+const everyPrefix = (t) => Array.from({ length: t.length + 1 }, (_, i) => i);
+function chunkAtWhitespace(t) {
+  const lens = [];
+  for (let i = 1; i <= t.length; i++) {
+    if (i === t.length || /\s/.test(t[i - 1])) lens.push(i);
+  }
+  return lens.length ? lens : [t.length];
+}
+
+const RENDERERS = [
+  ['mdToHtml', render],
+  ['mdToHtml∘squashOutsideCode (live-reply path)', renderLiveReply],
+  ['processWithThinking∘squashOutsideCode (main path)', renderMain],
+];
+
+for (const [rname, renderFn] of RENDERERS) {
+  for (const [name, text] of CORPUS) {
+    test(`invariant — ${rname} — char-by-char — ${name}`, () => {
+      simulate(text, everyPrefix(text), renderFn);
+    });
+    test(`invariant — ${rname} — whitespace-chunked — ${name}`, () => {
+      simulate(text, chunkAtWhitespace(text), renderFn);
+    });
+  }
+}
+
+// These samples carry <think> blocks (the corpus above is think-free), so they
+// specifically exercise the self-verifying local check refusing to finalize inside
+// or across a think block that processWithThinking floats to the top.
+const THINKING_CORPUS = [
+  ['leading think then answer', '<think>Let me reason about it.</think>\n\nThe answer is 42.'],
+  ['think with internal blank lines', '<think>Step one.\n\nStep two.\n\nStep three.</think>\n\nDone — the result follows.'],
+  ['think then several paragraphs', '<thinking>analyzing the request</thinking>\n\nFirst point made here.\n\nSecond point made here.\n\nThird and final point.'],
+  ['think then code block', '<think>I should show code.</think>\n\nHere:\n\n```python\nprint("hi")\n```\n\nThat is the snippet.'],
+];
+for (const [name, text] of THINKING_CORPUS) {
+  test(`invariant (processWithThinking) — char-by-char — ${name}`, () => {
+    simulate(text, everyPrefix(text), renderMain);
+  });
+}
+
+// A final-output check independent of chunking: streaming to completion must equal
+// a single full render.
+test('streamed-to-completion output equals full render for whole corpus', () => {
+  for (const [name, text] of CORPUS) {
+    let committed = 0;
+    let html = '';
+    for (let len = 1; len <= text.length; len++) {
+      const next = splitFinalized(text.slice(0, len), render, committed);
+      if (next > committed) {
+        html += render(text.slice(committed, next));
+        committed = next;
+      }
+    }
+    html += render(text.slice(committed));
+    assert.equal(normalizeRender(html), normalizeRender(render(text)), `final mismatch for ${name}`);
+  }
+});

tests/streaming/markdownHarness.mjs

@@ -0,0 +1,66 @@
+// Loads the real browser markdown renderer (static/js/markdown.js) under Node by
+// mocking the minimal browser globals it touches and stubbing its sibling imports.
+// This mirrors the loader in tests/test_markdown_rendering_js.py so the streaming
+// tests exercise the exact same renderer the browser runs.
+import fs from 'node:fs';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const REPO = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..', '..');
+
+export async function loadMarkdown() {
+  globalThis.window = { location: { origin: 'http://localhost' }, katex: null };
+  globalThis.document = {
+    readyState: 'loading',
+    addEventListener() {},
+    createElement(tag) {
+      if (tag !== 'template') throw new Error(`unsupported element: ${tag}`);
+      return {
+        _html: '',
+        content: { querySelectorAll() { return []; } },
+        set innerHTML(v) { this._html = v; },
+        get innerHTML() { return this._html; },
+      };
+    },
+  };
+  globalThis.MutationObserver = class { observe() {} };
+
+  let src = fs.readFileSync(path.join(REPO, 'static/js/markdown.js'), 'utf8');
+  src = src.replace(/import uiModule from ['"]\.\/ui\.js['"];/, '');
+  src = src.replace(
+    /import \{ splitTableRow \} from ['"]\.\/markdown\/tableRow\.js['"];/,
+    () => `function splitTableRow(row){return (row||'').replace(/^\\s*\\|/,'').replace(/\\|\\s*$/,'').split('|').map((c)=>c.trim());}`,
+  );
+  const emoji = fs
+    .readFileSync(path.join(REPO, 'static/js/emojiShortcodes.js'), 'utf8')
+    .replace(/^export default .*$/m, '')
+    .replace(/export const /g, 'const ')
+    .replace(/export function /g, 'function ');
+  src = src.replace(
+    /import \{ replaceEmojiShortcodes, hasEmojiShortcode \} from ['"]\.\/emojiShortcodes\.js['"];/,
+    () => emoji,
+  );
+  src = src.replace(
+    /var escapeHtml = uiModule\.esc;/,
+    () =>
+      `var escapeHtml = (v) => String(v ?? '').replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;').replace(/"/g,'&quot;').replace(/'/g,'&#39;');`,
+  );
+  const url = 'data:text/javascript;base64,' + Buffer.from(src).toString('base64');
+  return import(url);
+}
+
+// Canonicalize rendered HTML so two renders that produce the SAME DOM compare
+// equal. Collapses only newline-bearing whitespace BETWEEN tags (`>\n\n<` ->
+// `><`): it is insignificant in rendered HTML, and incremental finalization
+// legitimately emits `\n\n` between two blocks where a single full render emits
+// `\n`. Code whitespace is safe because code is HTML-escaped, so significant
+// newlines live inside <code> as text (never between a `>` and a `<`). Inline
+// single spaces between tags are left alone. Structural differences (two <ul> vs
+// one, <ol> vs <ul>) survive normalization and still fail, as they must.
+// Mermaid ids embed Date.now(), so they are normalized too.
+export function normalizeRender(html) {
+  return String(html)
+    .replace(/>\s*\n\s*</g, '><')
+    .trim()
+    .replace(/(mermaid|thinking)-\d+-\d+/g, '$1-X');
+}

tests/streaming/segmenter.test.mjs

@@ -0,0 +1,65 @@
+// Tests for the pure streaming-markdown segmenter.
+//
+// The segmenter's one job: given the full accumulated markdown text so far,
+// report how many leading characters are SAFE to finalize — i.e. freeze and
+// never re-render. "Safe" means: rendering the finalized prefix and the live
+// tail separately produces the same DOM as rendering the whole text at once.
+//
+// Invariant under test everywhere:  render(text[0:n]) + render(text[n:]) === render(text)
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { loadMarkdown, normalizeRender } from './markdownHarness.mjs';
+import { splitFinalized } from '../../static/js/streamingSegmenter.js';
+
+const md = await loadMarkdown();
+const render = (t) => md.mdToHtml(t);
+const splitOk = (text, n) =>
+  normalizeRender(render(text.slice(0, n)) + render(text.slice(n))) === normalizeRender(render(text));
+
+test('harness loads the real renderer', () => {
+  assert.match(render('hi'), /<p>hi<\/p>/);
+});
+
+test('nothing is finalized while a single block is still streaming', () => {
+  assert.equal(splitFinalized('an incomplete paragra', render), 0);
+});
+
+test('finalizes the first of two blank-line-separated paragraphs', () => {
+  const text = 'para one\n\npara two';
+  const n = splitFinalized(text, render);
+  assert.equal(n, 'para one\n\n'.length);
+  assert.ok(splitOk(text, n), 'split must be render-equivalent');
+});
+
+test('never finalizes the last (still-growing) block', () => {
+  // The trailing paragraph could still gain more characters, so it stays live.
+  const text = 'done\n\nstill going';
+  const n = splitFinalized(text, render);
+  assert.ok(n <= 'done\n\n'.length);
+  assert.ok(splitOk(text, n));
+});
+
+test('a closed code fence is finalized immediately, even as the last block', () => {
+  // This is the original flicker scenario: a completed code block must freeze
+  // so its hover buttons stop being recreated on every later token.
+  const text = 'Here:\n\n```python\nprint(1)\n```';
+  const n = splitFinalized(text, render);
+  assert.ok(n >= text.length - 1, `expected the whole closed fence finalized, got ${n} of ${text.length}`);
+  assert.ok(splitOk(text, n));
+});
+
+test('does NOT finalize across an OPEN code fence', () => {
+  const text = 'intro\n\n```python\nprint(1)\nprint(2)';
+  const n = splitFinalized(text, render);
+  // "intro" may finalize, but nothing inside the still-open fence may.
+  assert.ok(n <= 'intro\n\n'.length, `must not finalize into an open fence, got ${n}`);
+  assert.ok(splitOk(text, n));
+});
+
+test('does NOT split a loose list (blank line between items is not a boundary)', () => {
+  const text = '- a\n\n- b\n\nafter';
+  const n = splitFinalized(text, render);
+  assert.ok(splitOk(text, n), 'a wrong split here would turn one <ul> into two');
+  // The list must not be cut in the middle: either nothing or the whole list.
+  assert.ok(n === 0 || n >= '- a\n\n- b\n\n'.length, `loose list was cut at ${n}`);
+});

tests/test_calendar_list_range_aliases.py

@@ -0,0 +1,80 @@
+"""manage_calendar list_events should honor common range aliases.
+
+The agent prompt and schema prefer start/end, but model calls can emit
+start_date/end_date or from/to. Those aliases used to be ignored, causing the
+tool to fall back to its default 14-day window.
+"""
+
+import json
+import sys
+import tempfile
+import uuid
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
+
+from tests.helpers.import_state import clear_fake_database_modules
+
+clear_fake_database_modules()
+
+import core.database as cdb
+
+_TMPDB = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_ENGINE = create_engine(
+    f"sqlite:///{_TMPDB.name}",
+    connect_args={"check_same_thread": False},
+    poolclass=NullPool,
+)
+cdb.Base.metadata.create_all(_ENGINE)
+_TS = sessionmaker(bind=_ENGINE, autoflush=False, autocommit=False)
+
+
+@pytest.fixture(autouse=True)
+def _bind_temp_db(monkeypatch):
+    monkeypatch.setitem(sys.modules, "core.database", cdb)
+    parent = sys.modules.get("core")
+    if parent is not None:
+        monkeypatch.setattr(parent, "database", cdb, raising=False)
+    monkeypatch.setattr(cdb, "SessionLocal", _TS)
+    yield
+
+
+@pytest.mark.parametrize(
+    ("start_key", "end_key"),
+    [
+        ("start_date", "end_date"),
+        ("from", "to"),
+        ("range_start", "range_end"),
+    ],
+)
+async def test_list_events_honors_range_aliases(start_key, end_key):
+    from src.tool_implementations import do_manage_calendar
+
+    owner = "calendar-alias-" + uuid.uuid4().hex[:8]
+
+    inside = await do_manage_calendar(json.dumps({
+        "action": "create_event",
+        "summary": "Late June planning",
+        "dtstart": "2126-06-25T10:00:00Z",
+    }), owner=owner)
+    assert inside.get("exit_code", 0) == 0, inside
+
+    outside = await do_manage_calendar(json.dumps({
+        "action": "create_event",
+        "summary": "Outside July planning",
+        "dtstart": "2126-07-10T10:00:00Z",
+    }), owner=owner)
+    assert outside.get("exit_code", 0) == 0, outside
+
+    res = await do_manage_calendar(json.dumps({
+        "action": "list_events",
+        start_key: "2126-06-01T00:00:00Z",
+        end_key: "2126-07-01T00:00:00Z",
+    }), owner=owner)
+
+    assert res.get("exit_code", 0) == 0, res
+    summaries = [event["summary"] for event in res["events"]]
+    assert summaries == ["Late June planning"]
+    assert "between 2126-06-01 and 2126-07-01" in res["response"]

tests/test_fenced_invoke_no_raw_xml.py

@@ -34,3 +34,39 @@ def test_plain_fenced_python_block_still_parses_as_code():
     # No regression: an ordinary fenced python block (no <invoke>) still works.
     blocks = parse_tool_blocks('```python\nprint("hi")\n```')
     assert any(b.tool_type == "python" and 'print("hi")' in b.content for b in blocks), blocks
+
+
+def test_simple_web_search_call_inside_python_fence_runs_as_web_search():
+    blocks = parse_tool_blocks('```python\nweb_search("latest Python release")\n```')
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert blocks[0].content == "latest Python release"
+
+
+def test_google_search_alias_inside_bash_fence_preserves_freshness_args():
+    blocks = parse_tool_blocks(
+        '```bash\ngoogle_search(query="Qwen latest release", freshness="week", max_pages=7)\n```'
+    )
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "web_search"
+    assert '"query": "Qwen latest release"' in blocks[0].content
+    assert '"freshness": "week"' in blocks[0].content
+    assert '"max_pages": 7' in blocks[0].content
+
+
+def test_nontrivial_python_with_web_search_name_stays_python_code():
+    blocks = parse_tool_blocks('```python\nprint(web_search("latest Python release"))\n```')
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "python"
+
+
+def test_plain_search_function_inside_python_fence_stays_python_code():
+    blocks = parse_tool_blocks('```python\nsearch("private customer name")\n```')
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "python"
+
+
+def test_plain_fetch_function_inside_python_fence_stays_python_code():
+    blocks = parse_tool_blocks('```python\nfetch("internal-url")\n```')
+    assert len(blocks) == 1
+    assert blocks[0].tool_type == "python"

tests/test_streaming_segmenter_js.py

@@ -0,0 +1,38 @@
+"""Runs the Node-based streaming-render segmenter suite (tests/streaming/*.test.mjs).
+
+Covers the pure incremental-render segmenter (static/js/streamingSegmenter.js):
+unit boundaries plus a streaming-invariant fuzz that feeds a markdown corpus in
+token-by-token and asserts the freeze/tail split always matches a single full
+render. Pure JS — no DOM, no extra dependencies. Skipped when node is
+unavailable, mirroring tests/test_markdown_rendering_js.py.
+
+The renderer's DOM behavior (streamingRenderer.js) is exercised against a running
+app, not here, consistent with how this project tests browser-coupled code.
+"""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+_REPO = Path(__file__).resolve().parent.parent
+_HAS_NODE = shutil.which("node") is not None
+
+
+@pytest.mark.skipif(not _HAS_NODE, reason="node binary not on PATH")
+def test_streaming_segmenter_suite():
+    test_files = sorted(str(p) for p in (_REPO / "tests" / "streaming").glob("*.test.mjs"))
+    assert test_files, "no streaming test files found"
+
+    result = subprocess.run(
+        ["node", "--test", *test_files],
+        cwd=_REPO,
+        capture_output=True,
+        timeout=180,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise AssertionError(
+            f"node --test failed:\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
+        )