Kairos — Database Reference

Four persistence systems, each with a distinct responsibility. No system is interchangeable with another.

Aura note: the Neo4j database is named after the instance (e.g. 2016aa75), not neo4j — always open sessions with database=settings.NEO4J_DATABASE. Cloud Qdrant requires payload indexes on any filter field (asset_id, document_id, is_quarantine) — init_qdrant.py creates them.

1. Neo4j — Temporal Reality Graph

Architecture layer: Layer 4 (Temporal Reality Graph)

Neo4j is the primary knowledge store. It holds the authoritative, time-versioned graph of assets, documents, events, and the relationships between them. All reads requiring temporal context (as_of time-travel, authority pre-filtering, blast-radius traversal) go through Neo4j.

Node Labels

Regulation Concept Properties (Concept {type: 'Regulation'})

Seeded by backend/scripts/seed_regulations.py; read by GET /compliance/{gaps,dashboard,audit-pack}.

requires_document_type is what makes gap detection clause-specific. Without it the query could only ask "does this asset have any verified procedure", which ignored the clause entirely and — since nothing but manual quarantine promotion ever writes verification_status='verified' — reported every (regulation × asset) pair as a gap unconditionally. Values must match the document_type strings the loader actually assigns (see scripts/load_demo_dataset.py), or the gap becomes unclosable.

Asset Node Properties

KNOWLEDGE_EDGE — All 6 Properties, No Exceptions

Every relationship written to Neo4j uses the KNOWLEDGE_EDGE type. Six properties are mandatory on every write — Neo4j silently drops properties set to null, so missing any property breaks temporal queries and the "all 6" invariant.

valid_to sentinel rule: GraphService.create_knowledge_edge() defaults valid_to to datetime(9999, 12, 31, 23, 59, 59, tzinfo=timezone.utc) when no expiry is given. All "currently active" Cypher queries use (r.valid_to IS NULL OR r.valid_to > datetime()) to cover both legacy null edges and sentinel edges.

Supersession (never deletion): To retire an edge, call close_validity_window(edge_id, valid_to=now()). Never DELETE. The vault is permanent.

Relationship Types

Constraints and Indices

Uniqueness constraints (prevent duplicates):

Asset.asset_id · Document.document_id · Event.event_id
Person.person_id · Concept.concept_id · Organisation.org_id

Node property indices (hot-path lookups):

Asset: tag_number, site_id, equipment_class, criticality
Document: document_type, status, authority_level
Event: event_type, occurred_at

Edge property indices (temporal query performance):

KNOWLEDGE_EDGE: valid_from, authority_level, verification_status, document_id

Authority Pre-filter (non-negotiable)

Every graph traversal query must include this filter before any path expansion:

WHERE r.authority_level <= $max_level AND r.valid_from <= $as_of

This gates authority level and prevents future-dated edges from appearing in historical queries.

GraphService — Available Methods

All Neo4j access goes through backend/api/services/graph.py. Never write Cypher in routers.

Init

make init-neo4j
# or:
docker exec kairos-backend-api python scripts/init_neo4j.py

Schema file: db/neo4j/init_schema.cypher

2. Supabase — PostgreSQL Relational Store

Architecture layers: Layer 1 (Asset MDM) · Layer 2 (Vault registry) · Layer 3 (Pipeline tracking) · Layer 6 (Quarantine) · Layer 7 (Governance) · Layer 8 (Events + Briefs)

Supabase provides PostgreSQL 15, Auth (JWT via Supabase Auth), and Storage (the immutable vault bucket). The FastAPI backend always uses the service-role key (bypasses RLS). The anon/user key is reserved for future frontend direct-access flows.

Tables — Full Schema

assets — MDM mirror of Neo4j

asset_id        TEXT PRIMARY KEY
tag_number      TEXT NOT NULL
name            TEXT NOT NULL
equipment_class TEXT NOT NULL
criticality     TEXT CHECK IN ('safety_critical','critical','non_critical')
site_id         TEXT NOT NULL
facility_id     TEXT NOT NULL
parent_asset_id TEXT REFERENCES assets(asset_id)
eam_source      TEXT DEFAULT 'manual'
identity_confirmed      BOOLEAN DEFAULT FALSE
identity_confirmed_by   TEXT
identity_confirmed_at   TIMESTAMPTZ
status          TEXT CHECK IN ('active','decommissioned','under_review')
created_at      TIMESTAMPTZ DEFAULT NOW()
updated_at      TIMESTAMPTZ DEFAULT NOW()

Indices: site_id, equipment_class, tag_number

asset_alias_map — Tag alias resolution

id                  UUID PRIMARY KEY
canonical_asset_id  TEXT REFERENCES assets(asset_id)
alias               TEXT UNIQUE
alias_source        TEXT
confidence          NUMERIC(4,3) -- 0–1
confirmed           BOOLEAN DEFAULT FALSE
confirmed_by        TEXT
created_at          TIMESTAMPTZ

documents — Immutable vault registry

document_id     TEXT PRIMARY KEY
sha256_hash     TEXT UNIQUE   -- dedup gate
file_name       TEXT
file_size_bytes BIGINT
mime_type       TEXT
document_type   TEXT
authority_level INTEGER CHECK BETWEEN 1 AND 5
source_system   TEXT
vault_url       TEXT          -- Supabase Storage signed URL
status          TEXT CHECK IN ('active','superseded','archived','disputed')
version_chain   TEXT REFERENCES documents(document_id)  -- supersession chain
occurred_at     TIMESTAMPTZ   -- source document timestamp (migration 009)
ingested_at     TIMESTAMPTZ DEFAULT NOW()
ingested_by     TEXT

Indices: status, document_type, authority_level

Vault rule: Never UPDATE or DELETE. Supersede by setting status='superseded' on the old row and creating a new row linked via version_chain.

document_asset_links — Document ↔ Asset join

id          UUID PRIMARY KEY
document_id TEXT REFERENCES documents(document_id)
asset_id    TEXT REFERENCES assets(asset_id)
linked_at   TIMESTAMPTZ
UNIQUE (document_id, asset_id)

extraction_jobs — Temporal pipeline state tracking

job_id          UUID PRIMARY KEY  -- auto-generated by Postgres, read back after insert
document_id     TEXT REFERENCES documents(document_id)
pipeline_stage  TEXT DEFAULT 'queued'
progress_pct    INTEGER DEFAULT 0
ocr_confidence  NUMERIC(4,3)
entity_count    INTEGER
graph_edges     INTEGER
review_pending  INTEGER DEFAULT 0
error           TEXT
timestamp_drift_detected BOOLEAN DEFAULT FALSE  -- migration 009; cross-source skew only.
                                               -- The pipeline stopped setting it 2026-08-17: an
                                               -- occurred_at/ingested_at gap is ingest lag, not drift.
started_at      TIMESTAMPTZ
completed_at    TIMESTAMPTZ
created_at      TIMESTAMPTZ DEFAULT NOW()

Pipeline stages (in order): queued → ocr_running → ner_running → graph_linking → indexing → complete / review_required / failed

operational_events — Layer 8 event log

event_id         TEXT PRIMARY KEY
event_type       TEXT NOT NULL  -- work_order_created, ptw_generated, shift_handover, alarm, tag_out, inspection_complete, deviation_flag
source_system    TEXT
site_id          TEXT
asset_id         TEXT REFERENCES assets(asset_id)
payload          JSONB DEFAULT '{}'
occurred_at      TIMESTAMPTZ
received_at      TIMESTAMPTZ DEFAULT NOW()
redis_stream_id  TEXT
compound_event_id UUID         -- migration 010: links correlated same-asset events
event_subtype    TEXT          -- migration 012: 'recurring' when recurring failure detected

Indices: event_type, asset_id, occurred_at DESC, compound_event_id

There is no priority column here. GET /events derives it from payload.priority and defaults to "normal" when the key is absent (routers/events.py), so priority is unconstrained JSONB — no CHECK, no default, and present on only a subset of rows. models/event.py documents critical, high, normal, low on WorkOrderEvent, but nothing enforces that at the database. A UI filter must therefore derive its options from the data present rather than hard-code the four levels, or it renders dead options. (The priority TEXT NOT NULL DEFAULT 'normal' in db/schema.sql belongs to briefs, not to this table — an easy misread, and one that has already produced a wrong bug report.)

briefs — Proactive brief delivery log

brief_id          UUID PRIMARY KEY
trigger_event_id  TEXT
trigger_event_type TEXT NOT NULL
asset_id          TEXT REFERENCES assets(asset_id)
recipient_user_id TEXT NOT NULL
priority          TEXT DEFAULT 'normal'  -- critical, high, normal, medium, low
headline          TEXT NOT NULL
body              TEXT NOT NULL
action_items      JSONB DEFAULT '[]'     -- migration 003
warnings          JSONB DEFAULT '[]'     -- migration 003
quarantine_flags  JSONB DEFAULT '[]'     -- migration 003
sources           JSONB DEFAULT '[]'     -- array of SourceCitation objects
confidence        NUMERIC(4,3)
work_order_id     TEXT                   -- migration 003
ptw_id            TEXT                   -- migration 003
delivery_frozen   BOOLEAN DEFAULT FALSE  -- migration 008: frozen by deviation flag
requires_countersignature BOOLEAN DEFAULT FALSE
delivered_at      TIMESTAMPTZ
acknowledged_at   TIMESTAMPTZ
acknowledged_by   TEXT
countersigned_by  TEXT
countersigned_at  TIMESTAMPTZ
created_at        TIMESTAMPTZ DEFAULT NOW()

Indices: recipient_user_id, asset_id, delivered_at, (asset_id, delivery_frozen) WHERE delivery_frozen=TRUE

RLS: Each authenticated user sees only briefs where recipient_user_id = auth.uid(). Site-wide briefs use user_id = "site-{site_id}" and are visible to all users on that site.

brief_feedback

id           UUID PRIMARY KEY
brief_id     UUID REFERENCES briefs(brief_id)
rating       TEXT CHECK IN ('accurate','missing_context','incorrect')
notes        TEXT
submitted_by TEXT
submitted_at TIMESTAMPTZ DEFAULT NOW()

knowledge_conflicts — Dual-track governance

conflict_id   UUID PRIMARY KEY
track         TEXT CHECK IN ('administrative','engineering')
asset_id      TEXT REFERENCES assets(asset_id)
parameter     TEXT
source_a      JSONB   -- { document_id, value, authority_level }
source_b      JSONB
authority_a   INTEGER
authority_b   INTEGER
severity      TEXT
status        TEXT DEFAULT 'open' CHECK IN ('open','pending_moc','resolved')
sla_deadline  TIMESTAMPTZ
escalated_at  TIMESTAMPTZ    -- migration 015
escalated_to  TEXT           -- migration 015
resolved_by   TEXT
resolved_at   TIMESTAMPTZ
created_at    TIMESTAMPTZ DEFAULT NOW()

Indices: status, asset_id

SLA default: 5 days from creation. SLAService.check_and_escalate() runs lazily on GET /conflicts and GET /sla-report — idempotent, writes audit_log on first escalation only.

quarantine_items — Unverified knowledge gate

item_id        UUID PRIMARY KEY
asset_id       TEXT REFERENCES assets(asset_id)  -- NULL when asset unknown (never empty string)
content        TEXT NOT NULL
input_type     TEXT CHECK IN ('field_observation','voice_note','elicitation_response','deviation_flag','offboarding_response')
submitted_by   TEXT
submitted_at   TIMESTAMPTZ DEFAULT NOW()
reviewer_id    TEXT
review_status  TEXT DEFAULT 'pending' CHECK IN ('pending','promoted','disputed','archived')
reviewed_at    TIMESTAMPTZ
work_order_id  TEXT
session_context JSONB DEFAULT '{}'
sla_due_at     TIMESTAMPTZ DEFAULT (NOW() + INTERVAL '5 days')  -- migration 015
escalated_at   TIMESTAMPTZ                                       -- migration 015

Indices: review_status, asset_id

Quarantine is a one-way gate. Only human action via POST /governance/quarantine/{item_id}/promote can promote an item. No auto-promotion ever. Deviation flags override sla_due_at to 24h.

input_type CHECK constraint: Adding a new input_type requires dropping and re-adding the constraint. Current allowed values listed above.

moc_items — Management of Change

moc_id          TEXT PRIMARY KEY
conflict_id     UUID REFERENCES knowledge_conflicts(conflict_id)
asset_id        TEXT REFERENCES assets(asset_id)
description     TEXT
conflicting_sources JSONB DEFAULT '[]'
blast_radius    JSONB DEFAULT '[]'
status          TEXT DEFAULT 'draft' CHECK IN ('draft','pending_approval','approved','rejected')
approved_by     TEXT
approved_at     TIMESTAMPTZ
webhook_received_at TIMESTAMPTZ
created_at      TIMESTAMPTZ DEFAULT NOW()

audit_log — Immutable action record

id          BIGSERIAL PRIMARY KEY
action      TEXT NOT NULL
entity_type TEXT
entity_id   TEXT
performed_by TEXT NOT NULL
details     JSONB DEFAULT '{}'
timestamp   TIMESTAMPTZ DEFAULT NOW()   -- NOTE: column is 'timestamp', NOT 'created_at'

Indices: (entity_type, entity_id), timestamp DESC

Common pitfall: The time column is timestamp, not created_at. Any .order("created_at") call will fail.

elicitation_sessions — Micro-interview state

session_id    UUID PRIMARY KEY
work_order_id TEXT NOT NULL
asset_id      TEXT REFERENCES assets(asset_id)
questions     JSONB DEFAULT '[]'
status        TEXT CHECK IN ('pending','questions_ready','completed')
triggered_by  TEXT
created_at    TIMESTAMPTZ DEFAULT NOW()
updated_at    TIMESTAMPTZ DEFAULT NOW()

Index: work_order_id

ner_annotations — Active learning corpus

id             UUID PRIMARY KEY
document_id    TEXT REFERENCES documents(document_id)
entity_text    TEXT
entity_type    TEXT
corrected_type TEXT
is_correct     BOOLEAN
span_start     INTEGER
span_end       INTEGER
annotated_by   TEXT
created_at     TIMESTAMPTZ DEFAULT NOW()

Indices: document_id, created_at DESC, corrected_type WHERE is_correct=FALSE

extraction_overrides — SPC circuit breaker

id            UUID PRIMARY KEY
asset_class   TEXT
document_id   TEXT
override_type TEXT CHECK IN ('manual_correction','quarantine_rejection','annotation_correction')
created_at    TIMESTAMPTZ DEFAULT NOW()

Index: (asset_class, created_at DESC) — 7-day rolling window check

plant_operating_states — Push suppression gate

id          UUID PRIMARY KEY
site_id     TEXT
state       TEXT CHECK IN ('normal','turnaround','shutdown','emergency')
set_by      TEXT
set_at      TIMESTAMPTZ DEFAULT NOW()
expires_at  TIMESTAMPTZ

Index: (site_id, set_at DESC) — latest state per site

Effect: turnaround / shutdown / emergency suppress normal-priority briefs at the EEMUA 191 governor. Critical (PTW) briefs always pass.

validation_corpus — NER model gate ground truth

id           UUID PRIMARY KEY
document_id  TEXT
entity_text  TEXT
entity_type  TEXT
span_start   INTEGER
span_end     INTEGER
authority    TEXT CHECK IN ('human_promotion','annotation_correction')
promoted_by  TEXT
created_at   TIMESTAMPTZ DEFAULT NOW()

Indices: entity_type, document_id

offboarding_sessions — Knowledge capture series

id                   UUID PRIMARY KEY
personnel_id         TEXT
personnel_email      TEXT
retirement_date      DATE
total_sessions       INT DEFAULT 6
session_interval_days INT DEFAULT 12
status               TEXT CHECK IN ('scheduled','in_progress','completed','cancelled')
created_by           TEXT
created_at           TIMESTAMPTZ DEFAULT NOW()

offboarding_session_items

id                UUID PRIMARY KEY
session_id        UUID REFERENCES offboarding_sessions(id) ON DELETE CASCADE
session_number    INT
equipment_family  TEXT
focus_failure_modes TEXT[]
status            TEXT CHECK IN ('pending','questions_ready','completed')
questions         JSONB DEFAULT '[]'
scheduled_for     TIMESTAMPTZ
completed_at      TIMESTAMPTZ

Index: session_id

Row-Level Security

RLS is enabled on all 19 public tables in db/schema.sql: assets, documents, briefs, quarantine_items, audit_log (from the original schema) and the other 14 (asset_alias_map, document_asset_links, extraction_jobs, operational_events, brief_feedback, knowledge_conflicts, moc_items, elicitation_sessions, ner_annotations, extraction_overrides, plant_operating_states, validation_corpus, offboarding_sessions, offboarding_session_items), added by db/migrations/017_enable_rls_remaining_tables.sql. Supabase exposes every public table through PostgREST, so without RLS the public anon key could read, write and delete those rows directly, bypassing the API, OPA and the audit log. The 14 get no policies on purpose: with RLS on and no policy, anon and authenticated see nothing, and the backend (service-role key) is unaffected. Migration 017 is in the repo and schema.sql but has not been applied to the live project: it is a cloud write, so it waits for an explicit decision (status.md § Accepted risks and deploy checklist). It is idempotent, and the run must be recorded in db/maintenance/CHANGELOG.md.

The FastAPI backend always uses SUPABASE_SERVICE_ROLE_KEY — service role bypasses RLS entirely. Policies apply only to direct Supabase client access (anon/authenticated roles):

Storage Bucket — kairos-vault

Private bucket (no public access). All reads require a signed URL.

Immutability: Files are never overwritten. A new version creates a new document row linked via version_chain. The old file remains permanently.

Migrations

The full schema is consolidated into a single source-of-truth file, db/schema.sql — apply it to a fresh database to get the current schema. The 16 original ordered migrations were folded in and are no longer kept as separate files (a migration that has not yet been run on the live project is kept as a file under db/migrations/: currently only 017, which schema.sql already includes); the live applied history remains tracked by Supabase in supabase_migrations.schema_migrations (timestamp-versioned). Going forward, make a schema change directly in schema.sql (and record any ad-hoc live run in db/maintenance/CHANGELOG.md).

The table below is the schema-evolution changelog — what each of the 16 folded-in migrations contributed to schema.sql:

Maintenance & data hygiene

db/maintenance/ holds SQL you run directly against cloud Supabase — operations that make nuke cannot do, because make nuke only wipes the local Docker volumes (Neo4j, ES, Qdrant, Redis), never the hosted Supabase Postgres. Two files:

• reset_all_data.sql — TRUNCATE every public table (schema + auth kept). The only way to return the cloud database to a blank slate; run it via the Supabase SQL editor or MCP.
• CHANGELOG.md — a running log of every ad-hoc SQL/operation applied to the live project, so the hosted state is never a mystery.

For routine test-residue cleanup use the scripts, not raw SQL:

• make purge-test-data → scripts/purge_test_data.py clears test-prefixed rows (ASSET-TEST/DEDUP/EV/ACK-*, WO-*, DOC-*) from Neo4j + Supabase + Elasticsearch in one go.
• The integration suite also purges its own residue automatically on teardown (conftest.py).

3. Qdrant — Vector Store

Architecture layer: Layer 11 (Semantic Retrieval)

Qdrant stores 1024-dimensional embeddings produced by Jina AI jina-embeddings-v3. It powers semantic similarity search — finding knowledge fragments that are conceptually related to a query even when exact keywords don't match.

Collections

Vector Configuration

Point Schema (Payload)

Every Qdrant point carries this payload alongside the vector:

{
  "document_id": "DOC-XXXX",
  "asset_id": "P-101",
  "authority_level": 2,
  "confidence": 0.91,
  "is_quarantine": false,
  "content": "Four mechanical-seal failures recorded 2018–2026...",
  "document_type": "maintenance_record",
  "chunk_index": 0
}

Filtering

VectorStoreService.search() applies Qdrant payload filters before returning results:

Init

make init-qdrant
# or:
docker exec kairos-backend-api python scripts/init_qdrant.py

Collections are also created automatically at API startup via VectorStoreService.ensure_collections().

Access Pattern

from api.services.vector_store import VectorStoreService

# Upsert
await vector.upsert(
    collection=settings.QDRANT_COLLECTION_KNOWLEDGE,
    point_id=str(uuid.uuid4()),
    vector=embedding,            # List[float], 1024 dims
    payload={"document_id": ..., "asset_id": ..., "authority_level": 2, ...}
)

# Search
hits = await vector.search(
    collection=settings.QDRANT_COLLECTION_KNOWLEDGE,
    query_vector=query_embedding,
    limit=10,
    asset_id="P-101",           # optional filter
    authority_min=3,
    include_quarantine=False,
)

4. Elasticsearch — Full-Text Search

Architecture layer: Layer 11 (Exact Retrieval)

Elasticsearch handles keyword search, tag lookups, and structured field queries. It runs alongside Qdrant in parallel during hybrid search — ES for exact/keyword hits, Qdrant for semantic hits — and results are merged and re-ranked by authority level.

Indices

Document Shape — kairos_documents

{
  "document_id": "DOC-XXXX",
  "title": "EQ-101 Failure History Report",
  "content": "Four mechanical-seal failures...",
  "document_type": "maintenance_record",
  "authority_level": 2,
  "asset_ids": ["EQ-101", "P-101"],
  "ingested_at": "2026-07-01T10:00:00Z"
}

Hybrid Search Flow

GET /search?q=... runs three queries in parallel:

1. ES full-text on kairos_documents (keyword match)
2. Qdrant semantic on kairos_knowledge (cosine similarity)
3. Neo4j graph traversal (authority-filtered edge walk)

Results are merged, deduplicated by document_id, and re-ranked by authority_level (ascending — lower authority level = higher trust).

5. Redis — Streams, Governor, Cache, Broker

Architecture layer: Layer 8 (Event Bus) · Cross-cutting (Cache + Broker)

Redis is used for four distinct concerns, each isolated to a separate database number.

Database Allocation

Event Streams (Redis Streams on DB 0)

All streams use XADD. Workers and the brief engine consume via XREAD. Stream IDs are stored in operational_events.redis_stream_id for traceability.

EEMUA 191 Governor Keys

kairos:governor:{user_id}:hourly_count   TTL: 3600s (rolling hour)

EventBusService.check_governor(user_id) reads this counter before every brief delivery. Hard ceiling: 6 pushes/operator/hour. PTW briefs bypass the ceiling unconditionally. Plant state (turnaround/shutdown/emergency) suppresses normal-priority briefs regardless of count.

Delay Compensation Keys

kairos:brief_pending:{asset_id}:{work_order_id}  TTL: LATE_ARRIVAL_WINDOW + 60s
kairos:brief_pending:shift:{site_id}     TTL: LATE_ARRIVAL_WINDOW + 60s

The Celery task ID is stored in the {asset_id}:{work_order_id} key, so only a re-report of the same work order revokes and re-enqueues it. A PTW assembles its own brief immediately and never touches a pending slot.

Dedup Keys

kairos:dedup:{event_type}:{business_id or asset_id}     TTL: 600s (10 minutes)

The key is written by EventBusService.mark_seen only after an ingest has succeeded (the check itself is read-only). Identical (event_type, business_id) pairs, or (event_type, asset_id) for events with no business id, within 10 minutes collapse to {status: "deduplicated"}. Tests using shared_asset_id must use distinct asset IDs to avoid dedup collisions within the window.

Celery (DB 1)

Celery broker and result backend both use redis://kairos-redis:6379/1.

Active queues (all must be in the --queues flag of kairos-celery-worker):

6. Cross-System Patterns

Write Order for Document Ingestion

1. Supabase Storage → upload file, get vault_url
2. Supabase documents → insert row (sha256 dedup gate)
3. Supabase extraction_jobs → insert row, read back auto-generated job_id
4. Temporal → start DocumentIngestionWorkflow
5. Neo4j → MERGE Document node
6. Neo4j → CREATE KNOWLEDGE_EDGE (all 6 properties)
7. Qdrant → upsert vector point
8. Elasticsearch → index document text
9. Supabase extraction_jobs → update stage=complete

Confidence Routing

ocr_confidence < 0.5    → stage=review_required, stop pipeline
entity_confidence < 0.7 → quarantine_items (Supabase), no Neo4j write
entity_confidence ≥ 0.7 → KNOWLEDGE_EDGE verification_status=unverified
human promoted          → KNOWLEDGE_EDGE verification_status=verified

Never Do

• DELETE in Neo4j or Supabase — supersede/close instead
• Store time-series historian data in any Kairos store — ephemeral only
• Set valid_to=null on a KNOWLEDGE_EDGE — use the sentinel 9999-12-31T23:59:59Z
• Use asset_id="" in quarantine_items — use None / SQL NULL
• Call Supabase client directly in an async handler — always asyncio.to_thread(lambda: ...)
• Use audit_log.created_at — the column is timestamp