Merge pull request #1214 from arnoldwender/fix/kg-temporal-inversion-guard

fix(kg): reject inverted intervals in add_triple (valid_to < valid_from)
This commit is contained in:
Igor Lins e Silva
2026-05-06 01:32:16 -03:00
committed by GitHub
2 changed files with 43 additions and 0 deletions
+9
View File
@@ -171,6 +171,15 @@ class KnowledgeGraph:
add_triple("Max", "does", "swimming", valid_from="2025-01-01") add_triple("Max", "does", "swimming", valid_from="2025-01-01")
add_triple("Alice", "worried_about", "Max injury", valid_from="2026-01", valid_to="2026-02") add_triple("Alice", "worried_about", "Max injury", valid_from="2026-01", valid_to="2026-02")
""" """
# Reject inverted intervals: a triple with valid_to < valid_from
# would never satisfy `valid_from <= as_of AND valid_to >= as_of`,
# so it would be invisible to every query — silently corrupt.
if valid_from is not None and valid_to is not None and valid_to < valid_from:
raise ValueError(
f"valid_to={valid_to!r} is before valid_from={valid_from!r}; "
"an inverted interval would be invisible to every KG query"
)
sub_id = self._entity_id(subject) sub_id = self._entity_id(subject)
obj_id = self._entity_id(obj) obj_id = self._entity_id(obj)
pred = predicate.lower().replace(" ", "_") pred = predicate.lower().replace(" ", "_")
+34
View File
@@ -5,6 +5,8 @@ Covers: entity CRUD, triple CRUD, temporal queries, invalidation,
timeline, stats, and edge cases (duplicate triples, ID collisions). timeline, stats, and edge cases (duplicate triples, ID collisions).
""" """
import pytest
class TestEntityOperations: class TestEntityOperations:
def test_add_entity(self, kg): def test_add_entity(self, kg):
@@ -45,6 +47,38 @@ class TestTripleOperations:
tid2 = kg.add_triple("Alice", "works_at", "Acme") tid2 = kg.add_triple("Alice", "works_at", "Acme")
assert tid1 != tid2 # new triple since old one was closed assert tid1 != tid2 # new triple since old one was closed
def test_add_triple_rejects_inverted_interval(self, kg):
# valid_to before valid_from would never satisfy
# `valid_from <= as_of AND valid_to >= as_of` — silently invisible
# to every query. Reject at write time instead.
with pytest.raises(ValueError, match="before valid_from"):
kg.add_triple(
"Alice",
"worked_at",
"Acme",
valid_from="2026-03-01",
valid_to="2026-02-01",
)
def test_add_triple_accepts_equal_dates(self, kg):
# Same-day intervals are valid (point-in-time facts).
tid = kg.add_triple(
"Alice",
"joined",
"Acme",
valid_from="2026-03-15",
valid_to="2026-03-15",
)
assert tid.startswith("t_alice_joined_acme_")
def test_add_triple_allows_only_one_bound(self, kg):
# The guard only fires when BOTH bounds are set.
tid1 = kg.add_triple("Alice", "knows", "Bob", valid_from="2026-01-01")
assert tid1.startswith("t_alice_knows_bob_")
kg.invalidate("Alice", "knows", "Bob", ended="2026-02-01")
tid2 = kg.add_triple("Alice", "knew", "Bob", valid_to="2026-03-01")
assert tid2.startswith("t_alice_knew_bob_")
class TestQueries: class TestQueries:
def test_query_outgoing(self, seeded_kg): def test_query_outgoing(self, seeded_kg):