fix(docker): revert runtime base image to python:3.13-slim

#1290 bumped the runtime stage to python:3.14-slim. gensim is a base
(non-extras-gated) dependency and publishes no cp314 wheel on PyPI, so
pip falls back to building it from source, which needs a C compiler
this slim image doesn't carry:

  error: [Errno 2] No such file or directory: 'gcc'

This has broken the Docker build (and Container Security Scan) on
every push to main since #1290 merged. Confirmed via
`pip index versions gensim` / the PyPI files listing that gensim 4.4.0
ships cp313 wheels but no cp314 ones; the earlier lockfile fix in
#1290 only validated that `uv pip compile` could *resolve* dependencies
for 3.14, which reads sdist metadata and doesn't need a compiler --
it doesn't validate that `pip install` can actually *build* them,
which is what fails here.

Reverts to the exact base image digest that was building successfully
before #1290 and regenerates explorer-extra-py313.txt against today's
PyPI state. Once gensim (and anything else pulled in transitively)
ships cp314 wheels, the 3.14 bump can be retried.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.github/workflows/ci.yml

@@ -93,6 +93,7 @@ jobs:
           npm run test:graph-workspace
           npm run test:plugin-registry
           npm run test:deterministic-e2e
+          npm run test:graph-legend-e2e
       - name: Build Explorer frontend
         working-directory: explorer
         run: npm run build

.github/workflows/release.yml

@@ -67,6 +67,12 @@ jobs:
         run: |
           pip install -r .github/requirements/twine.txt --require-hashes
           twine check dist/*
+      # pypi-publish uploads everything under packages-dir (default: dist/) with
+      # no glob/include filter, so it must run before anything else writes a
+      # non-distribution file into dist/ - the Sigstore step below does exactly
+      # that (dist/*.sigstore.json), and pypi-publish fails on it with
+      # "InvalidDistribution: Unknown distribution format" if it runs after.
+      - uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # release/v1
       - name: Attest build provenance
         uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4
         with:
@@ -75,7 +81,7 @@ jobs:
       # OpenSSF Scorecard's Signed-Releases check does not inspect - it looks for
       # signature files attached as release assets. Sign here too so
       # `dist/*.sigstore.json` bundles ship alongside the wheel/sdist on the
-      # GitHub Release itself.
+      # GitHub Release itself. This must run after pypi-publish (see above).
       - name: Sign artifacts with Sigstore
         uses: sigstore/gh-action-sigstore-python@790bc6befb9d733738f18d8f895854b453640ec9 # v3.5.0
         with:
@@ -88,4 +94,3 @@ jobs:
             dist/*.whl
             dist/*.tar.gz
             dist/*.sigstore.json
-      - uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # release/v1

Dockerfile

@@ -20,6 +20,17 @@ RUN mkdir -p /app/semantica && npm run build
 # .github/dependabot.yml opens a PR bumping the digest pin above. Also: this
 # image only serves plain HTTP via uvicorn and never opens a QUIC listener,
 # so the bug isn't reachable here regardless.
+#
+# Pinned to 3.13, NOT 3.14: #1290 bumped this to python:3.14-slim and broke
+# the build outright (Container Security Scan, every run since) - gensim
+# (a base, non-extras-gated dependency) ships no cp314 wheel on PyPI yet, so
+# pip falls back to building it from source, which needs a C compiler this
+# slim image doesn't carry ("error: [Errno 2] No such file or directory:
+# 'gcc'"). Revisit the 3.14 bump once gensim (and anything else pulled in
+# transitively) publishes cp314 wheels - check with
+# `pip index versions gensim` / the project's PyPI files page, not just
+# whether `uv pip compile` resolves (resolution only reads sdist metadata,
+# it doesn't attempt the build that fails here).
 FROM python:3.13-slim@sha256:7ce4b6dfe35e55397b7cda544f8a13f191b7ae28dc5aad71fe664dbc9bc2623f AS runtime
 
 ENV PYTHONDONTWRITEBYTECODE=1 \

README.md

@@ -1406,6 +1406,9 @@ semantica-mcp
 | `get_graph_analytics` | Centrality, communities |
 | `export_graph` | Export to RDF/JSON/Parquet |
 | `get_graph_summary` | Graph statistics |
+| `query_graph` | Fetch a node, walk neighbours, keyword search |
+| `update_node` | Merge properties onto a node |
+| `delete_node` | Archive (soft-delete) a node |
 
 ### REST API
 

docs/guides/mcp-server.md

@@ -8,7 +8,7 @@ icon: "plug"
 
 MCP stands for the Model Context Protocol. It is an open standard that allows external AI assistants (like Claude Desktop, Cursor, or Windsurf) to securely access local tools and data sources. 
 
-The Semantica MCP server exposes your knowledge graph as 12 callable tools. By connecting it, any compatible AI client can traverse the graph live, record decisions, run analytics, and export results during a conversation — without you having to write custom tool wrappers.
+The Semantica MCP server exposes your knowledge graph as 15 callable tools. By connecting it, any compatible AI client can traverse the graph live, record decisions, run analytics, and export results during a conversation — without you having to write custom tool wrappers.
 
 <Info>
   The Semantica MCP server exposes 15 tools and 3 read-only resources. All tools accept and return JSON. No configuration beyond an optional environment variable for graph persistence is required.
@@ -40,7 +40,7 @@ Connecting your AI client follows a standard progression:
 1. **Install**: Install Semantica in your Python environment.
 2. **Configure Client**: Add the `semantica-mcp` command and absolute graph paths to your AI client's JSON configuration.
 3. **Start Client**: Launch Claude Desktop or Windsurf, which automatically spawns the MCP server.
-4. **Tool Calls**: Prompt the AI in natural language. The AI autonomously chains the 12 available tools.
+4. **Tool Calls**: Prompt the AI in natural language. The AI autonomously chains the 15 available tools.
 5. **Graph Updates**: The AI directly modifies your local graph, adding entities, edges, and decisions.
 
 ---

docs/reference/embeddings.md

@@ -1,6 +1,6 @@
 ---
 title: "Embeddings Module"
-description: "Text and graph embedding generation: FastEmbed, Sentence-Transformers, OpenAI, BGE: with pooling strategies and provider-agnostic API."
+description: "Text and graph embedding generation (FastEmbed, Sentence-Transformers, OpenAI, BGE) with pooling strategies and a provider-agnostic API."
 icon: "vector-square"
 ---
 
@@ -41,12 +41,12 @@ Semantica uses embeddings for:
 
 ## What You Get
 
-- **EmbeddingGenerator** — Main entry point: provider-agnostic, handles batching automatically across all backends.
-- **TextEmbedder** — Text-specific with automatic batching and progress tracking. Default method is FastEmbed.
-- **GraphEmbeddingManager** — Node and edge embeddings for graph databases: Neo4j, NetworkX, FalkorDB.
-- **VectorEmbeddingManager** — Prepare, normalize, and format embeddings for FAISS, Weaviate, Qdrant, and Milvus.
-- **Provider Stores** — `OpenAIStore`, `BGEStore`, `FastEmbedStore`, and `ProviderStoreFactory`.
-- **Pooling Strategies** — Mean, Max, CLS, Attention, and Hierarchical: control token-to-vector aggregation.
+- **EmbeddingGenerator**: provider-agnostic main entry point that handles batching automatically across all backends.
+- **TextEmbedder**: text-specific embedder with automatic batching and progress tracking. Default method is FastEmbed.
+- **GraphEmbeddingManager**: node and edge embeddings for graph databases (Neo4j, NetworkX, FalkorDB).
+- **VectorEmbeddingManager**: prepare, normalize, and format embeddings for FAISS, Weaviate, Qdrant, and Milvus.
+- **Provider Stores**: `OpenAIStore`, `BGEStore`, `FastEmbedStore`, and `ProviderStoreFactory`.
+- **Pooling Strategies**: Mean, Max, CLS, Attention, and Hierarchical control token-to-vector aggregation.
 
 ## Provider Setup
 
@@ -71,7 +71,7 @@ Semantica uses embeddings for:
     </Check>
 
     <Warning>
-      **FastEmbed ignores the `device` parameter.** FastEmbed uses ONNX Runtime and manages its own execution providers: passing `device="cuda"` has no effect. Switch to `method="sentence_transformers"` if you need explicit GPU control.
+      **FastEmbed ignores the `device` parameter.** FastEmbed uses ONNX Runtime and manages its own execution providers; passing `device="cuda"` has no effect. Switch to `method="sentence_transformers"` if you need explicit GPU control.
     </Warning>
   </Tab>
   <Tab title="Sentence-Transformers">
@@ -153,7 +153,7 @@ providers = check_available_providers()
 
 ## Getting Started
 
-`EmbeddingGenerator` is the fastest path to embeddings: the default method is FastEmbed (ONNX, no GPU needed):
+`EmbeddingGenerator` is the fastest path to embeddings. The default method is FastEmbed (ONNX, no GPU needed):
 
 ```python
 from semantica.embeddings import EmbeddingGenerator
@@ -173,7 +173,7 @@ print(f"Similarity: {score:.3f}")
 ```
 
 <Tip>
-  **Always use the same model for indexing and querying.** Vectors from different models are not comparable: they live in different vector spaces. Switching models requires re-embedding your entire corpus.
+  **Always use the same model for indexing and querying.** Vectors from different models are not comparable; they live in different vector spaces. Switching models requires re-embedding your entire corpus.
 </Tip>
 
 To switch provider after construction:
@@ -258,7 +258,7 @@ generator.set_text_model("sentence_transformers", "BAAI/bge-large-en-v1.5")
     similarity = generator.compare_embeddings(embeddings[0], embeddings[1])
     ```
 
-    **Best for:** CPU-only production, lowest latency without GPU. Default: works out of the box.
+    **Best for:** CPU-only production and lowest latency without GPU. The default works out of the box.
   </Tab>
   <Tab title="Sentence-Transformers">
     ```python
@@ -390,7 +390,7 @@ store = ProviderStoreFactory.create(provider="bge", model_name="BAAI/bge-large-e
 
 ## Pooling Strategies
 
-Pooling aggregates a set of embeddings into a single vector: useful when you have multiple chunk embeddings to combine:
+Pooling aggregates a set of embeddings into a single vector. Useful when you have multiple chunk embeddings to combine:
 
 <Tabs>
   <Tab title="MeanPooling (default)">
@@ -401,7 +401,7 @@ Pooling aggregates a set of embeddings into a single vector: useful when you hav
     pooled = pooler.pool(token_embeddings)   # shape: (hidden_dim,)
     ```
 
-    **Best for:** retrieval, semantic search, and clustering: averages all contributions.
+    **Best for:** retrieval, semantic search, and clustering. Averages all contributions.
   </Tab>
   <Tab title="MaxPooling">
     ```python
@@ -411,7 +411,7 @@ Pooling aggregates a set of embeddings into a single vector: useful when you hav
     pooled = pooler.pool(token_embeddings)
     ```
 
-    **Best for:** capturing the presence of any feature: takes the max activation per dimension.
+    **Best for:** capturing the presence of any feature. Takes the max activation per dimension.
   </Tab>
   <Tab title="CLSPooling">
     ```python
@@ -432,7 +432,7 @@ Pooling aggregates a set of embeddings into a single vector: useful when you hav
     pooled = pooler.pool(token_embeddings, chunk_size=10)
     ```
 
-    **Best for:** long documents: chunk-level mean pooling, then global mean pooling across chunks.
+    **Best for:** long documents (chunk-level mean pooling, then global mean pooling across chunks).
   </Tab>
   <Tab title="Strategy Comparison">
 
@@ -619,7 +619,7 @@ providers = check_available_providers()
 # → {"sentence_transformers": True, "fastembed": True, "openai": False}
 ```
 
-- [Vector Store](/reference/vector_store) — Store and search the generated embeddings.
-- [Split](/reference/split) — Chunk text before embedding for better retrieval quality.
-- [KG Module](/reference/kg) — Distance Intelligence uses graph embeddings for semantic neighbourhoods.
-- [Deduplication](deduplication) — Semantic deduplication uses embedding distance for entity resolution.
+- [Vector Store](/reference/vector_store): store and search the generated embeddings.
+- [Split](/reference/split): chunk text before embedding for better retrieval quality.
+- [KG Module](/reference/kg): Distance Intelligence uses graph embeddings for semantic neighbourhoods.
+- [Deduplication](/reference/deduplication): semantic deduplication uses embedding distance for entity resolution.

docs/reference/semantic_extract.md

@@ -191,6 +191,41 @@ trip = TripletExtractor(method=["llm", "pattern"])
 entities = ner.extract(text)
 ```
 
+### NER Merge Strategies
+
+`NERExtractor` uses `merge_strategy="fallback"` by default, so a method list remains an ordered fallback chain. To run several methods together, choose one of the explicit strategies below:
+
+| Strategy | Behavior |
+| :--- | :--- |
+| `fallback` | Return the first non-empty method result. |
+| `union` | Keep candidates from any method. Same-label boundary variants are aligned, while distinct labels remain available. |
+| `consensus` | Require cross-method support for an offset-aligned candidate. `min_votes` defaults to `2`. |
+
+```python
+from semantica.semantic_extract import NERExtractor
+
+ner = NERExtractor(
+    method=["spacy", "huggingface"],
+    merge_strategy="consensus",
+    min_votes=2,
+    min_agreement=0.75,  # optional support-ratio requirement
+    method_weights={"spacy": 0.8, "huggingface": 1.0},
+)
+entities = ner.extract(text)
+
+for entity in entities:
+    print(entity.metadata["supporting_methods"])
+    print(entity.metadata["vote_count"], entity.metadata["agreement"])
+```
+
+Consensus counts support against the configured eligible methods, not only methods that emitted a candidate. An empty or failed eligible method is therefore a non-supporting vote. Use `eligible_methods=[...]` to restrict the consensus denominator when the configured methods have different coverage, or use `merge_strategy="union"` for complementary rule extractors. `method_weights` only break an otherwise eligible exact-span cross-label tie; they never turn one method into multiple votes.
+
+Each merged entity includes `supporting_methods`, `vote_count`, `eligible_method_count`, `agreement`, and per-method `method_scores` in its metadata. Consensus treats compatible label aliases such as `PER`/`PERSON` and `ORGANIZATION`/`ORG` as the same vote. It resolves a cross-label conflict only when the final spans are identical, using method weight, vote count, confidence, and a stable label order; nested entities at different spans remain available. `ml` and `spacy` are one backend for both voting and weights, so their weights are interchangeable (conflicting values are rejected). Boundary candidates are matched one-to-one only when their span IoU is at least 0.5 with every existing vote in that candidate; equal-confidence variants prefer the longer span. If a provider omits offsets, Semantica resolves its entity text against whole-word document matches before merging. This keeps repeated mentions with the same text distinct and prevents one broad span from acting as a vote for multiple mentions.
+
+`ensemble_voting=True` is deprecated and maps to `merge_strategy="union"` during migration. Use `merge_strategy="consensus"` when method agreement is required.
+
+Unlike `fallback`, `union` and `consensus` never inject a pattern-derived entity after the configured methods return no candidates. An empty result is therefore meaningful in those strategies.
+
 
 ## Quick Start
 

docs_check.py

@@ -246,9 +246,12 @@ def _() -> list[str]:
         combined = (result.stdout or "") + (result.stderr or "")
 
         if result.returncode != 0:
-            # On Windows, npm cleanup raises EPERM on temp dirs — not a real
-            # export failure. Treat as a skip rather than a hard failure.
-            if sys.platform == "win32" and "EPERM" in combined and \
+            # On Windows, npm post-command cleanup can fail with EPERM/EBUSY on
+            # temp dirs — not a real export failure. Treat as a skip rather
+            # than a hard failure, unless a real Mintlify error signature is
+            # present.
+            if sys.platform == "win32" and \
+                    ("EPERM" in combined or "EBUSY" in combined) and \
                     "could not be generated" not in combined:
                 return []  # Windows temp-cleanup noise; real CI runs on Linux
 

explorer/package.json

@@ -9,9 +9,9 @@
     "lint": "eslint .",
     "preview": "vite preview",
     "test:graph-store": "node --test tests/graphStore.multi-edge.test.mjs",
-    "test:graph-workspace": "node --import tsx --test tests/markdownContentViewer.test.ts tests/markdownEditorInteraction.test.tsx tests/markdownEditorState.test.ts tests/nodeMarkdownSync.test.ts tests/graphSceneState.display.test.ts tests/temporalLifecycle.test.ts tests/deterministicExplorerRendering.test.ts tests/explorerCapabilities.test.tsx tests/smallGraphLayout.test.ts tests/realtimeGraphAttributes.test.ts",
-    "test:graph-workspace": "node --import tsx --test tests/markdownContentViewer.test.ts tests/graphSceneState.display.test.ts tests/temporalLifecycle.test.ts tests/deterministicExplorerRendering.test.ts tests/smallGraphLayout.test.ts tests/realtimeGraphAttributes.test.ts tests/ontologyEditorModel.test.ts",
+    "test:graph-workspace": "node --import tsx --test tests/markdownContentViewer.test.ts tests/markdownEditorInteraction.test.tsx tests/markdownEditorState.test.ts tests/nodeMarkdownSync.test.ts tests/graphSceneState.display.test.ts tests/temporalLifecycle.test.ts tests/deterministicExplorerRendering.test.ts tests/explorerCapabilities.test.tsx tests/smallGraphLayout.test.ts tests/realtimeGraphAttributes.test.ts tests/ontologyEditorModel.test.ts tests/graphColorLegend.test.ts",
     "test:deterministic-e2e": "node --import tsx --test tests/deterministicExplorerRendering.e2e.ts",
+    "test:graph-legend-e2e": "node --import tsx --test tests/graphColorLegend.e2e.ts",
     "test:plugin-registry": "node --import tsx --test tests/pluginRegistry.temporal.test.mjs"
   },
   "dependencies": {

explorer/src/store/graphStore.ts

@@ -26,6 +26,8 @@ export interface NodeAttributes {
   size: number;
   color: string;
   baseColor?: string;
+  /** Original semantic color when a display clone bakes interaction styling into baseColor. */
+  semanticBaseColor?: string;
   mutedColor?: string;
   glowColor?: string;
   baseSize?: number;

explorer/src/workspaces/GraphWorkspace/GraphWorkspace.tsx

@@ -28,7 +28,7 @@ import { useLoadGraph, useReloadGraph } from "./useLoadGraph";
 import { GraphLoadingOverlay } from "./GraphLoadingOverlay";
 import { createGraphLoadProgress, getGraphLoadTitle } from "./graphLoading";
 import { GRAPH_THEME, withAlpha } from "./graphTheme";
-import type { GraphEntityShapeVariant } from "./graphTheme";
+import { buildGraphColorLegend, type GraphColorLegendItem } from "./graphColorLegend";
 import { buildHeatmapRenderSnapshot, buildStructuralDistanceSnapshot, checkGroupedViewAvailability, getDistanceBandColor, resolveDisplayGraph, resolveDisplayStateSnapshot, resolveGroupedDisplayNodeId, resolveGroupedDisplayStateSnapshot, summarizeDistanceBuckets } from "./graphSceneState";
 import {
   type GraphPlugin,
@@ -169,14 +169,7 @@ const loadNeighborhoodPanelPlugin = () => import("./plugins/neighborhoodPanelPlu
 const loadTemporalOverlayPlugin = () => import("./plugins/temporalOverlayPlugin").then((module) => module.temporalOverlayPlugin);
 const EMPTY_PATH: string[] = [];
 const COMPACT_TOOLBAR_CLUSTER_IDS = new Set(["camera", "utility"]);
-const ENTITY_VISUAL_KEY: Array<{ shape: GraphEntityShapeVariant; label: string }> = [
-  { shape: "biomolecule", label: "Biomolecule" },
-  { shape: "condition", label: "Condition" },
-  { shape: "compound", label: "Compound" },
-  { shape: "process", label: "Process" },
-  { shape: "community", label: "Community" },
-  { shape: "entity", label: "Other" },
-];
+
 const DEBUG_GRAPH_WORKSPACE = import.meta.env.DEV;
 
 function debugGraphWorkspace(message: string, payload?: Record<string, unknown>) {
@@ -459,15 +452,21 @@ function SearchCommandBar({
   );
 }
 
-function EntityVisualKey() {
+function SemanticColorLegend({ items }: { items: GraphColorLegendItem[] }) {
+  if (!items.length) return null;
   return (
-    <div className="explore-entity-key" aria-label="Node visual key">
-      {ENTITY_VISUAL_KEY.map((item) => (
-        <div key={item.shape} className="explore-entity-key-item">
-          <span className="explore-entity-key-mark" data-shape={item.shape} />
-          <span>{item.label}</span>
-        </div>
-      ))}
+    <div className="explore-color-legend" role="group" aria-label="Node colors">
+      <span className="explore-color-legend-label" title="Base semantic colors; selection, zoom, and distance effects can change node appearance.">
+        Node colors
+      </span>
+      <ul className="explore-color-legend-items">
+        {items.map((item) => (
+          <li key={item.id} className="explore-color-legend-item" title={`${item.group}: ${item.count.toLocaleString()} nodes`}>
+            <span className="explore-color-legend-mark" style={{ backgroundColor: item.color }} aria-hidden="true" />
+            <span className="explore-color-legend-name">{item.group}</span>
+          </li>
+        ))}
+      </ul>
     </div>
   );
 }
@@ -906,55 +905,46 @@ const HUD_CSS = `
   .explore-tool-button[data-compact="true"] .explore-tool-button-label {
     display: none;
   }
-  .explore-entity-key {
+  .explore-color-legend {
     display: flex;
-    align-items: center;
-    gap: 10px;
-    flex-wrap: wrap;
+    align-items: baseline;
+    gap: 12px;
     padding: 2px 1px 0;
     color: ${GRAPH_THEME.ui.text.subtle};
-    font-size: 10px;
-    font-weight: 700;
-    letter-spacing: 0.02em;
+    font-size: 11px;
+    font-weight: 600;
   }
-  .explore-entity-key-item {
+  .explore-color-legend-label {
+    flex-shrink: 0;
+    color: ${GRAPH_THEME.ui.text.muted};
+  }
+  .explore-color-legend-items {
+    display: flex;
+    flex-wrap: wrap;
+    gap: 8px 14px;
+    min-width: 0;
+    max-height: 76px;
+    overflow-y: auto;
+    margin: 0;
+    padding: 0;
+    list-style: none;
+  }
+  .explore-color-legend-item {
     display: inline-flex;
     align-items: center;
     gap: 6px;
-    white-space: nowrap;
+    min-width: 0;
+    max-width: 100%;
   }
-  .explore-entity-key-mark {
-    width: 13px;
-    height: 13px;
-    display: inline-block;
-    border: 1px solid rgba(194, 214, 218, 0.42);
-    background: rgba(73, 154, 150, 0.58);
-    box-shadow: inset 0 1px 0 rgba(255,255,255,0.08);
+  .explore-color-legend-name {
+    overflow-wrap: anywhere;
   }
-  .explore-entity-key-mark[data-shape="entity"] {
-    border-radius: 999px;
-  }
-  .explore-entity-key-mark[data-shape="biomolecule"] {
-    clip-path: polygon(50% 7%, 86% 28%, 86% 72%, 50% 93%, 14% 72%, 14% 28%);
-  }
-  .explore-entity-key-mark[data-shape="condition"] {
-    border-radius: 5px;
-    transform: rotate(45deg) scale(0.88);
-  }
-  .explore-entity-key-mark[data-shape="compound"] {
-    width: 20px;
-    border-radius: 999px;
-  }
-  .explore-entity-key-mark[data-shape="process"] {
-    border-radius: 4px;
-    clip-path: polygon(0 0, 86% 0, 100% 16%, 100% 100%, 0 100%);
-  }
-  .explore-entity-key-mark[data-shape="community"] {
-    width: 15px;
-    height: 15px;
-    border-radius: 999px;
-    background: rgba(96, 190, 180, 0.16);
-    border-color: rgba(229, 213, 175, 0.54);
+  .explore-color-legend-mark {
+    width: 10px;
+    height: 10px;
+    flex-shrink: 0;
+    border-radius: 50%;
+    box-shadow: inset 0 0 0 1px rgba(255,255,255,0.16);
   }
   .explore-search-results {
     display: flex;
@@ -2280,6 +2270,11 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken, onDirt
     structuralSelectedNodeId,
     viewMode,
   ]);
+  const colorLegendItems = useMemo(() => {
+    // Store mutations can preserve graph identity while changing its attributes.
+    void graphVersion;
+    return buildGraphColorLegend(displayResult.graph);
+  }, [displayResult.graph, graphVersion]);
   const displayState = useMemo(
     () => (
       viewMode === "grouped"
@@ -3122,7 +3117,7 @@ export function GraphWorkspace({ externalFocusNodeId, externalFocusToken, onDirt
                     ))}
                   </div>
                 </div>
-                <EntityVisualKey />
+                {!showDistanceStatus ? <SemanticColorLegend items={colorLegendItems} /> : null}
               </div>
 
               {egoModeEnabled && (

explorer/src/workspaces/GraphWorkspace/MarkdownContentViewer.tsx

@@ -1,9 +1,12 @@
 import {
   useEffect,
+  useId,
+  useLayoutEffect,
   useMemo,
   useRef,
   useState,
   type CSSProperties,
+  type KeyboardEvent,
 } from "react";
 import ReactMarkdown, { type Components } from "react-markdown";
 import remarkGfm from "remark-gfm";
@@ -34,6 +37,26 @@ export interface MarkdownContentViewerProps {
   defaultMode?: "preview" | "source";
 }
 
+// Exported for unit-testing the roving-tabindex navigation logic without a DOM.
+// Given the ordered list of tab modes and the currently focused mode, returns
+// the mode that should receive focus for a given keyboard key.  Returns null if
+// the key is not a navigation key so callers can handle the default case.
+// eslint-disable-next-line react-refresh/only-export-components
+export function resolveTabNavigation(
+  current: "preview" | "source",
+  key: string,
+): "preview" | "source" | null {
+  const order = ["preview", "source"] as const;
+  const idx = order.indexOf(current);
+  switch (key) {
+    case "ArrowRight": return order[(idx + 1) % order.length];
+    case "ArrowLeft":  return order[(idx - 1 + order.length) % order.length];
+    case "Home":       return order[0];
+    case "End":        return order[order.length - 1];
+    default:           return null;
+  }
+}
+
 export function MarkdownContentViewer({
   content,
   resource,
@@ -70,6 +93,64 @@ export function MarkdownContentViewer({
     if (copied) setCopied(false);
   }
 
+  // useId (not hardcoded strings) so the ids stay unique if more than one viewer
+  // is ever mounted at once — the same pattern GraphWorkspace uses for its search
+  // combobox. Hardcoded ids would collide silently in that case.
+  const baseId = useId();
+  const previewTabId = `${baseId}-tab-preview`;
+  const sourceTabId = `${baseId}-tab-source`;
+  const panelId = `${baseId}-panel`;
+
+  // Roving tabindex: the tablist is one Tab stop and arrows move focus within it.
+  // Focus is tracked separately from selection because activation is manual (see
+  // handleTabKeyDown), so a tab can hold focus without being the selected one.
+  //
+  // focusedMode is stored in a ref rather than state so that moving focus with
+  // arrow keys does NOT trigger a React re-render. A re-render here is expensive:
+  // react-markdown@10 has no internal memoisation and calls processor.parse() +
+  // processor.runSync() unconditionally on every render — measured at 385ms for a
+  // 1000-row GFM table and 1.4s at 2000 rows (#1118). Using a ref means arrow-key
+  // navigation is free of Markdown re-parses while still keeping the DOM tabIndex
+  // attributes correct via direct mutation (the same pattern used by WAI-ARIA APG
+  // keyboard examples for roving tabindex).
+  //
+  // The JSX tabIndex props use activeMode (not the ref) to satisfy the
+  // react-hooks/refs lint rule that bars ref reads during render. JSX provides the
+  // correct value on initial render and after selectMode() calls (which always keep
+  // focusedModeRef.current === activeMode at React render boundaries). A
+  // useLayoutEffect (see below) corrects any JSX overwrite that occurs when focus
+  // and selection temporarily differ during arrow navigation.
+  const focusedModeRef = useRef<"preview" | "source">(defaultMode);
+  const previewTabRef = useRef<HTMLButtonElement>(null);
+  const sourceTabRef = useRef<HTMLButtonElement>(null);
+
+  const focusTab = (mode: "preview" | "source") => {
+    focusedModeRef.current = mode;
+    // Imperatively update tabIndex on both buttons so the roving tabindex
+    // DOM state is correct without scheduling a React re-render.
+    if (previewTabRef.current) previewTabRef.current.tabIndex = mode === "preview" ? 0 : -1;
+    if (sourceTabRef.current) sourceTabRef.current.tabIndex = mode === "source" ? 0 : -1;
+    (mode === "preview" ? previewTabRef : sourceTabRef).current?.focus();
+  };
+
+  const selectMode = (mode: "preview" | "source") => {
+    // Keep the ref in sync before setActiveMode so the upcoming re-render reads
+    // the correct focusedModeRef.current when evaluating JSX tabIndex props.
+    focusedModeRef.current = mode;
+    setActiveMode(mode);
+  };
+
+  // Manual activation (APG permits it, and here it is required): arrows move
+  // focus only, Enter/Space activates via the native button click. Automatic
+  // activation would re-run the full markdown parse on every arrow keypress —
+  // measured at 385ms for a 1000-row GFM table and 1.4s at 2000 rows (#1118).
+  const handleTabKeyDown = (event: KeyboardEvent<HTMLButtonElement>) => {
+    const next = resolveTabNavigation(focusedModeRef.current, event.key);
+    if (next === null) return;
+    event.preventDefault();
+    focusTab(next);
+  };
+
   const copyTimeoutRef = useRef<number | undefined>(undefined);
   useEffect(() => {
     return () => {
@@ -77,9 +158,44 @@ export function MarkdownContentViewer({
     };
   }, []);
 
+  // When the viewed resource/node changes, reset focusedModeRef to match the
+  // incoming defaultMode. The render-phase setActiveMode(defaultMode) above
+  // resets React state, but refs are not state and must be updated separately.
+  // useLayoutEffect fires synchronously before paint so the ref is correct before
+  // the no-deps tabIndex-correction effect (declared next) reads it.
+  // Using resourceKey as the dep means this runs exactly once per resource change,
+  // immediately after the render that detected the change.
+  useLayoutEffect(() => {
+    focusedModeRef.current = defaultMode;
+  }, [resourceKey, defaultMode]);
+
+  // After every render, restore the DOM tabIndex to match focusedModeRef.current.
+  // This is necessary because the JSX tabIndex props derive from activeMode, which
+  // is correct for initial render and for renders triggered by selectMode(). However,
+  // when focus and selection differ (i.e. after arrow-key navigation, before Enter/Space),
+  // any unrelated re-render (copy-button click, parent update, etc.) will reconcile JSX
+  // tabIndex={activeMode === X} and overwrite the imperative tabIndex values set by
+  // focusTab(). useLayoutEffect fires synchronously after React's DOM mutations, before
+  // paint, so it corrects any such overwrite before the user sees it. It does not
+  // schedule another render — the two property writes are pure DOM mutations.
+  // No deps array: intentional. The correction must run after every render, not just mount.
+  // SSR-safe: useLayoutEffect is silently skipped on the server; the JSX tabIndex from
+  // activeMode provides the correct initial value (focusedModeRef.current === activeMode
+  // at mount). Strict Mode: runs twice on remount — both runs write the same values,
+  // no state mutation, no render triggered.
+  useLayoutEffect(() => {
+    if (previewTabRef.current) {
+      previewTabRef.current.tabIndex = focusedModeRef.current === "preview" ? 0 : -1;
+    }
+    if (sourceTabRef.current) {
+      sourceTabRef.current.tabIndex = focusedModeRef.current === "source" ? 0 : -1;
+    }
+  });
+
   const rawContent = editor.editing
     ? editor.session?.draft ?? ""
     : (typeof content === "string" ? content : "");
+
   const previewContent = useMemo(() => {
     if (!editor.editing) return rawContent;
     const lines = rawContent.split(/\r?\n/);
@@ -87,7 +203,9 @@ export function MarkdownContentViewer({
     const closingIndex = lines.findIndex((line, index) => index > 0 && line === "---");
     return closingIndex < 0 ? rawContent : lines.slice(closingIndex + 1).join("\n").replace(/^\n/, "");
   }, [editor.editing, rawContent]);
+
   const hasContent = rawContent.trim().length > 0;
+
   const renderedMarkdown = useMemo(
     () => (
       <ReactMarkdown remarkPlugins={REMARK_PLUGINS} components={MARKDOWN_COMPONENTS}>
@@ -96,6 +214,7 @@ export function MarkdownContentViewer({
     ),
     [previewContent],
   );
+
   const handleCopy = async () => {
     if (!hasContent) return;
     try {
@@ -110,33 +229,37 @@ export function MarkdownContentViewer({
 
   const handleEdit = async () => {
     modeBeforeEditRef.current = activeMode;
-    setActiveMode("source");
+    selectMode("source");
     if (!await editor.beginEdit()) {
-      setActiveMode(modeBeforeEditRef.current);
+      selectMode(modeBeforeEditRef.current);
     }
   };
 
   const handleCancel = () => {
     editor.discard();
-    setActiveMode(modeBeforeEditRef.current);
+    selectMode(modeBeforeEditRef.current);
   };
 
   const handleApply = async () => {
     if (await editor.save()) {
-      setActiveMode("preview");
+      selectMode("preview");
     }
   };
 
   return (
     <div className={className} style={viewerContainerStyle}>
       <div style={viewerHeaderStyle}>
-        <div style={{ display: "flex", gap: 4 }} role="tablist" aria-label="Markdown view">
+        <div style={{ display: "flex", gap: 4 }} role="tablist" aria-label="Content view mode">
           <button
             type="button"
             role="tab"
+            id={previewTabId}
+            ref={previewTabRef}
             aria-selected={activeMode === "preview"}
-            aria-controls="markdown-viewer-panel"
-            onClick={() => setActiveMode("preview")}
+            aria-controls={panelId}
+            tabIndex={activeMode === "preview" ? 0 : -1}
+            onClick={() => selectMode("preview")}
+            onKeyDown={handleTabKeyDown}
             style={{ ...tabBtnStyle, ...(activeMode === "preview" ? activeTabBtnStyle : {}) }}
           >
             <Eye size={12} style={{ marginRight: 5 }} aria-hidden="true" />
@@ -145,9 +268,13 @@ export function MarkdownContentViewer({
           <button
             type="button"
             role="tab"
+            id={sourceTabId}
+            ref={sourceTabRef}
             aria-selected={activeMode === "source"}
-            aria-controls="markdown-viewer-panel"
-            onClick={() => setActiveMode("source")}
+            aria-controls={panelId}
+            tabIndex={activeMode === "source" ? 0 : -1}
+            onClick={() => selectMode("source")}
+            onKeyDown={handleTabKeyDown}
             style={{ ...tabBtnStyle, ...(activeMode === "source" ? activeTabBtnStyle : {}) }}
           >
             <Code2 size={12} style={{ marginRight: 5 }} aria-hidden="true" />
@@ -220,10 +347,20 @@ export function MarkdownContentViewer({
         </div>
       ) : null}
 
+      {/* Both tabs point aria-controls at this one panel: only the active view is
+          ever rendered inside (edit/preview/source/empty), so per-tab panel ids
+          would leave the inactive tab referencing an element not in the DOM.
+          Wrapping all branches — including empty state and editing textarea —
+          keeps every aria-controls reference resolvable at all times.
+          tabIndex=0 because the panel is a scroll container (viewerBodyStyle caps
+          its height), so keyboard users need to be able to focus and scroll it.
+          aria-busy signals to assistive tech that the content is loading/saving. */}
       <div
-        id="markdown-viewer-panel"
         role="tabpanel"
+        id={panelId}
+        aria-labelledby={activeMode === "preview" ? previewTabId : sourceTabId}
         aria-busy={saving || loading}
+        tabIndex={0}
         style={viewerBodyStyle}
       >
         {activeMode === "source" && editing ? (

explorer/src/workspaces/GraphWorkspace/graphColorLegend.ts

@@ -0,0 +1,36 @@
+import type Graph from "graphology";
+
+import type { NodeAttributes } from "../../store/graphStore";
+import { GRAPH_THEME, type GraphTheme } from "./graphTheme";
+
+export type GraphColorLegendItem = {
+  id: string;
+  group: string;
+  color: string;
+  count: number;
+};
+
+/** Normal semantic color, before selection, distance, or zoom styling. */
+export function getSemanticNodeColor(
+  attrs: Pick<NodeAttributes, "baseColor" | "color">,
+  theme: GraphTheme = GRAPH_THEME,
+) {
+  return String(attrs.baseColor || attrs.color || theme.palette.semantic[0]);
+}
+
+/** Use the displayed graph so synthetic groups and filtered views keep their colors. */
+export function buildGraphColorLegend(graph: Graph, theme: GraphTheme = GRAPH_THEME): GraphColorLegendItem[] {
+  const entries = new Map<string, GraphColorLegendItem>();
+  graph.forEachNode((_id, attrs) => {
+    if (attrs.hidden) return;
+    const group = String(attrs.semanticGroup || attrs.nodeType || "entity");
+    // Focused clones bake interaction colors into baseColor; keep those out of the semantic key.
+    const color = String(attrs.semanticBaseColor || getSemanticNodeColor(attrs as NodeAttributes, theme));
+    // A synthetic display node can share a semantic label with a different color.
+    const id = JSON.stringify([group, color]);
+    const current = entries.get(id);
+    if (current) current.count += 1;
+    else entries.set(id, { id, group, color, count: 1 });
+  });
+  return [...entries.values()].sort((a, b) => a.group.localeCompare(b.group) || a.color.localeCompare(b.color));
+}

explorer/src/workspaces/GraphWorkspace/graphSceneState.ts

@@ -19,6 +19,7 @@ import {
   withAlpha,
   zoomTierAtLeast,
 } from "./graphTheme";
+import { getSemanticNodeColor } from "./graphColorLegend";
 import { classifyEntityShape } from "./graphEntityShape";
 import { computeGraphAnalyticsBase } from "./graphAnalytics";
 import type {
@@ -1013,9 +1014,8 @@ function resolveNodeColor(
   state: GraphNodeVisualState,
   attrs: NodeAttributes,
   cameraRatio: number,
-  fallbackColor?: string,
 ) {
-  const semanticColor = String(attrs.baseColor || fallbackColor || theme.palette.semantic[0]);
+  const semanticColor = getSemanticNodeColor(attrs, theme);
   const isCommunityGroup = Boolean(attrs.isCommunityGroup);
   const entityShapeConfig = theme.nodes.entityShapes[resolveEntityShape(attrs)];
   const overviewTint = state === "neighbor"
@@ -1062,9 +1062,8 @@ function resolveNodeShellColor(
   state: GraphNodeVisualState,
   attrs: NodeAttributes,
   cameraRatio: number,
-  fallbackColor?: string,
 ) {
-  const semanticColor = String(attrs.baseColor || fallbackColor || theme.palette.semantic[0]);
+  const semanticColor = getSemanticNodeColor(attrs, theme);
   const isCommunityGroup = Boolean(attrs.isCommunityGroup);
   const entityShapeConfig = theme.nodes.entityShapes[resolveEntityShape(attrs)];
   const presenceBoost = getOverviewPresenceBoost(cameraRatio);
@@ -1633,8 +1632,8 @@ export function resolveNodeElementStyle(
   const isCommunityGroup = Boolean(attrs.isCommunityGroup);
   const baseSize = Number(attrs.baseSize || attrs.size || 4);
   const labelPriority = Number(attrs.labelPriority ?? 0);
-  const color = resolveNodeColor(theme, zoomTier, state, attrs, cameraRatio, attrs.color);
-  const shellColor = resolveNodeShellColor(theme, zoomTier, state, attrs, cameraRatio, attrs.color);
+  const color = resolveNodeColor(theme, zoomTier, state, attrs, cameraRatio);
+  const shellColor = resolveNodeShellColor(theme, zoomTier, state, attrs, cameraRatio);
   const sizeMultiplier = (state === "default" ? tierConfig.nodeScale : stateConfig.sizeMultiplier)
     * variantConfig.sizeMultiplier
     * (isCommunityGroup ? theme.grouped.style.nodeSizeScale : 1);
@@ -2570,6 +2569,7 @@ export function createFocusedGraph(
     color: selectedState.color,
     size: Math.max(selectedState.size, 22),
     baseColor: selectedState.color,
+    semanticBaseColor: getSemanticNodeColor(selectedAttrs),
     baseSize: Math.max(selectedState.size, 22),
     label: selectedState.label,
   });
@@ -2609,6 +2609,7 @@ export function createFocusedGraph(
       color: style.color,
       size: Math.max(style.size, 8.5),
       baseColor: style.color,
+      semanticBaseColor: getSemanticNodeColor(baseAttrs),
       baseSize: Math.max(style.size, 8.5),
       label: style.label,
     });

explorer/src/workspaces/GraphWorkspace/plugins/explorationEffectsPluginPhaseC.tsx

@@ -1,5 +1,6 @@
 import type { CSSProperties } from "react";
 
+import { buildGraphColorLegend } from "../graphColorLegend";
 import type {
   GraphAnalyticsSnapshot,
   GraphDiagnosticsSnapshot,
@@ -104,19 +105,7 @@ function renderAvailabilityText(availability: GraphEffectAvailability) {
 }
 
 function collectFallbackLegendItems(context: Parameters<NonNullable<GraphPlugin["renderPanel"]>>[0]) {
-  const groups = new Map<string, { count: number; color: string }>();
-  context.graph.forEachNode((_nodeId, attrs) => {
-    const semanticGroup = String(attrs.semanticGroup || attrs.nodeType || "entity");
-    const color = String(attrs.baseColor || context.theme.palette.semantic[0]);
-    const current = groups.get(semanticGroup);
-    groups.set(semanticGroup, {
-      count: (current?.count ?? 0) + 1,
-      color,
-    });
-  });
-
-  return [...groups.entries()]
-    .map(([group, data]) => ({ group, ...data }))
+  return buildGraphColorLegend(context.graph, context.theme)
     .sort((left, right) => right.count - left.count)
     .slice(0, context.theme.effects.legend.maxGroups);
 }
@@ -255,7 +244,7 @@ function renderRegionsAndSignals(
         <div style={{ display: "flex", flexDirection: "column", gap: 8 }}>
           <div style={subsectionTitleStyle}>Fallback semantic legend</div>
           {fallbackLegendItems.map((item) => (
-            <div key={item.group} style={legendRowStyle}>
+            <div key={item.id} style={legendRowStyle}>
               <span
                 style={{
                   ...legendSwatchStyle,

explorer/src/workspaces/GraphWorkspace/plugins/legendPlugin.tsx

@@ -1,5 +1,6 @@
 import type { CSSProperties } from "react";
 
+import { buildGraphColorLegend } from "../graphColorLegend";
 import type { GraphPlugin } from "./types";
 
 const LEGEND_PANEL_ID = "legend-panel";
@@ -25,19 +26,7 @@ export const legendPlugin: GraphPlugin = {
       return null;
     }
 
-    const groups = new Map<string, { count: number; color: string }>();
-    context.graph.forEachNode((_nodeId, attrs) => {
-      const semanticGroup = String(attrs.semanticGroup || attrs.nodeType || "entity");
-      const color = String(attrs.baseColor || context.theme.palette.semantic[0]);
-      const current = groups.get(semanticGroup);
-      groups.set(semanticGroup, {
-        count: (current?.count ?? 0) + 1,
-        color,
-      });
-    });
-
-    const items = [...groups.entries()]
-      .map(([group, data]) => ({ group, ...data }))
+    const items = buildGraphColorLegend(context.graph, context.theme)
       .sort((left, right) => right.count - left.count)
       .slice(0, MAX_GROUPS);
 
@@ -55,7 +44,7 @@ export const legendPlugin: GraphPlugin = {
           {items.length ? (
             <div style={{ display: "flex", flexDirection: "column", gap: 8 }}>
               {items.map((item) => (
-                <div key={item.group} style={legendRowStyle}>
+                <div key={item.id} style={legendRowStyle}>
                   <span
                     style={{
                       ...swatchStyle,

explorer/tests/graphColorLegend.e2e.ts

@@ -0,0 +1,112 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { existsSync } from "node:fs";
+import { setTimeout as delay } from "node:timers/promises";
+import test from "node:test";
+import { chromium, type Page } from "playwright";
+
+const BASE_URL = "http://127.0.0.1:4175";
+const initialNodes = [
+  { id: "alice", type: "Person", content: "Alice", properties: {} },
+  { id: "bob", type: "Person", content: "Bob", properties: {} },
+  { id: "acme", type: "Organization", content: "Acme", properties: {} },
+  { id: "london", type: "Location", content: "London", properties: {} },
+  { id: "research", type: "Project", content: "Research", properties: {} },
+  { id: "report", type: "Document", content: "Report", properties: {} },
+];
+const edges = [
+  ["alice", "acme", "WORKS_AT"], ["bob", "acme", "WORKS_AT"],
+  ["acme", "london", "LOCATED_IN"], ["alice", "research", "LEADS"],
+  ["bob", "report", "AUTHORED"], ["report", "research", "DESCRIBES"],
+].map(([source, target, type], i) => ({
+  id: `edge_${i}`, familyId: `edge_${i}`, source, target, type, weight: 1, properties: {},
+}));
+
+async function assertLegendMatchesGraph(page: Page, nodeIds?: string[]) {
+  const result = await page.evaluate(async (includedNodeIds) => {
+    const storePath = "/src/store/graphStore.ts";
+    const { graph } = await import(storePath);
+    const colors: Record<string, string> = {};
+    graph.forEachNode((id: string, attrs: { semanticGroup: string; baseColor: string }) => {
+      if (includedNodeIds && !includedNodeIds.includes(id)) return;
+      const hex = attrs.baseColor.replace("#", "");
+      colors[attrs.semanticGroup] = `rgb(${[0, 2, 4].map((i) => parseInt(hex.slice(i, i + 2), 16)).join(", ")})`;
+    });
+    const items = [...document.querySelectorAll(".explore-color-legend-item")].map((item) => ({
+      group: item.querySelector(".explore-color-legend-name")?.textContent,
+      color: getComputedStyle(item.querySelector(".explore-color-legend-mark")!).backgroundColor,
+    }));
+    return { colors, items };
+  }, nodeIds);
+  assert.equal(result.items.length, Object.keys(result.colors).length);
+  for (const item of result.items) {
+    assert.equal(item.color, result.colors[item.group!], `Swatch for ${item.group} must match the loaded canvas color`);
+  }
+}
+
+test("visible legend follows loaded data, reloads, focused views, and distance mode", async (t) => {
+  const server = spawn("npm", ["run", "dev", "--", "--host", "127.0.0.1", "--port", "4175", "--strictPort"], { stdio: "ignore" });
+  t.after(() => { server.kill(); });
+  let ready = false;
+  for (let i = 0; i < 100; i += 1) {
+    try { if ((await fetch(BASE_URL)).ok) { ready = true; break; } } catch { /* Starting Vite. */ }
+    await delay(100);
+  }
+  assert.ok(ready, "Vite must start");
+  const browser = await chromium.launch({
+    headless: true,
+    executablePath:
+      process.env.CHROMIUM_PATH || (existsSync("/usr/bin/chromium") ? "/usr/bin/chromium" : undefined),
+  });
+  t.after(() => browser.close());
+  const page = await browser.newPage({ viewport: { width: 1440, height: 1000 } });
+  page.setDefaultTimeout(10_000);
+  const errors: string[] = [];
+  page.on("pageerror", (error) => errors.push(error.message));
+  page.on("console", (message) => { if (message.type() === "error") errors.push(message.text()); });
+  let nodes = initialNodes;
+  await page.routeWebSocket("**/ws/graph-updates", () => {});
+  await page.route("**/api/**", async (route) => {
+    const path = new URL(route.request().url()).pathname;
+    let json: unknown = {};
+    if (path === "/api/info") json = { capabilities: { agent_memory: false } };
+    if (path === "/api/graph/stats") json = { node_count: nodes.length, edge_count: edges.length };
+    if (path === "/api/graph/nodes") json = { nodes, total: nodes.length, next_cursor: null };
+    if (path === "/api/graph/edges") json = { edges, total: edges.length, next_cursor: null };
+    if (path === "/api/temporal/bounds") json = { min: null, max: null };
+    if (path === "/api/temporal/snapshot") json = { active_node_ids: nodes.map((n) => n.id), active_node_count: nodes.length };
+    if (path === "/api/graph/search") json = { results: [{ node: nodes[0], score: 1 }] };
+    await route.fulfill({ json });
+  });
+  await page.goto(BASE_URL);
+  await page.getByRole("button", { name: "Open Semantica Explorer" }).click();
+  const legend = page.getByRole("group", { name: "Node colors" });
+  await legend.waitFor();
+  await page.locator("canvas").first().waitFor({ state: "visible" });
+  await assertLegendMatchesGraph(page);
+  assert.equal(await legend.getByText("Person", { exact: true }).count(), 1);
+  assert.equal(await legend.getByText("Biomolecule", { exact: true }).count(), 0);
+
+  nodes = initialNodes.map((node) => ({ ...node, type: node.type === "Person" ? "Researcher" : node.type }));
+  await page.getByRole("button", { name: "Reload graph data" }).click();
+  await legend.getByText("Researcher", { exact: true }).waitFor();
+  assert.equal(await legend.getByText("Person", { exact: true }).count(), 0);
+  await assertLegendMatchesGraph(page);
+
+  await page.getByPlaceholder("Search command, node, or concept").fill("Alice");
+  await page.getByRole("option").filter({ hasText: "Alice" }).click();
+  const heatmap = page.getByRole("button", { name: "Heatmap", exact: true });
+  await heatmap.click();
+  await legend.waitFor({ state: "hidden" });
+  await heatmap.click();
+  await legend.waitFor();
+  await assertLegendMatchesGraph(page);
+  await page.getByRole("button", { name: "Focused", exact: true }).click();
+  await legend.getByText("Document", { exact: true }).waitFor({ state: "hidden" });
+  await assertLegendMatchesGraph(page, ["alice", "acme", "research"]);
+  assert.equal(await legend.getByText("Researcher", { exact: true }).count(), 1);
+  await page.getByRole("button", { name: "Full Graph", exact: true }).click();
+  await legend.getByText("Document", { exact: true }).waitFor();
+  await assertLegendMatchesGraph(page);
+  assert.deepEqual(errors, []);
+});

explorer/tests/graphColorLegend.test.ts

@@ -0,0 +1,103 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import Graph from "graphology";
+
+import { clearGraph, graph as sourceGraph, type NodeAttributes } from "../src/store/graphStore.ts";
+import { buildGraphColorLegend } from "../src/workspaces/GraphWorkspace/graphColorLegend.ts";
+import { resolveDisplayGraph, resolveNodeElementStyle } from "../src/workspaces/GraphWorkspace/graphSceneState.ts";
+import { GRAPH_THEME, withAlpha } from "../src/workspaces/GraphWorkspace/graphTheme.ts";
+
+function attributes(overrides: Partial<NodeAttributes> = {}): NodeAttributes {
+  return {
+    label: "Example", content: "Example", x: 0, y: 0, size: 8,
+    nodeType: "Person", semanticGroup: "Person", color: "#123456",
+    properties: {}, ...overrides,
+  };
+}
+
+test("legend matches normal canvas colors, including color and theme fallbacks", () => {
+  const graph = new Graph();
+  const samples = [
+    attributes({ baseColor: "#abcdef" }),
+    attributes({ semanticGroup: "Organization" }),
+    attributes({ semanticGroup: "Location", color: "" }),
+  ];
+  samples.forEach((attrs, i) => graph.addNode(String(i), attrs));
+  const items = buildGraphColorLegend(graph);
+  for (const attrs of samples) {
+    const item = items.find((entry) => entry.group === attrs.semanticGroup)!;
+    const style = resolveNodeElementStyle(GRAPH_THEME, "inspection", "default", attrs, attrs.label);
+    assert.equal(style.color, withAlpha(item.color, GRAPH_THEME.nodes.entityShapes.entity.fillAlpha));
+  }
+  assert.equal(items.find((item) => item.group === "Person")?.color, "#abcdef");
+  assert.equal(items.find((item) => item.group === "Organization")?.color, "#123456");
+  assert.equal(items.find((item) => item.group === "Location")?.color, GRAPH_THEME.palette.semantic[0]);
+});
+
+test("semantic groups, not shape categories, determine labels and distinct entries", () => {
+  const graph = new Graph();
+  graph.addNode("one", attributes({ semanticGroup: "Research", entityShape: "compound" }));
+  graph.addNode("two", attributes({ semanticGroup: "Research", entityShape: "entity" }));
+  graph.addNode("synthetic", attributes({ semanticGroup: "Research", baseColor: "#654321", isCommunityGroup: true }));
+  graph.addNode("hidden", { ...attributes(), hidden: true });
+  assert.deepEqual(buildGraphColorLegend(graph).map(({ group, color, count }) => ({ group, color, count })), [
+    { group: "Research", color: "#123456", count: 2 },
+    { group: "Research", color: "#654321", count: 1 },
+  ]);
+  assert.equal(new Set(buildGraphColorLegend(graph).map((item) => item.id)).size, 2);
+});
+
+test("legend rebuilds after in-place changes and uses only the supplied display graph", () => {
+  const graph = new Graph();
+  graph.addNode("one", attributes());
+  graph.addNode("two", attributes({ semanticGroup: "Location" }));
+  const first = buildGraphColorLegend(graph);
+  graph.mergeNodeAttributes("one", { semanticGroup: "Project", baseColor: "#fedcba" });
+  graph.dropNode("two");
+  assert.equal(first.length, 2);
+  assert.deepEqual(buildGraphColorLegend(graph).map(({ group, color }) => ({ group, color })), [
+    { group: "Project", color: "#fedcba" },
+  ]);
+  graph.clear();
+  assert.deepEqual(buildGraphColorLegend(graph), []);
+});
+
+test("fallback labels and ordering are stable and no groups are silently dropped", () => {
+  const graph = new Graph();
+  graph.addNode("fallback", attributes({ semanticGroup: undefined, nodeType: "" }));
+  for (let i = 11; i >= 0; i -= 1) graph.addNode(String(i), attributes({ semanticGroup: undefined, nodeType: `Type ${i}` }));
+  const items = buildGraphColorLegend(graph);
+  assert.equal(items.length, 13);
+  assert.ok(items.some((item) => item.group === "entity"));
+  const reverse = new Graph();
+  graph.nodes().reverse().forEach((id) => reverse.addNode(id, graph.getNodeAttributes(id)));
+  assert.deepEqual(items, buildGraphColorLegend(reverse));
+});
+
+
+test("focused legend keeps semantic colors for selected, path, and neighbor clones", (t) => {
+  clearGraph();
+  t.after(clearGraph);
+  for (const id of ["selected", "path", "neighbor", "outside"]) {
+    sourceGraph.addNode(id, attributes({ label: id, baseColor: "#abcdef" }));
+  }
+  sourceGraph.addDirectedEdgeWithKey("path-edge", "selected", "path", { weight: 1 });
+  sourceGraph.addDirectedEdgeWithKey("neighbor-edge", "selected", "neighbor", { weight: 1 });
+  const focused = resolveDisplayGraph("selected", ["selected", "path"], ["path-edge"], "focused").graph;
+
+  // Verify the fixture exercises baked interaction colors, not ordinary clones.
+  assert.equal(focused.getNodeAttribute("selected", "baseColor"), GRAPH_THEME.palette.accent.selected);
+  assert.equal(focused.getNodeAttribute("path", "baseColor"), GRAPH_THEME.palette.accent.path);
+  assert.notEqual(focused.getNodeAttribute("neighbor", "baseColor"), "#abcdef");
+  assert.ok(!focused.hasNode("outside"));
+  assert.deepEqual(buildGraphColorLegend(focused).map(({ group, color, count }) => ({ group, color, count })), [
+    { group: "Person", color: "#abcdef", count: 3 },
+  ]);
+  assert.equal(sourceGraph.getNodeAttribute("selected", "baseColor"), "#abcdef");
+
+  // Changing focus retains the semantic swatch while following the displayed subset.
+  const nextFocus = resolveDisplayGraph("neighbor", [], [], "focused").graph;
+  assert.deepEqual(buildGraphColorLegend(nextFocus).map(({ color, count }) => ({ color, count })), [
+    { color: "#abcdef", count: 2 },
+  ]);
+});

explorer/tests/markdownContentViewer.test.ts

@@ -286,3 +286,335 @@ test("copy button always starts in un-copied state on initial render", () => {
   assert.equal(html.includes("Copy"), true, "Copy button must be present on initial render");
   assert.equal(html.includes("Copied"), false, "Copied indicator must NOT be present on initial render");
 });
+
+// ─── #1117: complete ARIA tab/tabpanel relationship ─────────────────────────
+// The tabs previously exposed role/aria-selected but never connected to the
+// panel, so assistive tech could not tell which content the tabs controlled.
+// These assertions read the rendered HTML, matching the aria-label precedent
+// used by the GFM footnote tests above.
+
+/** Pull an attribute value out of the element carrying a given marker attribute. */
+function attrOf(html: string, elementMarker: string, attr: string): string | null {
+  const idx = html.indexOf(elementMarker);
+  if (idx === -1) return null;
+  const tagStart = html.lastIndexOf("<", idx);
+  const tag = html.slice(tagStart, html.indexOf(">", idx) + 1);
+  const m = tag.match(new RegExp(`${attr}="([^"]*)"`));
+  return m ? m[1] : null;
+}
+
+test("each tab is wired to the panel and the panel back to the active tab", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Node\n\nBody text.",
+    defaultMode: "preview",
+  }));
+
+  const panelId = attrOf(html, 'role="tabpanel"', "id");
+  assert.ok(panelId, "panel must carry an id");
+
+  // Both tabs must reference the panel that actually exists in the DOM.
+  const controls = [...html.matchAll(/aria-controls="([^"]*)"/g)].map((m) => m[1]);
+  assert.equal(controls.length, 2, "both tabs must declare aria-controls");
+  for (const c of controls) {
+    assert.equal(c, panelId, "aria-controls must resolve to the rendered panel");
+  }
+
+  // The panel must be labelled by the *selected* tab.
+  const labelledBy = attrOf(html, 'role="tabpanel"', "aria-labelledby");
+  const selectedTabId = attrOf(html, 'aria-selected="true"', "id");
+  assert.ok(selectedTabId, "selected tab must carry an id");
+  assert.equal(labelledBy, selectedTabId, "panel must be labelled by the selected tab");
+});
+
+test("panel labelling follows the active tab in source mode", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Node\n\nBody text.",
+    defaultMode: "source",
+  }));
+  const labelledBy = attrOf(html, 'role="tabpanel"', "aria-labelledby");
+  const selectedTabId = attrOf(html, 'aria-selected="true"', "id");
+  // Assert both are present before comparing — otherwise null === null would
+  // make this pass against a component with no tab wiring at all.
+  assert.ok(labelledBy, "panel must declare aria-labelledby");
+  assert.ok(selectedTabId, "selected tab must carry an id");
+  assert.equal(labelledBy, selectedTabId);
+  assert.equal(selectedTabId.endsWith("-tab-source"), true, "source tab must be the selected one");
+});
+
+// The empty state is a third render branch. If the panel only existed on the two
+// content branches, aria-controls would dangle for empty nodes.
+test("tabpanel is still rendered, and aria-controls still resolves, when empty", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, { content: "" }));
+  assert.equal(html.includes("No content available for this node."), true);
+  const panelId = attrOf(html, 'role="tabpanel"', "id");
+  assert.ok(panelId, "empty state must still render the tabpanel");
+  const controls = [...html.matchAll(/aria-controls="([^"]*)"/g)].map((m) => m[1]);
+  assert.equal(controls.length, 2);
+  assert.deepEqual([...new Set(controls)], [panelId], "aria-controls must not dangle on the empty state");
+});
+
+test("tablist is a single tab stop via roving tabindex", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Node",
+    defaultMode: "preview",
+  }));
+  const tabIndexes = [...html.matchAll(/role="tab"[^>]*/g)].map((m) => m[0].match(/tabindex="(-?\d+)"/)?.[1]);
+  assert.equal(tabIndexes.filter((t) => t === "0").length, 1, "exactly one tab may be reachable via Tab");
+  assert.equal(tabIndexes.filter((t) => t === "-1").length, 1, "the other tab must be removed from tab order");
+});
+
+test("ids are unique per instance so two mounted viewers cannot collide", () => {
+  const one = renderToString(React.createElement(MarkdownContentViewer, { content: "# A" }));
+  const two = renderToString(React.createElement(
+    "div",
+    null,
+    React.createElement(MarkdownContentViewer, { content: "# A" }),
+    React.createElement(MarkdownContentViewer, { content: "# B" }),
+  ));
+  assert.ok(attrOf(one, 'role="tabpanel"', "id"));
+  const panelIds = [...two.matchAll(/role="tabpanel" id="([^"]*)"/g)].map((m) => m[1]);
+  assert.equal(panelIds.length, 2, "both viewers must render a panel");
+  assert.notEqual(panelIds[0], panelIds[1], "panel ids must differ between instances");
+});
+
+import { resolveTabNavigation } from "../src/workspaces/GraphWorkspace/MarkdownContentViewer.tsx";
+
+// ─── #1117 / Qodo: roving-tabindex navigation logic ─────────────────────────
+// resolveTabNavigation is the pure function that drives handleTabKeyDown.
+// Testing it directly gives us coverage of the navigation contract without
+// needing a live DOM or synthetic keyboard events.
+
+// ── ArrowRight moves focus forward, wraps at end ────────────────────────────
+test("ArrowRight from preview moves focus to source without wrapping", () => {
+  assert.equal(resolveTabNavigation("preview", "ArrowRight"), "source");
+});
+
+test("ArrowRight from source wraps back to preview", () => {
+  // With only two tabs the rightmost tab wraps to the first.
+  assert.equal(resolveTabNavigation("source", "ArrowRight"), "preview");
+});
+
+// ── ArrowLeft moves focus backward, wraps at start ──────────────────────────
+test("ArrowLeft from source moves focus to preview without wrapping", () => {
+  assert.equal(resolveTabNavigation("source", "ArrowLeft"), "preview");
+});
+
+test("ArrowLeft from preview wraps back to source", () => {
+  // The leftmost tab wraps to the last.
+  assert.equal(resolveTabNavigation("preview", "ArrowLeft"), "source");
+});
+
+// ── Home and End always resolve to the boundary tabs ────────────────────────
+test("Home always moves focus to the first tab (preview)", () => {
+  assert.equal(resolveTabNavigation("preview", "Home"), "preview", "Home on first tab stays at first");
+  assert.equal(resolveTabNavigation("source", "Home"), "preview", "Home on last tab jumps to first");
+});
+
+test("End always moves focus to the last tab (source)", () => {
+  assert.equal(resolveTabNavigation("source", "End"), "source", "End on last tab stays at last");
+  assert.equal(resolveTabNavigation("preview", "End"), "source", "End on first tab jumps to last");
+});
+
+// ── Non-navigation keys return null so the handler can bail out ─────────────
+test("non-navigation keys return null so keydown handler does not move focus", () => {
+  for (const key of ["Enter", "Space", " ", "Tab", "Escape", "a", "F1"]) {
+    assert.equal(
+      resolveTabNavigation("preview", key),
+      null,
+      `key "${key}" must return null`,
+    );
+    assert.equal(
+      resolveTabNavigation("source", key),
+      null,
+      `key "${key}" on source must return null`,
+    );
+  }
+});
+
+// ── Arrow navigation does NOT change activeMode (manual activation) ──────────
+// resolveTabNavigation only returns the target for focus movement. The caller
+// (focusTab) imperatively updates tabIndex and moves DOM focus without calling
+// setActiveMode. We verify the contract: resolveTabNavigation never returns a
+// value that could be interpreted as "activate" — it just returns a tab identity.
+// The absence of a setActiveMode call in focusTab is what enforces manual
+// activation; these tests confirm the logic layer does not accidentally activate.
+test("resolveTabNavigation return value is purely a focus target, never an activation signal", () => {
+  // A real activation calls setActiveMode. resolveTabNavigation just computes
+  // the next focused tab. If the caller only updates focusedModeRef + DOM tabIndex,
+  // activeMode remains unchanged. This test asserts the function's return contract.
+  const result = resolveTabNavigation("preview", "ArrowRight");
+  assert.equal(typeof result, "string", "returns a string tab name when key is a navigation key");
+  assert.notEqual(result, null, "non-null means 'move focus here'");
+  // The returned value is a valid tab mode, not a command to switch content.
+  assert.ok(result === "preview" || result === "source");
+});
+
+// ── Roving tabindex initial state for defaultMode='source' ──────────────────
+// The existing 'tablist is a single tab stop' test only checks defaultMode='preview'.
+// When the component starts in source mode the source tab must start at tabIndex 0.
+test("roving tabindex initial state is correct when defaultMode is source", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Node",
+    defaultMode: "source",
+  }));
+  const tabIndexes = [...html.matchAll(/role="tab"[^>]*/g)].map(
+    (m) => m[0].match(/tabindex="(-?\d+)"/)?.[1],
+  );
+  // There are exactly two tabs; one must be 0, the other -1.
+  assert.equal(tabIndexes.filter((t) => t === "0").length, 1, "exactly one tab is reachable via Tab");
+  assert.equal(tabIndexes.filter((t) => t === "-1").length, 1, "the other tab is removed from tab order");
+
+  // The source tab specifically must hold tabIndex 0 (it is the focused/active one).
+  // We identify the source tab by its id suffix and verify its tabindex.
+  const sourceTabMatch = [...html.matchAll(/role="tab"[^>]*/g)].find((m) =>
+    m[0].includes("-tab-source"),
+  );
+  assert.ok(sourceTabMatch, "source tab must be present in rendered HTML");
+  assert.equal(
+    sourceTabMatch[0].match(/tabindex="(-?\d+)"/)?.[1],
+    "0",
+    "source tab must have tabIndex 0 when defaultMode is source",
+  );
+});
+
+// ── aria-labelledby correctness for each defaultMode ────────────────────────
+// These tests verify the static wiring; the existing tests cover preview and
+// source modes, so these act as a consolidated regression check that both
+// directions of the panel labelling contract hold after the refactor.
+test("panel aria-labelledby matches the selected tab in preview mode after refactor", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Refactor check",
+    defaultMode: "preview",
+  }));
+  const labelledBy = attrOf(html, 'role="tabpanel"', "aria-labelledby");
+  const selectedTabId = attrOf(html, 'aria-selected="true"', "id");
+  assert.ok(labelledBy, "panel must carry aria-labelledby after refactor");
+  assert.ok(selectedTabId, "a tab must be aria-selected=true after refactor");
+  assert.equal(labelledBy, selectedTabId, "panel must be labelled by the selected tab");
+  assert.ok(selectedTabId.endsWith("-tab-preview"), "preview tab must be selected");
+});
+
+test("panel aria-labelledby matches the selected tab in source mode after refactor", () => {
+  const html = renderToString(React.createElement(MarkdownContentViewer, {
+    content: "# Refactor check",
+    defaultMode: "source",
+  }));
+  const labelledBy = attrOf(html, 'role="tabpanel"', "aria-labelledby");
+  const selectedTabId = attrOf(html, 'aria-selected="true"', "id");
+  assert.ok(labelledBy, "panel must carry aria-labelledby after refactor");
+  assert.ok(selectedTabId, "a tab must be aria-selected=true after refactor");
+  assert.equal(labelledBy, selectedTabId, "panel must be labelled by the selected tab");
+  assert.ok(selectedTabId.endsWith("-tab-source"), "source tab must be selected");
+});
+
+// ── Qodo performance regression: focusedModeRef is a ref, not state ──────────
+// The confirmed bug was: setFocusedMode (useState setter) caused a re-render
+// on every arrow keypress, which triggered react-markdown's full parse+runSync
+// cycle even though activeMode did not change.
+//
+// The fix uses useRef instead of useState for the focused-mode tracking. Refs
+// do not schedule re-renders when mutated. We cannot directly count React
+// renders inside renderToString (it runs synchronously, once). What we CAN
+// verify is the structural invariant that makes the fix work:
+//
+//   1. The component renders identically for the same props on successive
+//      renderToString calls (no hidden state that would differ if focusedMode
+//      were state vs ref — both start at defaultMode on fresh mount).
+//   2. The tabIndex JSX prop reads from focusedModeRef.current which equals
+//      defaultMode on initial render. This is the same output the old code
+//      produced, so no regression in SSR output.
+//
+// Full verification of "arrow key press does NOT trigger ReactMarkdown.parse()"
+// requires a live DOM + render-count instrumentation. That test belongs in an
+// interactive framework (Playwright component test or jsdom + Testing Library)
+// which is not installed in this project. The structural guarantee provided by
+// the ref-based implementation is documented here for that future test to pin.
+test("successive renderToString calls produce identical tabIndex output (ref parity with state)", () => {
+  const props = { content: "# Perf node\n\n" + "row. ".repeat(200), defaultMode: "preview" as const };
+  const first  = renderToString(React.createElement(MarkdownContentViewer, props));
+  const second = renderToString(React.createElement(MarkdownContentViewer, props));
+  // Both renders start with a fresh ref initialised to defaultMode, so output
+  // must be byte-for-byte identical (modulo React's useId counter which advances
+  // per call — we compare structure, not the specific id values).
+  const extractTabIndexes = (html: string) =>
+    [...html.matchAll(/role="tab"[^>]*/g)].map((m) => m[0].match(/tabindex="(-?\d+)"/)?.[1]);
+  assert.deepEqual(
+    extractTabIndexes(first),
+    extractTabIndexes(second),
+    "tabIndex values must be the same on every fresh mount with the same defaultMode",
+  );
+  // Verify content is actually rendered (not an empty-state shortcut).
+  assert.ok(first.includes("Perf node"), "markdown content must be rendered");
+});
+
+// ── Regression guard: tabIndex-reset-on-re-render (useLayoutEffect fix) ──────
+//
+// The adversarial review identified a concrete bug: after ArrowRight moves focus
+// to Source while Preview remains selected (activeMode='preview'), any subsequent
+// React re-render applied JSX tabIndex={activeMode === X} and overwrote the
+// imperative tabIndex values set by focusTab(), reverting focus tracking to the
+// selection state.
+//
+// Fix: useLayoutEffect(() => { ... }) with no deps array, which runs after every
+// React render and restores focusedModeRef.current to the DOM before paint.
+//
+// WHY THIS CANNOT BE TESTED WITH renderToString:
+//   The fix is a client-side DOM mutation applied by useLayoutEffect. On the
+//   server, useLayoutEffect is silently skipped (React design: effects do not run
+//   during SSR). renderToString produces only the initial HTML, which correctly
+//   reflects activeMode === focusedModeRef.current at mount time. It cannot
+//   simulate: (a) a keydown event that calls focusTab(), (b) a subsequent
+//   state-update re-render, or (c) the useLayoutEffect correction after that
+//   render. The full sequence requires a live DOM with React hydrated and event
+//   dispatch — either jsdom + React Testing Library, or Playwright component
+//   tests. Neither is installed in this project.
+//
+// WHAT WE CAN VERIFY (SSR-compatible proxies):
+//   1. The fix is mechanical: useLayoutEffect reads focusedModeRef.current and
+//      writes it unconditionally to the DOM. The only way it fails is if:
+//      (a) focusedModeRef.current is wrong — covered by the navigation logic tests.
+//      (b) useLayoutEffect is not called — impossible if it is in the component body
+//          unconditionally.
+//      (c) The ref assignment in focusTab() is skipped — covered by the imperative
+//          DOM update tests (focusTab sets the ref before calling .focus()).
+//   2. We verify the structural guarantee: on initial render focusedModeRef.current
+//      equals defaultMode, so JSX and useLayoutEffect agree, and no visible change
+//      occurs. This is the only SSR-observable aspect of the fix.
+//
+// TRACKING: Add a jsdom/Playwright test for the full sequence as a follow-up.
+// The specific scenario to pin:
+//   Preview selected → focusTab('source') → re-render (e.g. setCopied) →
+//   useLayoutEffect runs → sourceTab.tabIndex === 0 AND previewTab.tabIndex === -1.
+
+test("tabIndex regression (SSR proxy): initial focusedModeRef matches defaultMode so JSX and useLayoutEffect agree on mount", () => {
+  // On initial mount focusedModeRef.current = defaultMode and activeMode = defaultMode,
+  // so both the JSX tabIndex expression and the useLayoutEffect correction write
+  // identical values. There is no visible disagreement at first render.
+  // This confirms the static foundation the fix relies on.
+  for (const mode of ["preview", "source"] as const) {
+    const html = renderToString(React.createElement(MarkdownContentViewer, {
+      content: "# Node",
+      defaultMode: mode,
+    }));
+    const tabs = [...html.matchAll(/role="tab"[^>]*/g)];
+    assert.equal(tabs.length, 2, `${mode}: both tab buttons must be present`);
+
+    const focusedTab = tabs.find((m) => m[0].includes(`-tab-${mode}`));
+    const otherTab   = tabs.find((m) => !m[0].includes(`-tab-${mode}`));
+    assert.ok(focusedTab, `${mode}: the ${mode} tab must be present`);
+    assert.ok(otherTab,   `${mode}: the other tab must be present`);
+
+    // The tab matching defaultMode must have tabIndex=0 (focused/active at mount).
+    assert.equal(
+      focusedTab[0].match(/tabindex="(-?\d+)"/)?.[1],
+      "0",
+      `${mode}: ${mode} tab must start as the single Tab stop`,
+    );
+    // The other tab must have tabIndex=-1 (removed from tab order at mount).
+    assert.equal(
+      otherTab[0].match(/tabindex="(-?\d+)"/)?.[1],
+      "-1",
+      `${mode}: the other tab must be removed from tab order at mount`,
+    );
+  }
+});

integrations/openclaw/README.md

@@ -39,7 +39,7 @@ python -m semantica.mcp_server
 openclaw gateway restart
 ```
 
-All **12 Semantica tools** are now available to any OpenClaw agent:
+All **15 Semantica tools** are now available to any OpenClaw agent:
 
 | Tool | What it does |
 |---|---|
@@ -55,6 +55,9 @@ All **12 Semantica tools** are now available to any OpenClaw agent:
 | `get_graph_analytics` | Centrality, communities, topology stats |
 | `export_graph` | Export graph (JSON, RDF, GraphML, …) |
 | `get_graph_summary` | High-level graph overview |
+| `query_graph` | Fetch a node, walk neighbours, keyword search |
+| `update_node` | Merge properties onto a node |
+| `delete_node` | Archive (soft-delete) a node |
 
 **3 resources** are also exposed: `semantica://graph/summary`, `semantica://decisions/list`, `semantica://schema/info`.
 

integrations/openclaw/__init__.py

@@ -6,7 +6,7 @@ First-class integration between the Semantica semantic intelligence stack and
 `OpenClaw <https://openclaw.ai>`_ — the open-source personal AI agent platform.
 
 OpenClaw connects to external tools via MCP (Model Context Protocol).  This
-integration exposes the full Semantica MCP surface (12 tools, 3 resources) to
+integration exposes the full Semantica MCP surface (15 tools, 3 resources) to
 any OpenClaw agent and also ships a lightweight ``OpenClawKGTool`` that can be
 dropped directly into an OpenClaw SOUL.md tool-list as a native tool.
 
@@ -37,7 +37,7 @@ restart the OpenClaw Gateway::
 
     openclaw gateway restart
 
-All 12 Semantica tools are then available as native OpenClaw agent tools.
+All 15 Semantica tools are then available as native OpenClaw agent tools.
 
 Compatibility
 -------------

integrations/openclaw/mcp_tool.py

@@ -6,7 +6,7 @@ Two integration paths:
 
 1. **MCP (recommended)** — ``OpenClawMCPConfig`` emits the ``mcporter.json``
    snippet that wires Semantica's MCP server into the OpenClaw Gateway.
-   All 12 Semantica MCP tools become native OpenClaw agent tools with no
+   All 15 Semantica MCP tools become native OpenClaw agent tools with no
    extra code.
 
 2. **REST** — ``OpenClawKGTool`` is a plain Python class that calls the

semantica/cli.py

@@ -917,11 +917,11 @@ def doctor(cli_ctx: CLIContext, local_json: bool, deep_embeddings: bool) -> None
                     for lbl, st, note, hint in checks])
             return
 
-        tbl = Table(box=_TABLE_BOX, show_edge=False, padding=(0, 2))
-        tbl.add_column("Check",  style=_KEY, no_wrap=True, min_width=16)
-        tbl.add_column("Status", no_wrap=True, min_width=6)
-        tbl.add_column("Note",   style=_DIM)
-        tbl.add_column("Hint",   style=_DIM)
+        tbl = Table(box=_TABLE_BOX, show_edge=False, padding=(0, 2), expand=True)
+        tbl.add_column("Check",  style=_KEY, no_wrap=True, min_width=34)
+        tbl.add_column("Status", no_wrap=True, min_width=4)
+        tbl.add_column("Note",   style=_DIM, min_width=15, ratio=2, overflow="fold")
+        tbl.add_column("Hint",   style=_DIM, min_width=20, ratio=3, overflow="fold")
 
         icons = {"ok": f"[{_SUCCESS}] ✓[/{_SUCCESS}]",
                  "warn": f"[{_WARN_STY}] ⚠[/{_WARN_STY}]",
@@ -1155,6 +1155,57 @@ def _get_graph_store(cli_ctx: CLIContext) -> Any:
     return GraphStore(backend=backend, **graph_db)
 
 
+def _load_rule_definitions(path: str) -> List[str]:
+    """Load reasoning rule definitions from a YAML or plain-text rules file.
+
+    YAML files may hold a list of rule strings or a mapping with a ``rules``
+    list; anything else (e.g. Datalog) is read as one rule per non-comment
+    line. The strings are handed to ``Reasoner.add_rule()`` untouched.
+    """
+    text = Path(path).read_text(encoding="utf-8")
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError:
+        data = None
+    if isinstance(data, dict):
+        rules_value = data.get("rules")
+        if rules_value is None and "rules" not in data:
+            raise click.ClickException(
+                f"Rules file '{path}' is a YAML mapping but has no 'rules' key. "
+                "Expected either a YAML list or a mapping with a 'rules' list."
+            )
+        data = rules_value
+    if isinstance(data, list):
+        return [str(item) for item in data]
+    return [line.strip() for line in text.splitlines()
+            if line.strip() and not line.lstrip().startswith("#")]
+
+
+def _graph_store_facts(cli_ctx: CLIContext) -> List[str]:
+    """Read the configured graph store into Reasoner fact strings.
+
+    Follows the same conventions ``Reasoner.add_fact()`` applies to
+    KG-style dicts: nodes become ``Label(name)`` and relationships become
+    ``TYPE(source, target)``, with internal node ids resolved to names.
+    """
+    gs = _get_graph_store(cli_ctx)
+    nodes = gs.get_nodes(limit=sys.maxsize)
+    relationships = gs.get_relationships(limit=sys.maxsize)
+    names: Dict[Any, Any] = {}
+    facts: List[str] = []
+    for node in nodes:
+        props = node.get("properties") or {}
+        name = props.get("name") or props.get("id") or node.get("id")
+        names[node.get("id")] = name
+        for label in node.get("labels") or ["Entity"]:
+            facts.append(f"{label}({name})")
+    for rel in relationships:
+        source = names.get(rel.get("start_node_id"), rel.get("start_node_id"))
+        target = names.get(rel.get("end_node_id"), rel.get("end_node_id"))
+        facts.append(f"{rel.get('type', 'RELATED_TO')}({source}, {target})")
+    return facts
+
+
 # ─── Output helpers ──────────────────────────────────────────────────────────
 
 
@@ -1415,6 +1466,31 @@ _INGEST_TYPES = [
 ]
 
 _INGEST_FORMATS = ["pdf", "docx", "csv", "excel", "html", "json", "parquet", "xml", "rdf"]
+_GRAPH_STORE_ENV_BACKEND_HINTS = {
+    "GRAPH_STORE_NEO4J_URI": "neo4j",
+    "GRAPH_STORE_FALKORDB_HOST": "falkordb",
+    "GRAPH_STORE_NEPTUNE_ENDPOINT": "neptune",
+    "GRAPH_STORE_AGE_CONNECTION_STRING": "age",
+}
+
+
+def _configured_ingest_graph_backend(
+    cli_ctx: CLIContext, store_override: Optional[str]
+) -> Optional[str]:
+    graph_db = dict(cli_ctx.config.to_dict().get("graph_db", {}))
+    backend = store_override or cli_ctx.store_backend or graph_db.get("backend")
+    if backend:
+        return str(backend)
+
+    env_backend = os.environ.get("GRAPH_STORE_DEFAULT_BACKEND")
+    if env_backend:
+        return env_backend
+
+    for env_var, hinted_backend in _GRAPH_STORE_ENV_BACKEND_HINTS.items():
+        if os.environ.get(env_var):
+            return hinted_backend
+
+    return None
 
 
 @main.command()
@@ -1427,8 +1503,13 @@ _INGEST_FORMATS = ["pdf", "docx", "csv", "excel", "html", "json", "parquet", "xm
 @click.option("--watch", is_flag=True, default=False, help="Re-ingest on file changes.")
 @click.option("--batch-size", default=500, type=int, show_default=True)
 @click.option("--store", "store_override", default=None,
-              help="Target graph backend: neo4j falkordb age neptune")
-@click.option("--output", default=None, type=click.Path(), help="Write to file instead of graph store.")
+              help="Target graph backend: neo4j falkordb age neptune. "
+                   "Not yet implemented — ingest cannot persist to a graph "
+                   "store, so this only determines whether the command "
+                   "refuses to report false success; use --output instead.")
+@click.option("--output", default=None, type=click.Path(),
+              help="Write ingested content to a .json/.jsonl/.csv file "
+                   "instead of the (unimplemented) graph store.")
 @click.option("--dry-run", "local_dry", is_flag=True, default=False)
 @click.option("--json", "local_json", is_flag=True, default=False)
 @click.pass_obj
@@ -1452,6 +1533,19 @@ def ingest(
             _dry(cli_ctx, "ingest", json_out=_is_json(cli_ctx, local_json),
                  source=source, type=ingestor_type, format=fmt)
             return
+        graph_backend = _configured_ingest_graph_backend(cli_ctx, store_override)
+        if graph_backend and graph_backend.lower() != "memory" and not output:
+            raise click.ClickException(
+                f"A graph backend is configured ({graph_backend}), but "
+                "semantica ingest does not write to graph stores yet — no CLI "
+                "command currently does (tracked in issues #1351, #1352). "
+                "Pass --output <file>.json to save the ingested content "
+                "instead, or build a GraphStore/GraphBuilder directly in "
+                "Python."
+            )
+        # NOTE: --store/GRAPH_STORE_DEFAULT_BACKEND are read only to decide
+        # whether to raise the error above — nothing downstream of this point
+        # writes to a graph store, so neither is forwarded as an ingest kwarg.
         kwargs: Dict[str, Any] = {"batch_size": batch_size}
         if ingestor_type:
             kwargs["source_type"] = ingestor_type
@@ -1461,10 +1555,6 @@ def ingest(
             kwargs["recursive"] = True
         if watch:
             kwargs["watch"] = True
-        if store_override or cli_ctx.store_backend:
-            kwargs["store"] = store_override or cli_ctx.store_backend
-        if output:
-            kwargs["output"] = output
         try:
             from .ingest import ingest as _ingest
             label = Path(source).name if Path(source).exists() else source
@@ -1478,7 +1568,10 @@ def ingest(
                     result = _ingest(source, **kwargs)
         except ImportError as exc:
             raise click.ClickException(f"Ingest module not available: {exc}") from exc
-        if _is_json(cli_ctx, local_json):
+        if output:
+            _write_result_output(Path(output), result)
+            _ok(cli_ctx, f"Wrote {output}")
+        elif _is_json(cli_ctx, local_json):
             _jecho(result if isinstance(result, dict) else {"status": "ok"})
         else:
             _ok(cli_ctx, f"Ingested: {source}")
@@ -1735,9 +1828,19 @@ def embed(ctx: click.Context) -> None:
 def _json_default(obj) -> object:
     """JSON serialiser that converts NumPy scalars/arrays to native Python types.
 
+    Also expands dataclasses (e.g. ``FileObject`` from ``ingest``) to plain
+    dicts and decodes ``bytes`` as UTF-8 text where possible, so a domain
+    object round-trips through ``--output`` as data instead of a repr string.
     Falls back to ``str()`` for everything else so the writer never crashes on
-    unexpected types (e.g. ``datetime``, custom domain objects).
+    unexpected types (e.g. ``datetime``).
     """
+    if is_dataclass(obj) and not isinstance(obj, type):
+        return asdict(obj)
+    if isinstance(obj, bytes):
+        try:
+            return obj.decode("utf-8")
+        except UnicodeDecodeError:
+            return obj.hex()
     try:
         import numpy as np  # local import — only needed when result contains numpy
         if isinstance(obj, np.ndarray):
@@ -2160,17 +2263,43 @@ def reason_run(cli_ctx: CLIContext, engine: str, rules: Optional[str],
     cli_ctx = _require_ctx(cli_ctx)
 
     def _action() -> None:
+        # Only the forward-chaining production-rule engines run through
+        # Reasoner.infer_facts(); the other engines take different inputs
+        # (SPARQL/Datalog queries, observations, premises) and are not wired
+        # to this command yet. Fail honestly instead of silently
+        # forward-chaining under another engine's name.
+        if engine not in ("rete", "forward-chain"):
+            hint = (" Use 'semantica reason query' for SPARQL/Datalog queries."
+                    if engine in ("sparql", "datalog") else "")
+            raise click.ClickException(
+                f"Engine '{engine}' is not wired to 'reason run' yet; "
+                f"supported engines: rete, forward-chain.{hint}")
         try:
             from .reasoning import Reasoner
+            # Reasoner has no run() method (#1354); dispatch to its real
+            # API: facts from the configured graph store + rules from the
+            # optional --rules file into infer_facts().
             r = Reasoner(engine=engine, config=cli_ctx.config.to_dict())
+            rule_defs = _load_rule_definitions(rules) if rules else None
+            facts = _graph_store_facts(cli_ctx)
+
+            def _infer() -> Dict[str, Any]:
+                inferred = r.infer_facts(facts, rule_defs)
+                return {
+                    "engine": engine,
+                    "facts": len(facts),
+                    "inferred_count": len(inferred),
+                    "inferred_facts": inferred,
+                }
+
             if cli_ctx.quiet or cli_ctx.json_output:
-                result = r.run(rules_file=rules)
+                result = _infer()
             else:
                 with console.status(
                     f"[{_DIM}]Running {engine} reasoning engine…[/{_DIM}]",
                     spinner="dots",
                 ):
-                    result = r.run(rules_file=rules)
+                    result = _infer()
         except ImportError as exc:
             raise click.ClickException(f"Reasoning module not available: {exc}") from exc
         if _is_json(cli_ctx, local_json):
@@ -3661,14 +3790,17 @@ def store_connect(cli_ctx: CLIContext, backend: str, uri: Optional[str], local_j
 
     def _action() -> None:
         try:
-            from .graph_store import get_graph_store_method
-            store_cls = get_graph_store_method(backend)
+            # get_graph_store_method(task, method_name) is the method
+            # registry, not a backend factory (#1354); build the store
+            # through GraphStore, which resolves the backend by name.
+            from .graph_store import GraphStore
             cfg = dict(cli_ctx.config.to_dict().get("graph_db", {}))
+            cfg.pop("backend", None)
             if uri:
                 cfg["uri"] = uri
-            # Attempt instantiation as the minimal connectivity probe; backends
-            # that require a live connection will fail here if unreachable.
-            store_instance = store_cls(config=cfg)
+            # Instantiation only wires the backend; the probe below performs
+            # the live connectivity check and raises if unreachable.
+            store_instance = GraphStore(backend=backend, **cfg)
             for probe in ("health_check", "ping", "connect"):
                 fn = getattr(store_instance, probe, None)
                 if callable(fn):

semantica/context/context_graph.py

@@ -4791,14 +4791,18 @@ class ContextGraph:
 
                 # Find potential causes (decisions that influenced this one) via
                 # shared entities/timestamps - additive heuristic, skipping anything
-                # already covered by an explicit relationship above.
-                potential_causes = []
+                # already covered by an explicit relationship above. Deduplicate by
+                # decision id (dict preserves insertion order): a decision sharing
+                # several entities with the current one is one potential cause,
+                # not one per shared entity, otherwise the trace reports the same
+                # "influences" chain once per overlapping entity.
+                potential_causes = {}
                 for entity in current_decision["entities"]:
                     for other_decision_id in self._entity_index.get(entity, set()):
                         if other_decision_id != current_id and other_decision_id not in explicit_cause_ids:
                             other_decision = self._decisions[other_decision_id]
                             if other_decision["timestamp"] < current_decision["timestamp"]:
-                                potential_causes.append(other_decision_id)
+                                potential_causes[other_decision_id] = None
 
                 for cause_id in potential_causes:
                     cause_dec = self._decisions.get(cause_id, {})

semantica/context/erasure.py

@@ -14,10 +14,12 @@ not. It *composes* the existing public APIs; nothing in ``context_graph.py`` or
 ``agent_memory.py`` changes, and ``ContextGraph`` keeps its graph-scope
 contract.
 
-The property that matters is honest partial reporting. Three vector backends
-(FAISS, Milvus, Weaviate) expose no delete at all, so erasure is genuinely not
-completable on them today. The receipt says ``unsupported`` for those rather
-than reporting a success it did not achieve -- a receipt that reads
+The property that matters is honest partial reporting. FAISS exposes no delete
+at all -- a flat FAISS index cannot remove individual vectors without a full
+rebuild -- so erasure is genuinely not completable on it today. Milvus and
+Weaviate now expose ``delete_vectors`` and are fully supported. The receipt
+says ``unsupported`` for FAISS rather than reporting a success it did not
+achieve -- a receipt that reads
 "graph: erased, memory: 14 erased, vectors: unsupported on faiss" is
 actionable; a bare ``True`` is a compliance liability.
 
@@ -29,11 +31,12 @@ Example:
     ...     "customer-4471", reason="GDPR Art. 17 request #882"
     ... )
     >>> receipt.complete
-    False
+    True
     >>> receipt.stores["vectors"]["status"]
-    'unsupported'
+    'not_configured'
 """
 
+import copy
 import inspect
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -117,13 +120,21 @@ class ErasureReceipt:
         ]
 
     def to_dict(self) -> Dict[str, Any]:
-        """Serialize the receipt, deep-copying the per-store results."""
+        """Serialize the receipt, deep-copying the per-store results.
+
+        Each store result is copied recursively so the returned payload shares
+        no mutable objects with the live receipt: mutating
+        ``payload["stores"][name][...]`` (including nested dicts such as a
+        vector backend's ``backend_result``) cannot corrupt the audit record.
+        """
         return {
             "entity_id": self.entity_id,
             "reason": self.reason,
             "erased_at": self.erased_at,
             "complete": self.complete,
-            "stores": {name: dict(result) for name, result in self.stores.items()},
+            "stores": {
+                name: copy.deepcopy(result) for name, result in self.stores.items()
+            },
         }
 
 
@@ -368,8 +379,8 @@ class ErasureCoordinator:
 
         method_name, target = _vector_delete_capability(self.vector_store)
         if method_name is None:
-            # FAISS, Milvus and Weaviate expose no delete at all; FAISS in
-            # particular cannot remove from a flat index without a rebuild.
+            # FAISS exposes no delete at all; it cannot remove vectors from a
+            # flat index without a full rebuild.
             self.logger.warning(
                 "Vector backend %r exposes no delete; %d vector id(s) for %r "
                 "were not erased",

semantica/explorer/routes/ontology.py

@@ -35,6 +35,10 @@ router = APIRouter(prefix="/api/ontology", tags=["ontology"])
 _MAX_FETCH_BYTES = 20 * 1024 * 1024  # 20 MB
 _MAX_ANALYSIS_NODES = 5_000   # cap for health/suggest/shacl node scans to avoid OOM
 _MAX_ENTITIES_PER_SIDE = 500  # per-ontology cap for the O(n²) pairwise suggestion loop
+_GRAPH_TOO_LARGE_DETAIL = (
+    "Ontology editor graph exceeds the maximum size "
+    f"({_MAX_ANALYSIS_NODES} nodes or edges)."
+)
 
 
 class GraphTruncationError(Exception):
@@ -72,6 +76,20 @@ _ONTOLOGY_TYPES = frozenset({
 }) | _SCHEME_TYPES
 
 _SEARCHABLE_TYPES = _CLASS_TYPES | _PROPERTY_TYPES | _INDIVIDUAL_TYPES | _CONCEPT_TYPES | _SCHEME_TYPES
+_SCHEMA_NODE_TYPES = _CLASS_TYPES | _PROPERTY_TYPES | _CONCEPT_TYPES | _ONTOLOGY_TYPES
+_STRUCTURE_EDGE_TYPES = frozenset({
+    "rdf:type",
+    "rdfs:subClassOf",
+    "rdfs:domain",
+    "rdfs:range",
+    "owl:disjointWith",
+    "owl:equivalentClass",
+    "owl:equivalentProperty",
+    "owl:inverseOf",
+    "skos:broader",
+    "skos:narrower",
+    "skos:related",
+})
 
 _URI_PREFIX_MAP = {
     "http://www.w3.org/2002/07/owl#": "owl:",
@@ -1821,6 +1839,64 @@ async def search_entities(
     return results
 
 
+def _known_ontology_uris(
+    session: GraphSession, registry: Dict[str, OntologyEntry]
+) -> set[str]:
+    known = set(registry)
+    for node_type in _ONTOLOGY_TYPES:
+        for node in session.iter_nodes(node_type=node_type):
+            node_id = str(node.get("id", ""))
+            if node_id:
+                known.add(node_id)
+    return known
+
+
+def _collect_core_nodes(
+    session: GraphSession, uri: str, known_ontology_uris: set[str]
+) -> Dict[str, Dict[str, Any]]:
+    """Stream schema nodes, keeping only the ones this ontology owns.
+
+    Filtering as each node arrives makes _MAX_ANALYSIS_NODES bound the work and
+    not merely the response: foreign nodes are discarded instead of materialized,
+    and the scan stops once the owned ones pass the cap. The ownership filter has
+    to stay ahead of that check — thousands of *other* ontologies' nodes must
+    never make this one too large to open. Requesting pages instead would bound
+    nothing: paginate_nodes normalizes the whole matching set on every call.
+    """
+    core_nodes_by_id: Dict[str, Dict[str, Any]] = {}
+    for node_type in _SCHEMA_NODE_TYPES:
+        for node in session.iter_nodes(node_type=node_type):
+            node_id = str(node.get("id", ""))
+            if not node_id or not _node_belongs_to_ontology(
+                node, uri, known_ontology_uris
+            ):
+                continue
+            core_nodes_by_id[node_id] = node
+            if len(core_nodes_by_id) > _MAX_ANALYSIS_NODES:
+                raise GraphTruncationError(_GRAPH_TOO_LARGE_DETAIL)
+    return core_nodes_by_id
+
+
+def _select_structure_edges(
+    session: GraphSession, core_node_ids: set[str]
+) -> List[Dict[str, Any]]:
+    """Stream structural edges, keeping only those leaving a core node.
+
+    The requested ontology may reference outward (e.g. rdfs:range to an external
+    vocabulary), but an unrelated ontology's property pointing at a core class
+    must not leak inward.
+    """
+    selected_edges: List[Dict[str, Any]] = []
+    for edge_type in _STRUCTURE_EDGE_TYPES:
+        for edge in session.iter_edges(edge_type=edge_type):
+            if str(edge.get("source", "")) not in core_node_ids:
+                continue
+            selected_edges.append(edge)
+            if len(selected_edges) > _MAX_ANALYSIS_NODES:
+                raise GraphTruncationError(_GRAPH_TOO_LARGE_DETAIL)
+    return selected_edges
+
+
 @router.get("/graph", response_model=OntologyGraphResponse)
 async def get_ontology_graph(
     request: Request,
@@ -1828,88 +1904,39 @@ async def get_ontology_graph(
     session: GraphSession = Depends(get_session),
 ):
     """Return the editable schema subgraph for one registered ontology."""
-    registry = _get_registry(request)
-    ontology_nodes: List[Dict[str, Any]] = []
-    for node_type in _ONTOLOGY_TYPES:
-        nodes, _ = await asyncio.to_thread(
-            session.get_nodes, node_type=node_type, skip=0, limit=2**63 - 1
-        )
-        ontology_nodes.extend(nodes)
-    known_ontology_uris = set(registry) | {
-        str(node.get("id", "")) for node in ontology_nodes if node.get("id")
-    }
+    known_ontology_uris = await asyncio.to_thread(
+        _known_ontology_uris, session, _get_registry(request)
+    )
     if uri not in known_ontology_uris:
         raise HTTPException(status_code=404, detail="Ontology not found in registry.")
 
-    schema_types = _CLASS_TYPES | _PROPERTY_TYPES | _CONCEPT_TYPES | _ONTOLOGY_TYPES
-    candidates_by_id: Dict[str, Dict[str, Any]] = {}
-    for node_type in schema_types:
-        nodes, _ = await asyncio.to_thread(
-            session.get_nodes, node_type=node_type, skip=0, limit=2**63 - 1
+    try:
+        core_nodes_by_id = await asyncio.to_thread(
+            _collect_core_nodes, session, uri, known_ontology_uris
         )
-        candidates_by_id.update(
-            (str(node.get("id", "")), node) for node in nodes if node.get("id")
+        if not core_nodes_by_id:
+            raise HTTPException(status_code=404, detail="Ontology graph not found.")
+        core_node_ids = set(core_nodes_by_id)
+        selected_edges = await asyncio.to_thread(
+            _select_structure_edges, session, core_node_ids
         )
+    except GraphTruncationError as exc:
+        raise HTTPException(status_code=413, detail=str(exc)) from exc
 
-    core_node_ids = {
-        str(node.get("id", ""))
-        for node in candidates_by_id.values()
-        if _node_belongs_to_ontology(node, uri, known_ontology_uris)
-    }
-    if not core_node_ids:
-        raise HTTPException(status_code=404, detail="Ontology graph not found.")
+    # Invariant: the helpers raise the moment their accumulation passes
+    # _MAX_ANALYSIS_NODES, so core_nodes_by_id and selected_edges are both
+    # within the cap here; a post-filter re-check would be unreachable.
+    external_node_ids = {
+        node_id
+        for edge in selected_edges
+        for node_id in (str(edge.get("source", "")), str(edge.get("target", "")))
+    } - core_node_ids
+    external_nodes = await asyncio.gather(
+        *(asyncio.to_thread(session.get_node, node_id) for node_id in external_node_ids)
+    )
 
-    structure_edge_types = {
-        "rdf:type",
-        "rdfs:subClassOf",
-        "rdfs:domain",
-        "rdfs:range",
-        "owl:disjointWith",
-        "owl:equivalentClass",
-        "owl:equivalentProperty",
-        "owl:inverseOf",
-        "skos:broader",
-        "skos:narrower",
-        "skos:related",
-    }
-    selected_edges: List[Dict[str, Any]] = []
-    for edge_type in structure_edge_types:
-        edges, _ = await asyncio.to_thread(
-            session.get_edges,
-            edge_type=edge_type,
-            skip=0,
-            limit=2**63 - 1,
-        )
-        # Keep only edges whose source is a core node: the requested ontology
-        # may reference outward (e.g. rdfs:range to an external vocabulary),
-        # but an unrelated ontology's property pointing at a core class must
-        # not leak inward.
-        selected_edges.extend(
-            edge for edge in edges
-            if str(edge.get("source", "")) in core_node_ids
-        )
-    if (
-        len(core_node_ids) > _MAX_ANALYSIS_NODES
-        or len(selected_edges) > _MAX_ANALYSIS_NODES
-    ):
-        raise HTTPException(
-            status_code=413,
-            detail=(
-                "Ontology editor graph exceeds the maximum size "
-                f"({_MAX_ANALYSIS_NODES} nodes or edges)."
-            ),
-        )
-
-    selected_node_ids = set(core_node_ids)
-    for edge in selected_edges:
-        selected_node_ids.add(str(edge.get("source", "")))
-        selected_node_ids.add(str(edge.get("target", "")))
-
-    selected_nodes = [candidates_by_id[node_id] for node_id in core_node_ids]
-    for node_id in selected_node_ids - core_node_ids:
-        external = await asyncio.to_thread(session.get_node, node_id)
-        if external is not None:
-            selected_nodes.append(external)
+    selected_nodes = list(core_nodes_by_id.values())
+    selected_nodes.extend(node for node in external_nodes if node is not None)
     selected_nodes.sort(key=lambda node: str(node.get("id", "")))
     selected_edges.sort(
         key=lambda edge: (

semantica/explorer/session.py

@@ -9,7 +9,7 @@ import threading
 import time
 import uuid
 from datetime import UTC, datetime
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, Iterator, List, Optional
 
 from ..context.context_graph import ContextGraph, _resolve_edge_identity
 from .search_index import GraphSearchIndex
@@ -375,6 +375,52 @@ class GraphSession:
         )
         return page, total
 
+    def iter_nodes(self, node_type: Optional[str] = None) -> Iterator[Dict[str, Any]]:
+        """Yield matching nodes one at a time, in the same order as ``paginate_nodes``.
+
+        ``paginate_nodes`` normalizes and holds the entire matching set before it
+        slices out a page, so a caller that filters the result down itself cannot
+        bound its cost by asking for smaller pages — it would re-pay that full
+        cost per page. Streaming lets such a caller retain only what it selects
+        and stop scanning as soon as it has enough.
+
+        Only the id list is snapshotted under the lock; nodes are read one at a
+        time, so a concurrent mutation can be observed mid-iteration and ids that
+        disappear are skipped. ``paginate_nodes`` is the atomic alternative.
+        """
+        with self._lock:
+            source_ids = (
+                self.graph.node_type_index.get(node_type, set())
+                if node_type
+                else self.graph.nodes.keys()
+            )
+            node_ids = sorted(
+                (node_id for node_id in source_ids if node_id is not None),
+                key=lambda value: str(value),
+            )
+        for node_id in node_ids:
+            with self._lock:
+                raw = self.graph.find_node(node_id)
+            if raw is None:
+                continue
+            yield self.normalize_node(raw)
+
+    def iter_edges(self, edge_type: Optional[str] = None) -> Iterator[Dict[str, Any]]:
+        """Yield matching edges one at a time, in raw graph order.
+
+        Same rationale as ``iter_nodes``. Edge normalization derives an identity
+        hash per edge, which ``paginate_edges`` pays for every matching edge (and
+        then sorts) before paging; a filtering caller only needs it for the edges
+        it keeps. Callers that need a stable order sort the subset they select.
+        """
+        with self._lock:
+            raw_edges = self.graph.find_edges(edge_type=edge_type)
+        for edge in raw_edges:
+            normalized = self.normalize_edge(edge)
+            if not normalized["source"] or not normalized["target"]:
+                continue
+            yield normalized
+
     def get_raw_counts(self) -> tuple[int, int]:
         """O(1) node/edge counts from the raw collections, with no per-item
         normalization.

semantica/kg/config.py

@@ -50,7 +50,14 @@ class KGConfig:
         """Initialize configuration manager."""
         self.logger = get_logger("kg_config")
         self._configs: Dict[str, Any] = {}
-        self._method_configs: Dict[str, Dict] = {}
+        self._method_configs: Dict[str, Dict] = {
+            "build": {
+                # How GraphBuilder treats a relationship endpoint that points at
+                # an entity absent from the extracted set: "include" promotes a
+                # synthetic UNKNOWN entity (default), "reject" drops the edge.
+                "unknown_relation_endpoint": "include",
+            },
+        }
         self._load_config_file(config_file)
         self._load_env_vars()
 

semantica/kg/graph_builder.py

@@ -23,7 +23,7 @@ License: MIT
 """
 
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Set, Tuple, Union
 import time
 
 
@@ -89,7 +89,29 @@ class GraphBuilder:
         self.track_history = track_history
         self.version_snapshots = version_snapshots
         self.graph_store = graph_store
-        self.config = kwargs  # Store additional config for extractors
+        # The orchestrator builds with GraphBuilder(config=self.config.get("kg", {})).
+        # That lands as a nested "config" keyword, so fold it into the option
+        # mapping once here: every option ({entity_resolution,
+        # conflict_detection, unknown_relation_endpoint, ...}) is then read the
+        # same way whether it was given top-level or via the orchestrator path.
+        kwargs = dict(kwargs)
+        _nested = kwargs.pop("config", None)
+        if isinstance(_nested, dict):
+            for _key, _value in _nested.items():
+                kwargs.setdefault(_key, _value)
+        self.config = kwargs
+        # unknown_relation_endpoint lives in the per-module build config
+        # (semantica/kg/config.py); fall back to it so the option has a single
+        # documented home. Popped out of kwargs so it is not forwarded to
+        # extractors as an unused **config key.
+        from .config import kg_config
+
+        self.unknown_relation_endpoint = kwargs.pop(
+            "unknown_relation_endpoint",
+            kg_config.get_method_config("build").get(
+                "unknown_relation_endpoint", "include"
+            ),
+        )
         # Extractors are reused across texts: NERExtractor loads its spaCy model
         # eagerly in __init__, so constructing one per text would reload the
         # model on every source in a multi-document build.
@@ -102,6 +124,8 @@ class GraphBuilder:
             "extracted_relations": 0,
             "extracted_triplets": 0,
         }
+        # Counts relationships dropped by the reject policy per build() call.
+        self._rejected_relationships: int = 0
 
         # Initialize logging
         from ..utils.logging import get_logger
@@ -159,9 +183,81 @@ class GraphBuilder:
             }
             all_entities.append(entity_dict)
         elif hasattr(item, "subject") and hasattr(item, "predicate") and hasattr(item, "object"):
-            # It's likely a Relation object
+            # It's likely a Relation or Triplet object
             subj = item.subject
             obj = item.object
+            # Relation extraction synthesizes an UNKNOWN Entity for an endpoint
+            # that is absent from the NER entity list (metadata synthetic=True).
+            # Triplet extraction does the same, but its endpoints are strings, so
+            # the extractor records the absent endpoint texts instead. Only the
+            # id/text survives in the relationship today, so the synthetic object
+            # never reaches the entity collection and graph validation reports
+            # DANGLING_EDGE. Promote those endpoint entities into the graph here.
+            item_metadata = getattr(item, "metadata", None) or {}
+            # Triplet LLM path tags endpoint texts it could not match to an entity.
+            triplet_synthetic = [
+                text
+                for text in item_metadata.get("synthetic_endpoints", [])
+                if isinstance(text, str)
+            ]
+            synthetic_endpoints = []
+            for endpoint in (subj, obj):
+                if (
+                    not isinstance(endpoint, str)
+                    and isinstance(getattr(endpoint, "metadata", None), dict)
+                    and endpoint.metadata.get("synthetic")
+                ):
+                    synthetic_endpoints.append(endpoint)
+                elif isinstance(endpoint, str) and endpoint in triplet_synthetic:
+                    synthetic_endpoints.append(endpoint)
+            if synthetic_endpoints:
+                endpoint_policy = self.unknown_relation_endpoint
+                if endpoint_policy == "reject":
+                    self.logger.warning(
+                        "Dropping relationship %r->%r (%s): endpoint is synthetic and "
+                        "unknown_relation_endpoint='reject'",
+                        subj,
+                        obj,
+                        item.predicate,
+                    )
+                    self._rejected_relationships += 1
+                    return
+                # The promoted set is rebuilt only for relationships that actually
+                # carry a synthetic endpoint. Ordinary relationships (the common
+                # case) must not pay an O(all_entities) scan per item, which made
+                # build() quadratic on dense relation inputs.
+                existing_ids: Set[Any] = set()
+                for _ent in all_entities:
+                    if not isinstance(_ent, dict):
+                        continue
+                    for _key in ("id", "entity_id"):
+                        _cid = _ent.get(_key)
+                        if _cid is None:
+                            continue
+                        try:
+                            existing_ids.add(_cid)
+                        except TypeError:
+                            # Invalid/unhashable IDs are left for graph validation.
+                            continue
+                for endpoint in synthetic_endpoints:
+                    if isinstance(endpoint, str):
+                        endpoint_id = endpoint
+                        endpoint_text = endpoint
+                    else:
+                        endpoint_id = endpoint.id if hasattr(endpoint, "id") else endpoint.text
+                        endpoint_text = endpoint.text
+                    if endpoint_id in existing_ids:
+                        continue
+                    all_entities.append(
+                        {
+                            "id": endpoint_id,
+                            "name": endpoint_text,
+                            "type": "UNKNOWN",
+                            "confidence": 0.8,
+                            "metadata": {"synthetic": True},
+                        }
+                    )
+                    existing_ids.add(endpoint_id)
             subj_id = getattr(subj, "id", getattr(subj, "text", str(subj))) if not isinstance(subj, str) else subj
             obj_id = getattr(obj, "id", getattr(obj, "text", str(obj))) if not isinstance(obj, str) else obj
             rel_dict = {
@@ -543,6 +639,8 @@ class GraphBuilder:
             "extracted_relations": 0,
             "extracted_triplets": 0
         }
+        # Reset per-run rejection counter.
+        self._rejected_relationships = 0
         
         tracking_id = self.progress_tracker.start_tracking(
             module="kg",
@@ -782,6 +880,39 @@ class GraphBuilder:
                 if has_merged_entities:
                     self._remap_relationship_endpoints(resolved_entities, all_relationships)
 
+            # When a synthetic endpoint is promoted before the real entity arrives
+            # (e.g. relation data precedes NER entity data for the same text),
+            # the same id can appear once as synthetic and once as real. Prefer
+            # the real entity so the graph does not carry a duplicated,
+            # lower-confidence duplicate.
+            _real_ids: Set[Any] = set()
+            for _entity in resolved_entities:
+                if (
+                    isinstance(_entity, dict)
+                    and not (_entity.get("metadata") or {}).get("synthetic")
+                ):
+                    for _cid in (_entity.get("id"), _entity.get("entity_id")):
+                        if _cid is None:
+                            continue
+                        try:
+                            _real_ids.add(_cid)
+                        except TypeError:
+                            # Invalid/unhashable IDs are left for graph validation
+                            # to report rather than failing graph construction here.
+                            continue
+            if _real_ids:
+                filtered = []
+                for _entity in resolved_entities:
+                    if (
+                        isinstance(_entity, dict)
+                        and (_entity.get("metadata") or {}).get("synthetic")
+                    ):
+                        _eid = _entity.get("id")
+                        if _eid in _real_ids:
+                            continue
+                    filtered.append(_entity)
+                resolved_entities = filtered
+
             if input_relationships_count > 0 and len(all_relationships) == 0:
                 warning_msg = (
                     f"All relationships were dropped during graph building: "
@@ -801,6 +932,7 @@ class GraphBuilder:
                     "temporal_enabled": self.enable_temporal,
                     "timestamp": self._get_timestamp(),
                     "entity_resolution_applied": resolver_to_use is not None,
+                    "rejected_relationships": self._rejected_relationships,
                 },
             }
             structure_time = time.time() - structure_start

semantica/kg/kg_usage.md

@@ -132,6 +132,53 @@ kg1 = builder.build(initial_sources)
 kg2 = builder.build(additional_sources)
 ```
 
+### Unknown relation endpoints
+
+When a relationship endpoint names an entity that is absent from the extracted
+entity set (for example an LLM/HuggingFace extraction that synthesizes an
+endpoint), `GraphBuilder` promotes a synthetic `UNKNOWN` entity so the edge is
+not left dangling. This is the default (`"include"`). To drop such
+relationships instead, set `unknown_relation_endpoint="reject"`:
+
+```python
+builder = GraphBuilder(
+    unknown_relation_endpoint="reject",
+)
+kg = builder.build(sources)  # edges with unknown endpoints are dropped
+```
+
+Rejected relationships are observable at two levels:
+
+- A **`WARNING`-level log** is emitted for every dropped edge, including the
+  source/target and predicate, so it is visible without debug logging enabled.
+- `graph["metadata"]["rejected_relationships"]` holds the count of edges
+  dropped in that build call, giving callers a machine-readable signal:
+
+```python
+kg = builder.build(sources)
+if kg["metadata"]["rejected_relationships"]:
+    print(f"{kg['metadata']['rejected_relationships']} edge(s) were rejected "
+          f"because of unknown endpoints.")
+```
+
+The option can also be set process-wide via the module's build configuration,
+which is the documented home for it:
+
+```python
+from semantica.kg.config import kg_config
+
+kg_config.set_method_config("build", unknown_relation_endpoint="reject")
+# All subsequent GraphBuilder instances use "reject" as the default.
+```
+
+In a YAML/JSON/TOML config file, place it under the `kg_methods.build` key:
+
+```yaml
+kg_methods:
+  build:
+    unknown_relation_endpoint: reject
+```
+
 ## Graph Algorithms
 
 The knowledge graph module provides advanced algorithms for node embeddings, similarity calculations, path finding, link prediction, centrality measures, and community detection.

semantica/semantic_extract/methods.py

@@ -2316,7 +2316,7 @@ def extract_triplets_rules(
 
 
 def extract_triplets_huggingface(
-    text: str, model: str, device: Optional[str] = None, **kwargs
+    text: str, model: str, device: Optional[str] = None, entities: Optional[List[Entity]] = None, **kwargs
 ) -> List[Triplet]:
     """HuggingFace triplet extraction."""
     loader = HuggingFaceModelLoader(device=device)
@@ -2348,16 +2348,30 @@ def extract_triplets_huggingface(
                 tail = match.group("tail").strip()
                 
                 if head and relation and tail:
+                    # Head/tail are raw decoded strings from the model. Tag any
+                    # that match no known entity so the GraphBuilder promotes
+                    # them instead of leaving a dangling edge (#1463).
+                    # TripletExtractor dispatches with entities=..., so this is
+                    # the real NER list here, not a dead comparison.
+                    hf_entities = entities or []
+                    synthetic_endpoints = [
+                        endpoint_text
+                        for endpoint_text in (head, tail)
+                        if not match_entity(endpoint_text, hf_entities)
+                    ]
+                    hf_metadata = {
+                        "model": model,
+                        "extraction_method": "huggingface_rebel",
+                    }
+                    if synthetic_endpoints:
+                        hf_metadata["synthetic_endpoints"] = synthetic_endpoints
                     triplets.append(
                         Triplet(
                             subject=head,
                             predicate=relation,
                             object=tail,
                             confidence=0.9, # Model generation doesn't provide per-triplet confidence
-                            metadata={
-                                "model": model,
-                                "extraction_method": "huggingface_rebel"
-                            }
+                            metadata=hf_metadata
                         )
                     )
 
@@ -2517,16 +2531,27 @@ Text to extract from:
         # Convert back to internal Triplet format
         triplets = []
         for t_out in result_obj.triplets:
+            # An LLM triple may reference an endpoint that does not match any
+            # entity extracted by NER. Record those endpoints so the GraphBuilder
+            # can promote them as synthetic entities instead of leaving a
+            # dangling edge (see issue #1463).
+            synthetic_endpoints = []
+            for endpoint_text in (t_out.subject, t_out.object):
+                if not match_entity(endpoint_text, entities or []):
+                    synthetic_endpoints.append(endpoint_text)
+            metadata = {
+                "provider": provider,
+                "model": model,
+                "extraction_method": "llm_typed",
+            }
+            if synthetic_endpoints:
+                metadata["synthetic_endpoints"] = synthetic_endpoints
             triplets.append(Triplet(
                 subject=t_out.subject,
                 predicate=t_out.predicate,
                 object=t_out.object,
                 confidence=t_out.confidence,
-                metadata={
-                    "provider": provider, 
-                    "model": model, 
-                    "extraction_method": "llm_typed"
-                }
+                metadata=metadata,
             ))
         
         logger.info(f"Successfully extracted {len(triplets)} triplets using {provider}/{model} (typed)")

semantica/semantic_extract/ner_extractor.py

@@ -37,7 +37,7 @@ Key Features:
         * LLM-based: Large language model extraction
     - Fallback chain support: Try methods in order until one succeeds
     - Robust Fallbacks: Prevents empty results via ML -> Pattern -> Last Resort chain
-    - Ensemble voting: Combine results from multiple methods
+    - Explicit merge strategies: fallback, union, and consensus
     - Post-processing: Entity boundary validation
     - Multiple entity type support (PERSON, ORG, GPE, DATE, etc.)
     - Confidence scoring and filtering
@@ -62,15 +62,20 @@ Example Usage:
     >>> extractor = NERExtractor(method="huggingface", huggingface_model="dslim/bert-base-NER")
     >>> entities = extractor.extract_entities("Apple Inc. was founded in 1976.")
     >>> 
-    >>> # Using fallback chain
-    >>> extractor = NERExtractor(method=["llm", "ml", "pattern"], ensemble_voting=True)
+    >>> # Require agreement between multiple extraction methods
+    >>> extractor = NERExtractor(
+    ...     method=["llm", "ml"], merge_strategy="consensus", min_votes=2
+    ... )
     >>> entities = extractor.extract_entities("Apple Inc. was founded in 1976.")
 
 Author: Semantica Contributors
 License: MIT
 """
 
-from typing import Any, Dict, List, Optional, Tuple, Union
+import math
+import re
+import warnings
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 from ..utils.exceptions import ProcessingError
 from ..utils.helpers import safe_import
@@ -84,6 +89,31 @@ spacy, SPACY_AVAILABLE = safe_import("spacy")
 class NERExtractor:
     """Named Entity Recognition extractor."""
 
+    _VALID_MERGE_STRATEGIES = {"fallback", "union", "consensus"}
+    _MERGE_OPTION_KEYS = (
+        "merge_strategy",
+        "min_votes",
+        "min_agreement",
+        "method_weights",
+        "eligible_methods",
+    )
+    _MIN_SPAN_IOU = 0.5
+    _LABEL_ALIASES = {
+        "PER": "PERSON",
+        "PERSON": "PERSON",
+        "ORGANIZATION": "ORG",
+        "ORG": "ORG",
+        "LOCATION": "GPE",
+        "LOC": "GPE",
+        "GPE": "GPE",
+        "TIME": "DATE",
+        "DATE": "DATE",
+        "CURRENCY": "MONEY",
+        "MONEY": "MONEY",
+        "PERCENTAGE": "PERCENT",
+        "PERCENT": "PERCENT",
+    }
+
     def __init__(
         self, 
         method: Union[str, List[str]] = "ml", 
@@ -115,9 +145,16 @@ class NERExtractor:
                     third-party servers (Qwen, LLaMA gateways, etc.) that do
                     not implement the full function-calling protocol still
                     return correctly structured results.
-                - device: Device for HuggingFace models ("cuda" or "cpu")
-                - min_confidence: Minimum confidence threshold
-                - ensemble_voting: Enable ensemble voting (default: False)
+                 - device: Device for HuggingFace models ("cuda" or "cpu")
+                 - min_confidence: Minimum confidence threshold
+                - merge_strategy: "fallback" (default), "union", or "consensus"
+                - min_votes: Required supporting methods for consensus (default: 2)
+                - min_agreement: Optional minimum support ratio for consensus
+                - method_weights: Optional method weights for exact-span
+                  cross-label tie-breaking
+                - eligible_methods: Optional subset of configured methods to count
+                  as consensus voters
+                - ensemble_voting: Deprecated alias for merge_strategy="union"
                 - post_process: Enable post-processing (default: False)
         """
         self.logger = get_logger("ner_extractor")
@@ -133,6 +170,15 @@ class NERExtractor:
         self.language = config.get("language", "en")
         self.min_confidence = config.get("min_confidence", 0.5)
         self.ensemble_voting = config.get("ensemble_voting", False)
+        self.merge_strategy = self._resolve_merge_strategy(config)
+        self.min_votes = self._validate_min_votes(config.get("min_votes", 2))
+        self.min_agreement = self._validate_min_agreement(
+            config.get("min_agreement")
+        )
+        self.method_weights = self._validate_method_weights(
+            config.get("method_weights")
+        )
+        self.eligible_methods = config.get("eligible_methods")
         self.post_process = config.get("post_process", False)
         self.progress_tracker = get_progress_tracker()
         # Ensure progress tracker is enabled
@@ -164,6 +210,240 @@ class NERExtractor:
                     exc_info=True,
                 )
 
+    def _resolve_merge_strategy(self, config: Dict[str, Any]) -> str:
+        """Resolve the explicit merge strategy and the deprecated legacy flag."""
+        configured_strategy = config.get("merge_strategy")
+        if configured_strategy is None:
+            if self.ensemble_voting:
+                warnings.warn(
+                    "ensemble_voting is deprecated because it historically "
+                    "performed a union, not voting. Use merge_strategy='union' "
+                    "or merge_strategy='consensus' explicitly.",
+                    DeprecationWarning,
+                    stacklevel=3,
+                )
+                return "union"
+            return "fallback"
+
+        strategy = self._validate_merge_strategy(configured_strategy)
+        if self.ensemble_voting:
+            warnings.warn(
+                "ensemble_voting is deprecated and ignored when merge_strategy "
+                "is provided.",
+                DeprecationWarning,
+                stacklevel=3,
+            )
+        return strategy
+
+    @classmethod
+    def _validate_merge_strategy(cls, strategy: Any) -> str:
+        """Return a normalized merge strategy or raise a useful configuration error."""
+        if not isinstance(strategy, str):
+            raise ValueError(
+                "merge_strategy must be one of: fallback, union, consensus"
+            )
+
+        normalized = strategy.lower()
+        if normalized not in cls._VALID_MERGE_STRATEGIES:
+            raise ValueError(
+                "merge_strategy must be one of: fallback, union, consensus"
+            )
+        return normalized
+
+    @staticmethod
+    def _validate_min_votes(min_votes: Any) -> int:
+        """Validate the number of method votes required for consensus."""
+        if isinstance(min_votes, bool) or not isinstance(min_votes, int):
+            raise ValueError("min_votes must be a positive integer")
+        if min_votes < 1:
+            raise ValueError("min_votes must be a positive integer")
+        return min_votes
+
+    @staticmethod
+    def _validate_min_agreement(min_agreement: Any) -> Optional[float]:
+        """Validate an optional consensus support ratio."""
+        if min_agreement is None:
+            return None
+
+        try:
+            normalized = float(min_agreement)
+        except (TypeError, ValueError):
+            raise ValueError("min_agreement must be a number between 0 and 1")
+
+        if not math.isfinite(normalized) or not 0.0 <= normalized <= 1.0:
+            raise ValueError("min_agreement must be a number between 0 and 1")
+        return normalized
+
+    @classmethod
+    def _validate_method_weights(cls, method_weights: Any) -> Dict[str, float]:
+        """Validate optional positive method weights used for deterministic ties."""
+        if method_weights is None:
+            return {}
+        if not isinstance(method_weights, dict):
+            raise ValueError("method_weights must be a mapping of method names to weights")
+
+        normalized = {}
+        for method_name, weight in method_weights.items():
+            if not isinstance(method_name, str):
+                raise ValueError("method_weights keys must be method names")
+            try:
+                numeric_weight = float(weight)
+            except (TypeError, ValueError):
+                raise ValueError("method_weights values must be positive numbers")
+            if not math.isfinite(numeric_weight) or numeric_weight <= 0:
+                raise ValueError("method_weights values must be positive numbers")
+            identity = cls._method_identity(method_name)
+            existing_weight = normalized.get(identity)
+            if existing_weight is not None and existing_weight != numeric_weight:
+                raise ValueError(
+                    "method_weights assigns conflicting values to aliases for "
+                    f"backend '{identity}'"
+                )
+            normalized[identity] = numeric_weight
+        return normalized
+
+    @staticmethod
+    def _method_identity(method_name: str) -> str:
+        """Normalize aliases that share one extraction backend for vote counting."""
+        normalized = method_name.lower()
+        return "ml" if normalized in {"ml", "spacy"} else method_name
+
+    def _resolve_eligible_methods(
+        self,
+        methods: Sequence[str],
+        configured_methods: Any = None,
+    ) -> List[str]:
+        """Resolve the configured method names that are eligible consensus voters."""
+        available = []
+        seen = set()
+        for method_name in methods:
+            identity = self._method_identity(method_name)
+            if identity not in seen:
+                available.append((identity, method_name))
+                seen.add(identity)
+
+        configured = (
+            self.eligible_methods
+            if configured_methods is None
+            else configured_methods
+        )
+        if configured is None:
+            return [method_name for _, method_name in available]
+        if isinstance(configured, str):
+            configured = [configured]
+
+        try:
+            configured = list(configured)
+        except TypeError:
+            raise ValueError("eligible_methods must be a sequence of method names")
+
+        requested_identities = set()
+        for method_name in configured:
+            if not isinstance(method_name, str):
+                raise ValueError("eligible_methods must be a sequence of method names")
+            requested_identities.add(self._method_identity(method_name))
+
+        available_identities = {identity for identity, _ in available}
+        unknown_methods = [
+            method_name
+            for method_name in configured
+            if self._method_identity(method_name) not in available_identities
+        ]
+        if unknown_methods:
+            raise ValueError(
+                "eligible_methods contains methods not configured for extraction: "
+                + ", ".join(unknown_methods)
+            )
+
+        return [
+            method_name
+            for identity, method_name in available
+            if identity in requested_identities
+        ]
+
+    def _align_entities_to_text(
+        self, entities: List[Entity], text: str
+    ) -> List[Entity]:
+        """Resolve missing offsets before span-based methods are merged.
+
+        Some providers, notably typed LLM extraction, can return text and
+        labels without offsets. For a single method that is harmless, but a
+        span-based merge needs document locations. Missing spans are therefore
+        aligned by a deterministic, per-label text search. Valid provider
+        offsets are preserved; candidates that cannot be aligned are excluded
+        because union and consensus cannot safely merge them.
+        """
+        next_offsets = {}
+        occupied_offsets = {}
+        aligned = []
+
+        for entity in entities:
+            needle = entity.text
+            if not isinstance(needle, str) or not needle:
+                continue
+
+            key = (needle.casefold(), self._canonical_label(entity.label))
+            start_char = entity.start_char
+            end_char = entity.end_char
+            has_valid_span = (
+                isinstance(start_char, int)
+                and isinstance(end_char, int)
+                and 0 <= start_char < end_char <= len(text)
+                and text[start_char:end_char].casefold() == needle.casefold()
+            )
+            if has_valid_span:
+                aligned.append(entity)
+                next_offsets[key] = max(next_offsets.get(key, 0), end_char)
+                occupied_offsets.setdefault(key, set()).add((start_char, end_char))
+                continue
+
+            prior_offset = next_offsets.get(key, 0)
+            hinted_start = start_char if isinstance(start_char, int) else 0
+            search_start = max(prior_offset, min(max(hinted_start, 0), len(text)))
+            occupied = occupied_offsets.setdefault(key, set())
+            match = None
+            match_offset = 0
+            left_boundary = (
+                r"(?<!\w)" if needle[0].isalnum() or needle[0] == "_" else ""
+            )
+            right_boundary = (
+                r"(?!\w)" if needle[-1].isalnum() or needle[-1] == "_" else ""
+            )
+            pattern = re.compile(
+                left_boundary + re.escape(needle) + right_boundary,
+                re.IGNORECASE,
+            )
+            for segment, offset in ((text[search_start:], search_start), (text, 0)):
+                for candidate in pattern.finditer(segment):
+                    candidate_start = offset + candidate.start()
+                    candidate_end = offset + candidate.end()
+                    if (candidate_start, candidate_end) not in occupied:
+                        match = candidate
+                        match_offset = offset
+                        break
+                if match is not None:
+                    break
+
+            if match is None:
+                continue
+
+            resolved_start = match_offset + match.start()
+            resolved_end = match_offset + match.end()
+            aligned.append(
+                Entity(
+                    text=entity.text,
+                    label=entity.label,
+                    start_char=resolved_start,
+                    end_char=resolved_end,
+                    confidence=entity.confidence,
+                    metadata=dict(entity.metadata or {}),
+                )
+            )
+            next_offsets[key] = resolved_end
+            occupied.add((resolved_start, resolved_end))
+
+        return aligned
+
     def extract(self, text: Union[str, List[Dict[str, Any]], List[str]], pipeline_id: Optional[str] = None, **kwargs) -> Union[List[Entity], List[List[Entity]]]:
         """
         Alias for extract_entities.
@@ -347,11 +627,39 @@ class NERExtractor:
                 )
                 return []
 
-            # Use method from options if provided, otherwise use instance method
-            methods = options.get("method", self.method)
-            if isinstance(methods, str):
-                methods = [methods]
-            methods = self._filter_unusable_methods(methods)
+            # Use method from options if provided, otherwise use instance method.
+            # Keep the requested list separate from the executable list: in
+            # consensus mode, a configured method with no result is still an
+            # eligible non-supporting vote.
+            requested_methods = options.get("method", self.method)
+            if isinstance(requested_methods, str):
+                requested_methods = [requested_methods]
+
+            merge_strategy = self._validate_merge_strategy(
+                options.get("merge_strategy", self.merge_strategy)
+            )
+            if merge_strategy == "consensus":
+                eligible_methods = self._resolve_eligible_methods(
+                    requested_methods,
+                    options.get("eligible_methods", self.eligible_methods),
+                )
+            else:
+                # eligible_methods is a consensus-only setting. Union should
+                # retain every configured method's complementary output.
+                eligible_methods = self._resolve_eligible_methods(
+                    requested_methods, requested_methods
+                )
+            methods = self._filter_unusable_methods(requested_methods)
+
+            min_votes = self._validate_min_votes(
+                options.get("min_votes", self.min_votes)
+            )
+            min_agreement = self._validate_min_agreement(
+                options.get("min_agreement", self.min_agreement)
+            )
+            method_weights = self._validate_method_weights(
+                options.get("method_weights", self.method_weights)
+            )
 
             min_confidence = options.get("min_confidence", self.min_confidence)
             entity_types = options.get("entity_types", self.entity_types)
@@ -361,7 +669,9 @@ class NERExtractor:
             if entity_types:
                 all_options["entity_types"] = entity_types
 
-            # Try each method in order (fallback chain)
+            # Try each method in order. Fallback returns the first non-empty
+            # result; union and consensus keep empty method results so their
+            # denominators retain configured method provenance.
             all_entities = []
             for method_name in methods:
                 try:
@@ -373,6 +683,8 @@ class NERExtractor:
 
                     # Prepare method-specific options
                     method_options = all_options.copy()
+                    for merge_option in self._MERGE_OPTION_KEYS:
+                        method_options.pop(merge_option, None)
                     if method_name == "huggingface":
                         # Prioritize runtime options over config/defaults
                         method_options["model"] = (
@@ -400,6 +712,8 @@ class NERExtractor:
                                 method_options["api_key"] = api_key
 
                     entities = method_func(text, **method_options)
+                    if merge_strategy != "fallback":
+                        entities = self._align_entities_to_text(entities, text)
 
                     # Apply weighted scoring if entity_types are provided
                     if entity_types:
@@ -418,15 +732,14 @@ class NERExtractor:
                     # Filter by confidence
                     filtered = [e for e in entities if e.confidence >= min_confidence]
                     
-                    if filtered:
-                        all_entities.append((method_name, filtered))
-
-                        # If not using ensemble, return first successful result
-                        if not self.ensemble_voting:
+                    if merge_strategy == "fallback":
+                        if filtered:
                             # Ensure default metadata
                             for e in filtered:
-                                if e.metadata is None: e.metadata = {}
-                                if "batch_index" not in e.metadata: e.metadata["batch_index"] = 0
+                                if e.metadata is None:
+                                    e.metadata = {}
+                                if "batch_index" not in e.metadata:
+                                    e.metadata["batch_index"] = 0
 
                             self.progress_tracker.stop_tracking(
                                 tracking_id,
@@ -434,6 +747,8 @@ class NERExtractor:
                                 message=f"Extracted {len(filtered)} entities using {method_name}",
                             )
                             return filtered
+                    else:
+                        all_entities.append((method_name, filtered))
 
                 except Exception as e:
                     self.logger.warning(
@@ -441,15 +756,23 @@ class NERExtractor:
                     )
                     continue
 
-            # Ensemble voting if enabled
-            if self.ensemble_voting and len(all_entities) > 1:
+            if merge_strategy == "consensus":
                 entities = self._vote_entities(
-                    [entities for _, entities in all_entities]
+                    all_entities,
+                    eligible_methods=eligible_methods,
+                    min_votes=min_votes,
+                    min_agreement=min_agreement,
+                    method_weights=method_weights,
+                )
+            elif merge_strategy == "union":
+                entities = self._union_entities(
+                    all_entities,
+                    eligible_methods=eligible_methods,
+                    method_weights=method_weights,
                 )
-            elif all_entities:
-                entities = all_entities[0][1]  # Use first successful method
             else:
-                # Fallback to pattern-based extraction if all models fail
+                # Only the explicit fallback strategy may introduce its own
+                # pattern candidates after every configured method fails.
                 entities = self._extract_fallback(text)
 
             # Post-processing if enabled
@@ -488,30 +811,466 @@ class NERExtractor:
         return filtered
 
     def _vote_entities(
-        self, results: List[List[Entity]], threshold: float = 0.5
+        self,
+        results: Sequence[Union[List[Entity], Tuple[str, List[Entity]]]],
+        threshold: Optional[float] = None,
+        *,
+        eligible_methods: Optional[Sequence[str]] = None,
+        min_votes: Optional[int] = None,
+        min_agreement: Optional[float] = None,
+        method_weights: Optional[Dict[str, float]] = None,
     ) -> List[Entity]:
-        """Vote on entities across methods."""
-        entity_counts = {}
-        total_methods = len(results)
+        """Merge method results using span-aligned cross-method consensus.
 
-        for entities in results:
+        ``results`` accepts the historical ``List[List[Entity]]`` shape as
+        well as ``(method_name, entities)`` pairs. The latter retains method
+        provenance, while anonymous historical inputs receive stable generated
+        names. ``threshold`` remains a compatibility alias for
+        ``min_agreement``; confidence is never used as a substitute for votes.
+        """
+        resolved_min_votes = self._validate_min_votes(
+            self.min_votes if min_votes is None else min_votes
+        )
+        if min_agreement is None:
+            min_agreement = threshold if threshold is not None else self.min_agreement
+        resolved_min_agreement = self._validate_min_agreement(min_agreement)
+        resolved_method_weights = self._validate_method_weights(
+            self.method_weights if method_weights is None else method_weights
+        )
+
+        return self._merge_method_results(
+            results,
+            merge_strategy="consensus",
+            eligible_methods=eligible_methods,
+            min_votes=resolved_min_votes,
+            min_agreement=resolved_min_agreement,
+            method_weights=resolved_method_weights,
+        )
+
+    def _union_entities(
+        self,
+        results: Sequence[Union[List[Entity], Tuple[str, List[Entity]]]],
+        *,
+        eligible_methods: Optional[Sequence[str]] = None,
+        method_weights: Optional[Dict[str, float]] = None,
+    ) -> List[Entity]:
+        """Merge all method results while retaining single-method candidates."""
+        resolved_method_weights = self._validate_method_weights(
+            self.method_weights if method_weights is None else method_weights
+        )
+        return self._merge_method_results(
+            results,
+            merge_strategy="union",
+            eligible_methods=eligible_methods,
+            min_votes=1,
+            min_agreement=None,
+            method_weights=resolved_method_weights,
+        )
+
+    def _merge_method_results(
+        self,
+        results: Sequence[Union[List[Entity], Tuple[str, List[Entity]]]],
+        *,
+        merge_strategy: str,
+        eligible_methods: Optional[Sequence[str]],
+        min_votes: int,
+        min_agreement: Optional[float],
+        method_weights: Dict[str, float],
+    ) -> List[Entity]:
+        """Align overlapping mentions and merge them with a named strategy."""
+        method_results = self._normalize_method_results(results)
+        eligible_methods = self._normalize_eligible_method_names(
+            eligible_methods, method_results
+        )
+        if not eligible_methods:
+            return []
+
+        eligible_identities = {
+            self._method_identity(method_name) for method_name in eligible_methods
+        }
+        clusters = self._cluster_entities(method_results, eligible_identities)
+        merged = []
+
+        for cluster in clusters:
+            entity = self._build_merged_entity(
+                cluster,
+                eligible_methods=eligible_methods,
+                merge_strategy=merge_strategy,
+                min_votes=min_votes,
+                min_agreement=min_agreement,
+            )
+            if entity is None:
+                continue
+
+            merged.append(entity)
+
+        if merge_strategy == "consensus":
+            merged = self._resolve_consensus_label_conflicts(
+                merged, method_weights
+            )
+
+        return sorted(
+            merged,
+            key=lambda entity: (
+                entity.start_char,
+                entity.end_char,
+                entity.label,
+                entity.text.casefold(),
+            ),
+        )
+
+    def _normalize_method_results(
+        self,
+        results: Sequence[Union[List[Entity], Tuple[str, List[Entity]]]],
+    ) -> List[Tuple[str, List[Entity]]]:
+        """Coalesce alias methods so one backend cannot cast two votes."""
+        normalized = {}
+        for index, result in enumerate(results):
+            if (
+                isinstance(result, tuple)
+                and len(result) == 2
+                and isinstance(result[0], str)
+            ):
+                method_name, entities = result
+            else:
+                method_name, entities = f"method_{index + 1}", result
+
+            identity = self._method_identity(method_name)
+            if identity not in normalized:
+                normalized[identity] = {"name": method_name, "entities": []}
+            if entities:
+                normalized[identity]["entities"].extend(entities)
+
+        return [
+            (data["name"], data["entities"])
+            for data in normalized.values()
+        ]
+
+    def _normalize_eligible_method_names(
+        self,
+        eligible_methods: Optional[Sequence[str]],
+        method_results: Sequence[Tuple[str, List[Entity]]],
+    ) -> List[str]:
+        """Keep configured failed methods in the consensus denominator."""
+        if eligible_methods is None:
+            eligible_methods = [method_name for method_name, _ in method_results]
+
+        known_names = {
+            self._method_identity(method_name): method_name
+            for method_name, _ in method_results
+        }
+        normalized = []
+        seen = set()
+        for method_name in eligible_methods:
+            identity = self._method_identity(method_name)
+            if identity in seen:
+                continue
+            normalized.append(known_names.get(identity, method_name))
+            seen.add(identity)
+        return normalized
+
+    def _cluster_entities(
+        self,
+        method_results: Sequence[Tuple[str, List[Entity]]],
+        eligible_identities: set,
+    ) -> List[List[Tuple[str, Entity]]]:
+        """Align same-label mentions with deterministic one-to-one matching.
+
+        Each method is matched to existing candidates as a batch, ordered by
+        descending span IoU. This prevents an early, weaker boundary variant
+        from consuming a method's only vote before its exact match is seen.
+        Different labels stay separate here and are reconciled only after
+        each label's independent support has been counted.
+        """
+        clusters_by_label = {}
+        ordered_results = sorted(
+            method_results,
+            key=lambda result: (
+                self._method_identity(result[0]),
+                result[0],
+            ),
+        )
+
+        for method_name, entities in ordered_results:
+            method_identity = self._method_identity(method_name)
+            if method_identity not in eligible_identities:
+                continue
+
+            unique_entities = {}
             for entity in entities:
-                key = (entity.text.lower(), entity.label)
-                if key not in entity_counts:
-                    entity_counts[key] = {"entity": entity, "score": 0.0, "count": 0}
-                entity_counts[key]["score"] += entity.confidence
-                entity_counts[key]["count"] += 1
+                label = self._canonical_label(entity.label)
+                key = (label, entity.start_char, entity.end_char)
+                existing = unique_entities.get(key)
+                if existing is None or self._entity_order_key(
+                    entity
+                ) < self._entity_order_key(existing):
+                    unique_entities[key] = entity
 
-        # Return entities that meet threshold
-        voted = []
-        for key, data in entity_counts.items():
-            avg_score = data["score"] / data["count"]
-            if avg_score >= threshold:
-                entity = data["entity"]
-                entity.confidence = avg_score
-                voted.append(entity)
+            entities_by_label = {}
+            for entity in unique_entities.values():
+                label = self._canonical_label(entity.label)
+                entities_by_label.setdefault(label, []).append(entity)
 
-        return voted
+            for label in sorted(entities_by_label):
+                candidates = sorted(
+                    entities_by_label[label], key=self._entity_order_key
+                )
+                label_clusters = clusters_by_label.setdefault(label, [])
+                edges = []
+                for candidate_index, candidate in enumerate(candidates):
+                    for cluster_index, cluster in enumerate(label_clusters):
+                        if any(
+                            self._method_identity(cluster_method) == method_identity
+                            for cluster_method, _ in cluster
+                        ):
+                            continue
+                        # A cluster represents one consensus mention, so a
+                        # candidate must overlap *every* vote already in it.
+                        # Using a best-pair score here would let A~B and B~C
+                        # turn into a false A/B/C consensus when A !~ C.
+                        scores = [
+                            self._span_iou(candidate, clustered_entity)
+                            for _, clustered_entity in cluster
+                        ]
+                        score = min(scores)
+                        if score >= self._MIN_SPAN_IOU:
+                            edges.append((score, candidate_index, cluster_index))
+
+                matched_candidates = set()
+                matched_clusters = set()
+                for _, candidate_index, cluster_index in sorted(
+                    edges,
+                    key=lambda item: (
+                        -item[0],
+                        self._entity_order_key(candidates[item[1]]),
+                        item[2],
+                    ),
+                ):
+                    if (
+                        candidate_index in matched_candidates
+                        or cluster_index in matched_clusters
+                    ):
+                        continue
+                    label_clusters[cluster_index].append(
+                        (method_name, candidates[candidate_index])
+                    )
+                    matched_candidates.add(candidate_index)
+                    matched_clusters.add(cluster_index)
+
+                for candidate_index, candidate in enumerate(candidates):
+                    if candidate_index not in matched_candidates:
+                        label_clusters.append([(method_name, candidate)])
+
+        return [
+            cluster
+            for label in sorted(clusters_by_label)
+            for cluster in clusters_by_label[label]
+        ]
+
+    def _resolve_consensus_label_conflicts(
+        self,
+        entities: Sequence[Entity],
+        method_weights: Dict[str, float],
+    ) -> List[Entity]:
+        """Choose one deterministic label when candidates share one span.
+
+        Cross-label candidates only conflict when their final document spans
+        are identical. Nested entities at different spans remain distinct.
+        """
+        resolved = {}
+
+        def conflict_order_key(entity: Entity) -> Tuple[Any, ...]:
+            metadata = entity.metadata or {}
+            support_weight = sum(
+                self._method_weight(method_name, method_weights)
+                for method_name in metadata.get("supporting_methods", [])
+            )
+            confidence = self._numeric_confidence(entity.confidence)
+            confidence_key = -confidence if confidence is not None else float("inf")
+            return (
+                -support_weight,
+                -metadata.get("vote_count", 0),
+                confidence_key,
+                entity.label,
+                entity.text.casefold(),
+            )
+
+        for entity in entities:
+            key = (entity.start_char, entity.end_char)
+            existing = resolved.get(key)
+            if existing is None or conflict_order_key(entity) < conflict_order_key(
+                existing
+            ):
+                resolved[key] = entity
+
+        return list(resolved.values())
+
+    @staticmethod
+    def _span_iou(first: Entity, second: Entity) -> float:
+        """Return overlap-over-union for two document spans."""
+        intersection = max(
+            0,
+            min(first.end_char, second.end_char)
+            - max(first.start_char, second.start_char),
+        )
+        if not intersection:
+            return 0.0
+        union = max(first.end_char, second.end_char) - min(
+            first.start_char, second.start_char
+        )
+        return intersection / union if union else 0.0
+
+    @classmethod
+    def _canonical_label(cls, label: str) -> str:
+        """Normalize common NER aliases and BIO prefixes before label voting."""
+        normalized = str(label).strip().upper()
+        if "-" in normalized:
+            prefix, remainder = normalized.split("-", 1)
+            if prefix in {"B", "I", "L", "U", "E", "S"}:
+                normalized = remainder
+        return cls._LABEL_ALIASES.get(normalized, normalized)
+
+    @staticmethod
+    def _numeric_confidence(confidence: Any) -> Optional[float]:
+        """Convert a usable confidence score without treating missing scores as zero."""
+        if confidence is None:
+            return None
+        try:
+            normalized = float(confidence)
+        except (TypeError, ValueError):
+            return None
+        return normalized if math.isfinite(normalized) else None
+
+    @classmethod
+    def _entity_order_key(cls, entity: Entity) -> Tuple[Any, ...]:
+        """Provide a deterministic winner for boundary and confidence variants."""
+        confidence = cls._numeric_confidence(entity.confidence)
+        confidence_key = -confidence if confidence is not None else float("inf")
+        return (
+            confidence_key,
+            -(entity.end_char - entity.start_char),
+            entity.start_char,
+            entity.end_char,
+            entity.text.casefold(),
+            entity.label.casefold(),
+        )
+
+    def _method_weight(
+        self,
+        method_name: str,
+        method_weights: Dict[str, float],
+    ) -> float:
+        """Read a weight using the canonical backend name."""
+        identity = self._method_identity(method_name)
+        return method_weights.get(identity, 1.0)
+
+    def _build_merged_entity(
+        self,
+        cluster: Sequence[Tuple[str, Entity]],
+        *,
+        eligible_methods: Sequence[str],
+        merge_strategy: str,
+        min_votes: int,
+        min_agreement: Optional[float],
+    ) -> Optional[Entity]:
+        """Resolve one same-label, offset-aligned candidate."""
+        selected_by_method = {}
+        for method_name, entity in cluster:
+            identity = self._method_identity(method_name)
+            existing = selected_by_method.get(identity)
+            if existing is None or (
+                self._entity_order_key(entity)
+                < self._entity_order_key(existing[1])
+            ):
+                selected_by_method[identity] = (method_name, entity)
+
+        eligible_records = []
+        seen = set()
+        for method_name in eligible_methods:
+            identity = self._method_identity(method_name)
+            if identity not in seen:
+                eligible_records.append((identity, method_name))
+                seen.add(identity)
+        if not eligible_records:
+            return None
+
+        if not selected_by_method:
+            return None
+
+        supporting_entries = [
+            (identity, method_name, entity)
+            for identity, (method_name, entity) in selected_by_method.items()
+        ]
+        vote_count = len(supporting_entries)
+        agreement = vote_count / len(eligible_records)
+        if merge_strategy == "consensus" and (
+            vote_count < min_votes
+            or (min_agreement is not None and agreement < min_agreement)
+        ):
+            return None
+
+        representative = min(
+            (entity for _, _, entity in supporting_entries), key=self._entity_order_key
+        )
+        canonical_label = self._canonical_label(representative.label)
+
+        supporting_by_identity = {
+            identity: (method_name, entity)
+            for identity, method_name, entity in supporting_entries
+        }
+        supporting_methods = [
+            method_name
+            for identity, method_name in eligible_records
+            if identity in supporting_by_identity
+        ]
+        method_scores = {
+            method_name: (
+                self._numeric_confidence(supporting_by_identity[identity][1].confidence)
+                if identity in supporting_by_identity
+                else None
+            )
+            for identity, method_name in eligible_records
+        }
+
+        confidence_scores = []
+        for identity, method_name in eligible_records:
+            if identity not in supporting_by_identity:
+                continue
+            score = self._numeric_confidence(
+                supporting_by_identity[identity][1].confidence
+            )
+            if score is not None:
+                confidence_scores.append(score)
+
+        if confidence_scores:
+            confidence = sum(confidence_scores) / len(confidence_scores)
+        else:
+            confidence = representative.confidence
+
+        metadata = dict(representative.metadata or {})
+        metadata.update(
+            {
+                "merge_strategy": merge_strategy,
+                "supporting_methods": supporting_methods,
+                "vote_count": len(supporting_methods),
+                "eligible_method_count": len(eligible_records),
+                "agreement": agreement,
+                "method_scores": method_scores,
+            }
+        )
+
+        return Entity(
+            text=representative.text,
+            label=(
+                canonical_label
+                if merge_strategy == "consensus"
+                else representative.label
+            ),
+            start_char=representative.start_char,
+            end_char=representative.end_char,
+            confidence=confidence,
+            metadata=metadata,
+        )
 
     def _post_process_entities(self, entities: List[Entity], text: str) -> List[Entity]:
         """Post-process entities for refinement."""

semantica/vector_store/milvus_store.py

@@ -627,6 +627,46 @@ class MilvusStore:
             )
             raise
 
+    def delete_vectors(self, vector_ids: List[str], **options) -> Dict[str, Any]:
+        """Delete vectors from collection by their ids.
+
+        Args:
+            vector_ids: Vector ids to delete
+            **options: Additional options
+
+        Returns:
+            A dict with the number of matching entities that were deleted
+            (``delete_count``).
+        """
+        if self.collection is None:
+            raise ProcessingError(
+                "Collection not initialized. Call create_collection() or get_collection() first."
+            )
+
+        if not vector_ids:
+            return {"delete_count": 0}
+
+        try:
+            # Milvus DELETE deletes by expression. Escape each id so a quote or
+            # backslash in an id cannot break out of the string literal.
+            if len(vector_ids) == 1:
+                expr = f"id == {_format_milvus_value(vector_ids[0])}"
+            else:
+                formatted = ", ".join(_format_milvus_value(i) for i in vector_ids)
+                expr = f"id in [{formatted}]"
+            result = self.collection.collection.delete(expr=expr, **options)
+            delete_count = getattr(result, "delete_count", 0)
+            if delete_count is None:
+                delete_count = 0
+            elif isinstance(delete_count, (str, bytes)):
+                try:
+                    delete_count = int(delete_count)
+                except (TypeError, ValueError):
+                    delete_count = 0
+            return {"delete_count": delete_count}
+        except Exception as e:
+            raise ProcessingError(f"Failed to delete vectors: {str(e)}")
+
     def get_vector(self, vector_id: str) -> Optional[np.ndarray]:
         """Get vector by ID."""
         if not MILVUS_AVAILABLE or not self.collection:

tests/context/test_decision_causal_edge_regression.py

@@ -452,3 +452,42 @@ def test_find_precedents_sees_lowercase_precedent_edge():
     precedents = graph.find_precedents(later)
 
     assert [d.decision_id for d in precedents] == [precedent]
+
+
+def test_heuristic_cause_reported_once_per_shared_entity_pair():
+    """A potential cause found through the shared-entity heuristic must be
+    reported once, not once per shared entity.
+
+    ``trace_decision_causality()`` collects ``potential_causes`` by looping
+    over every entity of the current decision, so a decision sharing two
+    entities with an earlier one (e.g. the same customer and the same
+    property) used to be appended twice and produced two identical
+    "influences" chains.
+    """
+    graph = ContextGraph(advanced_analytics=True)
+    cause = graph.record_decision(
+        category="lending", scenario="earlier review", reasoning="r",
+        outcome="approved", confidence=0.9,
+        entities=["customer_123", "property_456"],
+    )
+    effect = graph.record_decision(
+        category="risk", scenario="later review", reasoning="r",
+        outcome="flagged", confidence=0.9,
+        entities=["customer_123", "property_456"],
+    )
+    # Pin timestamps so the earlier/later ordering is deterministic.
+    graph._decisions[cause]["timestamp"] = 100.0
+    graph._decisions[effect]["timestamp"] = 200.0
+
+    chains = graph.trace_decision_chain(effect)
+
+    influence_hops = [
+        (hop["from"], hop["to"])
+        for chain in chains
+        for hop in chain["hops"]
+        if hop["type"] == "influences"
+    ]
+    assert influence_hops, "shared-entity heuristic must find the earlier decision"
+    assert influence_hops.count((cause, effect)) == 1, (
+        "the same (cause, effect) pair must be reported once, not once per shared entity"
+    )

tests/context/test_erasure_coordinator.py

@@ -89,7 +89,7 @@ class _DeleteStore:
 
 
 class _NoDeleteStore:
-    """Backend shaped like FAISS/Milvus/Weaviate: no delete surface at all."""
+    """Backend shaped like FAISS: no delete surface at all."""
 
     backend = "faiss"
 
@@ -399,6 +399,121 @@ class TestReceipt(unittest.TestCase):
 
         self.assertEqual(receipt.stores["graph"]["status"], STATUS_ERASED)
 
+    def test_to_dict_deep_copies_nested_store_results(self):
+        """A nested dict in a store result is not shared with the live receipt.
+
+        ``backend_result`` from a vector backend is a dict of its own, so a
+        shallow per-store copy leaves it referenced by both the payload and the
+        receipt -- sanitizing the payload for a user-facing response would
+        silently corrupt the audit record.
+        """
+        receipt = ErasureReceipt(
+            entity_id="customer-4471",
+            stores={
+                "vectors": {
+                    "status": STATUS_ERASED,
+                    "backend": "qdrant",
+                    "backend_result": {"status": "completed"},
+                },
+            },
+        )
+
+        payload = receipt.to_dict()
+        payload["stores"]["vectors"]["backend_result"]["status"] = "redacted"
+
+        self.assertEqual(
+            receipt.stores["vectors"]["backend_result"]["status"], "completed"
+        )
+
+    def test_to_dict_store_results_are_distinct_objects(self):
+        """The per-store dict and any nested dict in the payload must not be
+        the same objects as the ones in the live receipt.
+
+        A mutation test proves *isolation* only when the copy actually
+        happened; identity checks prove *that* a copy was made.
+        """
+        receipt = ErasureReceipt(
+            entity_id="customer-4471",
+            stores={
+                "vectors": {
+                    "status": STATUS_ERASED,
+                    "backend": "qdrant",
+                    # Realistic Qdrant-shaped backend_result with rendered enum
+                    "backend_result": {"status": "UpdateStatus.COMPLETED", "points": 1},
+                    "vector_ids": 2,
+                    "via": "delete_vectors",
+                },
+                "graph": {"status": STATUS_ERASED, "nodes": 1, "edges": 3},
+            },
+        )
+
+        payload = receipt.to_dict()
+
+        # The stores container itself is a new dict.
+        self.assertIsNot(payload["stores"], receipt.stores)
+
+        # Each per-store result dict is a new object.
+        self.assertIsNot(
+            payload["stores"]["vectors"], receipt.stores["vectors"]
+        )
+        self.assertIsNot(
+            payload["stores"]["graph"], receipt.stores["graph"]
+        )
+
+        # The nested backend_result dict is also a new object.
+        self.assertIsNot(
+            payload["stores"]["vectors"]["backend_result"],
+            receipt.stores["vectors"]["backend_result"],
+        )
+
+        # Values are equal (correct copy), not just distinct references.
+        self.assertEqual(
+            payload["stores"]["vectors"]["backend_result"],
+            {"status": "UpdateStatus.COMPLETED", "points": 1},
+        )
+        self.assertEqual(payload["stores"]["graph"], {"status": STATUS_ERASED, "nodes": 1, "edges": 3})
+
+    def test_to_dict_isolation_across_multiple_stores(self):
+        """Mutations to any store in the payload must not affect any other
+        store in either the payload or the live receipt.
+
+        This catches a hypothetical implementation that shares a single deep
+        copy across all stores rather than copying each independently.
+        """
+        receipt = ErasureReceipt(
+            entity_id="e1",
+            stores={
+                "vectors": {
+                    "status": STATUS_UNSUPPORTED,
+                    "backend": "faiss",
+                    "vector_ids": 1,
+                    "detail": "backend exposes no delete()",
+                },
+                "memory": {"status": STATUS_ERASED, "items": 4},
+                "graph": {
+                    "status": STATUS_ERASED,
+                    "nodes": 1,
+                    "edges": 2,
+                },
+            },
+        )
+
+        payload = receipt.to_dict()
+
+        # Mutate every store in the payload.
+        payload["stores"]["vectors"]["status"] = "tampered"
+        payload["stores"]["memory"]["items"] = 0
+        payload["stores"]["graph"]["nodes"] = 99
+
+        # None of the live receipt's stores are affected.
+        self.assertEqual(receipt.stores["vectors"]["status"], STATUS_UNSUPPORTED)
+        self.assertEqual(receipt.stores["memory"]["items"], 4)
+        self.assertEqual(receipt.stores["graph"]["nodes"], 1)
+
+        # The other stores in the payload are also unaffected (no aliasing).
+        self.assertEqual(payload["stores"]["memory"]["items"], 0)   # our mutation
+        self.assertEqual(receipt.stores["memory"]["items"], 4)       # unchanged
+
     def test_receipt_and_tombstone_agree_on_when_the_erasure_happened(self):
         graph = _graph()
         receipt = ErasureCoordinator(graph=graph).erase_entity(

tests/explorer/test_ontology_subissue3.py

@@ -13,6 +13,7 @@ pytest.importorskip("fastapi")
 
 from semantica.explorer.app import create_app  # noqa: E402
 from semantica.explorer.routes.ontology import (  # noqa: E402
+    _MAX_ANALYSIS_NODES,
     OntologyEntry,
     _convert_ontology_to_graph,
     _node_belongs_to_ontology,
@@ -310,6 +311,66 @@ def test_ontology_graph_ignores_unrelated_data_when_enforcing_size_limit(client)
     }
 
 
+def test_ontology_graph_rejects_oversized_core_and_stops_scanning(client, monkeypatch):
+    graph = client.app.state.session.graph
+    for index in range(5_001):
+        graph.add_node(
+            f"http://example.org/onto-a#Bulk{index:05d}",
+            node_type="owl:Class",
+            content="Bulk",
+            scheme_uri="http://example.org/onto-a",
+        )
+    for index in range(3_000):
+        graph.add_node(
+            f"urn:unrelated:{index}",
+            node_type="owl:Class",
+            content="Unrelated",
+            scheme_uri="http://example.org/onto-b",
+        )
+
+    streamed = 0
+    original_iter_nodes = GraphSession.iter_nodes
+
+    def counting_iter_nodes(self, node_type=None):
+        nonlocal streamed
+        for node in original_iter_nodes(self, node_type=node_type):
+            streamed += 1
+            yield node
+
+    monkeypatch.setattr(GraphSession, "iter_nodes", counting_iter_nodes)
+
+    response = client.get(
+        "/api/ontology/graph",
+        params={"uri": "http://example.org/onto-a"},
+    )
+
+    assert response.status_code == 413
+    assert str(_MAX_ANALYSIS_NODES) in response.json()["detail"]
+    # The graph holds 8,001 owl:Class nodes and onto-a's own sort first, so a
+    # scan that abandons at the cap sees far fewer than the whole type.
+    assert streamed < 6_000
+
+
+def test_ontology_graph_hydrates_external_edge_targets_in_sorted_order(client):
+    graph = client.app.state.session.graph
+    external = "http://external.example/Thing"
+    also_external = "http://external.example/Aardvark"
+    graph.add_node(external, node_type="owl:Class", content="External Thing")
+    graph.add_node(also_external, node_type="owl:Class", content="External Aardvark")
+    graph.add_edge("http://example.org/onto-a#name", external, edge_type="rdfs:range")
+    graph.add_edge("http://example.org/onto-a#name", also_external, edge_type="rdfs:range")
+
+    response = client.get(
+        "/api/ontology/graph",
+        params={"uri": "http://example.org/onto-a"},
+    )
+
+    assert response.status_code == 200
+    node_ids = [node["id"] for node in response.json()["nodes"]]
+    assert {external, also_external} <= set(node_ids)
+    assert node_ids == sorted(node_ids)
+
+
 def test_shacl_generate_and_shapes(client):
     response = client.post(
         "/api/ontology/shacl/generate",

tests/kg/test_dangling_synthetic_endpoint.py

@@ -0,0 +1,460 @@
+"""Tests for reconciling synthetic relation endpoints into the graph.
+
+Relation extraction synthesizes ``UNKNOWN`` entities for endpoints that are
+absent from the NER entity list, but ``GraphBuilder._process_item`` kept only
+the endpoint string. The resulting graph then failed ``GraphValidator`` with
+``DANGLING_EDGE``. See issue #1463.
+"""
+
+import pytest
+
+from semantica.kg import GraphBuilder, GraphValidator
+from semantica.semantic_extract import Entity, Relation, Triplet
+
+
+def _known(text="Known Entity", label="CONCEPT"):
+    return Entity(text=text, label=label, start_char=0, end_char=len(text))
+
+
+def _synthetic(text):
+    return Entity(
+        text=text,
+        label="UNKNOWN",
+        start_char=0,
+        end_char=len(text),
+        confidence=0.8,
+        metadata={"synthetic": True},
+    )
+
+
+def _rel(subject, predicate, object_):
+    return Relation(subject=subject, predicate=predicate, object=object_)
+
+
+def _issues(graph):
+    return GraphValidator().validate(graph).to_dict()["issues"]
+
+
+def test_single_missing_endpoint_promoted():
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_known(), _rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    entity_ids = {e["id"] for e in graph["entities"]}
+    assert "Synthetic Entity" in entity_ids
+    assert not [i for i in _issues(graph) if i.get("code") == "DANGLING_EDGE"]
+
+
+def test_two_missing_endpoints_promoted():
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_rel(_synthetic("Org A"), "related_to", _synthetic("Org B"))],
+        extract=False,
+    )
+
+    entity_ids = {e["id"] for e in graph["entities"]}
+    assert {"Org A", "Org B"} <= entity_ids
+
+
+def test_promoted_default_confidence_is_0_8():
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    synthetic = next(e for e in graph["entities"] if e["id"] == "Synthetic Entity")
+    assert synthetic["confidence"] == 0.8
+    assert synthetic["metadata"].get("synthetic") is True
+    assert synthetic["type"] == "UNKNOWN"
+    assert synthetic["name"] == "Synthetic Entity"
+
+
+def test_no_duplicate_when_endpoint_already_present():
+    # A synthetic endpoint sharing an id with an existing entity must not
+    # create a duplicate node.
+    known = _known("Shared", label="CONCEPT")
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [known, _rel(known, "related_to", _synthetic("Shared"))],
+        extract=False,
+    )
+
+    ids = [e["id"] for e in graph["entities"] if e["id"] == "Shared"]
+    assert len(ids) == 1
+
+
+def test_merge_entities_true_no_dangling_edge():
+    # With entity merging explicitly enabled (and conflict resolution on, the
+    # default), a promoted synthetic endpoint still must not leave a dangling
+    # edge and must not duplicate the existing entity.
+    graph = GraphBuilder(merge_entities=True).build(
+        [_known(), _rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    assert not [i for i in _issues(graph) if i.get("code") == "DANGLING_EDGE"]
+    synthetic = [e for e in graph["entities"] if e["id"] == "Synthetic Entity"]
+    assert len(synthetic) == 1
+
+
+def test_reject_policy_drops_dangling_relationship():
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        unknown_relation_endpoint="reject",
+    )
+    graph = builder.build(
+        [_rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    assert graph["relationships"] == []
+    assert "Synthetic Entity" not in {e["id"] for e in graph["entities"]}
+
+
+def test_reject_policy_requires_config_to_promote():
+    # Same input as test_reject_policy but policies must be explicit; only
+    # "reject" disables promotion, the default remains permissive.
+    default_builder = GraphBuilder(resolve_conflicts=False)
+    graph = default_builder.build(
+        [_rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+    assert "Synthetic Entity" in {e["id"] for e in graph["entities"]}
+
+
+def test_dict_source_relationship_with_all_endpoints_present():
+    # Non-synthetic dict relationships with a legitimate entity set are left
+    # untouched; endpoints already resolve so no dangling edge appears.
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        {
+            "entities": [{"id": "Known Entity", "name": "Known Entity"}],
+            "relationships": [
+                {
+                    "source": "Known Entity",
+                    "target": "Known Entity",
+                    "type": "related_to",
+                }
+            ],
+        },
+        extract=False,
+    )
+
+    assert "Known Entity" in {e["id"] for e in graph["entities"]}
+    assert not [i for i in _issues(graph) if i.get("code") == "DANGLING_EDGE"]
+
+
+def test_reject_policy_via_nested_config():
+    # The orchestrator constructs GraphBuilder(config=self.config.get("kg", {})),
+    # which lands as a nested "config" keyword. The policy lookup must read
+    # through that nesting so the reject option works end to end.
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        config={"unknown_relation_endpoint": "reject"},
+    )
+    graph = builder.build(
+        [_rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    assert graph["relationships"] == []
+    assert "Synthetic Entity" not in {e["id"] for e in graph["entities"]}
+
+
+def test_real_entity_wins_over_prior_synthetic():
+    # A synthetic endpoint promoted early must yield to a real entity with the
+    # same id that arrives later (e.g. relation data precedes NER entity data).
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_rel(_synthetic("Shared"), "related_to", _known("Other"))],
+        extract=False,
+    )
+    # Force a real "Shared" entity to coexist with the promoted synthetic one and
+    # confirm only the real one survives.
+    real_graph = GraphBuilder(resolve_conflicts=False).build(
+        [
+            _synthetic("Shared"),
+            _known("Shared"),
+            _rel(_synthetic("Shared"), "related_to", _known("Other")),
+        ],
+        extract=False,
+    )
+
+    shared = [e for e in real_graph["entities"] if e["id"] == "Shared"]
+    assert len(shared) == 1
+    assert shared[0]["metadata"].get("synthetic") is not True
+
+
+def test_triplet_with_synthetic_endpoint_promoted():
+    # The LLM triplet path tags endpoint texts it could not match, and the
+    # GraphBuilder promotes them as synthetic entities (issue #1463).
+    triplet = Triplet(
+        subject="Missing Subj",
+        predicate="related_to",
+        object="Known Target",
+    )
+    triplet.metadata = {"synthetic_endpoints": ["Missing Subj"]}
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_known("Known Target"), triplet],
+        extract=False,
+    )
+
+    entity_ids = {e["id"] for e in graph["entities"]}
+    assert "Missing Subj" in entity_ids
+    assert not [i for i in _issues(graph) if i.get("code") == "DANGLING_EDGE"]
+
+
+def test_synthetic_endpoint_deduped_against_entity_id():
+    # A dict entity may carry its canonical id under "entity_id". A synthetic
+    # endpoint promoted with the same id must not create a duplicate node.
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [
+            {"entity_id": "Shared", "name": "Shared"},
+            _rel(_synthetic("Shared"), "related_to", _known("Other")),
+        ],
+        extract=False,
+    )
+
+    ids = [
+        e.get("id")
+        for e in graph["entities"]
+        if e.get("id") == "Shared" or e.get("entity_id") == "Shared"
+    ]
+    assert len(ids) == 1
+
+
+def test_dict_relationship_unknown_endpoint_untouched_by_design():
+    # Dictionary-form relationships stay out of the synthetic reconcile path by
+    # design; they have no synthetic marker so no entity is invented on their
+    # behalf.
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        {
+            "entities": [{"id": "Known"}],
+            "relationships": [
+                {"source": "Known", "target": "Absent", "type": "related_to"}
+            ],
+        },
+        extract=False,
+    )
+
+    ids = {e["id"] for e in graph["entities"]}
+    assert "Known" in ids
+    assert "Absent" not in ids
+
+
+def test_entity_with_none_metadata_does_not_crash_build():
+    # An entity carrying an explicit metadata=None must not raise in the
+    # synthetic-reconcile pass; .get("metadata", {}) returns None, not {}.
+    src = {
+        "entities": [
+            {"id": "a", "name": "a", "metadata": None},
+            {"id": "b", "name": "b"},
+        ],
+        "relationships": [{"source": "a", "target": "b", "type": "rel"}],
+    }
+    graph = GraphBuilder(resolve_conflicts=False).build(src, extract=False)
+
+    ids = {e["id"] for e in graph["entities"]}
+    assert {"a", "b"} <= ids
+    assert graph["relationships"][0]["type"] == "rel"
+
+
+def test_reject_policy_via_orchestrator_shaped_config_fold():
+    # The orchestrator builds with GraphBuilder(config=self.config.get("kg", {})),
+    # so config= carries flat kg options. Folding that dict into the option
+    # mapping must make reject effective and leave the nested
+    # entity_resolution/conflict_detection options readable too.
+    builder = GraphBuilder(
+        config={
+            "unknown_relation_endpoint": "reject",
+            "entity_resolution": {"enabled": False},
+            "conflict_detection": {"enabled": False},
+        },
+    )
+    graph = builder.build(
+        [_rel(_known(), "related_to", _synthetic("Synthetic Entity"))],
+        extract=False,
+    )
+
+    assert graph["relationships"] == []
+    assert "Synthetic Entity" not in {e["id"] for e in graph["entities"]}
+
+
+def test_real_entity_with_entity_id_only_wins_over_prior_synthetic():
+    # A real entity that carries its canonical id under ``entity_id`` (no ``id``
+    # field) must win the real-entity-wins dedup pass in build().
+    # The dedup collects real ids via both ``id`` and ``entity_id``, so a
+    # synthetic entity promoted with id='Shared' must be dropped when a real
+    # dict entity with entity_id='Shared' (no 'id') arrives.
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [
+            _rel(_synthetic("Shared"), "related_to", _known("Other")),
+            _known("Other"),
+            # real entity uses entity_id only — no 'id' key
+            {"entity_id": "Shared", "name": "Shared Real"},
+        ],
+        extract=False,
+    )
+
+    shared = [
+        e for e in graph["entities"]
+        if e.get("id") == "Shared" or e.get("entity_id") == "Shared"
+    ]
+    assert len(shared) == 1
+    # The real dict entity (entity_id-only) must survive, not the synthetic one.
+    assert shared[0].get("entity_id") == "Shared"
+    assert not (shared[0].get("metadata") or {}).get("synthetic")
+
+
+def test_llm_relation_synthetic_endpoints_promoted():
+    # _parse_relation_result marks unmatched relation endpoints as synthetic
+    # Entity objects (label=UNKNOWN, metadata={'synthetic': True}).
+    # GraphBuilder must promote those entities and produce a clean graph.
+    from semantica.semantic_extract.methods import _parse_relation_result
+    from semantica.semantic_extract import Entity as SEEntity
+
+    known_ent = SEEntity(text="Apple", label="ORG", start_char=0, end_char=5)
+    parsed = {
+        "relations": [
+            {
+                "subject": "Apple",
+                "predicate": "founded_by",
+                "object": "Steve Jobs",
+                "confidence": 0.95,
+            }
+        ]
+    }
+    relations = _parse_relation_result(
+        parsed,
+        [known_ent],
+        "Apple was founded by Steve Jobs.",
+        "openai",
+        "gpt-4",
+    )
+    assert len(relations) == 1
+    assert relations[0].object.metadata.get("synthetic") is True
+
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [known_ent, relations[0]],
+        extract=False,
+    )
+
+    entity_ids = {e["id"] for e in graph["entities"]}
+    assert "Steve Jobs" in entity_ids
+    assert not [i for i in _issues(graph) if i.get("code") == "DANGLING_EDGE"]
+
+
+def test_reject_policy_emits_warning_when_all_rels_dropped(caplog):
+    # When reject drops every relationship from a dict-style source, the
+    # build() 'all relationships dropped' warning must still fire (the
+    # input_relationships_count accounts for the dict source's list length
+    # before processing, so it sees the original count > 0).
+    import logging
+
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        unknown_relation_endpoint="reject",
+    )
+    rel = _rel(_known(), "related_to", _synthetic("S"))
+
+    with caplog.at_level(logging.WARNING, logger="graph_builder"):
+        graph = builder.build(
+            {
+                "entities": [{"id": "Known Entity", "name": "Known Entity"}],
+                "relationships": [rel],
+            },
+            extract=False,
+        )
+
+    assert graph["relationships"] == []
+    assert any(
+        "All relationships were dropped" in record.message
+        for record in caplog.records
+    )
+
+
+# ---------------------------------------------------------------------------
+# Observability of the reject policy (Qodo finding: "reject silently drops")
+# ---------------------------------------------------------------------------
+
+def test_reject_policy_count_in_metadata():
+    # graph["metadata"]["rejected_relationships"] must equal the number of
+    # relationships dropped by the reject policy so callers can observe the
+    # outcome without parsing logs.
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        unknown_relation_endpoint="reject",
+    )
+    graph = builder.build(
+        [
+            _known("A"),
+            _known("B"),
+            _rel(_known("A"), "clean", _known("B")),          # kept
+            _rel(_known("A"), "dirty1", _synthetic("S1")),    # rejected
+            _rel(_synthetic("S2"), "dirty2", _known("B")),    # rejected
+        ],
+        extract=False,
+    )
+
+    assert graph["metadata"]["rejected_relationships"] == 2
+    assert len(graph["relationships"]) == 1
+    assert graph["relationships"][0]["type"] == "clean"
+
+
+def test_include_policy_rejected_count_is_zero():
+    # With the default include policy the rejected count must be 0.
+    graph = GraphBuilder(resolve_conflicts=False).build(
+        [_known(), _rel(_known(), "r", _synthetic("S"))],
+        extract=False,
+    )
+    assert graph["metadata"]["rejected_relationships"] == 0
+
+
+def test_reject_policy_emits_warning_log_per_dropped_edge(caplog):
+    # Each dropped edge must produce a WARNING-level log entry (not just INFO).
+    # This makes the rejection visible to callers who configure standard
+    # WARNING-level logging without enabling DEBUG.
+    import logging
+
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        unknown_relation_endpoint="reject",
+    )
+
+    with caplog.at_level(logging.WARNING, logger="semantica.graph_builder"):
+        graph = builder.build(
+            [
+                _known(),
+                _rel(_known(), "dirty1", _synthetic("S1")),
+                _rel(_known(), "dirty2", _synthetic("S2")),
+            ],
+            extract=False,
+        )
+
+    assert graph["relationships"] == []
+    drop_records = [
+        r for r in caplog.records
+        if "Dropping relationship" in r.message and r.levelno == logging.WARNING
+    ]
+    assert len(drop_records) == 2, (
+        f"Expected 2 WARNING drop records, got {len(drop_records)}: {[r.message for r in caplog.records]}"
+    )
+
+
+def test_rejected_count_resets_between_build_calls():
+    # _rejected_relationships must be reset at the start of each build() call
+    # so successive builds on the same instance report independent counts.
+    builder = GraphBuilder(
+        resolve_conflicts=False,
+        unknown_relation_endpoint="reject",
+    )
+
+    g1 = builder.build(
+        [_known(), _rel(_known(), "r", _synthetic("S"))],
+        extract=False,
+    )
+    assert g1["metadata"]["rejected_relationships"] == 1
+
+    # Second build has no synthetic endpoints — count must be 0, not accumulated.
+    g2 = builder.build(
+        [_known("A"), _known("B"), _rel(_known("A"), "clean", _known("B"))],
+        extract=False,
+    )
+    assert g2["metadata"]["rejected_relationships"] == 0

tests/semantic_extract/test_gemini_model_override.py

@@ -0,0 +1,187 @@
+"""Regression tests for the per-call ``model=`` override on ``GeminiProvider``.
+
+``GeminiProvider.generate(model=...)`` and ``generate_structured(model=...)`` must
+select the model the same way as each other, on both supported Gemini SDK paths:
+
+* the new ``google.genai`` client (``_use_new_genai = True``), and
+* the legacy ``google.generativeai`` package (``_use_new_genai = False``), where a
+  per-call model has to be resolved through ``_legacy_client_for()`` because the
+  legacy ``GenerativeModel`` binds its model name at construction time.
+"""
+
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from semantica.semantic_extract.providers import GeminiProvider
+
+CONSTRUCTION_MODEL = "gemini-pro"
+OVERRIDE_MODEL = "gemini-1.5-flash"
+JSON_TEXT = '{"answer": 42}'
+
+
+def _make_provider():
+    """A GeminiProvider with the real SDK bootstrap skipped."""
+    with patch.object(GeminiProvider, "_init_client", return_value=None):
+        return GeminiProvider(api_key="fake-key", model=CONSTRUCTION_MODEL)
+
+
+class TestNewGenaiModelOverride:
+    """New ``google.genai`` client path (``_use_new_genai = True``)."""
+
+    @staticmethod
+    def _provider():
+        provider = _make_provider()
+        provider._use_new_genai = True
+        client = MagicMock()
+        response = MagicMock()
+        response.text = JSON_TEXT
+        client.models.generate_content.return_value = response
+        provider.client = client
+        return provider, client
+
+    def test_generate_honors_per_call_model(self):
+        provider, client = self._provider()
+
+        provider.generate("hello", model=OVERRIDE_MODEL)
+
+        assert (
+            client.models.generate_content.call_args.kwargs["model"] == OVERRIDE_MODEL
+        )
+
+    def test_generate_structured_honors_per_call_model(self):
+        provider, client = self._provider()
+
+        provider.generate_structured("hello", model=OVERRIDE_MODEL)
+
+        assert (
+            client.models.generate_content.call_args.kwargs["model"] == OVERRIDE_MODEL
+        )
+
+    def test_generate_falls_back_to_construction_model(self):
+        provider, client = self._provider()
+
+        provider.generate("hello")
+
+        assert (
+            client.models.generate_content.call_args.kwargs["model"]
+            == CONSTRUCTION_MODEL
+        )
+
+    def test_generate_structured_falls_back_to_construction_model(self):
+        provider, client = self._provider()
+
+        provider.generate_structured("hello")
+
+        assert (
+            client.models.generate_content.call_args.kwargs["model"]
+            == CONSTRUCTION_MODEL
+        )
+
+    def test_both_methods_select_model_consistently(self):
+        provider, client = self._provider()
+
+        provider.generate("hello", model=OVERRIDE_MODEL)
+        generate_model = client.models.generate_content.call_args.kwargs["model"]
+
+        client.models.generate_content.reset_mock()
+        provider.generate_structured("hello", model=OVERRIDE_MODEL)
+        structured_model = client.models.generate_content.call_args.kwargs["model"]
+
+        assert generate_model == structured_model == OVERRIDE_MODEL
+
+
+class TestLegacyModelOverride:
+    """Legacy ``google.generativeai`` path (``_use_new_genai = False``)."""
+
+    @pytest.fixture
+    def fake_legacy_genai(self, monkeypatch):
+        """Install a stand-in ``google.generativeai`` module.
+
+        ``GenerativeModel(name)`` returns a distinct mock per model name (cached,
+        mirroring ``_legacy_client_for``) so tests can assert which model name the
+        per-call override resolved to.
+        """
+        module = MagicMock()
+        module.created_models = {}
+
+        def make_model(name):
+            model = module.created_models.get(name)
+            if model is None:
+                model = MagicMock(name=f"GenerativeModel({name})")
+                response = MagicMock()
+                response.text = JSON_TEXT
+                model.generate_content.return_value = response
+                module.created_models[name] = model
+            return model
+
+        module.GenerativeModel.side_effect = make_model
+        monkeypatch.setitem(sys.modules, "google.generativeai", module)
+        return module
+
+    @staticmethod
+    def _provider():
+        provider = _make_provider()
+        provider._use_new_genai = False
+        construction_client = MagicMock(name="construction client")
+        response = MagicMock()
+        response.text = JSON_TEXT
+        construction_client.generate_content.return_value = response
+        provider.client = construction_client
+        return provider, construction_client
+
+    def test_generate_builds_client_for_per_call_model(self, fake_legacy_genai):
+        provider, construction_client = self._provider()
+
+        provider.generate("hello", model=OVERRIDE_MODEL)
+
+        fake_legacy_genai.GenerativeModel.assert_called_once_with(OVERRIDE_MODEL)
+        construction_client.generate_content.assert_not_called()
+
+    def test_generate_structured_builds_client_for_per_call_model(
+        self, fake_legacy_genai
+    ):
+        provider, construction_client = self._provider()
+
+        provider.generate_structured("hello", model=OVERRIDE_MODEL)
+
+        fake_legacy_genai.GenerativeModel.assert_called_once_with(OVERRIDE_MODEL)
+        construction_client.generate_content.assert_not_called()
+
+    def test_without_override_uses_construction_client(self, fake_legacy_genai):
+        provider, construction_client = self._provider()
+
+        provider.generate("hello")
+        provider.generate_structured("hello")
+
+        fake_legacy_genai.GenerativeModel.assert_not_called()
+        assert construction_client.generate_content.call_count == 2
+
+    def test_both_methods_select_model_consistently(self, fake_legacy_genai):
+        provider, construction_client = self._provider()
+
+        provider.generate("hello", model=OVERRIDE_MODEL)
+        provider.generate_structured("hello", model=OVERRIDE_MODEL)
+
+        assert set(fake_legacy_genai.created_models) == {OVERRIDE_MODEL}
+        assert (
+            fake_legacy_genai.created_models[OVERRIDE_MODEL].generate_content.call_count
+            == 2
+        )
+        construction_client.generate_content.assert_not_called()
+
+    def test_generate_structured_still_forwards_generation_config(
+        self, fake_legacy_genai
+    ):
+        """generate_structured() must forward generation params on the legacy path,
+        the same as generate() does."""
+        provider, _ = self._provider()
+
+        provider.generate_structured("hello", model=OVERRIDE_MODEL, temperature=0.2)
+
+        override_client = fake_legacy_genai.created_models[OVERRIDE_MODEL]
+        gen_config = override_client.generate_content.call_args.kwargs[
+            "generation_config"
+        ]
+        assert gen_config["temperature"] == 0.2

tests/semantic_extract/test_ner_merge_strategies.py

@@ -0,0 +1,763 @@
+"""Regression coverage for NER method merge strategies."""
+
+from unittest.mock import patch
+
+import pytest
+
+from semantica.semantic_extract.ner_extractor import NERExtractor
+from semantica.semantic_extract.types import Entity
+
+
+def entity(text, label, start, end, confidence=0.9):
+    """Create a compact entity fixture with realistic offsets."""
+    return Entity(text, label, start, end, confidence=confidence)
+
+
+def test_consensus_rejects_single_method_candidate_with_empty_peer():
+    """An empty eligible method must remain in the consensus denominator."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [("first", [entity("Apple", "ORG", 0, 5)]), ("second", [])],
+        eligible_methods=["first", "second"],
+    )
+
+    assert result == []
+
+
+def test_consensus_aligns_compatible_labels_and_exposes_provenance():
+    """Compatible labels vote together and preserve auditable method evidence."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("Apple", "ORGANIZATION", 0, 5, 0.8)]),
+            ("second", [entity("Apple", "ORG", 0, 5, 0.9)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert len(result) == 1
+    merged = result[0]
+    assert merged.label == "ORG"
+    assert merged.confidence == pytest.approx(0.85)
+    assert merged.metadata["supporting_methods"] == ["first", "second"]
+    assert merged.metadata["vote_count"] == 2
+    assert merged.metadata["eligible_method_count"] == 2
+    assert merged.metadata["agreement"] == 1.0
+    assert merged.metadata["method_scores"] == {"first": 0.8, "second": 0.9}
+
+
+def test_consensus_keeps_repeated_mentions_at_distinct_offsets():
+    """Matching text must not collapse separate document mentions."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+    first_mentions = [
+        entity("Apple", "ORG", 0, 5),
+        entity("Apple", "ORG", 10, 15),
+    ]
+    second_mentions = [
+        entity("Apple", "ORG", 0, 5),
+        entity("Apple", "ORG", 10, 15),
+    ]
+
+    result = extractor._vote_entities(
+        [("first", first_mentions), ("second", second_mentions)],
+        eligible_methods=["first", "second"],
+    )
+
+    assert [(item.text, item.start_char, item.end_char) for item in result] == [
+        ("Apple", 0, 5),
+        ("Apple", 10, 15),
+    ]
+
+
+def test_consensus_resolves_boundary_variants_deterministically():
+    """Equal-confidence overlapping spans prefer the most specific boundary."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("Apple", "ORG", 0, 5, 0.9)]),
+            ("second", [entity("Apple Inc.", "ORG", 0, 10, 0.9)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert len(result) == 1
+    assert (result[0].text, result[0].start_char, result[0].end_char) == (
+        "Apple Inc.",
+        0,
+        10,
+    )
+
+
+def test_consensus_resolves_conflicting_labels_deterministically():
+    """With equal vote counts, the higher-confidence label wins predictably."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=1
+    )
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("Apple", "ORG", 0, 5, 0.7)]),
+            ("second", [entity("Apple", "PRODUCT", 0, 5, 0.9)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert len(result) == 1
+    assert result[0].label == "PRODUCT"
+    assert result[0].metadata["supporting_methods"] == ["second"]
+
+
+def test_consensus_prefers_a_same_label_match_over_a_tied_conflict():
+    """A conflicting duplicate from one method cannot hide true agreement."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [
+            (
+                "first",
+                [
+                    entity("Apple", "PRODUCT", 0, 5),
+                    entity("Apple", "ORG", 0, 5),
+                ],
+            ),
+            ("second", [entity("Apple", "ORG", 0, 5)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert len(result) == 1
+    assert result[0].label == "ORG"
+    assert result[0].metadata["supporting_methods"] == ["first", "second"]
+
+
+def test_default_consensus_rejects_a_label_conflict_without_two_votes():
+    """Different labels do not turn two methods into two votes for either label."""
+    extractor = NERExtractor(method=["first", "second"], merge_strategy="consensus")
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("Apple", "ORG", 0, 5)]),
+            ("second", [entity("Apple", "PRODUCT", 0, 5)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert result == []
+
+
+def test_consensus_keeps_a_majority_label_despite_a_high_weight_single_vote():
+    """Weights break ties only after the configured vote requirements are met."""
+    extractor = NERExtractor(
+        method=["first", "second", "third"],
+        merge_strategy="consensus",
+        min_votes=2,
+        method_weights={"third": 100.0},
+    )
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("Apple", "ORG", 0, 5, 0.7)]),
+            ("second", [entity("Apple", "ORG", 0, 5, 0.7)]),
+            ("third", [entity("Apple", "PRODUCT", 0, 5, 0.99)]),
+        ],
+        eligible_methods=["first", "second", "third"],
+    )
+
+    assert len(result) == 1
+    assert result[0].label == "ORG"
+    assert result[0].metadata["supporting_methods"] == ["first", "second"]
+
+
+def test_method_weights_treat_ml_and_spacy_as_one_backend():
+    """A spaCy alias weight must apply when the configured method is called ml."""
+    extractor = NERExtractor(method="pattern", merge_strategy="consensus", min_votes=1)
+
+    result = extractor._vote_entities(
+        [
+            ("ml", [entity("Apple", "ORG", 0, 5, 0.7)]),
+            ("regex", [entity("Apple", "PRODUCT", 0, 5, 0.9)]),
+        ],
+        eligible_methods=["ml", "regex"],
+        min_votes=1,
+        method_weights={"spacy": 100.0, "regex": 1.0},
+    )
+
+    assert len(result) == 1
+    assert result[0].label == "ORG"
+
+
+def test_method_weights_reject_conflicting_ml_and_spacy_aliases():
+    """A single backend cannot receive two different alias weights."""
+    with pytest.raises(ValueError, match="conflicting values to aliases"):
+        NERExtractor(
+            method="pattern",
+            merge_strategy="consensus",
+            method_weights={"ml": 1.0, "spacy": 2.0},
+        )
+
+
+def test_consensus_does_not_merge_distant_mentions_through_a_broad_span():
+    """A broad span below the IoU threshold cannot fabricate cross-method support."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [
+            (
+                "first",
+                [
+                    entity("A B", "ORG", 0, 3),
+                    entity("C D", "ORG", 4, 7),
+                ],
+            ),
+            ("second", [entity("A B C D", "ORG", 0, 7)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert result == []
+
+
+def test_consensus_requires_each_vote_to_overlap_every_other_vote():
+    """A chain of pairwise overlaps must not fabricate three-way support."""
+    extractor = NERExtractor(
+        method=["first", "second", "third"],
+        merge_strategy="consensus",
+        min_votes=3,
+    )
+
+    result = extractor._vote_entities(
+        [
+            ("first", [entity("ABCD", "ORG", 0, 4)]),
+            ("second", [entity("ABCDEF", "ORG", 0, 6)]),
+            ("third", [entity("CDEF", "ORG", 2, 6)]),
+        ],
+        eligible_methods=["first", "second", "third"],
+    )
+
+    assert result == []
+
+
+def test_consensus_prefers_an_exact_boundary_match_within_one_method_batch():
+    """A weaker overlap cannot consume another method's exact boundary vote."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+
+    result = extractor._vote_entities(
+        [
+            (
+                "first",
+                [
+                    entity("Apple", "ORG", 0, 5, 0.99),
+                    entity("Apple Inc.", "ORG", 0, 10, 0.1),
+                ],
+            ),
+            ("second", [entity("Apple Inc.", "ORG", 0, 10, 0.9)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert len(result) == 1
+    assert (result[0].start_char, result[0].end_char) == (0, 10)
+    assert result[0].metadata["supporting_methods"] == ["first", "second"]
+
+
+def test_consensus_boundary_result_is_independent_of_method_result_order():
+    """Method result order cannot affect the selected consensus boundary."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+    first = [
+        entity("Apple", "ORG", 0, 5, 0.99),
+        entity("Apple Inc.", "ORG", 0, 10, 0.1),
+    ]
+    second = [entity("Apple Inc.", "ORG", 0, 10, 0.9)]
+
+    forward = extractor._vote_entities(
+        [("first", first), ("second", second)],
+        eligible_methods=["first", "second"],
+    )
+    reverse = extractor._vote_entities(
+        [("second", second), ("first", list(reversed(first)))],
+        eligible_methods=["first", "second"],
+    )
+
+    assert [
+        (item.label, item.start_char, item.end_char, item.metadata) for item in reverse
+    ] == [
+        (item.label, item.start_char, item.end_char, item.metadata) for item in forward
+    ]
+
+
+def test_consensus_keeps_nested_entities_with_different_labels():
+    """Exact-span label arbitration must not erase a nested entity."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+    mentions = [
+        entity("New York", "GPE", 0, 8),
+        entity("New York Times", "ORG", 0, 14),
+    ]
+
+    result = extractor._vote_entities(
+        [("first", mentions), ("second", mentions)],
+        eligible_methods=["first", "second"],
+    )
+
+    assert [(item.label, item.start_char, item.end_char) for item in result] == [
+        ("GPE", 0, 8),
+        ("ORG", 0, 14),
+    ]
+
+
+def test_union_keeps_complementary_single_method_entities():
+    """Explicit union preserves the legacy complementary-method behavior."""
+    extractor = NERExtractor(method=["first", "second"], merge_strategy="union")
+
+    result = extractor._union_entities(
+        [("first", [entity("SKU-12345", "PRODUCT_CODE", 10, 19)]), ("second", [])],
+        eligible_methods=["first", "second"],
+    )
+
+    assert [(item.text, item.label) for item in result] == [
+        ("SKU-12345", "PRODUCT_CODE")
+    ]
+    assert result[0].metadata["vote_count"] == 1
+    assert result[0].metadata["agreement"] == 0.5
+
+
+def test_union_preserves_distinct_labels_and_original_label_spelling():
+    """Union is complementary: it does not force label-conflict resolution."""
+    extractor = NERExtractor(method=["first", "second"], merge_strategy="union")
+
+    result = extractor._union_entities(
+        [
+            ("first", [entity("Apple", "ORG", 0, 5)]),
+            ("second", [entity("Apple", "PRODUCT", 0, 5)]),
+        ],
+        eligible_methods=["first", "second"],
+    )
+
+    assert [(item.text, item.label) for item in result] == [
+        ("Apple", "ORG"),
+        ("Apple", "PRODUCT"),
+    ]
+
+
+def test_ml_and_spacy_aliases_do_not_duplicate_an_entity_or_a_vote():
+    """Two names for spaCy represent one backend, not independent voters."""
+    extractor = NERExtractor(method=["ml", "spacy"], merge_strategy="union")
+
+    result = extractor._union_entities(
+        [
+            ("ml", [entity("Apple", "ORG", 0, 5)]),
+            ("spacy", [entity("Apple", "ORG", 0, 5)]),
+        ],
+        eligible_methods=["ml", "spacy"],
+    )
+
+    assert len(result) == 1
+    assert result[0].metadata["supporting_methods"] == ["ml"]
+    assert result[0].metadata["eligible_method_count"] == 1
+
+
+def test_extract_consensus_counts_a_successful_empty_method():
+    """The public extraction path must retain empty method results for voting."""
+    extractor = NERExtractor(
+        method=["pattern", "regex"], merge_strategy="consensus", min_votes=2
+    )
+    responses = {"pattern": [entity("Apple", "ORG", 0, 5)], "regex": []}
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: (
+            lambda _text, **_options: responses[method_name]
+        ),
+    ):
+        assert extractor.extract_entities("Apple") == []
+
+
+def test_extract_consensus_keeps_method_provenance_on_a_successful_vote():
+    """Public extraction must retain method names for provenance."""
+    extractor = NERExtractor(
+        method=["pattern", "regex"], merge_strategy="consensus", min_votes=2
+    )
+    responses = {
+        "pattern": [entity("Apple", "ORG", 0, 5, 0.8)],
+        "regex": [entity("Apple", "ORGANIZATION", 0, 5, 0.9)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert len(result) == 1
+    assert result[0].metadata["supporting_methods"] == ["pattern", "regex"]
+    assert result[0].metadata["method_scores"] == {"pattern": 0.8, "regex": 0.9}
+
+
+def test_merge_options_are_not_forwarded_to_custom_methods():
+    """Custom methods without **kwargs remain usable with the new merge API."""
+    extractor = NERExtractor(
+        method=["first", "second"], merge_strategy="consensus", min_votes=2
+    )
+    responses = {
+        "first": [entity("Apple", "ORG", 0, 5)],
+        "second": [entity("Apple", "ORG", 0, 5)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text: responses[method_name],
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert len(result) == 1
+
+
+def test_legacy_options_remain_available_to_custom_methods():
+    """Existing custom methods retain the legacy options they previously received."""
+    with pytest.warns(DeprecationWarning, match="ensemble_voting"):
+        extractor = NERExtractor(
+            method="custom", ensemble_voting=True, post_process=True
+        )
+    received = {}
+
+    def custom_method(_text, *, ensemble_voting, post_process):
+        received.update(
+            ensemble_voting=ensemble_voting,
+            post_process=post_process,
+        )
+        return [entity("Apple", "ORG", 0, 5)]
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        return_value=custom_method,
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert len(result) == 1
+    assert received == {"ensemble_voting": True, "post_process": True}
+
+
+def test_extract_consensus_recovers_missing_llm_offsets_before_voting():
+    """Typed LLM results with schema-default 0:0 spans can still vote safely."""
+    extractor = NERExtractor(
+        method=["llm", "regex"], merge_strategy="consensus", min_votes=2
+    )
+    responses = {
+        "llm": [entity("Apple", "ORG", 0, 0, 0.9)],
+        "regex": [entity("Apple", "ORG", 0, 5, 0.8)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert len(result) == 1
+    assert (result[0].start_char, result[0].end_char) == (0, 5)
+    assert result[0].metadata["supporting_methods"] == ["llm", "regex"]
+
+
+@pytest.mark.parametrize("start, end", [(0, 0), (None, None)])
+def test_union_discards_unresolved_invalid_offsets(start, end):
+    """Unresolvable default or nullable spans cannot leak into union output."""
+    extractor = NERExtractor(method="llm", merge_strategy="union")
+    responses = {"llm": [entity("Absent", "ORG", start, end, 0.9)]}
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        assert extractor.extract_entities("Apple") == []
+
+
+def test_union_recovers_nullable_offsets_when_text_matches():
+    """Nullable custom offsets remain usable when they can be aligned safely."""
+    extractor = NERExtractor(method="custom", merge_strategy="union")
+    responses = {"custom": [entity("Apple", "ORG", None, None, 0.9)]}
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert [(item.start_char, item.end_char) for item in result] == [(0, 5)]
+
+
+def test_missing_offsets_are_assigned_to_distinct_repeated_mentions():
+    """Text-only duplicate output is aligned in occurrence order before merging."""
+    extractor = NERExtractor(method="llm", merge_strategy="union")
+
+    aligned = extractor._align_entities_to_text(
+        [
+            entity("Apple", "ORG", 0, 0),
+            entity("Apple", "ORG", 0, 0),
+        ],
+        "Apple and Apple",
+    )
+
+    assert [(item.start_char, item.end_char) for item in aligned] == [(0, 5), (10, 15)]
+
+
+def test_missing_offsets_wrap_to_an_unoccupied_repeated_mention():
+    """A valid later mention must not make an earlier missing one unalignable."""
+    extractor = NERExtractor(method="llm", merge_strategy="union")
+
+    aligned = extractor._align_entities_to_text(
+        [
+            entity("Apple", "ORG", 10, 15),
+            entity("Apple", "ORG", 0, 0),
+        ],
+        "Apple and Apple",
+    )
+
+    assert [(item.start_char, item.end_char) for item in aligned] == [(10, 15), (0, 5)]
+
+
+def test_missing_offsets_do_not_align_a_substring_inside_a_larger_word():
+    """Offset recovery must not let Apple vote for the substring in Pineapple."""
+    extractor = NERExtractor(
+        method=["llm", "regex"], merge_strategy="consensus", min_votes=2
+    )
+    responses = {
+        "llm": [entity("Apple", "ORG", 0, 0, 0.9)],
+        "regex": [entity("Pineapple", "ORG", 0, 9, 0.8)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        assert extractor.extract_entities("Pineapple") == []
+
+
+def test_missing_offsets_preserve_original_unicode_character_positions():
+    """Case-insensitive matching preserves original Unicode offsets."""
+    extractor = NERExtractor(method="llm", merge_strategy="union")
+
+    aligned = extractor._align_entities_to_text(
+        [entity("Apple", "ORG", 0, 0)],
+        "İ Apple",
+    )
+
+    assert [(item.start_char, item.end_char) for item in aligned] == [(2, 7)]
+
+
+def test_consensus_eligible_methods_can_exclude_complementary_extractors():
+    """An explicit eligible subset controls the denominator used for agreement."""
+    extractor = NERExtractor(
+        method=["first", "second", "supplement"],
+        merge_strategy="consensus",
+        min_votes=2,
+        min_agreement=0.75,
+        eligible_methods=["first", "second"],
+    )
+    responses = {
+        "first": [entity("Apple", "ORG", 0, 5)],
+        "second": [entity("Apple", "ORG", 0, 5)],
+        "supplement": [],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert len(result) == 1
+    assert result[0].metadata["eligible_method_count"] == 2
+    assert result[0].metadata["agreement"] == 1.0
+
+
+def test_extract_consensus_does_not_inject_fallback_candidates():
+    """An all-empty consensus result must remain empty rather than fall back."""
+    extractor = NERExtractor(
+        method=["pattern", "regex"], merge_strategy="consensus", min_votes=2
+    )
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda _method_name: lambda _text, **_options: [],
+    ):
+        assert extractor.extract_entities("Apple") == []
+
+
+def test_consensus_counts_a_failed_eligible_method_in_agreement():
+    """A failed configured method is non-supporting rather than silently removed."""
+    extractor = NERExtractor(
+        method=["pattern", "regex"],
+        merge_strategy="consensus",
+        min_votes=1,
+        min_agreement=0.75,
+    )
+
+    def method_for(method_name):
+        if method_name == "pattern":
+            return lambda _text, **_options: [entity("Apple", "ORG", 0, 5)]
+        return lambda _text, **_options: (_ for _ in ()).throw(RuntimeError("offline"))
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=method_for,
+    ):
+        assert extractor.extract_entities("Apple") == []
+
+
+def test_union_ignores_consensus_eligible_method_subset():
+    """Complementary union must retain all methods even when consensus is scoped."""
+    extractor = NERExtractor(
+        method=["pattern", "regex"],
+        merge_strategy="union",
+        eligible_methods=["pattern"],
+    )
+    responses = {
+        "pattern": [entity("Apple", "ORG", 0, 5)],
+        "regex": [entity("SKU-12345", "PRODUCT_CODE", 10, 19)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple SKU-12345")
+
+    assert [(item.text, item.label) for item in result] == [
+        ("Apple", "ORG"),
+        ("SKU-12345", "PRODUCT_CODE"),
+    ]
+
+
+def test_issue_1283_consensus_requires_cross_method_agreement():
+    """The former ensemble union must not be mistaken for consensus."""
+    responses = {
+        "first": [entity("Apple", "ORG", 0, 5)],
+        "second": [entity("SKU-12345", "PRODUCT_CODE", 6, 15)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        with pytest.warns(DeprecationWarning, match="ensemble_voting"):
+            legacy = NERExtractor(method=["first", "second"], ensemble_voting=True)
+        consensus = NERExtractor(
+            method=["first", "second"],
+            merge_strategy="consensus",
+            min_votes=2,
+        )
+
+        legacy_entities = legacy.extract_entities("Apple SKU-12345")
+        consensus_entities = consensus.extract_entities("Apple SKU-12345")
+
+    assert [(item.text, item.label) for item in legacy_entities] == [
+        ("Apple", "ORG"),
+        ("SKU-12345", "PRODUCT_CODE"),
+    ]
+    assert consensus_entities == []
+
+
+def test_fallback_remains_first_nonempty_method():
+    """The default strategy remains the documented ordered fallback chain."""
+    extractor = NERExtractor(method=["first", "second"], merge_strategy="fallback")
+    responses = {"first": [], "second": [entity("Apple", "ORG", 0, 5)]}
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: (
+            lambda _text, **_options: responses[method_name]
+        ),
+    ):
+        result = extractor.extract_entities("Apple")
+
+    assert [(item.text, item.label) for item in result] == [("Apple", "ORG")]
+
+
+def test_default_strategy_short_circuits_at_the_first_nonempty_method():
+    """No merge strategy keeps the public ordered fallback behavior unchanged."""
+    extractor = NERExtractor(method=["first", "second"])
+    requested_methods = []
+    responses = {
+        "first": [entity("Apple", "ORG", 0, 5)],
+        "second": [entity("SKU-12345", "PRODUCT_CODE", 6, 15)],
+    }
+
+    def method_for(method_name):
+        requested_methods.append(method_name)
+        return lambda _text, **_options: responses[method_name]
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=method_for,
+    ):
+        result = extractor.extract_entities("Apple SKU-12345")
+
+    assert [(item.text, item.label) for item in result] == [("Apple", "ORG")]
+    assert requested_methods == ["first"]
+
+
+def test_legacy_ensemble_flag_maps_to_deprecated_union_strategy():
+    """Existing callers keep their union behavior while receiving migration guidance."""
+    with pytest.warns(DeprecationWarning, match="ensemble_voting"):
+        extractor = NERExtractor(method=["first", "second"], ensemble_voting=True)
+
+    assert extractor.merge_strategy == "union"
+
+
+def test_legacy_ensemble_flag_still_runs_the_union_path():
+    """The deprecated flag retains single-method candidates during migration."""
+    with pytest.warns(DeprecationWarning, match="ensemble_voting"):
+        extractor = NERExtractor(method=["first", "second"], ensemble_voting=True)
+    responses = {
+        "first": [entity("Apple", "ORG", 0, 5)],
+        "second": [entity("SKU-12345", "PRODUCT_CODE", 6, 15)],
+    }
+
+    with patch(
+        "semantica.semantic_extract.methods.get_entity_method",
+        side_effect=lambda method_name: lambda _text, **_options: responses[
+            method_name
+        ],
+    ):
+        result = extractor.extract_entities("Apple SKU-12345")
+
+    assert [(item.text, item.label) for item in result] == [
+        ("Apple", "ORG"),
+        ("SKU-12345", "PRODUCT_CODE"),
+    ]

tests/test_cli_commands.py

@@ -69,6 +69,19 @@ def _json_output(result) -> Any:
     return json.loads(result.output.strip())
 
 
+def _flatten(output: str) -> str:
+    """Undo Rich panel wrapping for substring assertions on error text.
+
+    Rich wraps long messages across multiple bordered lines (each with its
+    own leading/trailing "│"), so a naive whitespace join still leaves those
+    border characters between words that were split across lines. Strip the
+    box-drawing characters first, then collapse whitespace.
+    """
+    for ch in "┌┐└┘│─":
+        output = output.replace(ch, " ")
+    return " ".join(output.split())
+
+
 # ─── Global flags ─────────────────────────────────────────────────────────────
 
 
@@ -336,6 +349,79 @@ class TestIngest:
         assert captured["sources"] == "README.md"
         assert captured["kwargs"]["method"] == "file"
 
+    def test_configured_graph_backend_does_not_report_false_success(
+        self, runner, monkeypatch
+    ):
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+
+        result = runner.invoke(cli_module.main, ["ingest", "README.md"])
+        output = _flatten(result.output)
+
+        assert result.exit_code != 0
+        assert "does not write to graph stores" in output
+        assert "Ingested:" not in output
+
+    def test_configured_graph_backend_error_does_not_recommend_broken_kg_build(
+        self, runner, monkeypatch
+    ):
+        # kg build also does not persist to a configured graph store
+        # (tracked separately as #1352), so the error must not send users to
+        # a command that will silently no-op the same way.
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+
+        result = runner.invoke(cli_module.main, ["ingest", "README.md"])
+        output = _flatten(result.output)
+
+        assert result.exit_code != 0
+        assert "kg build" not in output
+
+    def test_output_flag_writes_real_content_and_bypasses_graph_error(
+        self, runner, monkeypatch, tmp_path
+    ):
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+        out_path = tmp_path / "out.json"
+
+        result = runner.invoke(
+            cli_module.main, ["ingest", "README.md", "--output", str(out_path)]
+        )
+
+        _ok(result, substr="Wrote")
+        written = json.loads(out_path.read_text(encoding="utf-8"))
+        assert written == {"files": [{"path": "README.md"}]}
+
+    def test_store_and_output_are_not_forwarded_to_ingest_backend(
+        self, runner, monkeypatch, tmp_path
+    ):
+        captured = {}
+
+        def fake_ingest_file(sources, **kwargs):
+            captured["kwargs"] = kwargs
+            return [{"path": sources}]
+
+        monkeypatch.setattr("semantica.ingest.methods.ingest_file", fake_ingest_file)
+        out_path = tmp_path / "out.json"
+
+        result = runner.invoke(
+            cli_module.main,
+            ["ingest", "README.md", "--store", "neo4j", "--output", str(out_path)],
+        )
+
+        _ok(result)
+        assert "store" not in captured["kwargs"]
+        assert "output" not in captured["kwargs"]
+
     def test_import_error_is_clean(self, runner, monkeypatch):
         monkeypatch.setattr(cli_module, "__import__", _import_side_effect, raising=False)
         original_import = __import__
@@ -771,6 +857,156 @@ class TestReason:
         assert result.exit_code != 0
         assert "Traceback" not in result.output
 
+    def test_run_infers_from_graph_store_facts(self, runner, monkeypatch, tmp_path):
+        # reason run used to call Reasoner.run(), which does not exist
+        # (#1354); it must feed graph store facts + --rules into
+        # Reasoner.infer_facts().
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+        rules_file = tmp_path / "rules.yaml"
+        rules_file.write_text(
+            '- IF Person(?x) THEN Human(?x)\n'
+            '- IF MANAGES(?x, ?y) THEN Manager(?x)\n'
+            '- IF Employee(?x) THEN Staff(?x)\n',
+            encoding="utf-8")
+
+        class _FakeStore:
+            # Same dict schema as the real backends: nodes carry
+            # labels/properties, relationships carry start_node_id/end_node_id.
+            def get_nodes(self, limit=None):
+                return [{"id": 1, "labels": ["Person"],
+                         "properties": {"name": "Alice"}},
+                        {"id": 2, "labels": ["Person", "Employee"],
+                         "properties": {"name": "Bob"}}]
+
+            def get_relationships(self, limit=None):
+                return [{"id": 9, "type": "MANAGES",
+                         "start_node_id": 1, "end_node_id": 2}]
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _FakeStore())
+        result = runner.invoke(
+            cli_module.main,
+            ["--json", "reason", "run", "--rules", str(rules_file)],
+        )
+        _ok(result)
+        data = json.loads(result.output.strip())
+        # Person(Alice), Person(Bob), Employee(Bob), MANAGES(Alice, Bob)
+        assert data["facts"] == 4
+        assert "Human(Alice)" in data["inferred_facts"]
+        # Relationship endpoints resolve node ids to names.
+        assert "Manager(Alice)" in data["inferred_facts"]
+        # Secondary labels also become facts.
+        assert "Staff(Bob)" in data["inferred_facts"]
+        assert data["inferred_count"] == len(data["inferred_facts"])
+
+    def test_run_rejects_unwired_engine(self, runner):
+        result = runner.invoke(cli_module.main,
+                               ["reason", "run", "--engine", "sparql"])
+        assert result.exit_code != 0
+        assert "not wired" in result.output
+        assert "reason query" in result.output
+        assert "Traceback" not in result.output
+
+    def test_load_rule_definitions_formats(self, tmp_path):
+        yaml_list = tmp_path / "list.yaml"
+        yaml_list.write_text('- IF A(?x) THEN B(?x)\n- IF B(?x) THEN C(?x)\n',
+                             encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(yaml_list)) == [
+            "IF A(?x) THEN B(?x)", "IF B(?x) THEN C(?x)"]
+
+        yaml_map = tmp_path / "map.yaml"
+        yaml_map.write_text('rules:\n  - IF A(?x) THEN B(?x)\n', encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(yaml_map)) == [
+            "IF A(?x) THEN B(?x)"]
+
+        plain = tmp_path / "rules.dl"
+        plain.write_text('# comment\nIF A(?x) THEN B(?x)\n\n', encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(plain)) == [
+            "IF A(?x) THEN B(?x)"]
+
+    def test_run_empty_graph_returns_zero_facts(self, runner, monkeypatch):
+        """reason run with an empty graph store should not crash and report 0 facts."""
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+
+        class _EmptyStore:
+            def get_nodes(self, limit=None): return []
+            def get_relationships(self, limit=None): return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _EmptyStore())
+        result = runner.invoke(cli_module.main, ["--json", "reason", "run"])
+        _ok(result)
+        data = json.loads(result.output.strip())
+        assert data["facts"] == 0
+        assert data["inferred_count"] == 0
+        assert data["inferred_facts"] == []
+
+    def test_run_graph_store_error_surfaces_cleanly(self, runner, monkeypatch):
+        """A graph-store connectivity error must surface as a clean error, not a Traceback."""
+
+        def _bad_store(ctx):
+            raise RuntimeError("connection refused")
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", _bad_store)
+        result = runner.invoke(cli_module.main, ["reason", "run"])
+        assert result.exit_code != 0
+        assert "Traceback" not in result.output
+        assert "connection refused" in result.output
+
+    def test_run_no_rules_uses_empty_ruleset(self, runner, monkeypatch):
+        """reason run without --rules should still succeed (zero rules -> zero inferences)."""
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+
+        class _FakeStore:
+            def get_nodes(self, limit=None):
+                return [{"id": 1, "labels": ["Person"], "properties": {"name": "Alice"}}]
+
+            def get_relationships(self, limit=None):
+                return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _FakeStore())
+        result = runner.invoke(cli_module.main, ["--json", "reason", "run"])
+        _ok(result)
+        data = json.loads(result.output.strip())
+        assert data["facts"] == 1
+        assert data["inferred_count"] == 0
+
+    def test_load_rule_definitions_yaml_mapping_without_rules_key_raises(self, tmp_path):
+        """A YAML mapping with no 'rules' key must raise ClickException, not silently
+        pass the raw YAML lines as rules."""
+        bad = tmp_path / "bad.yaml"
+        bad.write_text("some_key: some_value\nother_key: other_value\n", encoding="utf-8")
+        import click as _click
+        with pytest.raises(_click.ClickException, match="no 'rules' key"):
+            cli_module._load_rule_definitions(str(bad))
+
+    def test_load_rule_definitions_empty_file_returns_empty_list(self, tmp_path):
+        empty = tmp_path / "empty.yaml"
+        empty.write_text("", encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(empty)) == []
+
+    def test_load_rule_definitions_yaml_rules_null_falls_to_plaintext(self, tmp_path):
+        """rules: null is valid YAML with the key present; the null value is
+        not a list, so the function falls through to plain-text parsing and
+        returns the literal line (one no-op rule).  This documents the edge
+        case rather than asserting a specific useful behaviour."""
+        f = tmp_path / "null_rules.yaml"
+        f.write_text("rules: null\n", encoding="utf-8")
+        result = cli_module._load_rule_definitions(str(f))
+        # Plain-text fallback: the non-comment, non-blank line becomes a rule.
+        assert result == ["rules: null"]
+
+    def test_run_rejects_deductive_engine(self, runner, monkeypatch):
+        """Engines other than rete/forward-chain must be rejected with a helpful message."""
+
+        class _EmptyStore:
+            def get_nodes(self, limit=None): return []
+            def get_relationships(self, limit=None): return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _EmptyStore())
+        result = runner.invoke(cli_module.main, ["reason", "run", "--engine", "deductive"])
+        assert result.exit_code != 0
+        assert "not wired" in result.output
+        assert "Traceback" not in result.output
+
     def test_explain_requires_conclusion(self, runner):
         result = runner.invoke(cli_module.main, ["reason", "explain"])
         assert result.exit_code != 0
@@ -1350,6 +1586,65 @@ class TestStore:
         result = runner.invoke(cli_module.main, ["store", "connect", "--backend", "neo4j"])
         _ok(result)
 
+    def test_connect_dispatches_through_graph_store(self, runner, monkeypatch):
+        # store connect used to call get_graph_store_method(backend) — the
+        # method registry, which needs (task, method_name) — so it raised a
+        # TypeError before any connection attempt (#1354).
+        calls = {}
+
+        class _FakeGraphStore:
+            def __init__(self, backend=None, **cfg):
+                calls["backend"] = backend
+                calls["cfg"] = cfg
+
+            def connect(self):
+                calls["connected"] = True
+                return True
+
+        import semantica.graph_store as gs_mod
+        monkeypatch.setattr(gs_mod, "GraphStore", _FakeGraphStore)
+        result = runner.invoke(cli_module.main, [
+            "store", "connect", "--backend", "neo4j",
+            "--uri", "bolt://example:7687", "--json"])
+        _ok(result)
+        data = _json_output(result)
+        assert data == {"backend": "neo4j", "connected": True}
+        assert calls["backend"] == "neo4j"
+        assert calls["cfg"].get("uri") == "bolt://example:7687"
+        assert calls.get("connected") is True
+
+    def test_connect_invalid_backend_reports_error_not_dispatch_error(self, runner):
+        """An unknown backend name must produce a meaningful backend error, not a
+        Python TypeError from the old get_graph_store_method() dispatch (#1354)."""
+        result = runner.invoke(cli_module.main,
+                               ["store", "connect", "--backend", "does-not-exist"])
+        # Exit 0 because store_connect always catches and reports errors gracefully.
+        _ok(result)
+        # The output must mention the backend, not a Python internal error.
+        assert "does-not-exist" in result.output
+        assert "TypeError" not in result.output
+        assert "Traceback" not in result.output
+
+    def test_connect_backend_error_surfaces_in_json(self, runner, monkeypatch):
+        """A connect() failure must appear in JSON output as connected=False with an error field."""
+
+        class _FailingStore:
+            def __init__(self, backend=None, **cfg):
+                pass
+
+            def connect(self):
+                raise RuntimeError("auth failed")
+
+        import semantica.graph_store as gs_mod
+        monkeypatch.setattr(gs_mod, "GraphStore", _FailingStore)
+        result = runner.invoke(cli_module.main, [
+            "store", "connect", "--backend", "neo4j", "--json"])
+        _ok(result)
+        data = _json_output(result)
+        assert data["connected"] is False
+        assert "auth failed" in data.get("error", "")
+        assert data["backend"] == "neo4j"
+
     def test_migrate_dry_run(self, runner):
         result = runner.invoke(cli_module.main, ["store", "migrate",
                                       "--from", "faiss", "--to", "qdrant", "--dry-run"])
@@ -2071,6 +2366,160 @@ class TestDoctorEmbeddingHintsAndEnv:
         assert "hash fallback" in st["note"], "padded/caps env value must enable deep mode"
 
 
+class TestDoctorTableLayout:
+    """#1428 + Qodo review: doctor table must keep Check labels and Hint text
+    readable at a normal 80-column terminal.
+
+    These tests render the *human-readable* (non-JSON) doctor table into a
+    captured 80-column Rich console so they cover the actual column-width
+    arithmetic, not just the JSON data.
+
+    Two regressions are protected:
+
+    A. #1428 — Hint (and Note) columns must not collapse into unreadable
+       single-character fragments or be silently truncated with a layout '…'.
+       overflow="fold" on both columns ensures content wraps across lines while
+       remaining fully present.
+
+    B. Qodo — Long Check labels such as "Embeddings (sentence-transformers)"
+       must not be truncated/ellipsized.  Assigning ratio=1 to the Check column
+       (as the original PR did) caused Rich to squeeze it below its min_width
+       at narrow terminals, so the fix removes ratio from the fixed-size columns.
+    """
+
+    def _render_doctor_at_80(self, runner, monkeypatch):
+        """Return the plain-text (ANSI-stripped) doctor table rendered at 80 cols."""
+        import io
+        import re
+        from rich.console import Console
+
+        # Unset LLM-provider env vars so the warn rows (with hints) are always present.
+        for var in ("OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GROQ_API_KEY"):
+            monkeypatch.delenv(var, raising=False)
+
+        buf = io.StringIO()
+        narrow_console = Console(
+            file=buf, width=80, highlight=False, force_terminal=True, no_color=True
+        )
+        monkeypatch.setattr(cli_module, "console", narrow_console)
+
+        result = runner.invoke(cli_module.main, ["doctor"])
+        assert result.exit_code == 0, f"doctor exited non-zero: {result.output!r}"
+
+        return re.sub(r"\x1b\[[0-9;]*m", "", buf.getvalue())
+
+    def _hint_column_parts(self, output: str) -> "list[str]":
+        """Extract non-blank Hint-column segments from each rendered line.
+
+        Locates the Hint column start from the header row and slices that
+        suffix from every subsequent line, so the test is insensitive to the
+        exact widths of the other columns.
+        """
+        lines = output.splitlines()
+        # Line 0 is blank (console.print() blank line before table).
+        hdr = next((l for l in lines if "Hint" in l and "Check" in l), None)
+        assert hdr is not None, "Could not find table header in doctor output"
+        hint_start = hdr.index("Hint")
+
+        parts = []
+        for line in lines:
+            if len(line) > hint_start:
+                seg = line[hint_start:].rstrip()
+                if seg and seg != "Hint" and not set(seg).issubset({"─", " "}):
+                    parts.append(seg)
+        return parts
+
+    # ── B: Qodo regression ────────────────────────────────────────────────────
+
+    def test_long_check_label_not_truncated_at_80_cols(self, runner, monkeypatch):
+        """'Embeddings (sentence-transformers)' must appear verbatim at 80 cols.
+
+        Before the fix, ratio=1 on the Check column let Rich squeeze it below
+        its min_width, turning the label into 'Embedd…' or similar.
+        """
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        assert "Embeddings (sentence-transformers)" in output, (
+            "Check label 'Embeddings (sentence-transformers)' was truncated in "
+            "the 80-column doctor table — the ratio= constraint on the Check "
+            "column must be removed so min_width=34 is always honoured."
+        )
+
+    def test_all_check_labels_not_truncated_at_80_cols(self, runner, monkeypatch):
+        """Every standard Check label must appear verbatim at 80 cols."""
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        for label in (
+            "Python",
+            "semantica",
+            "rich",
+            "Graph store",
+            "Vector store",
+            "Embeddings (sentence-transformers)",
+            "Embeddings (fastembed)",
+            "OpenAI",
+            "Anthropic",
+            "Groq",
+            "Config file",
+            "Log directory",
+        ):
+            assert label in output, (
+                f"Check label {label!r} was truncated or missing in the "
+                "80-column doctor table."
+            )
+
+    # ── A: #1428 regression ───────────────────────────────────────────────────
+
+    def test_hint_content_fully_present_at_80_cols(self, runner, monkeypatch):
+        """The LLM-provider hints must be fully present (folded, not ellipsized).
+
+        With overflow='fold' the full hint text wraps across lines; no
+        characters are discarded.  Joining the Hint-column segments (stripping
+        whitespace) must reconstruct each complete hint string.
+        """
+        import re
+
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        parts = self._hint_column_parts(output)
+        hint_joined = re.sub(r"\s+", "", "".join(parts))
+
+        # Each LLM-provider hint must be fully recoverable from the folded lines.
+        for expected in (
+            "exportOPENAI_API_KEY=\u2026",       # export OPENAI_API_KEY=…
+            "exportANTHROPIC_API_KEY=\u2026",    # export ANTHROPIC_API_KEY=…
+            "exportGROQ_API_KEY=\u2026",         # export GROQ_API_KEY=…
+        ):
+            assert expected in hint_joined, (
+                f"Hint content {expected!r} is missing from the 80-column "
+                "doctor table — overflow='fold' must be set on the Hint column "
+                "so no content is silently discarded."
+            )
+
+    def test_hint_column_has_no_single_char_fragments_at_80_cols(
+        self, runner, monkeypatch
+    ):
+        """No Hint-column line must be a single alphabetic character.
+
+        The original #1428 bug produced outputs like:
+            export
+            O
+            P
+            E
+            N
+            A
+            I
+            ...
+        because Rich allocated the Hint column only 1–2 characters of content
+        width.  overflow='fold' on a properly-wide column eliminates this.
+        """
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        parts = self._hint_column_parts(output)
+        single_char_alpha = [p for p in parts if len(p.strip()) == 1 and p.strip().isalpha()]
+        assert not single_char_alpha, (
+            f"Hint column contains single-character lines {single_char_alpha!r} "
+            "at 80 columns — the Hint column is too narrow; check min_width and "
+            "ratio settings."
+        )
+
+
 class TestEmbedGenerateOutput:
     """#994: `embed generate --output` must write files `embed index` can read."""
 

tests/test_llm_gemini.py

@@ -51,6 +51,20 @@ def test_generate_structured_forwards_to_the_real_provider():
     gemini.provider.generate_structured.assert_called_once_with("hello")
 
 
+def test_per_call_model_override_forwarded_for_both_methods():
+    gemini = Gemini(api_key="fake-key")
+    gemini.provider = MagicMock()
+    gemini.provider.is_available.return_value = True
+
+    gemini.generate("hello", model="gemini-1.5-flash")
+    gemini.generate_structured("hello", model="gemini-1.5-flash")
+
+    gemini.provider.generate.assert_called_once_with("hello", model="gemini-1.5-flash")
+    gemini.provider.generate_structured.assert_called_once_with(
+        "hello", model="gemini-1.5-flash"
+    )
+
+
 def test_generate_typed_forwards_schema_and_max_retries():
     gemini = Gemini(api_key="fake-key")
     gemini.provider = MagicMock()

tests/test_ner_configurations.py

@@ -236,11 +236,11 @@ class TestNERConfigurations(unittest.TestCase):
         mock_spacy.load.return_value = mock_nlp
         
         with patch('semantica.semantic_extract.methods.SPACY_AVAILABLE', True):
-            # Init extractor with list of methods
-            extractor = NERExtractor(method=["llm", "ml"], ensemble_voting=True)
+            # Explicit union retains complementary single-method entities.
+            extractor = NERExtractor(method=["llm", "ml"], merge_strategy="union")
             entities = extractor.extract_entities(self.text)
             
-            # Since ensemble_voting=True (implied merge), we expect unique entities
+            # Union keeps unique entities from every successful method.
             # Apple Inc (from both) + Steve Jobs (from ML)
             
             texts = [e.text for e in entities]

tests/vector_store/test_milvus_delete_vectors.py

@@ -0,0 +1,123 @@
+"""Tests for MilvusStore.delete_vectors (#1374)."""
+
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from semantica.context.erasure import STATUS_ERASED, ErasureCoordinator
+from semantica.utils.exceptions import ProcessingError
+from semantica.vector_store import VectorStore
+from semantica.vector_store.milvus_store import MilvusStore
+
+
+class MilvusStoreDeleteVectorsTest(TestCase):
+    def setUp(self):
+        self.patches = [
+            patch("semantica.vector_store.milvus_store.MILVUS_AVAILABLE", True)
+        ]
+        for p in self.patches:
+            p.start()
+
+    def tearDown(self):
+        for p in reversed(self.patches):
+            p.stop()
+
+    def _store(self, result=None, error=None):
+        """Return (store, exprs) where delete() records exprs, returns result."""
+        exprs = []
+        coll = MagicMock()
+
+        def _delete(expr, **kwargs):
+            exprs.append(expr)
+            if error is not None:
+                raise error
+            if result is None:
+                return MagicMock(delete_count=0)
+            return result
+
+        coll.collection.delete.side_effect = _delete
+        store = MilvusStore()
+        store.collection = coll
+        return store, exprs
+
+    def test_delete_single_id_uses_equality_expr(self):
+        store, exprs = self._store(result=MagicMock(delete_count=1))
+        ret = store.delete_vectors(["abc"])
+        self.assertEqual(exprs, ['id == "abc"'])
+        self.assertEqual(ret, {"delete_count": 1})
+
+    def test_delete_many_ids_uses_in_expr(self):
+        store, exprs = self._store(result=MagicMock(delete_count=2))
+        ret = store.delete_vectors(["a", "b"])
+        self.assertEqual(exprs, ['id in ["a", "b"]'])
+        self.assertEqual(ret, {"delete_count": 2})
+
+    def test_delete_escapes_quote_and_backslash_in_id(self):
+        store, exprs = self._store()
+        store.delete_vectors(['he said "hi"', "a\\b"])
+        self.assertEqual(exprs, ['id in ["he said \\"hi\\"", "a\\\\b"]'])
+
+    def test_delete_empty_ids_is_noop(self):
+        store, _ = self._store()
+        ret = store.delete_vectors([])
+        self.assertEqual(ret, {"delete_count": 0})
+        store.collection.collection.delete.assert_not_called()
+
+    def test_delete_without_collection_raises(self):
+        store = MilvusStore()
+        with self.assertRaises(ProcessingError):
+            store.delete_vectors(["a"])
+
+    def test_delete_backend_error_raises_processing_error(self):
+        store, _ = self._store(error=RuntimeError("connection reset"))
+        with self.assertRaises(ProcessingError):
+            store.delete_vectors(["a"])
+
+    def test_delete_string_delete_count_is_parsed(self):
+        store, _ = self._store(result=MagicMock(delete_count="3"))
+        ret = store.delete_vectors(["a", "b", "c"])
+        self.assertEqual(ret, {"delete_count": 3})
+
+    def test_delete_none_delete_count_defaults_zero(self):
+        store, _ = self._store(result=MagicMock(delete_count=None))
+        ret = store.delete_vectors(["a"])
+        self.assertEqual(ret, {"delete_count": 0})
+
+
+class MilvusErasureIntegrationTest(TestCase):
+    """ErasureCoordinator reaches the real MilvusStore.delete_vectors path."""
+
+    def _bind_milvus_as_vector_store(self):
+        vs = VectorStore(backend="milvus", config={"dimension": 3})
+        milvus = MilvusStore()
+        coll = MagicMock()
+        coll.collection.delete.return_value = MagicMock(delete_count=0)
+        milvus.collection = coll
+        vs._backend_store = milvus
+        return vs, coll
+
+    def test_erasure_reports_erased_when_delete_runs(self):
+        vs, coll = self._bind_milvus_as_vector_store()
+        coord = ErasureCoordinator(vector_store=vs)
+        receipt = coord.erase_entity("customer-4471")
+        coll.collection.delete.assert_called()
+        self.assertEqual(receipt.stores["vectors"]["status"], STATUS_ERASED)
+
+    def test_erasure_backend_name_is_milvus(self):
+        vs, _ = self._bind_milvus_as_vector_store()
+        coord = ErasureCoordinator(vector_store=vs)
+        receipt = coord.erase_entity("customer-4471")
+        self.assertEqual(receipt.stores["vectors"]["backend"], "milvus")
+
+    def test_facade_delete_vectors_forwards_to_milvus(self):
+        """VectorStore.delete_vectors() delegates to MilvusStore and returns its dict.
+
+        ErasureCoordinator probes _backend_store directly, so this test
+        exercises the public VectorStore facade path that other callers use.
+        """
+        vs, coll = self._bind_milvus_as_vector_store()
+        coll.collection.delete.return_value = MagicMock(delete_count=2)
+
+        ret = vs.delete_vectors(["id-1", "id-2"])
+
+        coll.collection.delete.assert_called_once()
+        self.assertEqual(ret, {"delete_count": 2})