Entity History & Audit
Every change to a canonical (golden) entity is recorded as an audit event. You can view the full history of an entity, lock it to prevent automatic changes, revert it to a previous state, and configure temporal matching to handle time-varying source records.
Audit Events
When an entity is created, merged, split, or updated, the system records an AuditEvent:
| Field | Type | Description |
|---|---|---|
id | UUID | Event ID |
tenant_id | UUID | Tenant this event belongs to |
actor_id | UUID | Who made the change (user ID or system ID) |
actor_type | string | "user", "system", or "api_key" |
action | string | What happened: "create", "merge", "update", "revert", "lock", "unlock" |
resource_type | string | Type of resource changed (e.g., "canonical_entity") |
resource_id | UUID | ID of the entity that changed |
before_state | JSON | Entity data before the change (null for create) |
after_state | JSON | Entity data after the change |
reason | string | Optional human-readable reason |
timestamp | datetime | When the change occurred (UTC) |
The before_state and after_state fields capture the full canonical_data JSONB at the time of the event. This gives you a complete snapshot — not just what fields changed, but what the entire record looked like before and after.
Viewing Entity History
Retrieve the audit trail for any canonical entity:
import httpx
entity_id = "550e8400-e29b-41d4-a716-446655440000"
resp = httpx.get(f"https://api.kanoniv.com/v1/canonical/{entity_id}/history",
headers={"X-API-Key": "kn_..."})
events = resp.json()
for event in events:
print(f"[{event['timestamp']}] {event['action']} by {event['actor_type']}")
if event.get("reason"):
print(f" Reason: {event['reason']}")curl "https://api.kanoniv.com/v1/canonical/550e8400-e29b-41d4-a716-446655440000/history" \
-H "X-API-Key: kn_..."Example response:
[
{
"id": "a1b2c3d4-...",
"action": "merge",
"actor_type": "system",
"before_state": {"name": "John Doe", "email": "[email protected]"},
"after_state": {"name": "John Doe", "email": "[email protected]", "phone": "+1-555-0123"},
"timestamp": "2026-02-10T14:30:00Z"
},
{
"id": "e5f6a7b8-...",
"action": "create",
"actor_type": "system",
"before_state": null,
"after_state": {"name": "John Doe", "email": "[email protected]"},
"timestamp": "2026-02-08T09:15:00Z"
}
]Events are returned in reverse chronological order (newest first).
Entity Locking
Lock an entity to prevent automatic merges and updates. Locked entities are skipped during reconciliation and queued for manual review instead.
# Lock an entity
httpx.post(f"https://api.kanoniv.com/v1/canonical/{entity_id}/lock",
headers={"X-API-Key": "kn_...", "Content-Type": "application/json"},
json={"locked": True})
# Unlock it later
httpx.post(f"https://api.kanoniv.com/v1/canonical/{entity_id}/lock",
headers={"X-API-Key": "kn_...", "Content-Type": "application/json"},
json={"locked": False})# Lock
curl -X POST "https://api.kanoniv.com/v1/canonical/550e8400-.../lock" \
-H "X-API-Key: kn_..." \
-H "Content-Type: application/json" \
-d '{"locked": true}'
# Unlock
curl -X POST "https://api.kanoniv.com/v1/canonical/550e8400-.../lock" \
-H "X-API-Key: kn_..." \
-H "Content-Type: application/json" \
-d '{"locked": false}'The is_locked flag on CanonicalEntity controls this behavior. When is_locked is true:
- Reconciliation will not modify the entity's
canonical_data - New matching records that would merge into this entity are held for review
- Manual overrides and explicit API updates still work
Reverting to a Historical State
If a merge or update produced bad results, you can revert an entity to the state it was in at any previous audit event.
Step-by-Step
- Get the entity history to find the event you want to revert to:
resp = httpx.get(f"https://api.kanoniv.com/v1/canonical/{entity_id}/history",
headers={"X-API-Key": "kn_..."})
events = resp.json()
# Find the event before the bad merge
for event in events:
print(f"{event['id']}: [{event['timestamp']}] {event['action']}")- Choose the target event — the revert restores the
after_stateof the selected event:
a1b2c3d4-...: [2026-02-10T14:30:00Z] merge ← the bad merge
e5f6a7b8-...: [2026-02-08T09:15:00Z] create ← revert to this- Revert:
target_event_id = "e5f6a7b8-..."
resp = httpx.post(
f"https://api.kanoniv.com/v1/canonical/{entity_id}/revert/{target_event_id}",
headers={"X-API-Key": "kn_..."})
restored = resp.json()
print(f"Restored to: {restored['canonical_data']}")curl -X POST \
"https://api.kanoniv.com/v1/canonical/550e8400-.../revert/e5f6a7b8-..." \
-H "X-API-Key: kn_..."The revert updates the entity's canonical_data to the after_state of the selected event. A new audit event is created recording the revert itself.
Lock After Reverting
After reverting, consider locking the entity to prevent the same bad merge from happening again on the next reconciliation run. Fix the underlying rule or threshold first, then unlock.
Temporal Matching
Source records can have time-validity windows via valid_from and valid_to fields. The temporal matching strategy controls how the engine handles these windows during reconciliation.
Configuration
Define temporal fields on a source in your spec YAML:
sources:
- name: crm
adapter: postgres
location: customers
primary_key: id
attributes:
name: name
email: email
temporal:
valid_from_field: effective_date
valid_to_field: expiry_date
strategy: point_in_timeStrategies
| Strategy | Behavior | Use Case |
|---|---|---|
latest_only | Only the most recent record per entity is considered | Default. Good when you only care about current state |
point_in_time | Records are matched based on overlapping validity windows | Regulatory reporting, historical audits |
bi_temporal | Both transaction time and valid time are tracked | Full audit trail with corrections — know what was known at any point |
How It Affects Matching
With latest_only (the default), each entity contributes one record to matching. With point_in_time, a single entity may have multiple valid records at different times — the engine matches records whose validity windows overlap.
Example: a customer record that was valid from Jan 1 to Mar 31 will only match against other records that overlap that window. A record valid from Apr 1 onward would be treated as a separate observation.
bi_temporal adds a second time axis — when the record was entered into the system. This lets you answer questions like "what did we know about this customer as of February 1st?" even if the record was later corrected.
Full Workflow: Investigating a Bad Merge
Here's a complete example: discovering a bad merge, investigating it, reverting, and preventing recurrence.
import httpx
API = "https://api.kanoniv.com"
HEADERS = {"X-API-Key": "kn_...", "Content-Type": "application/json"}
entity_id = "550e8400-e29b-41d4-a716-446655440000"
# 1. Get the entity and notice something is wrong
entity = httpx.get(f"{API}/v1/canonical/{entity_id}",
headers=HEADERS).json()
print(f"Current: {entity['canonical_data']}")
# Shows merged data from two unrelated people
# 2. View history to understand what happened
events = httpx.get(f"{API}/v1/canonical/{entity_id}/history",
headers=HEADERS).json()
for e in events:
print(f"[{e['timestamp']}] {e['action']}")
if e['before_state']:
print(f" Before: {e['before_state']}")
print(f" After: {e['after_state']}")
# 3. Lock the entity to prevent further damage
httpx.post(f"{API}/v1/canonical/{entity_id}/lock",
headers=HEADERS, json={"locked": True})
# 4. Revert to the state before the bad merge
good_event_id = events[1]["id"] # the event before the bad merge
restored = httpx.post(
f"{API}/v1/canonical/{entity_id}/revert/{good_event_id}",
headers=HEADERS).json()
print(f"Restored: {restored['canonical_data']}")
# 5. Investigate: was the matching rule too aggressive?
# (Fix the rule/threshold, then unlock)See Also
- Audit Logs (Cloud) — Audit streaming, retention, and SIEM integration
- Spec Versioning — Track changes to the matching logic that produced these events
- Entities API Reference — Full endpoint documentation for entity operations