Bi-Temporal Versioning
Bi-Temporal Versioning & Time-Travel Queries
SmartMemory tracks two independent time dimensions for memory items, so you can answer both "what did we believe was true on date X?" and "what did the system have stored on date X?".
The two times
| Dimension | Semantics |
|---|---|
| Valid time | When the fact is or was true in the real world |
| Transaction time | When the system recorded the version |
Each version is represented by the real TemporalVersion dataclass:
from dataclasses import fields
from smartmemory.temporal import TemporalVersion
print([field.name for field in fields(TemporalVersion)])Source: smart-memory-core/smartmemory/temporal/queries.py.
A version's transaction-time interval [transaction_time_start, transaction_time_end) is closed on the left and exclusive on the right. The current version has transaction_time_end = None. The same shape applies to valid time: an open valid_time_end means "still believed to be true."
How versions are created
Manual supersession records that one memory replaces another. The default search path returns the current value only. Passing include_superseded=True returns the old and new values.
as_of_date depends on a preserved creation time for the old value. In lite mode, put that time in metadata={"reference_time": <iso datetime>} before superseding. Without reference_time, the old value may be anchored at supersession time, so a query for an earlier moment returns the new value.
Querying current state and history
from smartmemory.models.memory_item import MemoryItem
from smartmemory.pipeline.config import PipelineConfig
from smartmemory.tools.factory import lite_context
lite = dict(pipeline_profile=PipelineConfig.lite(llm_enabled=False))
with lite_context(**lite) as memory:
old = memory.add(
MemoryItem(
content="The incident channel is #ops-old",
memory_type="semantic",
)
)
new = memory.add(
MemoryItem(
content="The incident channel is #ops-war-room",
memory_type="semantic",
)
)
memory.supersede(old, new, reason="Team renamed the channel")
print([r.content for r in memory.search("incident channel")])
print([
r.content
for r in memory.search("incident channel", include_superseded=True)
])Querying what was believed before a change
from datetime import datetime, timezone
from smartmemory.models.memory_item import MemoryItem
from smartmemory.pipeline.config import PipelineConfig
from smartmemory.tools.factory import lite_context
lite = dict(pipeline_profile=PipelineConfig.lite(llm_enabled=False))
with lite_context(**lite) as memory:
old = memory.add(
MemoryItem(
content="The billing provider is Stripe",
memory_type="semantic",
metadata={
"reference_time": datetime.now(timezone.utc).isoformat(),
},
)
)
t_before_change = datetime.now(timezone.utc)
new = memory.add(
MemoryItem(
content="The billing provider is Paddle",
memory_type="semantic",
metadata={
"reference_time": datetime.now(timezone.utc).isoformat(),
},
)
)
memory.supersede(old, new, reason="Pricing model changed")
historical = memory.search(
"billing provider",
as_of_date=t_before_change,
include_superseded=True,
)
current = memory.search("billing provider")
print([r.content for r in historical])
print([r.content for r in current])Service query API
Service routes live under /memory/temporal/* and are wired in smart-memory-service/memory_service/api/routes/temporal.py. The JS SDK mirrors them on client.memories.*:
| Endpoint | SDK method | Returns |
|---|---|---|
GET /memory/temporal/{id}/history | getHistory(id, opts) | All versions of id |
GET /memory/temporal/at/{timestamp} | timeTravel(ts) | The state of every item at ts |
GET /memory/temporal/{id}/at/{timestamp} | getItemAtTime(id, ts) | The version of id valid at ts |
GET /memory/temporal/{id}/changes | getChanges(id, opts) | Change events between since and until |
POST /memory/temporal/{id}/compare?v1=&v2= | compareVersions(id, ...) | Field-level diff between two versions |
POST /memory/temporal/{id}/rollback | rollback(id, opts) | Replay a past version as the new HEAD |
GET /memory/temporal/{id}/audit | (REST only) | Full audit trail with provenance |
GET /memory/temporal/search/during | (REST only) | All items valid during a window |
GET /memory/temporal/compliance/report | (REST only) | Aggregate retention and change report |
GET /memory/temporal/relationships/{rel_id}/history etc. | (REST only) | Bi-temporal queries on edges |
JS SDK examples
const past = await client.memories.timeTravel('2024-09-01T00:00:00Z', {
limit: 100,
});const diff = await client.memories.compareVersions('item123', 3, 5);await client.memories.rollback('item123', { toVersion: 3 });
await client.memories.rollback('item123', { toTime: '2024-09-15T12:00:00Z' });A rollback writes a new version with the chosen prior content. It does not delete the intervening versions, so the audit trail stays intact.
Use cases
- Audit and compliance: answer "what did this user see on date X" exactly.
- Backdated facts: set
valid_time_startin the past when ingesting historical records. - Belief revision: when a fact changes, the old version stays queryable.
- Diff and blame:
compareVersionsshows which field changed between any two versions.
Caveats
- Time-travel queries iterate version chains in FalkorDB. For items with hundreds of versions, prefer
getChanges()with a tightsince/untilwindow overgetHistory()with no bound. as_of_dateover supersession needs the old record's creation time. In lite examples, preserve it withmetadata["reference_time"].- Rollback is best-effort for relations. Edges follow the same bi-temporal model, but rollback only restores node properties.
Key code surfaces
smartmemory/temporal/queries.py:TemporalQueriessmartmemory/temporal/version_tracker.py: version writessmartmemory/temporal/relationships.py: bi-temporal edgesmemory_service/api/routes/temporal.py: REST endpoints