ae6f3e9a95
Implement Phase 28: Sovereign Knowledge Graph 'Time Travel' - agent/temporal_knowledge_graph.py: SQLite-backed temporal triple store with versioning, validity periods, and temporal query operators (BEFORE, AFTER, DURING, OVERLAPS, AT) - agent/temporal_reasoning.py: Temporal reasoning engine supporting historical queries, fact evolution tracking, and worldview snapshots - tools/temporal_kg_tool.py: Tool integration with functions for storing facts with time, querying historical state, generating temporal summaries, and natural language temporal queries - tests/test_temporal_kg.py: Comprehensive test coverage including storage tests, query operators, historical summaries, and integration tests
474 lines
18 KiB
Python
474 lines
18 KiB
Python
"""Tests for Temporal Knowledge Graph implementation.
|
|
|
|
Tests cover:
|
|
- Temporal storage tests
|
|
- Query operator tests (BEFORE, AFTER, DURING, OVERLAPS)
|
|
- Historical summary tests
|
|
- Integration with tools
|
|
"""
|
|
|
|
import pytest
|
|
import tempfile
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from agent.temporal_knowledge_graph import (
|
|
TemporalTripleStore, TemporalTriple, TemporalOperator
|
|
)
|
|
from agent.temporal_reasoning import (
|
|
TemporalReasoner, ChangeType, HistoricalSummary
|
|
)
|
|
from tools.temporal_kg_tool import (
|
|
store_fact_with_time,
|
|
query_historical_state,
|
|
get_fact_history,
|
|
generate_temporal_summary,
|
|
when_did_we_learn,
|
|
how_has_it_changed,
|
|
query_with_temporal_operator,
|
|
get_worldview_at_time
|
|
)
|
|
|
|
|
|
class TestTemporalTripleStore:
|
|
"""Tests for the TemporalTripleStore class."""
|
|
|
|
@pytest.fixture
|
|
def store(self):
|
|
"""Create a temporary store for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
store = TemporalTripleStore(db_path)
|
|
yield store
|
|
# Cleanup
|
|
os.unlink(db_path)
|
|
|
|
def test_store_fact(self, store):
|
|
"""Test storing a basic fact."""
|
|
triple = store.store_fact("Timmy", "has_feature", "sovereignty")
|
|
|
|
assert triple.subject == "Timmy"
|
|
assert triple.predicate == "has_feature"
|
|
assert triple.object == "sovereignty"
|
|
assert triple.version == 1
|
|
assert triple.valid_until is None
|
|
|
|
def test_store_fact_with_validity_period(self, store):
|
|
"""Test storing a fact with validity bounds."""
|
|
valid_from = "2026-01-01T00:00:00"
|
|
valid_until = "2026-12-31T23:59:59"
|
|
|
|
triple = store.store_fact(
|
|
"Hermes",
|
|
"status",
|
|
"active",
|
|
valid_from=valid_from,
|
|
valid_until=valid_until
|
|
)
|
|
|
|
assert triple.valid_from == valid_from
|
|
assert triple.valid_until == valid_until
|
|
|
|
def test_fact_versioning(self, store):
|
|
"""Test that facts are properly versioned."""
|
|
# Store initial fact
|
|
triple1 = store.store_fact("Timmy", "version", "1.0")
|
|
assert triple1.version == 1
|
|
|
|
# Store updated fact
|
|
triple2 = store.store_fact("Timmy", "version", "2.0")
|
|
assert triple2.version == 2
|
|
|
|
# Check that first fact was superseded
|
|
history = store.get_fact_history("Timmy", "version")
|
|
assert len(history) == 2
|
|
assert history[0].superseded_by == triple2.id
|
|
|
|
def test_query_at_time(self, store):
|
|
"""Test querying facts at a specific time."""
|
|
# Store facts at different times
|
|
store.store_fact("Timmy", "status", "alpha", valid_from="2026-01-01T00:00:00")
|
|
store.store_fact("Timmy", "status", "beta", valid_from="2026-03-01T00:00:00")
|
|
store.store_fact("Timmy", "status", "stable", valid_from="2026-06-01T00:00:00")
|
|
|
|
# Query at different points
|
|
feb_facts = store.query_at_time("2026-02-01T00:00:00", subject="Timmy")
|
|
assert len(feb_facts) == 1
|
|
assert feb_facts[0].object == "alpha"
|
|
|
|
may_facts = store.query_at_time("2026-05-01T00:00:00", subject="Timmy")
|
|
assert len(may_facts) == 1
|
|
assert may_facts[0].object == "beta"
|
|
|
|
jul_facts = store.query_at_time("2026-07-01T00:00:00", subject="Timmy")
|
|
assert len(jul_facts) == 1
|
|
assert jul_facts[0].object == "stable"
|
|
|
|
def test_query_temporal_operators(self, store):
|
|
"""Test temporal query operators."""
|
|
# Store some facts
|
|
store.store_fact("A", "rel", "1", valid_from="2026-01-01T00:00:00")
|
|
store.store_fact("B", "rel", "2", valid_from="2026-03-01T00:00:00")
|
|
store.store_fact("C", "rel", "3", valid_from="2026-06-01T00:00:00")
|
|
|
|
# Test BEFORE
|
|
before_april = store.query_temporal(
|
|
TemporalOperator.BEFORE, "2026-04-01T00:00:00"
|
|
)
|
|
assert len(before_april) == 2 # A and B
|
|
|
|
# Test AFTER
|
|
after_feb = store.query_temporal(
|
|
TemporalOperator.AFTER, "2026-02-01T00:00:00"
|
|
)
|
|
assert len(after_feb) == 2 # B and C
|
|
|
|
# Test DURING (at a specific time)
|
|
during_may = store.query_temporal(
|
|
TemporalOperator.DURING, "2026-05-01T00:00:00"
|
|
)
|
|
assert len(during_may) == 1 # Only B is valid in May
|
|
assert during_may[0].object == "2"
|
|
|
|
def test_get_fact_history(self, store):
|
|
"""Test retrieving fact version history."""
|
|
# Create multiple versions
|
|
store.store_fact("Feature", "status", "planned", valid_from="2026-01-01T00:00:00")
|
|
store.store_fact("Feature", "status", "in_progress", valid_from="2026-02-01T00:00:00")
|
|
store.store_fact("Feature", "status", "completed", valid_from="2026-03-01T00:00:00")
|
|
|
|
history = store.get_fact_history("Feature", "status")
|
|
|
|
assert len(history) == 3
|
|
assert history[0].object == "planned"
|
|
assert history[1].object == "in_progress"
|
|
assert history[2].object == "completed"
|
|
|
|
# Check versions
|
|
assert history[0].version == 1
|
|
assert history[1].version == 2
|
|
assert history[2].version == 3
|
|
|
|
def test_get_entity_changes(self, store):
|
|
"""Test getting entity changes in a time range."""
|
|
store.store_fact("Codebase", "feature", "auth", valid_from="2026-01-01T00:00:00")
|
|
store.store_fact("Codebase", "feature", "logging", valid_from="2026-02-01T00:00:00")
|
|
store.store_fact("Codebase", "feature", "metrics", valid_from="2026-03-01T00:00:00")
|
|
|
|
changes = store.get_entity_changes(
|
|
"Codebase",
|
|
"2026-01-15T00:00:00",
|
|
"2026-03-15T00:00:00"
|
|
)
|
|
|
|
# Should include logging and metrics
|
|
assert len(changes) >= 2
|
|
|
|
def test_export_import(self, store):
|
|
"""Test exporting and importing data."""
|
|
# Store some data
|
|
store.store_fact("Test", "data", "value1")
|
|
store.store_fact("Test", "data", "value2")
|
|
|
|
# Export
|
|
json_data = store.export_to_json()
|
|
assert "Test" in json_data
|
|
assert "value1" in json_data
|
|
assert "value2" in json_data
|
|
|
|
# Create new store and import
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path2 = f.name
|
|
|
|
try:
|
|
store2 = TemporalTripleStore(db_path2)
|
|
store2.import_from_json(json_data)
|
|
|
|
# Verify imported data
|
|
facts = store2.query_at_time(datetime.now().isoformat(), subject="Test")
|
|
assert len(facts) >= 1
|
|
finally:
|
|
os.unlink(db_path2)
|
|
|
|
|
|
class TestTemporalReasoner:
|
|
"""Tests for the TemporalReasoner class."""
|
|
|
|
@pytest.fixture
|
|
def reasoner(self):
|
|
"""Create a temporary reasoner for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
store = TemporalTripleStore(db_path)
|
|
reasoner = TemporalReasoner(store)
|
|
yield reasoner
|
|
os.unlink(db_path)
|
|
|
|
def test_what_did_we_believe(self, reasoner):
|
|
"""Test "what did we believe" queries."""
|
|
# Set up facts
|
|
reasoner.store.store_fact("Timmy", "view", "optimistic", valid_from="2026-01-01T00:00:00")
|
|
reasoner.store.store_fact("Timmy", "view", "cautious", valid_from="2026-03-01T00:00:00")
|
|
|
|
# Query before March
|
|
beliefs = reasoner.what_did_we_believe("Timmy", "2026-02-15T00:00:00")
|
|
assert len(beliefs) == 1
|
|
assert beliefs[0].object == "optimistic"
|
|
|
|
def test_when_did_we_learn(self, reasoner):
|
|
"""Test "when did we learn" queries."""
|
|
timestamp = "2026-02-15T10:30:00"
|
|
reasoner.store.store_fact(
|
|
"MLX",
|
|
"integrated_with",
|
|
"Hermes",
|
|
valid_from=timestamp
|
|
)
|
|
|
|
when = reasoner.when_did_we_learn("MLX", "integrated_with")
|
|
assert when == timestamp
|
|
|
|
def test_how_has_it_changed(self, reasoner):
|
|
"""Test "how has it changed" queries."""
|
|
reasoner.store.store_fact("Security", "level", "low", valid_from="2026-01-01T00:00:00")
|
|
reasoner.store.store_fact("Security", "level", "medium", valid_from="2026-02-01T00:00:00")
|
|
reasoner.store.store_fact("Security", "level", "high", valid_from="2026-03-01T00:00:00")
|
|
|
|
changes = reasoner.how_has_it_changed("Security", "2026-01-15T00:00:00")
|
|
|
|
assert len(changes) >= 2
|
|
# Check that changes are properly categorized
|
|
change_types = [c.change_type for c in changes]
|
|
assert ChangeType.MODIFIED in change_types or ChangeType.ADDED in change_types
|
|
|
|
def test_generate_temporal_summary(self, reasoner):
|
|
"""Test generating historical summaries."""
|
|
# Create a history of changes
|
|
reasoner.store.store_fact("Project", "status", "planning", valid_from="2026-01-01T00:00:00")
|
|
reasoner.store.store_fact("Project", "status", "development", valid_from="2026-02-01T00:00:00")
|
|
reasoner.store.store_fact("Project", "milestone", "alpha", valid_from="2026-02-15T00:00:00")
|
|
reasoner.store.store_fact("Project", "status", "testing", valid_from="2026-03-01T00:00:00")
|
|
|
|
summary = reasoner.generate_temporal_summary(
|
|
"Project",
|
|
"2026-01-01T00:00:00",
|
|
"2026-04-01T00:00:00"
|
|
)
|
|
|
|
assert summary.entity == "Project"
|
|
assert summary.total_changes >= 3
|
|
assert len(summary.evolution_timeline) >= 3
|
|
assert len(summary.current_state) >= 1
|
|
|
|
def test_get_worldview_at_time(self, reasoner):
|
|
"""Test getting complete worldview at a time."""
|
|
reasoner.store.store_fact("Timmy", "mood", "happy", valid_from="2026-01-01T00:00:00")
|
|
reasoner.store.store_fact("Timmy", "task", "coding", valid_from="2026-01-01T00:00:00")
|
|
reasoner.store.store_fact("Hermes", "status", "active", valid_from="2026-01-01T00:00:00")
|
|
|
|
worldview = reasoner.get_worldview_at_time("2026-01-15T00:00:00")
|
|
|
|
assert "Timmy" in worldview
|
|
assert "Hermes" in worldview
|
|
assert len(worldview["Timmy"]) == 2
|
|
|
|
def test_infer_temporal_relationship(self, reasoner):
|
|
"""Test temporal relationship inference."""
|
|
triple_a = reasoner.store.store_fact("A", "rel", "1", valid_from="2026-01-01T00:00:00")
|
|
triple_a.valid_until = "2026-02-01T00:00:00"
|
|
|
|
triple_b = reasoner.store.store_fact("B", "rel", "2", valid_from="2026-02-15T00:00:00")
|
|
|
|
rel = reasoner.infer_temporal_relationship(triple_a, triple_b)
|
|
assert "before" in rel.lower()
|
|
|
|
|
|
class TestTemporalKGTools:
|
|
"""Tests for the temporal KG tool functions."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_singleton(self):
|
|
"""Reset singleton instances before each test."""
|
|
import tools.temporal_kg_tool as tool_module
|
|
tool_module._store = None
|
|
tool_module._reasoner = None
|
|
yield
|
|
tool_module._store = None
|
|
tool_module._reasoner = None
|
|
|
|
def test_store_fact_with_time(self):
|
|
"""Test the store_fact_with_time tool function."""
|
|
result = store_fact_with_time(
|
|
subject="Hermes Agent",
|
|
predicate="has_feature",
|
|
object="input_sanitizer",
|
|
valid_from="2026-04-01T01:00:00"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["triple"]["subject"] == "Hermes Agent"
|
|
assert result["triple"]["predicate"] == "has_feature"
|
|
assert result["triple"]["object"] == "input_sanitizer"
|
|
|
|
def test_query_historical_state(self):
|
|
"""Test the query_historical_state tool function."""
|
|
# Store a fact first
|
|
store_fact_with_time(
|
|
subject="Timmy",
|
|
predicate="view_on_sovereignty",
|
|
object="strong",
|
|
valid_from="2026-02-01T00:00:00"
|
|
)
|
|
|
|
# Query it
|
|
result = query_historical_state("Timmy", "2026-03-01T00:00:00")
|
|
|
|
assert result["success"] is True
|
|
assert result["subject"] == "Timmy"
|
|
assert result["fact_count"] == 1
|
|
assert result["facts"][0]["object"] == "strong"
|
|
|
|
def test_get_fact_history(self):
|
|
"""Test the get_fact_history tool function."""
|
|
# Create version history
|
|
store_fact_with_time("Feature", "status", "planned", valid_from="2026-01-01T00:00:00")
|
|
store_fact_with_time("Feature", "status", "done", valid_from="2026-02-01T00:00:00")
|
|
|
|
result = get_fact_history("Feature", "status")
|
|
|
|
assert result["success"] is True
|
|
assert result["version_count"] == 2
|
|
assert len(result["versions"]) == 2
|
|
|
|
def test_when_did_we_learn(self):
|
|
"""Test the when_did_we_learn tool function."""
|
|
store_fact_with_time(
|
|
"MLX",
|
|
"integrated_with",
|
|
"Hermes",
|
|
valid_from="2026-03-15T12:00:00"
|
|
)
|
|
|
|
result = when_did_we_learn("MLX", "integrated_with")
|
|
|
|
assert result["success"] is True
|
|
assert result["first_known"] == "2026-03-15T12:00:00"
|
|
|
|
def test_how_has_it_changed(self):
|
|
"""Test the how_has_it_changed tool function."""
|
|
store_fact_with_time("Codebase", "feature_count", "10", valid_from="2026-01-01T00:00:00")
|
|
store_fact_with_time("Codebase", "feature_count", "20", valid_from="2026-02-01T00:00:00")
|
|
|
|
result = how_has_it_changed("Codebase", "2026-01-15T00:00:00")
|
|
|
|
assert result["success"] is True
|
|
assert result["change_count"] >= 1
|
|
|
|
def test_query_with_temporal_operator(self):
|
|
"""Test the query_with_temporal_operator tool function."""
|
|
store_fact_with_time("A", "rel", "1", valid_from="2026-01-01T00:00:00")
|
|
store_fact_with_time("B", "rel", "2", valid_from="2026-03-01T00:00:00")
|
|
|
|
result = query_with_temporal_operator("BEFORE", "2026-02-01T00:00:00")
|
|
|
|
assert result["success"] is True
|
|
assert result["fact_count"] == 1
|
|
assert result["facts"][0]["subject"] == "A"
|
|
|
|
def test_get_worldview_at_time(self):
|
|
"""Test the get_worldview_at_time tool function."""
|
|
store_fact_with_time("Timmy", "mood", "good", valid_from="2026-01-01T00:00:00")
|
|
store_fact_with_time("Hermes", "status", "running", valid_from="2026-01-01T00:00:00")
|
|
|
|
result = get_worldview_at_time("2026-01-15T00:00:00")
|
|
|
|
assert result["success"] is True
|
|
assert result["entity_count"] == 2
|
|
|
|
def test_generate_temporal_summary(self):
|
|
"""Test the generate_temporal_summary tool function."""
|
|
store_fact_with_time("Security", "level", "low", valid_from="2026-01-01T00:00:00")
|
|
store_fact_with_time("Security", "level", "high", valid_from="2026-03-01T00:00:00")
|
|
|
|
result = generate_temporal_summary("Security", "2026-01-01T00:00:00", "2026-04-01T00:00:00")
|
|
|
|
assert result["success"] is True
|
|
assert result["entity"] == "Security"
|
|
assert result["summary"]["total_changes"] >= 1
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration tests for the complete temporal KG system."""
|
|
|
|
@pytest.fixture
|
|
def system(self):
|
|
"""Create a complete temporal KG system."""
|
|
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
|
|
db_path = f.name
|
|
|
|
store = TemporalTripleStore(db_path)
|
|
reasoner = TemporalReasoner(store)
|
|
|
|
yield {"store": store, "reasoner": reasoner}
|
|
|
|
os.unlink(db_path)
|
|
|
|
def test_full_workflow(self, system):
|
|
"""Test a complete temporal knowledge workflow."""
|
|
store = system["store"]
|
|
reasoner = system["reasoner"]
|
|
|
|
# 1. Store initial facts about a security audit
|
|
store.store_fact("SecurityAudit", "status", "scheduled", valid_from="2026-01-01T00:00:00")
|
|
store.store_fact("SecurityAudit", "auditor", "ExternalFirm", valid_from="2026-01-01T00:00:00")
|
|
|
|
# 2. Update as audit progresses
|
|
store.store_fact("SecurityAudit", "status", "in_progress", valid_from="2026-02-01T00:00:00")
|
|
store.store_fact("SecurityAudit", "findings", "none_yet", valid_from="2026-02-01T00:00:00")
|
|
|
|
# 3. Complete audit
|
|
store.store_fact("SecurityAudit", "status", "completed", valid_from="2026-03-01T00:00:00")
|
|
store.store_fact("SecurityAudit", "findings", "5_minor_issues", valid_from="2026-03-01T00:00:00")
|
|
store.store_fact("SecurityAudit", "recommendation", "address_within_30_days", valid_from="2026-03-01T00:00:00")
|
|
|
|
# 4. Query historical state
|
|
jan_state = reasoner.get_worldview_at_time("2026-01-15T00:00:00", ["SecurityAudit"])
|
|
assert jan_state["SecurityAudit"][0]["predicate"] == "status"
|
|
assert jan_state["SecurityAudit"][0]["object"] == "scheduled"
|
|
|
|
feb_state = reasoner.get_worldview_at_time("2026-02-15T00:00:00", ["SecurityAudit"])
|
|
status_fact = [f for f in feb_state["SecurityAudit"] if f["predicate"] == "status"][0]
|
|
assert status_fact["object"] == "in_progress"
|
|
|
|
# 5. Generate summary
|
|
summary = reasoner.generate_temporal_summary(
|
|
"SecurityAudit",
|
|
"2026-01-01T00:00:00",
|
|
"2026-04-01T00:00:00"
|
|
)
|
|
|
|
assert summary.total_changes >= 5
|
|
assert any(f["predicate"] == "status" for f in summary.key_facts)
|
|
|
|
# 6. Check when we learned about findings
|
|
when = reasoner.when_did_we_learn("SecurityAudit", "findings")
|
|
assert when is not None
|
|
|
|
def test_temporal_inference(self, system):
|
|
"""Test temporal inference capabilities."""
|
|
store = system["store"]
|
|
reasoner = system["reasoner"]
|
|
|
|
# Store facts with temporal relationships
|
|
triple_a = store.store_fact("EventA", "happened", "yes", valid_from="2026-01-01T00:00:00")
|
|
triple_a.valid_until = "2026-01-31T23:59:59"
|
|
|
|
triple_b = store.store_fact("EventB", "happened", "yes", valid_from="2026-02-01T00:00:00")
|
|
|
|
# Infer relationship
|
|
rel = reasoner.infer_temporal_relationship(triple_a, triple_b)
|
|
assert "before" in rel.lower()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|