SmartMemory
Concepts

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

DimensionSemantics
Valid timeWhen the fact is or was true in the real world
Transaction timeWhen 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.*:

EndpointSDK methodReturns
GET /memory/temporal/{id}/historygetHistory(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}/changesgetChanges(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}/rollbackrollback(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_start in the past when ingesting historical records.
  • Belief revision: when a fact changes, the old version stays queryable.
  • Diff and blame: compareVersions shows 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 tight since/until window over getHistory() with no bound.
  • as_of_date over supersession needs the old record's creation time. In lite examples, preserve it with metadata["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: TemporalQueries
  • smartmemory/temporal/version_tracker.py: version writes
  • smartmemory/temporal/relationships.py: bi-temporal edges
  • memory_service/api/routes/temporal.py: REST endpoints

On this page