Merge branch 'main' into main

2026-08-29 04:26:20 +00:00 · 2026-04-19 20:10:22 +05:30
parent 983f5301e8 5a852169be
commit f6198039fa
9 changed files with 729 additions and 88 deletions
@@ -111,5 +111,9 @@ sample_data/
 # Test Results
 test_results.txt

+# Frontend workspace artifacts
+semantica-explorer/
+node_modules/
+
 # Frontend build artifacts (generated by Vite — do not track in git)
 semantica/static/
@@ -19,7 +19,12 @@ from .ws import ConnectionManager


 def _install_mutation_bridge(app: FastAPI, session: GraphSession) -> None:
+    previous_callback = getattr(session.graph, "mutation_callback", None)
+
    def on_mutation(event_type: str, entity_id: str, payload: dict) -> None:
+        session.handle_graph_mutation(event_type, entity_id, payload)
+        if callable(previous_callback):
+            previous_callback(event_type, entity_id, payload)
        loop = getattr(app.state, "event_loop", None)
        manager = getattr(app.state, "ws_manager", None)
        if loop is None or manager is None or loop.is_closed():
@@ -150,6 +155,17 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI:
            "status": "active",
        }

+    @app.get("/", include_in_schema=False)
+    async def root():
+        index_path = Path(__file__).resolve().parent.parent / "static" / "index.html"
+        if index_path.is_file():
+            return FileResponse(index_path)
+        return HTMLResponse(
+            '<!doctype html><html lang="en"><head><meta charset="UTF-8">'
+            '<title>Semantica Knowledge Explorer</title></head>'
+            '<body><div id="root"></div></body></html>'
+        )
+
    static_dir = Path(__file__).resolve().parent.parent / "static"
    if static_dir.is_dir():
        assets_dir = static_dir / "assets"
@@ -136,11 +136,11 @@ def _apply_inferred_edges(
            continue
        source, target = args
        if session.get_node(source) is None:
-            session.graph.add_node(source, "entity", content=source)
+            session.add_node(source, "entity", content=source)
        if session.get_node(target) is None:
-            session.graph.add_node(target, "entity", content=target)
+            session.add_node(target, "entity", content=target)
        edge_type = body.inferred_edge_type or predicate
-        session.graph.add_edge(
+        session.add_edge(
            source,
            target,
            edge_type=edge_type,
@@ -354,4 +354,6 @@ async def merge_nodes(
        return removed, edges_updated

    removed_ids, edges_updated = await asyncio.to_thread(_do_merge)
+    if removed_ids:
+        await asyncio.to_thread(session.rebuild_search_index)
    return MergeResponse(merged_into=primary_id, removed_ids=removed_ids, edges_updated=edges_updated)
@@ -1,4 +1,4 @@
-"""
+"""
 Provenance routes for lineage visualization and exportable reports.
 """

@@ -9,33 +9,13 @@ from typing import Any, Dict, List, Optional
 import networkx as nx
 from fastapi import APIRouter, Depends, Query
 from fastapi.responses import PlainTextResponse, Response
-from pydantic import BaseModel

 from ..dependencies import get_session
+from ..schemas import ProvenanceEdge, ProvenanceNode, ProvenanceResponse
 from ..session import GraphSession

 router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"])

-
-class ProvenanceNode(BaseModel):
-    id: str
-    label: str
-    prov_type: str
-    parent_id: str
-
-
-class ProvenanceEdge(BaseModel):
-    id: str
-    source: str
-    target: str
-    label: str
-
-
-class ProvenanceResponse(BaseModel):
-    nodes: List[ProvenanceNode]
-    edges: List[ProvenanceEdge]
-
-
 _AGENT_TYPES = {"person", "organization", "system", "agent"}
 _ACTIVITY_TYPES = {"action", "event", "process", "activity", "decision", "publication"}

@@ -67,7 +47,7 @@ def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> d
        if edge.source_id in hop_nodes or edge.target_id in hop_nodes:
            graph.add_edge(edge.source_id, edge.target_id, label=edge.edge_type)

-    subgraph = nx.ego_graph(graph, node_id, radius=2, undirected=False)
+    subgraph = nx.ego_graph(graph, node_id, radius=2, undirected=True)
    provenance_nodes: List[Dict[str, Any]] = []
    for graph_node_id in subgraph.nodes():
        node = session.graph.nodes.get(graph_node_id)
@@ -85,12 +65,19 @@ def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> d

    provenance_edges: List[Dict[str, Any]] = []
    for source, target, data in subgraph.edges(data=True):
+        if target == node_id:
+            direction = "upstream"
+        elif source == node_id:
+            direction = "downstream"
+        else:
+            direction = "lateral"
        provenance_edges.append(
            {
                "id": f"{source}-{target}",
                "source": source,
                "target": target,
                "label": data.get("label", "related_to"),
+                "direction": direction,
            }
        )

@@ -104,7 +91,7 @@ def _build_report(session: GraphSession, node_id: str) -> Dict[str, Any]:
        "node_id": node_id,
        "label": node.get("content", node_id) if node else node_id,
        "type": node.get("type", "entity") if node else "entity",
-        "properties": node.get("properties", {}) if node else {},
+        "properties": node.get("metadata", node.get("properties", {})) if node else {},
        "lineage": provenance,
    }

@@ -129,9 +116,29 @@ def _render_markdown(report: Dict[str, Any]) -> str:
    for node in report.get("lineage", {}).get("nodes", []):
        lines.append(f"- `{node['id']}` ({node['prov_type']}): {node['label']}")

-    lines.extend(["", "## Lineage Edges"])
-    for edge in report.get("lineage", {}).get("edges", []):
-        lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+    edges = report.get("lineage", {}).get("edges", [])
+    grouped_edges: Dict[str, List] = {"upstream": [], "downstream": [], "lateral": []}
+    for edge in edges:
+        direction = edge.get("direction", "lateral")
+        if direction not in grouped_edges:
+            direction = "lateral"
+        grouped_edges[direction].append(edge)
+
+    if grouped_edges["upstream"]:
+        lines.extend(["", "## Upstream"])
+        for edge in grouped_edges["upstream"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
+    if grouped_edges["downstream"]:
+        lines.extend(["", "## Downstream"])
+        for edge in grouped_edges["downstream"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
+    if grouped_edges["lateral"]:
+        lines.extend(["", "## Lateral"])
+        for edge in grouped_edges["lateral"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
    return "\n".join(lines)


@@ -288,3 +288,23 @@ class MergeResponse(BaseModel):
    merged_into: str
    removed_ids: List[str]
    edges_updated: int
+
+
+class ProvenanceNode(BaseModel):
+    id: str
+    label: str
+    prov_type: str
+    parent_id: Optional[str] = None
+
+
+class ProvenanceEdge(BaseModel):
+    id: str
+    source: str
+    target: str
+    label: str
+    direction: str
+
+
+class ProvenanceResponse(BaseModel):
+    nodes: List[ProvenanceNode]
+    edges: List[ProvenanceEdge]
@@ -0,0 +1,398 @@
+"""
+Explorer-local in-memory node search index.
+"""
+
+from __future__ import annotations
+
+import bisect
+import heapq
+import re
+from collections import OrderedDict, defaultdict
+from dataclasses import dataclass
+from typing import Any, DefaultDict, Dict, Iterable, List, Optional, Tuple
+
+_TOKEN_RE = re.compile(r"[a-z0-9]+")
+_WHITESPACE_RE = re.compile(r"\s+")
+_CURATED_ALIAS_KEYS = (
+    "label",
+    "name",
+    "title",
+    "pref_label",
+    "preferred_label",
+    "prefLabel",
+    "aliases",
+    "alias",
+    "synonyms",
+    "synonym",
+    "symbol",
+    "display_name",
+    "displayName",
+    "text",
+    "content",
+)
+
+
+def _normalize_text(value: Any) -> str:
+    if value is None:
+        return ""
+    text = str(value).strip().lower()
+    if not text:
+        return ""
+    return _WHITESPACE_RE.sub(" ", text)
+
+
+def _tokenize(text: str) -> Tuple[str, ...]:
+    if not text:
+        return ()
+    return tuple(dict.fromkeys(_TOKEN_RE.findall(text)))
+
+
+def _collect_text_fragments(value: Any, fragments: List[str], *, limit: int = 64) -> None:
+    if value is None or len(fragments) >= limit:
+        return
+    if isinstance(value, dict):
+        for nested in value.values():
+            _collect_text_fragments(nested, fragments, limit=limit)
+            if len(fragments) >= limit:
+                return
+        return
+    if isinstance(value, (list, tuple, set)):
+        for nested in value:
+            _collect_text_fragments(nested, fragments, limit=limit)
+            if len(fragments) >= limit:
+                return
+        return
+
+    normalized = _normalize_text(value)
+    if normalized:
+        fragments.append(normalized)
+
+
+def _coerce_float(value: Any) -> Optional[float]:
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+@dataclass(frozen=True)
+class IndexedNodeDocument:
+    node_id: str
+    normalized_id: str
+    node_type: str
+    exact_terms: frozenset[str]
+    tokens: frozenset[str]
+    primary_text: str
+    secondary_text: str
+    confidence: Optional[float]
+    tags: Tuple[str, ...]
+
+
+class GraphSearchIndex:
+    def __init__(
+        self,
+        *,
+        cache_size: int = 128,
+        prefix_min_length: int = 2,
+        prefix_max_length: int = 12,
+        secondary_scan_limit: int = 12000,
+    ) -> None:
+        self.cache_size = cache_size
+        self.prefix_min_length = prefix_min_length
+        self.prefix_max_length = prefix_max_length
+        self.secondary_scan_limit = secondary_scan_limit
+        self._documents: Dict[str, IndexedNodeDocument] = {}
+        self._exact_index: DefaultDict[str, set[str]] = defaultdict(set)
+        self._token_index: DefaultDict[str, set[str]] = defaultdict(set)
+        self._prefix_index: DefaultDict[str, set[str]] = defaultdict(set)
+        self._ordered_node_ids: List[str] = []
+        self._cache: OrderedDict[Tuple[Any, ...], List[Tuple[str, float]]] = OrderedDict()
+
+    def rebuild(self, nodes: Iterable[Dict[str, Any]]) -> None:
+        self._documents.clear()
+        self._exact_index.clear()
+        self._token_index.clear()
+        self._prefix_index.clear()
+        self._ordered_node_ids = []
+        self.clear_cache()
+
+        for node in nodes:
+            self.upsert(node, clear_cache=False)
+
+        self._ordered_node_ids.sort()
+
+    def clear_cache(self) -> None:
+        self._cache.clear()
+
+    def remove(self, node_id: str, *, clear_cache: bool = True) -> None:
+        existing = self._documents.pop(node_id, None)
+        if existing is None:
+            return
+
+        for term in existing.exact_terms:
+            bucket = self._exact_index.get(term)
+            if bucket is None:
+                continue
+            bucket.discard(node_id)
+            if not bucket:
+                self._exact_index.pop(term, None)
+
+        for token in existing.tokens:
+            bucket = self._token_index.get(token)
+            if bucket is None:
+                continue
+            bucket.discard(node_id)
+            if not bucket:
+                self._token_index.pop(token, None)
+
+            for length in range(self.prefix_min_length, min(len(token), self.prefix_max_length) + 1):
+                prefix = token[:length]
+                prefix_bucket = self._prefix_index.get(prefix)
+                if prefix_bucket is None:
+                    continue
+                prefix_bucket.discard(node_id)
+                if not prefix_bucket:
+                    self._prefix_index.pop(prefix, None)
+
+        pos = bisect.bisect_left(self._ordered_node_ids, node_id)
+        if pos < len(self._ordered_node_ids) and self._ordered_node_ids[pos] == node_id:
+            self._ordered_node_ids.pop(pos)
+
+        if clear_cache:
+            self.clear_cache()
+
+    def upsert(self, node: Dict[str, Any], *, clear_cache: bool = True) -> None:
+        node_id = str(node.get("id", "")).strip()
+        if not node_id:
+            return
+
+        self.remove(node_id, clear_cache=False)
+        document = self._build_document(node)
+        self._documents[node_id] = document
+
+        for term in document.exact_terms:
+            self._exact_index[term].add(node_id)
+
+        for token in document.tokens:
+            self._token_index[token].add(node_id)
+            for length in range(self.prefix_min_length, min(len(token), self.prefix_max_length) + 1):
+                self._prefix_index[token[:length]].add(node_id)
+
+        bisect.insort(self._ordered_node_ids, node_id)
+
+        if clear_cache:
+            self.clear_cache()
+
+    def search(
+        self,
+        query: str,
+        *,
+        limit: int = 20,
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> tuple[List[Tuple[str, float]], Dict[str, Any]]:
+        normalized_query = _normalize_text(query)
+        filters = filters or {}
+        diagnostics: Dict[str, Any] = {
+            "cache_hit": False,
+            "path": "empty",
+            "candidates": 0,
+        }
+        if not normalized_query:
+            return [], diagnostics
+
+        cache_key = self._cache_key(normalized_query, limit, filters)
+        cached = self._cache.get(cache_key)
+        if cached is not None:
+            self._cache.move_to_end(cache_key)
+            diagnostics.update({"cache_hit": True, "path": "cache", "candidates": len(cached)})
+            return list(cached), diagnostics
+
+        query_tokens = _tokenize(normalized_query)
+        exact_ids = set(self._exact_index.get(normalized_query, set()))
+        token_sets: List[set[str]] = []
+        prefix_sets: List[set[str]] = []
+        for token in query_tokens:
+            exact_token_ids = set(self._token_index.get(token, set()))
+            prefix_ids = set(self._prefix_index.get(token, set())) if len(token) >= self.prefix_min_length else set()
+            if exact_token_ids:
+                token_sets.append(exact_token_ids)
+            if prefix_ids:
+                prefix_sets.append(prefix_ids)
+
+        candidate_ids: set[str] = set(exact_ids)
+        if token_sets:
+            intersected = set.intersection(*token_sets)
+            candidate_ids.update(intersected if intersected else set().union(*token_sets))
+        if prefix_sets:
+            candidate_ids.update(set().union(*prefix_sets))
+
+        diagnostics["path"] = "index"
+
+        if not candidate_ids:
+            diagnostics["path"] = "secondary_scan"
+            candidate_ids = self._secondary_scan(normalized_query, limit)
+
+        diagnostics["candidates"] = len(candidate_ids)
+
+        scored: List[Tuple[float, int, int, str]] = []
+        for node_id in candidate_ids:
+            document = self._documents.get(node_id)
+            if document is None or not self._passes_filters(document, filters):
+                continue
+            score = self._score_document(document, normalized_query, query_tokens)
+            if score <= 0:
+                continue
+            token_hits = sum(1 for token in query_tokens if token in document.tokens)
+            exactness = 1 if normalized_query == document.normalized_id or normalized_query in document.exact_terms else 0
+            scored.append((score, exactness, token_hits, node_id))
+
+        top_matches = heapq.nlargest(limit, scored, key=lambda item: (item[0], item[1], item[2], item[3]))
+        results = [(node_id, round(score, 4)) for score, _, _, node_id in top_matches]
+        self._store_cache(cache_key, results)
+        return results, diagnostics
+
+    def _secondary_scan(self, normalized_query: str, limit: int) -> set[str]:
+        matches: set[str] = set()
+        max_hits = max(limit * 20, 200)
+        scanned = 0
+        for node_id in self._ordered_node_ids:
+            if scanned >= self.secondary_scan_limit or len(matches) >= max_hits:
+                break
+            scanned += 1
+            document = self._documents.get(node_id)
+            if document is None:
+                continue
+            if normalized_query in document.primary_text or normalized_query in document.secondary_text:
+                matches.add(node_id)
+        return matches
+
+    def _score_document(
+        self,
+        document: IndexedNodeDocument,
+        normalized_query: str,
+        query_tokens: Tuple[str, ...],
+    ) -> float:
+        score = 0.0
+        if normalized_query == document.normalized_id:
+            score = max(score, 140.0)
+        elif normalized_query in document.exact_terms:
+            score = max(score, 120.0)
+
+        if normalized_query and normalized_query in document.primary_text:
+            score = max(score, 78.0 + min(len(normalized_query), 24) / 10.0)
+        elif normalized_query and normalized_query in document.secondary_text:
+            score = max(score, 26.0 + min(len(normalized_query), 24) / 20.0)
+
+        token_hits = 0
+        prefix_hits = 0
+        for token in query_tokens:
+            if token in document.tokens:
+                token_hits += 1
+            elif len(token) >= self.prefix_min_length and any(candidate.startswith(token) for candidate in document.tokens):
+                prefix_hits += 1
+
+        score += token_hits * 18.0
+        score += prefix_hits * 10.0
+
+        if len(query_tokens) > 1 and token_hits:
+            score += token_hits * 4.0
+
+        return score
+
+    def _passes_filters(self, document: IndexedNodeDocument, filters: Dict[str, Any]) -> bool:
+        filter_type = filters.get("type") or filters.get("node_type")
+        if filter_type and document.node_type != str(filter_type):
+            return False
+
+        min_confidence = _coerce_float(filters.get("min_confidence"))
+        if min_confidence is not None:
+            if document.confidence is None or document.confidence < min_confidence:
+                return False
+
+        tags_filter = filters.get("tags")
+        if tags_filter:
+            if isinstance(tags_filter, str):
+                required_tags = {_normalize_text(tags_filter)}
+            else:
+                required_tags = {
+                    normalized
+                    for normalized in (_normalize_text(tag) for tag in tags_filter)
+                    if normalized
+                }
+            if required_tags and not required_tags.issubset(set(document.tags)):
+                return False
+
+        return True
+
+    def _cache_key(
+        self,
+        normalized_query: str,
+        limit: int,
+        filters: Dict[str, Any],
+    ) -> Tuple[Any, ...]:
+        serialized_filters: List[Tuple[str, Any]] = []
+        for key in sorted(filters.keys()):
+            value = filters[key]
+            if isinstance(value, (list, tuple, set)):
+                serialized_filters.append((key, tuple(sorted(str(item) for item in value))))
+            else:
+                serialized_filters.append((key, str(value)))
+        return normalized_query, limit, tuple(serialized_filters)
+
+    def _store_cache(self, cache_key: Tuple[Any, ...], results: List[Tuple[str, float]]) -> None:
+        self._cache[cache_key] = list(results)
+        self._cache.move_to_end(cache_key)
+        while len(self._cache) > self.cache_size:
+            self._cache.popitem(last=False)
+
+    def _build_document(self, node: Dict[str, Any]) -> IndexedNodeDocument:
+        node_id = str(node.get("id", "")).strip()
+        node_type = str(node.get("type", "entity"))
+        properties = dict(node.get("properties", {}) or {})
+
+        primary_terms: List[str] = []
+        for candidate in (node_id, node.get("content", "")):
+            normalized = _normalize_text(candidate)
+            if normalized:
+                primary_terms.append(normalized)
+
+        for alias_key in _CURATED_ALIAS_KEYS:
+            _collect_text_fragments(properties.get(alias_key), primary_terms, limit=32)
+
+        deduped_primary_terms = tuple(dict.fromkeys(term for term in primary_terms if term))
+        primary_text = " ".join(deduped_primary_terms)
+        tokens = frozenset(_tokenize(primary_text))
+
+        secondary_fragments: List[str] = []
+        for key, value in properties.items():
+            if key in _CURATED_ALIAS_KEYS or key in {"content", "valid_from", "valid_until"}:
+                continue
+            _collect_text_fragments(value, secondary_fragments, limit=48)
+            if len(secondary_fragments) >= 48:
+                break
+
+        secondary_text = " ".join(dict.fromkeys(fragment for fragment in secondary_fragments if fragment))
+        confidence = _coerce_float(properties.get("confidence"))
+
+        raw_tags = properties.get("tags") or []
+        if isinstance(raw_tags, str):
+            raw_tags = [raw_tags]
+        tags = tuple(
+            dict.fromkeys(
+                normalized for normalized in (_normalize_text(tag) for tag in raw_tags) if normalized
+            )
+        )
+
+        return IndexedNodeDocument(
+            node_id=node_id,
+            normalized_id=_normalize_text(node_id),
+            node_type=node_type,
+            exact_terms=frozenset(deduped_primary_terms),
+            tokens=tokens,
+            primary_text=primary_text,
+            secondary_text=secondary_text,
+            confidence=confidence,
+            tags=tags,
+        )
@@ -4,12 +4,15 @@ Semantica Explorer session helpers.

 import base64
 import json
+import logging
 import threading
+import time
 import uuid
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 from typing import Any, Dict, Iterable, List, Optional

 from ..context.context_graph import ContextGraph, _resolve_edge_identity
+from .search_index import GraphSearchIndex

 _KG_AVAILABLE = False
 try:
@@ -28,6 +31,8 @@ try:
 except ImportError:
    pass

+logger = logging.getLogger(__name__)
+

 class GraphSession:
    """Thread-safe session wrapper around a loaded ``ContextGraph``."""
@@ -35,6 +40,7 @@ class GraphSession:
    def __init__(self, graph: ContextGraph) -> None:
        self.graph = graph
        self._lock = threading.RLock()
+        self._search_index = GraphSearchIndex()

        self.annotations: Dict[str, Dict[str, Any]] = {}

@@ -46,6 +52,7 @@ class GraphSession:
        self._similarity: Any = None
        self._link_predictor: Any = None
        self._validator: Any = None
+        self.rebuild_search_index()

    @classmethod
    def from_file(cls, path: str) -> "GraphSession":
@@ -390,6 +397,28 @@ class GraphSession:
        with self._lock:
            return self.graph.get_neighbors(node_id, hops=depth)

+    def rebuild_search_index(self) -> None:
+        with self._lock:
+            normalized_nodes = [
+                self.normalize_node(node.to_dict())
+                for node in self.graph.nodes.values()
+                if node is not None
+            ]
+        self._search_index.rebuild(normalized_nodes)
+
+    def handle_graph_mutation(self, event_type: str, entity_id: str, payload: Dict[str, Any]) -> None:
+        normalized_event = str(event_type or "").upper()
+        if normalized_event in {"ADD_NODE", "UPDATE_NODE"}:
+            normalized_node = self.normalize_node(payload or {})
+            if normalized_node.get("id"):
+                with self._lock:
+                    self._search_index.upsert(normalized_node)
+        elif normalized_event in {"REMOVE_NODE", "DELETE_NODE"}:
+            with self._lock:
+                self._search_index.remove(str(entity_id))
+        elif normalized_event in {"RELOAD_GRAPH", "RESET_GRAPH"}:
+            self.rebuild_search_index()
+
    def search(
        self,
        query: str,
@@ -397,64 +426,34 @@ class GraphSession:
        filters: Optional[Dict[str, Any]] = None,
    ) -> List[Dict[str, Any]]:
        filters = filters or {}
-        try:
-            with self._lock:
-                raw = self.graph.query(query)[:limit]
-        except Exception:
-            raw = []
+        started_at = time.perf_counter()
+        matches, diagnostics = self._search_index.search(query, limit=limit, filters=filters)

-        if not raw:
-            nodes, _ = self.get_nodes(search=query, skip=0, limit=max(limit * 5, limit))
-            scored = []
-            lowered_query = query.lower().strip()
-            for node in nodes:
-                haystacks = [
-                    str(node.get("id", "")),
-                    str(node.get("content", "")),
-                    json.dumps(node.get("properties", {}), default=str),
-                ]
-                best_score = 0.0
-                for haystack in haystacks:
-                    lowered = haystack.lower()
-                    if lowered == lowered_query:
-                        best_score = max(best_score, 1.0)
-                    elif lowered_query in lowered:
-                        best_score = max(best_score, min(0.9, len(lowered_query) / max(len(lowered), 1)))
-                if best_score > 0:
-                    scored.append({"node": node, "score": round(best_score, 4)})
-            raw = sorted(scored, key=lambda item: item["score"], reverse=True)[:limit]
-
-        normalized = []
-        for result in raw:
-            result_node = result.get("node", {})
-            node = (
-                self.normalize_node(result_node)
-                if "properties" in result_node or "metadata" in result_node or "content" in result_node
-                else result_node
-            )
-
-            filter_type = filters.get("type") or filters.get("node_type")
-            if filter_type and node["type"] != filter_type:
-                continue
-
-            min_confidence = self._coerce_float(filters.get("min_confidence"))
-            node_confidence = self._coerce_float(node["properties"].get("confidence"))
-            if min_confidence is not None and (
-                node_confidence is None or node_confidence < min_confidence
-            ):
-                continue
-
-            tags_filter = filters.get("tags")
-            if tags_filter:
-                node_tags = node["properties"].get("tags") or []
-                if isinstance(node_tags, str):
-                    node_tags = [node_tags]
-                if not set(tags_filter).issubset(set(node_tags)):
+        normalized_results: List[Dict[str, Any]] = []
+        with self._lock:
+            for node_id, score in matches:
+                raw_node = self.graph.find_node(node_id)
+                if raw_node is None:
                    continue
+                node_payload = raw_node.to_dict() if hasattr(raw_node, "to_dict") else raw_node
+                normalized_results.append(
+                    {
+                        "node": self.normalize_node(node_payload),
+                        "score": score,
+                    }
+                )

-            normalized.append({"node": node, "score": result.get("score", 0.0)})
-
-        return normalized[:limit]
+        duration_ms = round((time.perf_counter() - started_at) * 1000, 2)
+        logger.debug(
+            "Explorer search query=%r limit=%s cache_hit=%s path=%s candidates=%s duration_ms=%s",
+            query,
+            limit,
+            diagnostics.get("cache_hit"),
+            diagnostics.get("path"),
+            diagnostics.get("candidates"),
+            duration_ms,
+        )
+        return normalized_results[:limit]

    def get_stats(self) -> Dict[str, Any]:
        with self._lock:
@@ -594,8 +593,51 @@ class GraphSession:

    def add_nodes(self, nodes: List[Dict[str, Any]]) -> int:
        with self._lock:
-            return self.graph.add_nodes(nodes)
+            added = self.graph.add_nodes(nodes)
+            has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
+        if added and not has_mutation_callback:
+            self.rebuild_search_index()
+        return added

    def add_edges(self, edges: List[Dict[str, Any]]) -> int:
        with self._lock:
-            return self.graph.add_edges(edges)
+            added = self.graph.add_edges(edges)
+            has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
+        if added and not has_mutation_callback:
+            self.rebuild_search_index()
+        return added
+
+    def add_node(
+        self,
+        node_id: str,
+        node_type: str,
+        content: Optional[str] = None,
+        **properties: Any,
+    ) -> bool:
+        with self._lock:
+            added = self.graph.add_node(node_id, node_type, content=content, **properties)
+            has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
+        if added and not has_mutation_callback:
+            normalized = self.get_node(node_id)
+            if normalized is not None:
+                self._search_index.upsert(normalized)
+        return added
+
+    def add_edge(
+        self,
+        source_id: str,
+        target_id: str,
+        edge_type: str = "related_to",
+        weight: float = 1.0,
+        **properties: Any,
+    ) -> bool:
+        with self._lock:
+            added = self.graph.add_edge(
+                source_id,
+                target_id,
+                edge_type=edge_type,
+                weight=weight,
+                **properties,
+            )
+            has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None))
+        return added
@@ -36,6 +36,16 @@ def _build_sample_graph() -> ContextGraph:
    graph.add_node("javascript", node_type="language", content="JavaScript programming language", x=100, y=120)
    graph.add_node("web_dev", node_type="concept", content="Web Development", x=24, y=30)
    graph.add_node("ml", node_type="concept", content="Machine Learning", x=45, y=60)
+    graph.add_node(
+        "metformin",
+        node_type="drug",
+        content="Metformin",
+        aliases=["Glucophage"],
+        confidence="0.97",
+        tags=["drug", "featured"],
+        x=22,
+        y=33,
+    )
    graph.add_node(
        "decision_1",
        node_type="decision",
@@ -244,6 +254,74 @@ class TestSearchAndStats:
        assert payload["total"] >= 1
        assert all(item["node"]["type"] == "language" for item in payload["results"])

+    def test_search_exact_and_prefix(self, client):
+        exact_response = client.post(
+            "/api/graph/search",
+            json={"query": "Metformin", "limit": 5},
+        )
+        assert exact_response.status_code == 200
+        exact_payload = exact_response.json()
+        assert exact_payload["results"][0]["node"]["id"] == "metformin"
+
+        prefix_response = client.post(
+            "/api/graph/search",
+            json={"query": "metf", "limit": 5},
+        )
+        assert prefix_response.status_code == 200
+        prefix_payload = prefix_response.json()
+        assert any(item["node"]["id"] == "metformin" for item in prefix_payload["results"])
+
+    def test_search_filters_and_cache_stability(self, client):
+        body = {
+            "query": "framework",
+            "filters": {"type": "decision", "min_confidence": 0.8},
+            "limit": 5,
+        }
+        first_response = client.post("/api/graph/search", json=body)
+        second_response = client.post("/api/graph/search", json=body)
+
+        assert first_response.status_code == 200
+        assert second_response.status_code == 200
+        assert first_response.json() == second_response.json()
+        results = first_response.json()["results"]
+        assert [item["node"]["id"] for item in results] == ["decision_1"]
+
+    def test_search_sees_new_nodes_after_mutation(self, client):
+        session = client.app.state.session
+        assert session.add_node(
+            "metformin_hcl",
+            "drug",
+            content="Metformin Hydrochloride",
+            aliases=["Glucophage XR"],
+            confidence="0.93",
+        )
+
+        response = client.post(
+            "/api/graph/search",
+            json={"query": "glucophage", "limit": 10},
+        )
+        assert response.status_code == 200
+        result_ids = [item["node"]["id"] for item in response.json()["results"]]
+        assert "metformin" in result_ids
+        assert "metformin_hcl" in result_ids
+
+    def test_search_secondary_scan_fallback_matches_non_curated_properties(self, client):
+        session = client.app.state.session
+        assert session.add_node(
+            "fallback_node",
+            "entity",
+            content="Alpha",
+            description="rareterm",
+        )
+
+        response = client.post(
+            "/api/graph/search",
+            json={"query": "rareterm", "limit": 10},
+        )
+        assert response.status_code == 200
+        result_ids = [item["node"]["id"] for item in response.json()["results"]]
+        assert "fallback_node" in result_ids
+
    def test_stats(self, client):
        response = client.get("/api/graph/stats")
        assert response.status_code == 200
@@ -0,0 +1,74 @@
+"""Unit tests for explorer provenance route helpers."""
+
+from types import SimpleNamespace
+
+from semantica.explorer.routes.provenance import _build_provenance, _render_markdown
+
+
+def _make_session_with_chain() -> SimpleNamespace:
+    """Build a minimal session-like object for Source -> Intermediate -> node_id."""
+    nodes = {
+        "Source": SimpleNamespace(node_type="entity", content="Source"),
+        "Intermediate": SimpleNamespace(node_type="entity", content="Intermediate"),
+        "node_id": SimpleNamespace(node_type="entity", content="Target"),
+    }
+    edges = [
+        SimpleNamespace(source_id="Source", target_id="Intermediate", edge_type="related_to"),
+        SimpleNamespace(source_id="Intermediate", target_id="node_id", edge_type="related_to"),
+    ]
+    graph = SimpleNamespace(nodes=nodes, edges=edges)
+    return SimpleNamespace(graph=graph)
+
+
+def test_build_provenance_direction_classification_chain():
+    session = _make_session_with_chain()
+
+    data = _build_provenance(session, "node_id")
+
+    node_ids = {node["id"] for node in data["nodes"]}
+    assert "Source" in node_ids
+    assert "Intermediate" in node_ids
+
+    edge_by_pair = {(edge["source"], edge["target"]): edge for edge in data["edges"]}
+
+    assert edge_by_pair[("Intermediate", "node_id")]["direction"] == "upstream"
+    assert edge_by_pair[("Source", "Intermediate")]["direction"] != "downstream"
+
+
+def test_render_markdown_groups_edges_by_direction():
+    report = {
+        "node_id": "node_id",
+        "label": "Target",
+        "type": "entity",
+        "properties": {},
+        "lineage": {
+            "nodes": [
+                {"id": "Source", "prov_type": "Entity", "label": "Source"},
+                {"id": "Intermediate", "prov_type": "Entity", "label": "Intermediate"},
+                {"id": "node_id", "prov_type": "Entity", "label": "Target"},
+            ],
+            "edges": [
+                {
+                    "id": "Intermediate-node_id",
+                    "source": "Intermediate",
+                    "target": "node_id",
+                    "label": "related_to",
+                    "direction": "upstream",
+                },
+                {
+                    "id": "Source-Intermediate",
+                    "source": "Source",
+                    "target": "Intermediate",
+                    "label": "related_to",
+                    "direction": "lateral",
+                },
+            ],
+        },
+    }
+
+    markdown = _render_markdown(report)
+
+    assert "## Upstream" in markdown
+    assert "## Lateral" in markdown
+    assert "`Intermediate` -[related_to]-> `node_id`" in markdown
+    assert "`Source` -[related_to]-> `Intermediate`" in markdown