Temporal Data Lineage for Auditing

Destructive updates in traditional relational databases overwrite history, causing immediate spoliation of evidence during regulatory discovery. When an auditor queries a system state, standard log files provide fragmented temporal approximations rather than deterministic database snapshots.

We enforce deterministic data lineage across four statutory domain environments where automated decisions intersect with legal liability:

• Financial AML & Fraud Detection: Under regulatory frameworks such as the EU Anti-Money Laundering Directives and the US BSA/Patriot Act, financial institutions must prove the precise historical entity relationships and transaction context present when an automated Suspicious Activity Report (SAR) was triggered or suppressed.
• Government Registries & Real Estate: Automating beneficial ownership tracking under statutory mandates—such as the German Transparency Register (Transparenzregister) governed by the Money Laundering Act (Geldwäschegesetz – GwG)—requires an unalterable history of ownership chain transformations across decades.
• Healthcare & Life Sciences: AI-driven clinical decision support tools and automated diagnostic pipelines regulated under FDA and EMA guidelines must trace every generated output back to the exact version of the medical literature and patient state active at the microsecond of care delivery.
• RegTech & AI Legal Synthesis: When Retrieval-Augmented Generation (RAG) systems ingest and synthesize regulatory guidance for automated contract drafting or compliance checking, organizations must prove the LLM retrieved verified, un-tampered document snapshots valid at transaction time.

flowchart TD
    Auditor["Client / Legal Auditor"] --> FastAPI["FastAPI Layer<br/>(Point-in-Time APIs)"]
    FastAPI --> Validation["Pydantic Validation & Audit Context<br/>(User ID, Legal Basis, Correlation ID)"]
    
    subgraph Postgres["PostgreSQL Engine & Vector Layer"]
        Active["Active Knowledge Chunks<br/><code>[chunk_id, sys_period, vec]</code>"]
        History["Historical Chunk Archive<br/><code>[chunk_id, sys_period, vec]</code>"]
        Ledger["Cryptographic Hash Chain<br/>(WAL / pg_audit / Ledger)"]
        
        Active -- "PL/pgSQL Trigger / WAL" --> History
    end
    
    FastAPI --> Active

Layer 1: System-Versioned Temporal Logging (sys_period) in Relational Stores

Updating a vector index or replacing a chunk in a knowledge base without temporal tracing creates an un-auditable system state. If an AI agent provides compliance advice based on a document chunk at $T_1$, and that chunk is re-embedded or modified at $T_2$, traditional databases overwrite the past. During legal discovery, the system loses the ability to prove what knowledge context the model retrieved.

By implementing sys_period logging on knowledge stores, time becomes an explicit dimension within the database engine itself.

Half-Open Temporal Ranges (tstzrange)

Instead of tracking a single updated_at timestamp, every document chunk, system prompt version, and embedding reference contains a sys_period column defined as a PostgreSQL tstzrange (timestamp with time zone range).

This captures the continuous interval during which a specific knowledge state was active:

We define interval boundaries using four exact rules:

• Inclusive lower bound ($t_{\text{start}}$): The microsecond transaction timestamp (transaction_timestamp()) generated when the record was ingested or modified.
• Exclusive upper bound ($t_{\text{end}}$): The microsecond transaction timestamp generated when the record was superseded, soft-deleted, or re-embedded.
• Unbounded upper state: Active records available for live retrieval carry an upper bound of infinity.
• Half-open boundary isolation: Representing bounds as $[t_{\text{start}}, t_{\text{end}})$ prevents duplicate retrievals at boundary microseconds. If Version 1 closes at 08:30:00.000 and Version 2 opens at 08:30:00.000, an automated retrieval query executed at 08:30:00.000 evaluates to exactly one record state.

Active Chunk Version:     [ 2026-01-15 10:00:00+00,  infinity )
Deprecated Chunk Version: [ 2026-01-15 10:00:00+00,  2026-06-01 08:30:00+00 )

Temporal Exclusion Constraints

Modifying a document chunk without strict concurrency controls causes race conditions and overlapping temporal ranges. In a temporal framework, uniqueness is enforced across both entity identity and valid time windows.

We configure PostgreSQL Exclusion Constraints backed by Generalized Search Tree (GiST) indexes via the btree_gist extension:

CREATE EXTENSION IF NOT EXISTS btree_gist;
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE rag_knowledge_chunks (
    chunk_id UUID NOT NULL,
    document_id UUID NOT NULL,
    chunk_index INT NOT NULL,
    chunk_content TEXT NOT NULL,
    embedding vector(1536),
    metadata JSONB DEFAULT '{}'::jsonb,
    sys_period TSTZRANGE NOT NULL DEFAULT tstzrange(transaction_timestamp(), 'infinity', '[)'),
    
    PRIMARY KEY (chunk_id, sys_period),
    
    -- Enforce: No duplicate chunk_ids can exist with overlapping temporal ranges
    CONSTRAINT no_overlapping_chunk_periods 
    EXCLUDE USING gist (chunk_id WITH =, sys_period WITH &&)
);

The operator chunk_id WITH = isolates checks to the exact semantic entity, while sys_period WITH && evaluates range overlaps. If an ingestion worker attempts to write a chunk version whose temporal range overlaps an existing active range for that chunk_id, PostgreSQL rejects the transaction at the database kernel boundary.

Continuous temporal updates cause severe GiST spatial tree degradation over time. While PostgreSQL autovacuum cleans dead tuple slots, it fails to rebalance GiST internal bounding boxes as timestamp ranges advance continuously. We observed severe index bloat and read latency degradation on high-throughput write paths. Executing a standard REINDEX acquires an ACCESS EXCLUSIVE lock that halts active incoming read and write transactions. We have not fully resolved online GiST index compaction without incurring significant background CPU contention, so we are currently hacking around this issue by executing REINDEX INDEX CONCURRENTLY inside scheduled off-peak maintenance cron jobs.

Layer 2: PL/pgSQL Trigger Architecture & State Evolution

The database operational layout couples the active query table with an immutable history archive through a kernel-level PL/pgSQL trigger execution boundary.

erDiagram
    rag_knowledge_chunks {
        uuid chunk_id PK
        uuid document_id
        int chunk_index
        text chunk_content
        vector embedding
        jsonb metadata
        tstzrange sys_period PK
    }

    rag_knowledge_chunks_history {
        uuid chunk_id
        uuid document_id
        int chunk_index
        text chunk_content
        vector embedding
        jsonb metadata
        tstzrange sys_period
    }

    trg_rag_chunks_sys_period {
        function process_rag_chunk_sys_period
    }

    rag_knowledge_chunks ||--o{ trg_rag_chunks_sys_period : "Fires BEFORE UPDATE or DELETE"
    trg_rag_chunks_sys_period ||--|{ rag_knowledge_chunks_history : "Appends closed-interval OLD state"

Knowledge Lifecycle Mechanics

The relational engine manages historical state updates through a twin-table structure:

flowchart TD
    UpdateOp["UPDATE Chunk Text / 
Embedding at Time T2"]
    
    subgraph ActiveState["Active Table: rag_knowledge_chunks"]
        ActiveRow["chunk_id: 101<br/>sys_period: [T2, ∞)"]
    end
    
    subgraph HistState["History Table: rag_knowledge_chunks_history"]
        HistRow["chunk_id: 101<br/>sys_period: [T1, T2)"]
    end
    
    UpdateOp -->|"1. Inserts New Version"| ActiveRow
    UpdateOp -->|"2. Archives Closed Version"| HistRow

We define state transformations across four operational phases:

• State Ingestion (INSERT): A regulatory document chunk is embedded at T1. The database initializes sys_period to (T,infty).
• State Mutation (UPDATE): The underlying document is re-chunked at T2. The active row’s upper boundary is capped at T2 and copied to rag_knowledge_chunks_history. A new row containing updated text and vector embeddings is inserted into rag_knowledge_chunks with {sys_period} = [T_2, infty].
• State Deprecation (DELETE): When a document is retracted at T3, the database removes the row from rag_knowledge_chunks and writes a closed record to rag_knowledge_chunks_history with {sys_period} = [T1, T3].
• Deterministic Read Selection (SELECT): Active queries read rag_knowledge_chunks directly without traversing historical record archives.

PL/pgSQL Trigger Function

To keep backend application code free of temporal versioning logic, state archiving is offloaded entirely to database triggers:

-- Mirror table for historical storage
CREATE TABLE rag_knowledge_chunks_history (LIKE rag_knowledge_chunks);

-- State management trigger function
CREATE OR REPLACE FUNCTION process_rag_chunk_sys_period()
RETURNS TRIGGER AS $$
DECLARE
    sys_now TIMESTAMPTZ := transaction_timestamp();
BEGIN
    -- Handle Chunk Updates (Re-embedding / Text Edits)
    IF (TG_OP = 'UPDATE') THEN
        OLD.sys_period := tstzrange(lower(OLD.sys_period), sys_now, '[)');
        INSERT INTO rag_knowledge_chunks_history SELECT OLD.*;

        NEW.sys_period := tstzrange(sys_now, 'infinity', '[)');
        RETURN NEW;
        
    -- Handle Chunk Deprecation / Removal
    ELSIF (TG_OP = 'DELETE') THEN
        OLD.sys_period := tstzrange(lower(OLD.sys_period), sys_now, '[)');
        INSERT INTO rag_knowledge_chunks_history SELECT OLD.*;
        RETURN OLD;
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Attach invariant trigger
CREATE TRIGGER trg_rag_chunks_sys_period
BEFORE UPDATE OR DELETE ON rag_knowledge_chunks
FOR EACH ROW EXECUTE FUNCTION process_rag_chunk_sys_period();

We explicitly set temporal markers using transaction_timestamp() rather than clock_timestamp(). transaction_timestamp() guarantees that all state mutations occurring within a single explicit transaction block receive identical time boundaries, preserving transaction isolation across complex batch updates.

Point-in-Time State Reconstruction Mechanics

To reproduce the exact database state available during an automated decision on March 15, 2026, at 14:30:00 UTC, we execute a point-in-time (AS OF) reconstruction query using the PostgreSQL containment operator @>:

SELECT 
    chunk_id, 
    document_id, 
    chunk_content, 
    metadata,
    sys_period
FROM (
    SELECT * FROM rag_knowledge_chunks
    UNION ALL
    SELECT * FROM rag_knowledge_chunks_history
) AS continuous_knowledge_base
WHERE sys_period @> '2026-03-15 14:30:00+00'::timestamptz
  AND document_id = 'c7a42b10-8e12-4c91-9e05-2d4e8b31411f';

Range Partitioning for Historical Scales

As historical archives exceed tens of millions of rows, GiST traversal latencies increase. To maintain flat sub-millisecond lookup times, we enforce declarative range partitioning on rag_knowledge_chunks_history using the lower bound of sys_period:

CREATE TABLE rag_knowledge_chunks_history (
    chunk_id UUID NOT NULL,
    document_id UUID NOT NULL,
    chunk_index INT NOT NULL,
    chunk_content TEXT NOT NULL,
    embedding vector(1536),
    metadata JSONB DEFAULT '{}'::jsonb,
    sys_period TSTZRANGE NOT NULL
) PARTITION BY RANGE (lower(sys_period));

CREATE TABLE rag_knowledge_chunks_history_y2026m03 
PARTITION OF rag_knowledge_chunks_history
FOR VALUES FROM ('2026-03-01 00:00:00+00') TO ('2026-04-01 00:00:00+00');

CREATE TABLE rag_knowledge_chunks_history_y2026m04 
PARTITION OF rag_knowledge_chunks_history
FOR VALUES FROM ('2026-04-01 00:00:00+00') TO ('2026-05-01 00:00:00+00');

Partition pruning excludes non-relevant partition tables from the execution plan during point-in-time queries, keeping working sets inside the active PostgreSQL buffer pool.

API Exposure via Asynchronous FastAPI Services

The service layer exposes temporal query interfaces over HTTP, mapping incoming point-in-time requests to underlying PostgreSQL queries using SQLAlchemy Async sessions.

from datetime import datetime
from typing import Optional
from uuid import UUID
from fastapi import FastAPI, Depends, Query, HTTPException, Header
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text

app = FastAPI(title="Deterministic Lineage API", version="1.0.0")

@app.get("/api/v1/knowledge/chunks/{chunk_id}")
async def get_chunk_as_of(
    chunk_id: UUID,
    as_of: Optional[datetime] = Query(None, description="Point-in-time timestamp for historical reconstruction"),
    x_legal_basis_id: str = Header(..., description="Mandatory legal audit justification ID"),
    x_operator_id: str = Header(..., description="Authenticated operator ID"),
    db: AsyncSession = Depends(get_db_session)
):
    # Set localized session variables for PostgreSQL WAL/pg_audit logging
    await db.execute(
        text("SET LOCAL audit.context = :ctx"), 
        {"ctx": f"operator:{x_operator_id};legal_basis:{x_legal_basis_id}"}
    )
    
    target_time = as_of or datetime.now()
    
    query = text("""
        SELECT chunk_id, document_id, chunk_content, metadata, lower(sys_period) as valid_from
        FROM (
            SELECT * FROM rag_knowledge_chunks
            UNION ALL
            SELECT * FROM rag_knowledge_chunks_history
        ) AS unified_chunks
        WHERE chunk_id = :chunk_id 
          AND sys_period @> :target_time::timestamptz
    """)
    
    result = await db.execute(query, {"chunk_id": chunk_id, "target_time": target_time})
    record = result.mappings().first()
    
    if not record:
        raise HTTPException(status_code=404, detail="No active chunk state found at specified timestamp")
        
    return record

We integrate four security and audit specifications into the API layer:

• Auditor Session Context: Executing SET LOCAL audit.context writes operator identity and statutory justification codes directly into PostgreSQL Write-Ahead Logs (pg_audit).
• Schema Validation: Pydantic serialization models validate all payload parameters, requiring explicit statutory authorization codes on write and point-in-time read operations.
• Access Control Isolation: OAuth2/OIDC integration enforces Role-Based Access Control (RBAC) and Row-Level Security (RLS) policies based on auditor jurisdiction.
• OpenAPI Metadata Generation: The endpoint schema auto-generates compliance documentation exposed directly to internal regulatory tools.

Executing SET LOCAL audit.context inside asynchronous connection pools (Asyncpg via SQLAlchemy) presents severe connection contamination risks. If an unhandled exception aborts a request before an explicit transaction rollback or commit occurs, residual session variables remain attached to the pooled database connection. Subsequent requests re-using that connection inherit stale audit metadata. We have not fully resolved native connection state reset handling in Asyncpg, so we are currently hacking around this issue by executing RESET ALL; DISCARD TEMP; via a connection checkout event hook, adding 0.4ms to total request overhead.

Layer 3: Grounding Non-Deterministic RAG Outputs in Relational Lineage

Non-deterministic machine learning model outputs must anchor directly to deterministic relational records. Re-running vector searches without temporal constraints causes context drift when knowledge collections undergo updates.

flowchart LR
    GenResponse["Generated Response R"] --> EmbeddingMeta["Vector Embedding Metadata & Seed"]
    EmbeddingMeta --> TemporalRecord["PostgreSQL 
Temporal Record @ Snapshot T"]

Vector embeddings stored via pgvector map directly to primary keys in system-versioned temporal tables at microsecond $T$. To establish complete model auditability, we log an immutable execution metadata tuple alongside relational primary keys:

where H is the SHA-256 hash of the system prompt template, V is the model version string, T is the transaction microsecond, S is the fixed seed integer (Temperature = 0), and K is the array of chunk primary keys valid at timestamp T.

import hashlib
from dataclasses import dataclass
from typing import List
from uuid import UUID

@dataclass
class DeterministicRAGAuditLog:
    prompt_hash: str
    model_version: str
    timestamp_epoch_us: int
    seed: int
    chunk_keys: List[UUID]

def build_audit_record(
    prompt_template: str,
    model_version: str,
    timestamp_us: int,
    seed: int,
    retrieved_chunk_ids: List[UUID]
) -> DeterministicRAGAuditLog:
    p_hash = hashlib.sha256(prompt_template.encode("utf-8")).hexdigest()
    
    return DeterministicRAGAuditLog(
        prompt_hash=p_hash,
        model_version=model_version,
        timestamp_epoch_us=timestamp_us,
        seed=seed,
        chunk_keys=retrieved_chunk_ids
    )

Replaying K through the @> containment query engine reconstructs the raw statutory context retrieved at timestamp T, eliminating context drift across model execution runs.

Architectural Metric Matrix