fix(mcp): report duplicate kg_add reuse

mempalace/mcp_server.py

@@ -3787,8 +3787,20 @@ def tool_kg_add(
         },
     )
 
-    triple_id = _call_kg(
-        lambda kg: kg.add_triple(
+    def _add_with_status(kg):
+        # KG identities are normalized case-insensitively and collapse
+        # spaces/apostrophes. Mirror that key here so the MCP response
+        # reports reuse even when callers spell the same entity differently.
+        normalized_predicate = predicate.lower().replace(" ", "_")
+        normalized_object = object.lower().replace(" ", "_").replace("'", "")
+        already_exists = any(
+            fact.get("current") is True
+            and fact.get("predicate") == normalized_predicate
+            and str(fact.get("object") or "").lower().replace(" ", "_").replace("'", "")
+            == normalized_object
+            for fact in kg.query_entity(subject, direction="outgoing")
+        )
+        triple_id = kg.add_triple(
             subject,
             predicate,
             object,
@@ -3798,8 +3810,17 @@ def tool_kg_add(
             source_file=source_file,
             source_drawer_id=source_drawer_id,
         )
-    )
-    return {"success": True, "triple_id": triple_id, "fact": f"{subject} → {predicate} → {object}"}
+        return triple_id, already_exists
+
+    triple_id, already_exists = _call_kg(_add_with_status)
+    result = {
+        "success": True,
+        "triple_id": triple_id,
+        "fact": f"{subject} → {predicate} → {object}",
+    }
+    if already_exists:
+        result["reason"] = "already_exists"
+    return result
 
 
 def tool_kg_invalidate(subject: str, predicate: str, object: str, ended: str = None):

tests/test_kg_add_duplicate_reason.py

@@ -0,0 +1,85 @@
+"""Regression coverage for issue #2268: truthful duplicate KG responses."""
+
+import pytest
+
+from mempalace import mcp_server
+from mempalace.knowledge_graph import KnowledgeGraph
+
+
+@pytest.fixture
+def isolated_kg(monkeypatch, tmp_path):
+    db_path = tmp_path / "knowledge_graph.sqlite3"
+    kg = KnowledgeGraph(db_path=str(db_path))
+    monkeypatch.setattr(mcp_server, "_resolve_kg_path", lambda: str(db_path))
+    monkeypatch.setattr(mcp_server, "_get_kg", lambda *args, **kwargs: kg)
+    monkeypatch.setattr(mcp_server, "_wal_log", lambda *args, **kwargs: None)
+    try:
+        yield kg
+    finally:
+        kg.close()
+
+
+def test_kg_add_duplicate_reports_already_exists_and_preserves_original_provenance(isolated_kg):
+    first = mcp_server.tool_kg_add(
+        subject="Dax",
+        predicate="likes",
+        object="tea",
+        source_closet="cap_1",
+    )
+    second = mcp_server.tool_kg_add(
+        subject="Dax",
+        predicate="likes",
+        object="tea",
+        source_closet="cap_2",
+    )
+
+    assert first["success"] is True
+    assert "reason" not in first
+    assert second["success"] is True
+    assert second["reason"] == "already_exists"
+    assert second["triple_id"] == first["triple_id"]
+
+    facts = isolated_kg.query_entity("Dax", direction="outgoing")
+    current = [
+        fact
+        for fact in facts
+        if fact["current"] and fact["predicate"] == "likes" and fact["object"] == "tea"
+    ]
+    assert len(current) == 1
+    assert current[0]["source_closet"] == "cap_1"
+
+
+def test_kg_add_duplicate_reason_respects_entity_identity_normalization(isolated_kg):
+    first = mcp_server.tool_kg_add(
+        subject="Dax Rider",
+        predicate="visits",
+        object="Tea Shop",
+    )
+    second = mcp_server.tool_kg_add(
+        subject="dax rider",
+        predicate="visits",
+        object="tea shop",
+    )
+
+    assert second["reason"] == "already_exists"
+    assert second["triple_id"] == first["triple_id"]
+
+
+def test_kg_add_after_invalidation_is_a_fresh_insert(isolated_kg):
+    first = mcp_server.tool_kg_add(
+        subject="Dax",
+        predicate="likes",
+        object="tea",
+    )
+    isolated_kg.invalidate("Dax", "likes", "tea", ended="2026-08-15")
+
+    second = mcp_server.tool_kg_add(
+        subject="Dax",
+        predicate="likes",
+        object="tea",
+        valid_from="2026-08-16",
+    )
+
+    assert second["success"] is True
+    assert "reason" not in second
+    assert second["triple_id"] != first["triple_id"]