From 4d3259df319a2cb9abd4943772fc42e8d0f0aa2c Mon Sep 17 00:00:00 2001
From: T1mn <136770748@qq.com>
Date: Wed, 19 Aug 2026 20:15:20 +0800
Subject: [PATCH] fix(kg): validate entity_id aliases

---
 semantica/kg/graph_validator.py  | 47 ++++++++++++++++++++------------
 tests/kg/test_graph_validator.py | 34 +++++++++++++++++++++++
 2 files changed, 63 insertions(+), 18 deletions(-)
 create mode 100644 tests/kg/test_graph_validator.py

diff --git a/semantica/kg/graph_validator.py b/semantica/kg/graph_validator.py
index 9d6a2cbb..a8d126ca 100644
--- a/semantica/kg/graph_validator.py
+++ b/semantica/kg/graph_validator.py
@@ -31,6 +31,7 @@ from dataclasses import dataclass, field
 from enum import Enum
 import networkx as nx
 
+from ..utils.entity_ids import get_entity_id
 from ..utils.logging import get_logger
 
 class ValidationSeverity(Enum):
@@ -147,19 +148,23 @@ class GraphValidator:
         # 2. Entity Validation
         entity_ids = set()
         for entity in entities:
+            eid = get_entity_id(entity)
+
             # Check required fields
             missing = self.required_entity_fields - set(entity.keys())
+            missing.discard("id")
+            if eid is None:
+                missing.add("id or entity_id")
             if missing:
                 issues.append(ValidationIssue(
                     code="MISSING_FIELD",
                     message=f"Entity missing required fields: {missing}",
                     severity=ValidationSeverity.ERROR,
-                    element_id=entity.get("id", "unknown"),
+                    element_id=eid or "unknown",
                     element_type="entity"
                 ))
             
             # Check ID uniqueness
-            eid = entity.get("id")
             if eid:
                 if eid in entity_ids:
                     issues.append(ValidationIssue(
@@ -183,18 +188,29 @@ class GraphValidator:
                         element_type="entity"
                     ))
 
+        def get_relationship_endpoint(rel, field, alias):
+            endpoint = rel.get(field)
+            if endpoint is None:
+                endpoint = rel.get(alias)
+            return endpoint
+
+        def is_valid_id(node_id):
+            if node_id is None:
+                return False
+            try:
+                return node_id in entity_ids
+            except TypeError:
+                # Not hashable, so it can't be in the set of string IDs
+                return False
+
         # 3. Relationship Validation
         for i, rel in enumerate(relationships):
             # Endpoints may use either the legacy ``source``/``target`` keys or
             # the canonical ``source_id``/``target_id`` keys emitted by
             # ``ContextGraph.to_kg_dict()``. Accept either variant so both
             # representations validate consistently.
-            src = rel.get("source")
-            if src is None:
-                src = rel.get("source_id")
-            tgt = rel.get("target")
-            if tgt is None:
-                tgt = rel.get("target_id")
+            src = get_relationship_endpoint(rel, "source", "source_id")
+            tgt = get_relationship_endpoint(rel, "target", "target_id")
 
             # Check required fields: ``type`` plus a resolvable source/target.
             missing = set()
@@ -215,15 +231,6 @@ class GraphValidator:
                 continue
             
             # Check Dangling Edges
-            def is_valid_id(node_id):
-                if node_id is None:
-                    return False
-                try:
-                    return node_id in entity_ids
-                except TypeError:
-                    # Not hashable, so it can't be in the set of string IDs
-                    return False
-
             if not is_valid_id(src):
                 issues.append(ValidationIssue(
                     code="DANGLING_EDGE",
@@ -259,7 +266,11 @@ class GraphValidator:
         try:
             nx_graph = nx.DiGraph()
             nx_graph.add_nodes_from(entity_ids)
-            nx_graph.add_edges_from([(r["source"], r["target"]) for r in relationships if r.get("source") in entity_ids and r.get("target") in entity_ids])
+            for relationship in relationships:
+                src = get_relationship_endpoint(relationship, "source", "source_id")
+                tgt = get_relationship_endpoint(relationship, "target", "target_id")
+                if is_valid_id(src) and is_valid_id(tgt):
+                    nx_graph.add_edge(src, tgt)
             
             # Check for cycles
             try:
diff --git a/tests/kg/test_graph_validator.py b/tests/kg/test_graph_validator.py
new file mode 100644
index 00000000..a0c57add
--- /dev/null
+++ b/tests/kg/test_graph_validator.py
@@ -0,0 +1,34 @@
+from semantica.kg.graph_builder import GraphBuilder
+from semantica.kg.graph_validator import GraphValidator
+
+
+def test_entity_id_aliases_are_validated_across_graph_structure():
+    """Entity aliases should work for schema and structural validation."""
+    graph = GraphBuilder(
+        merge_entities=True,
+        entity_resolution_strategy="exact",
+        resolve_conflicts=False,
+    ).build(
+        {
+            "entities": [
+                {"entity_id": "alice:1", "name": "Alice", "type": "Person"},
+                {"entity_id": "alice:2", "name": " Alice ", "type": "Person"},
+                {"entity_id": "org:1", "name": "Acme", "type": "Organization"},
+            ],
+            "relationships": [
+                {
+                    "source_id": "alice:2",
+                    "target_id": "org:1",
+                    "type": "WORKS_FOR",
+                }
+            ],
+        }
+    )
+
+    result = GraphValidator().validate(graph)
+
+    assert result.is_valid
+    assert not any(
+        issue.code in {"MISSING_FIELD", "DANGLING_EDGE", "ORPHAN_NODES"}
+        for issue in result.issues
+    )