Bitemporal Versioning

The BiTemporalStore provides PostgreSQL + TimescaleDB storage with bi-temporal semantics for the Context Graph. It tracks both when facts were true in the real world (valid_time) and when they were recorded in the system (transaction_time), enabling point-in-time queries, historical auditing, and decision versioning.

Overview

Bi-temporal data management is critical for enterprise AI platforms where regulatory compliance, auditability, and the ability to answer "what did we know and when did we know it" are requirements.

Source: data-plane/ai-service/src/context_graph/storage/bitemporal_store.py

Temporal Dimensions

Dimension	Column	Meaning
Valid Time	`valid_time` / `valid_time_start`	When the fact was true in the real world
Transaction Time	`transaction_time` / `transaction_time_start`	When the fact was recorded in the system

Query Types

Query Type	Description	Example
`CURRENT`	Latest known state	"What is the current schema for this dataset?"
`AS_OF_VALID`	State at a specific real-world time	"What was the model accuracy on January 1st?"
`AS_OF_TRANSACTION`	State as known at a specific system time	"What did we know about this entity at noon yesterday?"
`BITEMPORAL`	Both valid and transaction time specified	"What was the entity state as of Jan 1 according to what we knew on Feb 1?"
`HISTORY`	Full version history	"Show all changes to this entity"

Schema

Context Graph Events Table

CREATE TABLE context_graph_events (
    event_id UUID PRIMARY KEY,
    event_type VARCHAR(100) NOT NULL,
    entity_urn VARCHAR(500) NOT NULL,
    entity_type VARCHAR(100) NOT NULL,
    valid_time TIMESTAMPTZ NOT NULL,
    transaction_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    actor_type VARCHAR(50) NOT NULL,
    actor_id VARCHAR(200) NOT NULL,
    context JSONB NOT NULL,
    outcome JSONB,
    lineage JSONB,
    facets JSONB,
    version INT NOT NULL DEFAULT 1,
    previous_version_id UUID,
    is_current BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    created_by VARCHAR(200)
);

Decisions Table

CREATE TABLE decisions (
    decision_id UUID PRIMARY KEY,
    decision_urn VARCHAR(500) UNIQUE NOT NULL,
    decision_type VARCHAR(100) NOT NULL,
    subject_urn VARCHAR(500) NOT NULL,
    tenant_id VARCHAR(100) NOT NULL,
    decision_time TIMESTAMPTZ NOT NULL,
    valid_time TIMESTAMPTZ NOT NULL,
    transaction_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    initiated_by JSONB NOT NULL,
    decided_by JSONB NOT NULL,
    approved_by JSONB,
    choice JSONB NOT NULL,
    rationale JSONB NOT NULL,
    outcome JSONB,
    embedding_id VARCHAR(200),
    version INT NOT NULL DEFAULT 1,
    is_current BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Entity Operations

Store an Entity

entity_id = await store.store_entity(
    entity_urn="urn:matih:dataset:acme:sales",
    entity_type="dataset",
    tenant_id="acme",
    data={"name": "sales", "schema": {...}},
    valid_from=datetime(2025, 1, 15),
    created_by="admin@acme.com",
)

When an entity already exists, the store supersedes the previous version by setting transaction_time_end on the old record and inserting a new record with an incremented version number.

Get Entity at Point in Time

entity = await store.get_entity(
    entity_urn="urn:matih:dataset:acme:sales",
    valid_time=datetime(2025, 6, 1),
    transaction_time=datetime(2025, 7, 1),
)

Get Entity Version History

history = await store.get_entity_history(
    entity_urn="urn:matih:dataset:acme:sales",
    tenant_id="acme",
    limit=100,
)

Returns a VersionHistory object with all versions, first/latest timestamps, and the current version number.

Decision Operations

Store a Decision

decision_id = await store.store_decision(decision)

Uses upsert semantics on decision_urn to handle updates.

Query Decisions

decisions = await store.query_decisions(
    DecisionQuery(
        tenant_id="acme",
        decision_types=[DecisionType.MODEL_DEPLOYMENT],
        start_time=datetime(2025, 1, 1),
        limit=20,
    )
)

Search Precedent Decisions

precedents = await store.search_precedents(
    tenant_id="acme",
    decision_type="model_deployment",
    characteristics={"model_type": "classification"},
    limit=5,
)

TimescaleDB Integration

When TimescaleDB is available, the events table is converted to a hypertable partitioned by transaction_time for improved time-series query performance. The store gracefully falls back to standard PostgreSQL if TimescaleDB is not installed.

Indexing Strategy

Index	Table	Purpose
`idx_events_entity_urn`	`context_graph_events`	Entity lookup
`idx_events_bitemporal`	`context_graph_events`	Bi-temporal queries
`idx_events_current`	`context_graph_events`	Current state (partial)
`idx_decisions_tenant`	`decisions`	Tenant-scoped queries
`idx_decisions_bitemporal`	`decisions`	Bi-temporal decision queries
`idx_entities_current`	`entities`	Current entity versions

Health Check

health = await store.health_check()
# {"backend": "postgresql", "status": "healthy", "mode": "live", "timescale_enabled": true}

Vector Stores (Pinecone/Qdrant)Search Capabilities Overview