fix(langchain): enforce node.id as merge key; stable document IDs

- props["name"] = node.id overwrites incoming properties["name"] so the
  MERGE predicate and stored name always match; prevents duplicate nodes
  and broken relationship MATCH on re-ingest
- Add _stable_document_id(): SHA-256(page_content + metadata) makes
  include_source=True idempotent across Python processes

test(adapters): tighten integration assertions for idempotency and rels

- test_add_graph_documents_creates_relationship: assert LC_RESEARCHES
  edge exists, not just source node presence
- test_add_graph_documents_idempotent: cnt == 1 to catch duplicates
- test_upsert_nodes_idempotent: len == 1 to catch duplicates
- test_get_rel_map: assert len >= 1 to verify non-empty result

Author: polaz

langchain-coordinode/langchain_coordinode/graph.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import re
 from typing import Any
 
@@ -104,7 +105,9 @@ def add_graph_documents(
             for node in doc.nodes:
                 label = _cypher_ident(node.type or "Entity")
                 props = dict(node.properties or {})
-                props.setdefault("name", node.id)
+                # Always enforce node.id as the merge key; incoming
+                # properties["name"] must not drift from the MERGE predicate.
+                props["name"] = node.id
                 self._client.cypher(
                     f"MERGE (n:{label} {{name: $name}}) SET n += $props",
                     params={"name": node.id, "props": props},
@@ -138,7 +141,7 @@ def add_graph_documents(
 
             # ── Optionally link source document ───────────────────────────
             if include_source and doc.source:
-                src_id = getattr(doc.source, "id", None) or str(id(doc.source))
+                src_id = getattr(doc.source, "id", None) or _stable_document_id(doc.source)
                 self._client.cypher(
                     "MERGE (d:__Document__ {id: $id}) SET d.page_content = $text",
                     params={"id": src_id, "text": doc.source.page_content or ""},
@@ -198,6 +201,19 @@ def __exit__(self, *args: Any) -> None:
 # ── Schema parser ─────────────────────────────────────────────────────────
 
 
+def _stable_document_id(source: Any) -> str:
+    """Return a deterministic ID for a LangChain Document.
+
+    Combines ``page_content`` and sorted ``metadata`` items so the same
+    document produces the same node across different Python processes,
+    making ``include_source=True`` re-ingest truly idempotent.
+    """
+    content = getattr(source, "page_content", "") or ""
+    metadata = getattr(source, "metadata", {}) or {}
+    stable = content + "|" + "|".join(f"{k}={v}" for k, v in sorted(metadata.items()))
+    return hashlib.sha256(stable.encode()).hexdigest()[:32]
+
+
 def _cypher_ident(name: str) -> str:
     """Escape a label/type name for use as a Cypher identifier."""
     # If already safe (alphanumeric + underscore, not starting with digit) keep as-is

tests/integration/adapters/test_langchain.py

@@ -93,12 +93,13 @@ def test_add_graph_documents_creates_relationship(graph, unique_tag):
 
     graph.add_graph_documents([doc])
 
-    # Verify both nodes exist
+    # Verify the relationship was created, not just the source node.
     result = graph.query(
-        "MATCH (n:LCPerson2 {name: $name}) RETURN n.name AS name",
-        params={"name": f"Charlie-{unique_tag}"},
+        "MATCH (a:LCPerson2 {name: $src})-[r:LC_RESEARCHES]->(b:LCConcept {name: $dst}) "
+        "RETURN count(r) AS cnt",
+        params={"src": f"Charlie-{unique_tag}", "dst": f"GraphRAG-{unique_tag}"},
     )
-    assert len(result) >= 1, f"source node not found: {result}"
+    assert result[0]["cnt"] >= 1, f"relationship not found: {result}"
 
 
 def test_add_graph_documents_idempotent(graph, unique_tag):
@@ -113,7 +114,7 @@ def test_add_graph_documents_idempotent(graph, unique_tag):
         "MATCH (n:LCIdempotent {name: $name}) RETURN count(n) AS cnt",
         params={"name": f"Idempotent-{unique_tag}"},
     )
-    assert result[0]["cnt"] >= 1
+    assert result[0]["cnt"] == 1
 
 
 def test_schema_refreshes_after_add(graph, unique_tag):

tests/integration/adapters/test_llama_index.py

@@ -66,7 +66,7 @@ def test_upsert_nodes_idempotent(store, tag):
     store.upsert_nodes([node])  # second call must not raise
 
     found = store.get(properties={"name": f"Idem-{tag}"})
-    assert len(found) >= 1
+    assert len(found) == 1
 
 
 def test_get_by_id(store, tag):
@@ -116,6 +116,7 @@ def test_get_rel_map(store, tag):
 
     result = store.get_rel_map([src], depth=1, limit=10)
     assert isinstance(result, list)
+    assert len(result) >= 1
 
 
 # ── Delete ────────────────────────────────────────────────────────────────────