fix(kg): reject inverted intervals in add_triple (valid_to < valid_from)

A triple with valid_to < valid_from satisfies neither of the temporal
filter clauses in query_entity():

    valid_from <= as_of AND valid_to >= as_of

so the triple is invisible to every query — silently corrupt. Reject
at write time with a clear error instead of letting bad data pile up
in the SQLite store.

The guard only fires when both bounds are present; open intervals
(only valid_from or only valid_to) are still accepted, and same-day
intervals (valid_from == valid_to, point-in-time facts) are explicitly
allowed.
This commit is contained in:
Arnold Wender
2026-04-26 13:00:27 +02:00
parent 6890948e09
commit 0b8c2c158f
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):