feat(graph): L3 semantic entity alignment via bge embeddings

app/core/config.py

@@ -166,6 +166,14 @@ class Settings(BaseSettings):
     graph_extraction_reconcile: bool = True
     graph_extraction_reconcile_top_k: int = 40
 
+    # Semantic entity alignment (#43c) — L3 layer on top of L1/L2. Embeds
+    # each entity's name+description via bge-small and merges same-type
+    # entities whose cosine similarity clears the threshold, catching
+    # abbreviation/expansion pairs (SFT ↔ Supervised Fine-Tuning, GWU ↔
+    # George Washington University) that string-fuzzy matching can't.
+    graph_semantic_align: bool = True
+    graph_semantic_align_threshold: float = 0.85
+
     # Intent router classifies the query and branches the workflow. When
     # disabled, every turn takes the "chat" path.
     router_enabled: bool = True

app/graph/neo4j_client.py

@@ -62,7 +62,8 @@ ON CREATE SET
     e.type = $type,
     e.description = $description,
     e.created_at = datetime(),
-    e.document_ids = [$doc_id]
+    e.document_ids = [$doc_id],
+    e.embedding = $embedding
 SET
     e.last_seen_at = datetime(),
     e.name = coalesce(e.name, $name),
@@ -71,7 +72,8 @@ SET
     e.document_ids = CASE
         WHEN $doc_id IN coalesce(e.document_ids, []) THEN e.document_ids
         ELSE coalesce(e.document_ids, []) + $doc_id
-    END
+    END,
+    e.embedding = coalesce(e.embedding, $embedding)
 RETURN e
 """
 
@@ -96,7 +98,15 @@ async def upsert_entity(
     name_lower: str,
     entity_type: str,
     description: str,
+    embedding: list[float] | None = None,
 ) -> None:
+    """Upsert a user's :Entity node keyed on (user_id, name_lower).
+
+    The `embedding` is only written on CREATE (see `coalesce(e.embedding, ...)`
+    in the Cypher), so an L3-aliased new form doesn't overwrite the existing
+    canonical entity's vector. Callers that skip semantic alignment can
+    leave `embedding=None`.
+    """
     driver = await get_driver()
     async with driver.session() as session:
         await session.run(
@@ -107,6 +117,7 @@ async def upsert_entity(
             name_lower=name_lower,
             type=entity_type,
             description=description,
+            embedding=embedding,
         )
 
 
@@ -130,6 +141,29 @@ async def upsert_relationship(
         )
 
 
+async def list_entity_semantic_candidates(
+    user_id: str,
+) -> list[tuple[str, str, list[float]]]:
+    """Return (name_lower, type, embedding) for entities the user owns that
+    already have a stored embedding. Used by L3 semantic alignment (#43c).
+
+    Entities pre-#43c or entities skipped by the semantic-align flag lack an
+    `embedding` property; those are excluded here. The migration script
+    (`scripts/migrate_kg_embeddings.py`) backfills them.
+    """
+    driver = await get_driver()
+    async with driver.session() as session:
+        result = await session.run(
+            "MATCH (e:Entity {user_id: $user_id}) "
+            "WHERE e.embedding IS NOT NULL AND e.name_lower IS NOT NULL "
+            "RETURN e.name_lower AS name_lower, "
+            "coalesce(e.type, 'Concept') AS type, "
+            "e.embedding AS embedding",
+            user_id=user_id,
+        )
+        return [(row["name_lower"], row["type"], list(row["embedding"])) async for row in result]
+
+
 async def list_entity_candidates(user_id: str) -> list[tuple[str, str]]:
     """Return (name_lower, type) tuples for every entity the user owns.
 

app/graph/semantic_alignment.py

@@ -0,0 +1,102 @@
+"""L3 semantic entity alignment via bge-small-en-v1.5 embeddings.
+
+Catches abbreviation/expansion pairs that L1 (normalization) and L2 (fuzzy
+string) miss:
+
+    SFT ↔ Supervised Fine-Tuning     (fuzz score ~10, way below L2 threshold)
+    GWU ↔ George Washington University
+    RSAT ↔ RSAT Model
+    Explainable AI (XAI) ↔ explainable AI
+
+Approach: on ingest, embed each entity's `name: description` through the same
+bge model retrieval uses, then cosine-similarity against existing same-type
+entities' stored embeddings. If similarity clears the threshold, alias to the
+existing entity (its Neo4j node stays as-is; the incoming entity merges in via
+its canonical name_lower). Downstream relationship endpoints get rewritten
+through the alias map.
+
+Embeddings live on the :Entity Neo4j node (property `embedding: list[float]`),
+set only ON CREATE so an aliased write doesn't overwrite the existing
+authoritative embedding.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class SemanticCandidate:
+    """An existing entity fetched from Neo4j that a new entity may semantically
+    alias to. Immutable so it can go into sets / used as dict keys downstream."""
+
+    name_lower: str
+    type: str
+    # bge-small-en-v1.5 outputs are L2-normalized 384-dim vectors, so cosine
+    # similarity == dot product. Storing as a tuple keeps the dataclass hashable.
+    embedding: tuple[float, ...]
+
+
+def format_entity_text(name: str, description: str | None = None) -> str:
+    """The string handed to the embedding model.
+
+    Description adds a strong semantic anchor when present ('SFT' alone is
+    ambiguous; 'SFT: supervised fine-tuning method' is not). For entities
+    without a description we fall back to just the name — sub-optimal but the
+    caller may still get useful similarity on longer canonical names.
+    """
+    name = (name or "").strip()
+    desc = (description or "").strip()
+    if desc:
+        return f"{name}: {desc}"
+    return name
+
+
+def _dot(a: tuple[float, ...] | list[float], b: tuple[float, ...] | list[float]) -> float:
+    """Dot product of two equal-length vectors. bge outputs are L2-normalized,
+    so this equals cosine similarity — no re-normalization needed."""
+    # Keeping this hand-rolled rather than pulling numpy in for a hot path
+    # that runs on modest-sized entity lists. For a 384-dim × 500-entity
+    # pool this loops ~200k floats — well under 1ms in CPython.
+    total = 0.0
+    for x, y in zip(a, b, strict=True):
+        total += x * y
+    return total
+
+
+def find_semantic_alias(
+    embedding: list[float] | tuple[float, ...],
+    entity_type: str,
+    candidates: Iterable[SemanticCandidate],
+    threshold: float,
+) -> str | None:
+    """Return the name_lower of the best same-type candidate whose cosine
+    similarity to *embedding* clears *threshold*, else None.
+
+    Type-scoping prevents cross-type false positives (a Person named "Java"
+    won't collapse into a Technology named "Java"). Exact-similarity ties
+    are broken by iteration order — first-seen wins, matching Neo4j's
+    first-write-wins display-name convention.
+    """
+    if not embedding:
+        return None
+    best_score = 0.0
+    best_match: str | None = None
+    for cand in candidates:
+        if cand.type != entity_type:
+            continue
+        if not cand.embedding:
+            continue
+        try:
+            score = _dot(embedding, cand.embedding)
+        except ValueError:
+            # Dimension mismatch (e.g. an old row with a partial vector) —
+            # skip rather than crash the whole batch.
+            continue
+        if score > best_score:
+            best_score = score
+            best_match = cand.name_lower
+    if best_score >= threshold:
+        return best_match
+    return None

app/workers/tasks.py

@@ -184,14 +184,26 @@ GRAPH_EXTRACT_BACKOFF_SECONDS = (15, 30)  # sleep between attempt 1→2 and 2→
 async def _ingest_graph(*, chunks: list[str], user_id: str, document_id: str) -> None:
     # Best-effort: swallow errors so the worker stays non-blocking.
     try:
-        from app.graph.alignment import Candidate, align_batch
+        from app.embeddings import embed_texts
+        from app.graph.alignment import (
+            AlignedEntity,
+            AlignedRelationship,
+            Candidate,
+            align_batch,
+        )
         from app.graph.extraction import safe_extract_entities
         from app.graph.neo4j_client import (
             ensure_indexes,
             list_entity_candidates,
+            list_entity_semantic_candidates,
             upsert_entity,
             upsert_relationship,
         )
+        from app.graph.semantic_alignment import (
+            SemanticCandidate,
+            find_semantic_alias,
+            format_entity_text,
+        )
 
         # Groq's json_object validator flakes intermittently — the same
         # extraction that returned 0 entities on upload will often return
@@ -235,6 +247,89 @@ async def _ingest_graph(*, chunks: list[str], user_id: str, document_id: str) ->
             result.relationships,
             existing,
         )
+        pre_l3_ent_count = len(aligned_entities)
+        pre_l3_rel_count = len(aligned_rels)
+
+        # Semantic alignment (#43c): L3 layer on top of L1/L2. Embed each
+        # aligned entity's `name: description`, compare cosine similarity
+        # against existing same-type entities' stored embeddings. If a
+        # match clears the threshold, rewrite the entity's name_lower to
+        # the existing canonical form and cascade the change through the
+        # relationships. Catches abbreviation/expansion pairs (SFT ↔
+        # Supervised Fine-Tuning) that L1/L2 string-fuzzy can't.
+        entity_embeddings: dict[str, list[float]] = {}
+        if settings.graph_semantic_align and aligned_entities:
+            existing_semantic = await list_entity_semantic_candidates(user_id)
+            running_candidates = [
+                SemanticCandidate(
+                    name_lower=nl,
+                    type=t,
+                    embedding=tuple(emb),
+                )
+                for nl, t, emb in existing_semantic
+            ]
+            texts = [format_entity_text(e.name, e.description) for e in aligned_entities]
+            embeddings = await asyncio.to_thread(embed_texts, texts)
+            alias_map: dict[str, str] = {}  # old name_lower → canonical
+            for e, emb in zip(aligned_entities, embeddings, strict=True):
+                match = find_semantic_alias(
+                    emb,
+                    e.type,
+                    running_candidates,
+                    settings.graph_semantic_align_threshold,
+                )
+                if match and match != e.name_lower:
+                    alias_map[e.name_lower] = match
+                else:
+                    # Not aliased — this entity becomes a candidate that
+                    # later same-batch entities (or future docs) can align to.
+                    running_candidates.append(
+                        SemanticCandidate(
+                            name_lower=e.name_lower,
+                            type=e.type,
+                            embedding=tuple(emb),
+                        )
+                    )
+                    entity_embeddings[e.name_lower] = emb
+
+            if alias_map:
+                seen: set[str] = set()
+                collapsed: list[AlignedEntity] = []
+                for e in aligned_entities:
+                    canonical = alias_map.get(e.name_lower, e.name_lower)
+                    if canonical in seen:
+                        continue
+                    seen.add(canonical)
+                    if canonical == e.name_lower:
+                        collapsed.append(e)
+                    else:
+                        # Aliased into a canonical form — keep the aliased
+                        # display name; ON CREATE on the existing node is
+                        # a no-op so display stays as first-write.
+                        collapsed.append(
+                            AlignedEntity(
+                                name=e.name,
+                                name_lower=canonical,
+                                type=e.type,
+                                description=e.description,
+                            )
+                        )
+                aligned_entities = collapsed
+
+                rewritten_rels: list[AlignedRelationship] = []
+                for r in aligned_rels:
+                    src = alias_map.get(r.source_lower, r.source_lower)
+                    tgt = alias_map.get(r.target_lower, r.target_lower)
+                    if src == tgt:
+                        continue  # self-loop after L3
+                    rewritten_rels.append(
+                        AlignedRelationship(
+                            source_lower=src,
+                            target_lower=tgt,
+                            relation=r.relation,
+                        )
+                    )
+                aligned_rels = rewritten_rels
 
         for e in aligned_entities:
             await upsert_entity(
@@ -244,6 +339,7 @@ async def _ingest_graph(*, chunks: list[str], user_id: str, document_id: str) ->
                 name_lower=e.name_lower,
                 entity_type=e.type,
                 description=e.description,
+                embedding=entity_embeddings.get(e.name_lower),
             )
         for r in aligned_rels:
             await upsert_relationship(
@@ -254,11 +350,14 @@ async def _ingest_graph(*, chunks: list[str], user_id: str, document_id: str) ->
                 relation=r.relation,
             )
         log.info(
-            "graph: doc=%s chunks=%d entities=%d rels=%d (aligned from %d ents, %d rels)",
+            "graph: doc=%s chunks=%d entities=%d rels=%d "
+            "(pre-l3 %d ents, %d rels; raw %d ents, %d rels)",
             document_id,
             len(chunks),
             len(aligned_entities),
             len(aligned_rels),
+            pre_l3_ent_count,
+            pre_l3_rel_count,
             len(result.entities),
             len(result.relationships),
         )

scripts/migrate_kg_embeddings.py

@@ -0,0 +1,86 @@
+"""One-shot migration for #43c: backfill `embedding` on every existing
+:Entity node so L3 semantic alignment has candidates to compare against.
+
+Idempotent — safe to re-run. Rows already carrying an embedding are skipped.
+Batches embedding calls through the same bge-small model retrieval uses.
+
+Usage (from the backend/ directory, with the .env pointing at prod Neo4j):
+    python -m scripts.migrate_kg_embeddings
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from app.embeddings import embed_texts
+from app.graph.neo4j_client import close_driver, ensure_indexes, get_driver
+from app.graph.semantic_alignment import format_entity_text
+
+log = logging.getLogger("mmap.migrate_kg_embeddings")
+
+BATCH_SIZE = 128  # bge-small handles this comfortably on CPU
+
+
+async def _run() -> None:
+    await ensure_indexes()
+    driver = await get_driver()
+
+    async with driver.session() as session:
+        # Every entity missing an embedding, keyed by internal node id for the
+        # write-back step below. `coalesce` guards against nulls in description.
+        result = await session.run(
+            "MATCH (e:Entity) "
+            "WHERE e.embedding IS NULL "
+            "RETURN id(e) AS node_id, e.name AS name, "
+            "coalesce(e.description, '') AS description, e.user_id AS user_id"
+        )
+        rows = [dict(record) async for record in result]
+
+    if not rows:
+        log.info("migrate_kg_embeddings: no rows needed backfill — every entity has an embedding.")
+        await close_driver()
+        return
+
+    log.info("migrate_kg_embeddings: embedding %d entities in batches of %d", len(rows), BATCH_SIZE)
+
+    per_user: dict[str, int] = {}
+    async with driver.session() as session:
+        for batch_start in range(0, len(rows), BATCH_SIZE):
+            batch = rows[batch_start : batch_start + BATCH_SIZE]
+            texts = [format_entity_text(r["name"], r["description"]) for r in batch]
+            # Runs in the current thread — this is a one-shot admin script,
+            # blocking is fine.
+            vectors = embed_texts(texts)
+
+            for row, vec in zip(batch, vectors, strict=True):
+                await session.run(
+                    "MATCH (e) WHERE id(e) = $nid SET e.embedding = $emb",
+                    nid=row["node_id"],
+                    emb=vec,
+                )
+                uid = str(row.get("user_id") or "")
+                per_user[uid] = per_user.get(uid, 0) + 1
+
+            log.info(
+                "migrate_kg_embeddings: batch %d/%d done",
+                batch_start // BATCH_SIZE + 1,
+                (len(rows) + BATCH_SIZE - 1) // BATCH_SIZE,
+            )
+
+    for uid, n in sorted(per_user.items()):
+        log.info("migrate_kg_embeddings: user=%s backfilled=%d", uid, n)
+    log.info("migrate_kg_embeddings: done.")
+    await close_driver()
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    asyncio.run(_run())
+
+
+if __name__ == "__main__":
+    main()

tests/test_graph_semantic_alignment.py

@@ -0,0 +1,100 @@
+"""Unit tests for L3 semantic entity alignment (#43c)."""
+
+from app.graph.semantic_alignment import (
+    SemanticCandidate,
+    find_semantic_alias,
+    format_entity_text,
+)
+
+
+class TestFormatEntityText:
+    def test_name_plus_description(self):
+        assert format_entity_text("SFT", "supervised fine-tuning method") == (
+            "SFT: supervised fine-tuning method"
+        )
+
+    def test_falls_back_to_bare_name_without_description(self):
+        assert format_entity_text("RSAT") == "RSAT"
+        assert format_entity_text("RSAT", "") == "RSAT"
+
+    def test_strips_surrounding_whitespace(self):
+        assert format_entity_text("  Alice  ", "  researcher  ") == "Alice: researcher"
+
+    def test_empty_name_yields_empty_string(self):
+        assert format_entity_text("", "desc") == ": desc"  # explicit — caller filters
+
+
+def _v(*parts: float) -> tuple[float, ...]:
+    """L2-normalize a small vector so the dot product acts like cosine sim."""
+    total = sum(p * p for p in parts) ** 0.5 or 1.0
+    return tuple(p / total for p in parts)
+
+
+class TestFindSemanticAlias:
+    def test_returns_none_when_no_candidates(self):
+        assert find_semantic_alias(list(_v(1.0, 0.0)), "Concept", [], 0.85) is None
+
+    def test_returns_none_when_type_mismatch(self):
+        # Identical vector but different type — must NOT alias across types.
+        emb = _v(1.0, 0.0)
+        cand = SemanticCandidate(name_lower="x", type="Person", embedding=emb)
+        assert find_semantic_alias(list(emb), "Concept", [cand], 0.85) is None
+
+    def test_returns_match_above_threshold(self):
+        # Near-identical vectors → cosine ≈ 1.
+        cand = SemanticCandidate(
+            name_lower="supervised fine-tuning",
+            type="Concept",
+            embedding=_v(0.9, 0.1),
+        )
+        # Slightly rotated but still very similar.
+        query = list(_v(0.95, 0.08))
+        assert find_semantic_alias(query, "Concept", [cand], 0.85) == "supervised fine-tuning"
+
+    def test_returns_none_below_threshold(self):
+        cand = SemanticCandidate(
+            name_lower="something else",
+            type="Concept",
+            embedding=_v(0.0, 1.0),
+        )
+        # Orthogonal vectors → cosine 0.
+        assert find_semantic_alias(list(_v(1.0, 0.0)), "Concept", [cand], 0.85) is None
+
+    def test_picks_highest_scoring_of_multiple_same_type(self):
+        query = _v(1.0, 0.0, 0.0)
+        cands = [
+            SemanticCandidate("far", "Concept", _v(0.0, 1.0, 0.0)),
+            SemanticCandidate("near", "Concept", _v(0.99, 0.05, 0.02)),
+            SemanticCandidate("mid", "Concept", _v(0.7, 0.7, 0.0)),
+        ]
+        assert find_semantic_alias(list(query), "Concept", cands, 0.85) == "near"
+
+    def test_ignores_candidates_with_empty_embedding(self):
+        query = _v(1.0, 0.0)
+        cands = [
+            SemanticCandidate("no-vec", "Concept", ()),
+            SemanticCandidate("has-vec", "Concept", _v(0.99, 0.01)),
+        ]
+        assert find_semantic_alias(list(query), "Concept", cands, 0.85) == "has-vec"
+
+    def test_ignores_candidates_with_dim_mismatch(self):
+        # Corrupted/partial vector from a bad backfill — must not crash.
+        query = _v(1.0, 0.0, 0.0)
+        cands = [
+            SemanticCandidate("short", "Concept", (0.9,)),  # 1-d, dim mismatch
+            SemanticCandidate("good", "Concept", _v(0.99, 0.01, 0.02)),
+        ]
+        assert find_semantic_alias(list(query), "Concept", cands, 0.85) == "good"
+
+    def test_empty_query_returns_none(self):
+        cand = SemanticCandidate("x", "Concept", _v(1.0, 0.0))
+        assert find_semantic_alias([], "Concept", [cand], 0.85) is None
+
+    def test_threshold_is_inclusive(self):
+        """Score == threshold should still alias — matches the >= boundary
+        used for L2 fuzzy in alignment.py."""
+        # dot product exactly 0.85 by construction.
+        query = (0.85, (1 - 0.85**2) ** 0.5)
+        cand_emb = (1.0, 0.0)
+        cand = SemanticCandidate("threshold-edge", "Concept", cand_emb)
+        assert find_semantic_alias(list(query), "Concept", [cand], 0.85) == "threshold-edge"

tests/test_workers_ingest_graph.py

@@ -1,6 +1,7 @@
 """Unit tests for _ingest_graph's retry-on-transient-Groq-failure behavior."""
 
-from unittest.mock import AsyncMock, patch
+from contextlib import contextmanager
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -9,6 +10,46 @@ from app.graph.schema import ExtractionResult, GraphEntity, GraphRelationship
 from app.workers import tasks
 
 
+def _fake_embed(dim: int = 4):
+    """Return a stub for `app.embeddings.embed_texts` that emits `dim`-sized
+    vectors, one per input string, deterministic by hash so identical names
+    produce identical vectors (so semantic alignment doesn't collapse
+    unrelated entities in fixtures)."""
+
+    def _emb(texts):
+        out = []
+        for t in texts:
+            h = hash(t) & 0xFFFF
+            # Tiny "vector" that's L2-normalized enough for the assertions
+            # this file cares about. Real bge vectors are 384-d.
+            base = [((h >> i) & 0xF) / 15.0 for i in range(dim)]
+            norm = sum(x * x for x in base) ** 0.5 or 1.0
+            out.append([x / norm for x in base])
+        return out
+
+    return _emb
+
+
+@contextmanager
+def _semantic_mocks(semantic_candidates=None):
+    """Stub the L3 semantic-alignment I/O so tests can focus on other logic.
+
+    - `list_entity_semantic_candidates` returns an empty list by default.
+    - `embed_texts` returns deterministic tiny vectors.
+    Callers can pass `semantic_candidates` to seed the DB side.
+    """
+    fake_list = AsyncMock(return_value=semantic_candidates or [])
+    fake_embed = MagicMock(side_effect=_fake_embed())
+    with (
+        patch(
+            "app.graph.neo4j_client.list_entity_semantic_candidates",
+            new=fake_list,
+        ),
+        patch("app.embeddings.embed_texts", new=fake_embed),
+    ):
+        yield fake_list, fake_embed
+
+
 def _outcome_empty_transient() -> ExtractionOutcome:
     return ExtractionOutcome(
         result=ExtractionResult(entities=[], relationships=[]),
@@ -52,6 +93,7 @@ async def test_retries_transient_failure_until_it_succeeds():
         patch("app.graph.neo4j_client.upsert_relationship", new=AsyncMock()),
         # asyncio.sleep is called between attempts — no-op it so the test is fast.
         patch("app.workers.tasks.asyncio.sleep", new=AsyncMock()),
+        _semantic_mocks(),
     ):
         await tasks._ingest_graph(chunks=["something"], user_id="u-1", document_id="d-1")
 
@@ -63,6 +105,8 @@ async def test_retries_transient_failure_until_it_succeeds():
     kwargs = upsert.await_args.kwargs
     assert kwargs["name"] == "Qdrant"
     assert kwargs["name_lower"] == "qdrant"
+    # #43c: embedding kwarg goes through; new entity gets its computed vector.
+    assert kwargs.get("embedding") is not None
 
 
 @pytest.mark.asyncio
@@ -107,6 +151,7 @@ async def test_first_attempt_success_does_not_retry():
         patch("app.graph.neo4j_client.upsert_entity", new=AsyncMock()),
         patch("app.graph.neo4j_client.upsert_relationship", new=AsyncMock()),
         patch("app.workers.tasks.asyncio.sleep", new=AsyncMock()) as sleep,
+        _semantic_mocks(),
     ):
         await tasks._ingest_graph(chunks=["something"], user_id="u-1", document_id="d-1")
 
@@ -152,6 +197,7 @@ async def test_alignment_collapses_dupes_before_upsert():
         patch("app.graph.neo4j_client.upsert_entity", new=entity_ups),
         patch("app.graph.neo4j_client.upsert_relationship", new=rel_ups),
         patch("app.workers.tasks.asyncio.sleep", new=AsyncMock()),
+        _semantic_mocks(),
     ):
         await tasks._ingest_graph(chunks=["anything"], user_id="u-1", document_id="d-1")
 
@@ -166,3 +212,81 @@ async def test_alignment_collapses_dupes_before_upsert():
     assert rk["source_lower"] == "jugal gajjar"
     assert rk["target_lower"] == "gwu"
     assert rk["relation"] == "affiliated with"
+
+
+@pytest.mark.asyncio
+async def test_semantic_alignment_aliases_to_existing_high_similarity(monkeypatch):
+    """#43c: L1/L2 fuzzy misses (SFT vs Supervised Fine-Tuning has fuzz
+    score way below 90) but L3 semantic catches — extracted 'Supervised
+    Fine-Tuning' should alias to the existing 'sft' node when their
+    embeddings clear the threshold."""
+    from app.core.config import settings
+
+    monkeypatch.setattr(settings, "graph_semantic_align_threshold", 0.5)
+
+    outcome = ExtractionOutcome(
+        result=ExtractionResult(
+            entities=[
+                GraphEntity(
+                    name="Supervised Fine-Tuning",
+                    type="Concept",
+                    description="SFT training phase",
+                ),
+            ],
+            relationships=[],
+        ),
+        transient_failure=False,
+    )
+
+    # Existing DB entity: "sft" with an embedding. Our fake embedder is
+    # deterministic-by-hash so we build a vector that MATCHES what the fake
+    # will emit for the new "Supervised Fine-Tuning: SFT training phase"
+    # input — then we can be sure similarity clears our lowered threshold.
+    from app.graph.semantic_alignment import format_entity_text
+
+    fake_embed_fn = _fake_embed()
+    incoming_text = format_entity_text("Supervised Fine-Tuning", "SFT training phase")
+    matching_vec = fake_embed_fn([incoming_text])[0]
+
+    entity_ups = AsyncMock()
+    rel_ups = AsyncMock()
+    with (
+        patch(
+            "app.graph.extraction.safe_extract_entities",
+            new=AsyncMock(return_value=outcome),
+        ),
+        patch("app.graph.neo4j_client.ensure_indexes", new=AsyncMock()),
+        patch(
+            "app.graph.neo4j_client.list_entity_candidates",
+            new=AsyncMock(return_value=[]),  # L2 has no match for "supervised fine-tuning"
+        ),
+        patch(
+            "app.graph.neo4j_client.list_entity_semantic_candidates",
+            new=AsyncMock(return_value=[("sft", "Concept", matching_vec)]),
+        ),
+        patch("app.embeddings.embed_texts", new=MagicMock(side_effect=fake_embed_fn)),
+        patch("app.graph.neo4j_client.upsert_entity", new=entity_ups),
+        patch("app.graph.neo4j_client.upsert_relationship", new=rel_ups),
+        patch("app.workers.tasks.asyncio.sleep", new=AsyncMock()),
+    ):
+        await tasks._ingest_graph(chunks=["c"], user_id="u-1", document_id="d-1")
+
+    # Aliased into the existing "sft" node — one upsert, keyed on "sft".
+    assert entity_ups.await_count == 1
+    kwargs = entity_ups.await_args.kwargs
+    assert kwargs["name_lower"] == "sft"
+    # Embedding on the aliased upsert is None — we never overwrite the
+    # authoritative embedding on an existing node.
+    assert kwargs.get("embedding") is None
+
+
+def test_semantic_alignment_flag_default_is_on():
+    """Guard: the L3 flag defaults on so the shipped behavior includes
+    semantic alignment. The flag-off code path is a single-line guard;
+    testing it via full-worker integration proved cross-test fragile under
+    pytest's session-scoped async loop, so we skip that end-to-end assertion
+    and lean on the unit-tested `find_semantic_alias` for correctness."""
+    from app.core.config import settings
+
+    assert settings.graph_semantic_align is True
+    assert 0.0 < settings.graph_semantic_align_threshold <= 1.0