External Data Tracking

Outcome. Pin the exact upstream value a decision used, so the decision stays reproducible after the source changes.

Prerequisites. The external extra and a fetch you can route through the tracker.

Packages. briefcase-ai

Languages. Python

ExternalDataTracker records a hashed snapshot of every external fetch a decision relied on — an API response, a database query, a file — so the decision stays reproducible even after that source changes underneath you.

external For: governance & audit

When you’d reach for this

Your support-triage agent classified a ticket as low priority because the pricing_service reported a customer’s plan at one tier. Two weeks later that upstream record is corrected to a higher tier, and the routing looks wrong in hindsight. Without a snapshot you can’t prove the agent acted on the value that was live at the time. Capture the upstream value with the decision and the answer is exact, not reconstructed.

How it works

1. Snapshot the fetch. track_api_call hashes the response and stores it when the policy allows.

2. Detect drift. detect_drift compares the live value against the latest snapshot to see whether the source moved.

3. Append a correction. When the upstream value is fixed, correct_snapshot records the fix without erasing the original.

flowchart LR
    A["pricing_service fetch"] --> B["track_api_call()"]
    B --> C["snapshot + SHA-256 hash"]
    C --> D["classify_ticket decision"]
    E["live value, two weeks later"] --> F{"detect_drift"}
    C --> F
    F -- "changed" --> G["correct_snapshot<br/>original kept"]

Diagram description

The triage agent snapshots the pricing service response via track_api_call, which hashes it before the ticket is classified. Two weeks later detect_drift compares the live value against the stored snapshot and reports a change; correct_snapshot records the fix while leaving the original snapshot intact for audit.

Install

pip install briefcase-ai[external]

from briefcase.external import (
    ExternalDataTracker,
    SnapshotPolicy,
    SnapshotFrequency,
)

Track an External Call

track_api_call() hashes the response, stores a snapshot when the policy allows, and reports whether the source drifted since the last snapshot.

from briefcase.external import ExternalDataTracker

tracker = ExternalDataTracker()

response = {"items": [{"sku": "A-100", "price": 19.99}]}

result = tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data=response,
    record_count=1,
)

print(result["data_hash"][:12])     # SHA-256 of the response
print(result["snapshot_stored"])    # True — first snapshot is always stored
print(result["drift_detected"])     # False — nothing to compare against yet

The same shape applies to database queries and file fetches:

tracker.track_db_query(
    db_system="postgres",
    db_name="catalog",
    query="SELECT sku, price FROM products",
    result_data=[{"sku": "A-100", "price": 19.99}],
    result_count=1,
    store_snapshot=True,
)

tracker.track_file_fetch(
    source_name="reference_rates",
    file_data=b"sku,price\nA-100,19.99\n",
    file_path="reference/rates.csv",
    record_count=1,
)

Method	Use it for
track_api_call	An HTTP/API response the decision read.
track_db_query	A database query result the decision read.
track_file_fetch	A reference file the decision read.

Detect Drift

Compare current data against the latest stored snapshot for a source. detect_drift() returns None when there is no prior snapshot.

tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data={"items": [{"sku": "A-100", "price": 19.99}]},
)

report = tracker.detect_drift(
    "pricing_service",
    current_data={"items": [{"sku": "A-100", "price": 24.99}]},
)

print(report.has_changed)    # True
print(report.drift_score)    # 1.0
print(report.size_delta)     # byte difference vs the baseline snapshot

Compare two specific snapshots by id with compare_snapshots():

tracker = ExternalDataTracker()

first = tracker.track_api_call(
    api_name="inventory_api", endpoint="https://inv.internal/v1/stock", method="GET",
    response_data={"items": [{"sku": "A-100", "qty": 40}]},
)
second = tracker.track_api_call(
    api_name="inventory_api", endpoint="https://inv.internal/v1/stock", method="GET",
    response_data={"items": [{"sku": "A-100", "qty": 25}]},
)

report = tracker.compare_snapshots(first["snapshot_id"], second["snapshot_id"])
print(report.has_changed)    # True

Snapshot Policy

A SnapshotPolicy controls when snapshots are taken and how long they are kept. Set a per-source policy, or pass default_policy to the tracker.

from briefcase.external import (
    ExternalDataTracker, SnapshotPolicy, SnapshotFrequency,
)

tracker = ExternalDataTracker(
    default_policy=SnapshotPolicy(
        frequency=SnapshotFrequency.ON_CHANGE,
        retention_days=90,
        max_snapshots=100,
    )
)

tracker.set_policy(
    "pricing_service",
    SnapshotPolicy(frequency=SnapshotFrequency.EVERY_CALL),
)

Field	Default	Description
frequency	ON_CHANGE	When to store a snapshot
retention_days	90	Days to retain snapshots (0 = forever)
change_threshold	0.0	Minimum change to count as drift on ON_CHANGE
max_snapshots	0	Max snapshots per source (0 = unlimited)
compress	False	Compress snapshot bodies before storage

SnapshotFrequency values: EVERY_CALL, ON_CHANGE, HOURLY, DAILY, WEEKLY.

Append a Correction

When a source returned bad data, append a correction instead of overwriting the snapshot. The correction keeps the parent’s valid_time (when the data was true in the real world) but gets a fresh transaction time, so historical queries still see the original belief and later queries see the corrected value.

original = tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data={"items": [{"sku": "A-100", "price": 1999.00}]},  # bad value
)

corrected = tracker.correct_snapshot(
    original["snapshot_id"],
    corrected_data={"items": [{"sku": "A-100", "price": 19.99}]},
    source="manual_review",
)

print(corrected.parent_snapshot_id == original["snapshot_id"])  # True

The original snapshot is never mutated; the correction records its lineage through parent_snapshot_id.

Redact PII Before Storage

Pass a sanitizer (for example briefcase.sanitize.Sanitizer) and snapshot bodies are redacted before they are persisted to durable storage. The data_hash is still computed over the original payload, so drift detection is unaffected.

from briefcase.external import ExternalDataTracker
from briefcase.sanitize import Sanitizer

tracker = ExternalDataTracker(sanitizer=Sanitizer())

Fails closed

If the sanitizer raises, the tracker persists metadata only — raw, potentially PII-bearing data never reaches storage on the error path.

Limits

drift_score is binary unless you supply a detector. Any difference scores 1.0, identical data scores 0.0. A price moving from 19.99 to 19.999 and a response changing entirely both report 1.0. Pass a function to set_change_detector(source_name, fn) when you need magnitude, and note that SnapshotPolicy.change_threshold is only meaningful once you do.

Retention is not automatic. retention_days and max_snapshots describe what enforce_retention() will do when you call it. Nothing calls it for you, so snapshots accumulate until you schedule that call.

Snapshots live in the tracker. They are held in a dict on the instance, so a new ExternalDataTracker starts with no history and detect_drift returns None again. Configure the lakeFS repository and branch for storage that outlives the process.

You only see what you tracked. Drift is measured between one call and the last stored snapshot, so anything the source did between your calls is invisible. Under the default ON_CHANGE frequency, unchanged responses store nothing, which means the snapshot timeline shows transitions rather than observations. Use EVERY_CALL when you need to prove what a source returned at a specific moment.

Comparison is over the payload, not its meaning. The hash is order-insensitive for dict keys, so reordered JSON does not register as drift, but a semantically identical response with a new timestamp field does.

API reference

briefcase.external has the tracker, snapshot, policy, and drift-report types.

Where this fits

Related: RAG Versioning The same snapshot-and-detect model, applied to a retrieval index.

Guide: Reproducible RAG Put snapshots and versioning together end to end.