Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions mempalace/knowledge_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,15 @@ def add_triple(
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")
"""
# 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)
obj_id = self._entity_id(obj)
pred = predicate.lower().replace(" ", "_")
Expand Down
34 changes: 34 additions & 0 deletions tests/test_knowledge_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
timeline, stats, and edge cases (duplicate triples, ID collisions).
"""

import pytest


class TestEntityOperations:
def test_add_entity(self, kg):
Expand Down Expand Up @@ -45,6 +47,38 @@ def test_invalidated_triple_allows_re_add(self, kg):
tid2 = kg.add_triple("Alice", "works_at", "Acme")
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:
def test_query_outgoing(self, seeded_kg):
Expand Down