Kairos — Backend Reference

For AI coding agents: This document covers every service, worker, data model, and configuration parameter in the Kairos backend. Read alongside ARCHITECTURE.md (layer design), docs/API.md (endpoint reference), docs/INFRA.md (containers, ports, Redis/Qdrant/ES, observability, dev commands), and docs/FIXTURES.md (mock data fallbacks). All 34 implementation tasks are verified complete.

Table of Contents

1. What Kairos Does
2. Repository Layout
3. Infrastructure Stack
4. FastAPI Application
5. Data Models (Pydantic)
6. Services Layer
7. Temporal Workflows and Activities
8. Celery Workers
9. Go OT Connector
10. Database Schemas
11. Configuration Reference
12. Auth and Authorization
13. Observability
14. Non-Negotiable Rules

1. What Kairos Does

Kairos is an Industrial Operational Intelligence Platform. It continuously monitors the operational pulse of asset-intensive facilities (oil & gas, power, pharma, steel, mining) and delivers the right knowledge to the right person at the exact moment it is needed — without being asked.

Three-phase architecture:

• Phase 1 (live): Retrieval — ingest documents, build a temporal knowledge graph, answer queries with source-cited results.
• Phase 2 (live): LLM Synthesis — POST /search/synthesize assembles retrieved facts into provenance-backed answers via NVIDIA NIM (cloud-only; the local Ollama fallback is disabled — see §6 LLMService). POST /search/synthesize/stream returns the same answer as SSE for progressive render; it is a separate endpoint so the parse contract and its measured answer-quality figure carry no risk.
• Phase 3 (live): Proactive Push — event-driven brief delivery to operators (work orders, PTWs, shift handovers, alarms, tag-outs, inspections) via Redis Streams and the EEMUA 191 governor.

2. Repository Layout

kairos/                          # repo root
├── backend/                     # Python app (Docker build context)
│   ├── api/
│   │   ├── main.py                  # FastAPI app factory + lifespan
│   │   ├── config.py                # All settings (pydantic-settings, env vars)
│   │   ├── dependencies.py          # DI: Neo4j, Qdrant, ES, Redis, Supabase, Auth
│   │   ├── middleware/
│   │   │   ├── opa.py               # OPA policy enforcement middleware
│   │   │   ├── ratelimit.py         # Per-IP Redis rate limit
│   │   │   └── telemetry.py         # OTEL tracing + metrics setup
│   │   ├── models/                  # Pydantic request/response schemas
│   │   ├── routers/                 # One file per domain layer
│   │   ├── services/                # All business logic (no logic in routers)
│   │   └── utils/
│   │       └── failure_families.py  # Shared FAILURE_FAMILIES dict (recurring detection)
│   ├── workers/
│   │   ├── celery_app.py            # Celery app definition
│   │   ├── attribution.py           # Outcome attribution worker
│   │   ├── brief_assembly.py        # Delayed brief assembly worker
│   │   ├── model_validation.py      # NER model gate evaluation worker
│   │   ├── offboarding.py           # Off-boarding interview question generation worker
│   │   ├── temporal_worker.py       # Temporal activity worker (ingestion pipeline)
│   │   ├── elicitation_worker.py    # Temporal worker (elicitation workflows)
│   │   └── voice_transcription.py   # Groq Whisper transcription + NER Celery task
│   ├── workflows/
│   │   ├── document_pipeline.py     # DocumentIngestionWorkflow + all 7 activities
│   │   └── elicitation_workflow.py  # MicroInterviewWorkflow + question generation (Neo4j + LLM)
│   ├── connectors/                  # Go service — OT historian + EAM sync
│   │   ├── cmd/connector/main.go    # Entry point, all HTTP handlers
│   │   ├── cmd/connector/auth.go    # X-Connector-Secret middleware + boot-time config validation
│   │   ├── cmd/connector/auth_test.go  # Go tests: secret check, config refusal, body cap, no upstream echo
│   │   ├── internal/ot/client.go    # PIWebAPIClient + MockHistorianClient
│   │   ├── internal/ot/client_test.go  # Go test: tag escaping, URL scrubbed from errors
│   │   └── fixtures/sample_assets.json  # 5 demo assets for EAM sync
│   ├── scripts/
│   │   ├── seed_users.py            # Creates 6 Supabase auth users (admin, engineer, field_worker, reliability, compliance, demo). Passwords come from KAIROS_SEED_PASSWORD_* in .env.
│   │   │                             # Passwords from KAIROS_SEED_PASSWORD_* in .env (refuses to run if any is blank);
│   │   │                             # role + site_id are written to app_metadata, display name to user_metadata
│   │   ├── migrate_roles_to_app_metadata.py  # One-off: copy role/site_id/name from user_metadata to app_metadata for
│   │   │                             # existing users. DRY RUN by default, --apply writes cloud Supabase Auth.
│   │   │                             # Must run BEFORE deploying code that reads app_metadata only
│   │   ├── seed_regulations.py      # Seeds 12 regulations into Neo4j
│   │   ├── init_neo4j.py            # Neo4j schema constraints + indices
│   │   ├── init_qdrant.py           # Qdrant collection creation
│   │   ├── run_model_validation.py  # Entity-extraction F1 vs validation_corpus (Layer-0 model gate).
│   │   │                             # NOTE: prints only — it does NOT write audit_log, so its
│   │   │                             # results never appear on /governance/model-gate. Only the
│   │   │                             # Celery task (POST /governance/model-gate/run) persists.
│   │   ├── seed_validation_corpus.py # Seed NER ground-truth entities from canon (entity-F1 labels)
│   │   ├── load_demo_dataset.py     # Load dataset/ via the real API pipeline; seeds aliases + NER corpus (`make load-dataset`)
│   │   ├── audit_submission_patterns.py  # ARCHITECTURE §8 mitigation 3: submission-rate outliers.
│   │   │                             # Median-based, reports only, silent below 5 accounts.
│   │   ├── verify_graph_perf.py     # ARCHITECTURE §7 query-perf regression check (`make graph-perf`).
│   │   │                             # Asserts plan SHAPE, not timings — dbHits move with the data.
│   │   ├── backfill_graph_nodes.py  # Layer-4 corpus backfill: Event (free) + Person/Organisation (NIM).
│   │   │                             # DRY RUN by default — pass --apply to write. Idempotent (MERGE).
│   │   │                             # Stamps Document.entity_backfill_at on a model-backed pass so a
│   │   │                             # doc mentioning nobody is not re-extracted forever; --force resets.
│   │   ├── redate_demo.py           # Shift demo event dates so the newest is "yesterday" (`make redate-demo`).
│   │   │                             # Golden events only (matched via the loader's mapping). DRY RUN by
│   │   │                             # default; --apply writes cloud Supabase + Neo4j. Safe to re-run.
│   │   ├── generate_showcase.py     # Write the showcase_* files into dataset/ from the generator (`make generate-showcase`). No store.
│   │   ├── load_showcase.py         # Load the showcase_* files in dataset/ (about 160 assets, about 470 documents, 90 days of history)
│   │   │                             # into the CURRENT stores (`make load-showcase`). DRY RUN by default; --apply
│   │   │                             # also needs SHOWCASE_CONFIRM. Fails if any row lacks a showcase marker or the
│   │   │                             # real-mode counts change.
│   │   ├── redate_showcase.py       # Shift the showcase's dates forward (`make redate-showcase`); time columns of the
│   │   │                             # loader's own rows only. DRY RUN by default. SHOWCASE_AUTO_REDATE runs it daily.
│   │   ├── reset_showcase.py        # Put the showcase's state back to its seed (`make reset-showcase`). The only
│   │   │                             # delete: marker filters only, never the vault. DRY RUN by default.
│   │   ├── showcase/                # The generator: spec.py (sites, people, assets), docs.py (the documents),
│   │   │                             # history.py (events, briefs, governance, knowledge capture), redate.py;
│   │   │                             # files.py writes and reads the showcase files in dataset/ and binds it to load time
│   │   ├── purge_test_data.py       # Delete test-prefixed rows from all stores (`make purge-test-data`)
│   │   └── wipe_local_stores.py     # Empty Neo4j + ES + Qdrant entirely (`make wipe-local` / `reset-local`)
│   └── requirements.txt
├── benchmark/                   # Evaluation harness + evidence (mounted at /app/benchmark)
│   ├── run_benchmark.py         # Retrieval · answer quality · provenance · KG-linkage · time-to-answer (`make benchmark`)
│   ├── verify_layers.py         # Per-layer smoke + latency (`make verify`)
│   ├── questions.json           # 25 domain-expert Q&A across 15 categories (grounded in the dataset canon)
│   └── RESULTS.md               # Raw output of both scripts (methodology → docs/BENCHMARKS.md)
├── db/                          # Database schemas (mounted into Python containers)
│   ├── schema.sql               # Consolidated Supabase schema — single source of truth (001–016 folded in, plus the 017 RLS block)
│   ├── migrations/              # Hand-applied SQL not yet run on the live project
│   │   └── 017_enable_rls_remaining_tables.sql  # RLS on the 14 tables that lacked it (live RLS was already on, verified 2 Oct; idempotent safety net, see DATABASE.md)
│   ├── maintenance/             # Cloud-Supabase reset SQL (reset_all_data.sql) + CHANGELOG.md (tracked runs)
│   └── neo4j/init_schema.cypher # Neo4j constraints + indices
├── fixtures/                    # Shared mock data (mounted into Python containers)
│   └── pid_topology_mock.json
├── infra/                       # Infrastructure configs
│   ├── policies/kairos.rego     # OPA RBAC rules (active — mounted by kairos-opa)
│   ├── temporal/dynamicconfig.yaml  # Temporal server config (active)
│   ├── caddy/Caddyfile          # HTTPS reverse proxy (active under --profile prod): 30 MB body cap + security headers
│   └── grafana/dashboards-import/  # Grafana Cloud dashboard JSONs (obs is Grafana Cloud; not mounted by any container)
├── frontend/                    # Next.js UI (separate Docker build context)
└── tests/                       # Pytest test suite (mounted into backend-api); the test_sec_*.py files cover the 2026-10-01 security pass

3. Infrastructure Stack

Moved to docs/INFRA.md. That file is the single source of truth for all Docker containers, ports, Redis DB allocation, Redis Streams, Qdrant collections, Elasticsearch indices, OTEL pipeline, Grafana dashboards, and dev commands.

4. FastAPI Application

Entry point: backend/api/main.py

create_app() registers:

1. CORS middleware (CORS_ORIGINS from settings)
2. OPAMiddleware (enforces RBAC via OPA on writes and sensitive reads)
3. RateLimitMiddleware (per-IP; zero, meaning off, only when Settings.is_development)
4. OTEL instrumentation (setup_telemetry(app))
5. All routers with prefix/tag

Every middleware reads request.scope["path"], never request.url.path: the URL is rebuilt from the Host header, so Host: x/health turned /events/work-order into /health/events/work-order, which the OPA skip list treated as exempt. The scope path is what the router matches.

Database errors reaching the global handler return fixed messages (foreign key: A referenced record does not exist.; unique: A record with these identifiers already exists.), never the upstream detail.

Lifespan: On startup, VectorStoreService.ensure_collections() and SearchEngineService.ensure_indices() create Qdrant collections and ES indices if missing.

Dependency Injection (api/dependencies.py)

All backends are injected as FastAPI dependencies. Each is a singleton (module-level global, lazy-initialized).

Auth flow in get_current_user:

1. If no Authorization header and Settings.dev_bypass_allowed (APP_DEBUG=True and APP_ENV=development) → returns {user_id: "dev-user", role: "engineer"} (dev only). APP_ENV is trimmed and lower-cased, and only the exact value development counts: production, staging or a typo are all non-development (Settings.is_development).
2. If token matches INTERNAL_API_KEY (hmac.compare_digest) → returns service account {role: "admin"} (Go connector bypass).
3. Otherwise → calls supabase.auth.get_user(token) using a fresh anon client (never the service-role client). Role and site are read from user.app_metadata (users can edit user_metadata, never app_metadata). POST /auth/logout drops the token from the verified-token cache (_auth_cache_drop) and revokes the session. site_scope(user, requested) returns the caller's own site (None for admin) and 403s a request for another one; it is how every site-scoped read and write in the routers is bounded.

5. Data Models (Pydantic)

All models live in backend/api/models/.

Asset (models/asset.py)

Key constraint: confirmed_by_user_id is mandatory in AssetCreate — AI-inferred identities are never accepted.

Document (models/document.py)

Brief (models/brief.py)

Event (models/event.py)

All events inherit from BaseEvent (event_id, source_system, site_id, occurred_at, received_at).

close_notes on WorkOrderEvent is used by the attribution worker for execution compliance keyword matching.

6. Services Layer

All business logic lives in backend/api/services/. Routers call services; services never call routers.

GraphService (services/graph.py)

Interface to Neo4j. All writes use MERGE, never CREATE for asset nodes.

Key methods:

• topology_node_id(document_id, element_id) — graph node id of a P&ID element, {document_id}:{element_id}. The vision model numbers elements per drawing (TOPO-EQ-001), so a bare id merged two drawings onto one Concept. Supabase rows keep the raw id; verify_topology_element matches both the scoped and the legacy bare form, pinned to the document's own edge.
• create_asset_node(props) — MERGE Asset node with all properties
• get_asset(asset_id) — single asset lookup
• list_assets(site_id, equipment_class, skip, limit) — paginated
• get_asset_hierarchy(asset_id) — PARENT_OF traversal up to 10 levels
• get_asset_knowledge_at(asset_id, as_of) — KNOWLEDGE_EDGE traversal with optional time-travel. Deduped by edge_id — the graph can hold multiple physical relationships sharing one logical edge_id (Cypher DISTINCT can't collapse them), so the same fact would otherwise repeat.
• create_knowledge_edge(asset_id, document_id, rel_type, props) — writes edge with all 6 required properties. valid_to defaults to the open-ended sentinel _OPEN_VALID_TO (see below) when not supplied; it is never stored as null.
• merge_document_node(document_id, props) — MERGE Document node
• detect_conflict(source_id, source_label, relationship_type, new_document_id, new_authority_level) — dual-track conflict detection. Returns None immediately for any type in NON_ASSERTING_RELATIONSHIPS (see below), before the Cypher runs.
• get_blast_radius(document_id) — traverses (source)-[r:KNOWLEDGE_EDGE {document_id}]->(target) and returns each {edge, source, target}. The affected entity is the source (e.g. the asset), not the target (the document node). Deduped by edge_id (re-runs leave duplicate relationships).
• close_validity_window(edge_id, closed_at) — sets valid_to on a KNOWLEDGE_EDGE (used by MoC webhook and the in-app MoC approve endpoint, via the shared _resolve_moc_conflict helper)
• create_concept_node(props) — Concept:Regulation seed
• link_concept_to_asset(concept_id, asset_id, props) — compliance framework linkage
• get_event_timeline(asset_id, window_start_iso) — Event nodes linked to an asset within a date window for RCA pack assembly

NON_ASSERTING_RELATIONSHIPS — what may be called a conflict: DOCUMENTED_BY, MENTIONS_PERSON, MENTIONS_ORGANISATION, CONTAINS_TOPOLOGY_ELEMENT record provenance or structure, never a claim, so two of them can never contradict each other — an archive holding several documents about one pump is the normal state, not a disagreement. is_asserting_relationship() is the single predicate; routers/governance.py applies the same set when listing conflicts, because rows written before this guard live in a cloud store and cannot be deleted.

The honest check would compare the two asserted values, but a KNOWLEDGE_EDGE has no value property — the six mandatory props are validity, authority, document, confidence and verification status. That is also why auto-detected conflicts carry no value in source_a/source_b. Adding one is a schema change plus a backfill of every existing edge, so the correct move at this scale is to stop asking the question of edges that cannot answer it.

All 6 properties required on every KNOWLEDGE_EDGE write: valid_from, valid_to, authority_level, document_id, confidence, verification_status

_OPEN_VALID_TO sentinel (datetime(9999, 12, 31, 23, 59, 59, tzinfo=timezone.utc)): Neo4j silently drops properties set to null. Storing valid_to=None would cause the key to disappear from edge.keys(), silently violating the six-property rule. Instead, create_knowledge_edge() stores (valid_to or self._OPEN_VALID_TO).isoformat() — the far-future sentinel signals "currently active / no supersession date" without dropping the property.

"Active edge" query pattern: All Cypher queries that filter for currently-active edges use (r.valid_to IS NULL OR r.valid_to > datetime()), not r.valid_to IS NULL alone. The IS NULL branch covers legacy edges written before the sentinel was introduced; the > datetime() branch covers all new sentinel edges. Affected files: graph.py (detect_conflict, close_validity_windows), compliance.py (3 call sites), brief_engine.py, offboarding.py, elicitation_workflow.py.

SearchService (services/search_service.py)

Orchestrates hybrid search across three engines.

hybrid_search(query, collection, asset_id, authority_min, include_quarantine, as_of, limit):

1. Neo4j graph traversal — only when asset_id provided, and run first: its edges name every document linked to the asset
2. ES exact search — tag numbers, document IDs, clause refs
3. Qdrant semantic search — 1024-dim embedding via LLMService.embed()

ES and Qdrant run in parallel. With asset_id, both match a document filed under that asset or one of the graph-linked documents from step 1. A document is indexed under a single primary asset_id, but often concerns several assets (the EQ-1xx work-order CSV is filed under EQ-102 and records EQ-101's seal failures; SOP-HE-GEN-11 is filed under no asset). Without the widened scope an asset-scoped search could never reach them. If the graph lookup fails, search degrades to the primary-asset scope rather than failing.

Fusion — Reciprocal Rank Fusion (_RRF_K = 60), then authority ordering. Each source's results are scored 1/(60 + rank) and summed, so a document more than one source agrees on ranks higher. RRF replaced a direct comparison of ES relevance against Qdrant cosine similarity: BM25 is unbounded and cosine is 0–1, so comparing them numerically ranked by whichever source emitted bigger numbers.

Authority remains the primary sort key — level 1 (Regulatory) outranks level 5 (Field) — because that is a deliberate safety property, not a relevance artefact. Within an authority level, an asset-scoped search ranks the asset's own documents ahead of documents that are only graph-linked to it; otherwise a linked shift log could push the asset's own closeout form out of limit. RRF orders results after that, which is where the scale mismatch actually did damage. The fused score is written back to relevance_score.

Provenance edges never take a slot on their own (_rankable_graph_hits). A DOCUMENTED_BY or MENTIONS_* edge states no fact, so alone it renders a content-free stub ranked by the edge's authority. With extraction linking every document an asset appears in, such stubs (a level-1 regulation, a level-4 PTW) filled the top of an EQ-101 search and pushed out the closeout form holding the answer. They still widen the text-search scope and still boost a document text search also found.

Merging duplicates (_better): when several sources return the same document_id, the most authoritative record wins but the longest snippet is kept — collapsing by document_id used to discard the losing record's text, so a semantic chunk containing the answer could be replaced by a shorter exact-match excerpt and synthesis never saw the fact. retrieval_method records every method that surfaced it (e.g. exact+semantic).

Graph hits carry text (_edge_snippet): a graph result used to have snippet="", so it entered the ranking but gave synthesis nothing to read — a fact existing only as an edge was invisible to the answer. Edges now render as a readable fact line including validity, verification status and confidence, and an unverified edge is flagged is_quarantine.

Covered by tests/test_search_fusion.py.

SearchEngineService (services/search_engine.py)

Wraps Elasticsearch.

• ensure_indices() — creates kairos_documents, kairos_assets, and kairos_events indices on startup
• search(query, index, asset_id, limit, include_superseded=False) — ES full-text search with highlight
• index_document(doc_id, content, metadata) — index a document chunk

Superseded documents are excluded by default (must_not on status: "superseded"). It is must_not superseded rather than must active deliberately: the query also spans kairos_assets, whose documents carry no status field, and requiring active would drop every asset hit. Time-travel callers pass include_superseded=True.

VectorStoreService (services/vector_store.py)

Wraps Qdrant.

• ensure_collections() — creates kairos_documents and kairos_knowledge on startup
• upsert(collection, point_id, vector, payload) — upsert a single vector
• search(collection, query_vector, asset_id, limit, include_superseded=False) — ANN search with optional asset filter
• mark_superseded(collection, document_id) — payload update flagging every chunk of a document superseded. Never a delete: the vault is immutable and time-travel still has to reach the chunks.

Same must_not reasoning as ES — points indexed before status existed have no such key, and Qdrant treats a missing key as non-matching, so requiring active would silently drop them.

LLMService (services/llm.py)

LLM synthesis + embedding. Never originates knowledge — only assembles retrieved context.

• synthesize(query, context, query_category, aliases=None) — nvidia/nemotron-3-ultra-550b-a55b on the first configured tier (NIM by default), falling through the cascade below. A safety gate runs twice: on the evidence before synthesis, and on the result after it (an honest "not specified in the sources" must not render as a hedged answer).
• evidence_gate(...) / result_gate(...) — those two gates, as separate methods returning a refusal dict or None. Extracted so synthesize() and synthesize_stream() call the same code: a second copy of a refusal rule would drift, and whichever an operator hit would be wrong.
• synthesize_stream(...) — async generator of (event, payload) behind POST /search/synthesize/stream. Safety-critical categories emit no answer text at all: CONFIDENCE: arrives after ANSWER:, and result_gate can convert a finished answer into a refusal, so streaming it would show text the gate is about to retract. Everything else streams ANSWER: as it arrives. Streams tier 1 only; a mid-stream failure falls back to the cascade, skipping the tier whose stream just failed (retrying it would cost a second full timeout and overrun the frontend's 90 s budget), and emits restart so the client discards the partial text.
• rca_synthesize(query, context) — RCA-specific prompt, returns timeline + hypotheses
• embed(text, task) — Jina jina-embeddings-v3 (1024-dim), bounded LRU cached (_LRU, 512 entries, keyed on (task, text)). Every search embeds its query before touching Qdrant, so a repeated or polled query previously paid a Jina round-trip each time. Embeddings are deterministic per (task, text) for a fixed model, so caching is safe; a failed embedding ([]) is never cached. Process-local — move to Redis if cross-replica hit rate matters.
• classify_query_category(query) — deterministic classifier mapping free text onto a safety-critical category, or None. Called by POST /search/synthesize (and its stream) for every request: the client's query_category is ignored. It fails closed: keyword lists and unit regexes first, then a catch-all safety_parameter_unspecified for a question that reads like a limit, rating or setting (a number with an engineering unit, a tag plus a parameter word) and no list named. A false positive costs a refusal that points at the sources; a miss costs an ungated answer.
• query_asset_tags(query, aliases) — the assets a question names, by tag or by confirmed alias (resolved to the canonical id); standards and document prefixes such as OISD-117 or SOP-114 are not assets. _gate_evidence(...) then requires every named asset to be vouched for by same-asset evidence (worst-covered asset's best authority, lowest best confidence).
• Prompt hardening. _prompt(user, system) returns a str subclass carrying the system message, so the rules ride in the system channel and the user message holds only the escaped <query> and <document index authority_level document_id> blocks (_untrusted escapes text so it cannot close a tag or forge a [Source n | ...] header). Escaping reduces injection, it does not eliminate it, so the gates never rely on what the model says about authority.
• parse_synthesis_response(answer) — extracts structured fields from LLM output. The model's own trailing lines win (SOURCES_USED last match), CONFIDENCE takes the lowest value, so nothing echoed from a document can raise it; valid_citations(cited, n) drops numbers outside the evidence.
• result_gate fails closed: no parseable CONFIDENCE, or a citation outside the evidence, is a refusal.
• parse_rca_response(answer, allowed_sources) — extracts hypothesis|evidence_weight|sources lines; a hypothesis may cite only the document ids the model was shown, and weights are clamped to 1.0

Every tier failing on 429 is reported as such. Each provider tags a rate_limited result; when the whole cascade fails and any tier was rate-limited, the response gets rate_limited: true plus a message naming them. The second tier is a free-tier Gemini key, so under sustained load answer quality degrades toward zero while retrieval keeps working — configure Ollama as a genuine third tier, move Gemini to paid, or drop the cascade so a NIM failure surfaces as an honest error.

Provider cascade (_synthesize_cascade): Nebius Token Factory → NIM → OpenRouter → Gemini → Ollama. The order and the per-tier settings live in one place, services/model_providers.py (Provider, all_tiers, synthesis_cascade). all_tiers is every tier, keyed or not, and is what the /health/model probe looks names up in; synthesis_cascade is the keyed subset that _synthesize_cascade iterates. A tier with no API key is not in the list at all, so defaults ship with only NVIDIA_NIM_API_KEY set and the cascade is NIM-only, identical to before the registry existed. Every OpenAI-compatible tier goes through one call path, _synthesize_provider(provider, …): same retry on 502/503/504, same verify_served_model check, same failure shape. Ollama keeps its own method because /api/generate is a different wire format, not a different URL.

Token Factory is wired but unkeyed, so it is inert until NEBIUS_TOKEN_FACTORY_API_KEY is set. It serves the same Nemotron build as NIM, so promoting it changes who serves the model rather than which model answers, and the benchmark counts tokenfactory and nim alike as the pinned model. Fill GEMINI_API_KEY (Google's OpenAI-compatible endpoint) to add a generous free-tier cloud fallback; set OLLAMA_BASE_URL to add the local air-gapped tier (intentionally empty by default, because an accidental local Ollama once consumed ~6 GB RAM). Embeddings stay Jina → Ollama.

Adding a provider is one entry in all_tiers() plus its settings. extra_body carries provider-specific keys (Nemotron's enable_thinking: false), so a key meant for one vendor never reaches another that would reject it.

Turning on Nebius Token Factory

Everything is built; only the key is missing. Steps, locally or on the server:

1. Set NEBIUS_TOKEN_FACTORY_API_KEY in .env.
2. Recreate every container that calls a model, because env changes are read at start: docker compose up -d --force-recreate kairos-backend-api kairos-celery-worker kairos-temporal-activity-worker kairos-elicitation-worker
3. System Health → AI models → turn on Nebius Token Factory. The probe sends exactly what a real answer sends, including the thinking-off flag, and reports the model Token Factory says it served.
   • OK and no model mismatch: done.
   • 400 mentioning chat_template_kwargs: Token Factory rejects the flag. Set NEBIUS_TOKEN_FACTORY_DISABLE_THINKING=false, repeat step 2. NIM keeps its own switch.
4. Ask the Copilot an ordinary question. The line under the answer should read Nebius Token Factory. If it reads NVIDIA NIM, Token Factory failed and the cascade fell through; the API log names why (synthesis.provider_failed, provider=tokenfactory).
5. Re-run benchmark/run_benchmark.py before quoting any quality figure, because the published one was measured through NIM. A run answered by tokenfactory counts as the pinned model.

Safety-critical categories: max_allowable_pressure, isolation_interlock_sequence, torque_specification, electrical_rating, pressure_relief_setting, safety_shutdown_setpoint, safety_parameter_unspecified

The refusal gate clears on either high confidence (≥ 0.7) or an authoritative source (authority_level ≤ 3 = regulatory / engineering / OEM), and refuses only when both fail. Both signals are needed: hybrid search and graph facts carry authority_level but no confidence, so a confidence-only gate read them as 0.0 and would refuse every safety-critical query. Covered by tests/test_query_category.py and tests/test_sec_llm.py.

BriefEngine (services/brief_engine.py)

Assembles operator briefs from 5 parallel graph+vector+ES+Supabase queries.

• assemble_work_order_brief(event) — pulls failure history, open conflicts, procedures, quarantine flags; appends correlated DCS alarms / PTW context. Headline + body are operator-readable prose (grouped record counts, named source documents, ⚠ lines for conflicts/quarantine) — not a raw edge dump. No LLM call (phase discipline); authority/verification stay in the source badges.
• assemble_ptw_brief(event) — adds isolation topology, regulatory requirements
• assemble_shift_handover_brief(event) — pulls active WOs, alarms, open PTWs
• assemble_recurring_failure_brief(event) — triggered when same-family WO detected in 90-day window; priority=high headline includes recurrence count
• assemble_tag_out_brief(event) — pulls downstream topology dependencies for the tagged-out asset
• assemble_inspection_brief(event) — triggered on result=failed or non-empty findings; includes finding text and referenced document
• deliver(brief, redis) — saves to briefs table, publishes to REDIS_STREAM_BRIEFS, records kairos.briefs.delivered metric. 4-hour cool-down per (recipient, asset, trigger event type).
• _get_correlated_events(event_id) — fetches all events sharing the same compound_event_id via Supabase

EventBusService (services/event_bus.py)

Redis Streams producer + EEMUA 191 push governor.

• publish(stream, payload) — XADD to any stream
• publish_work_order(payload) / publish_ptw(payload) — typed publish helpers
• is_duplicate(asset_id, event_type, business_id=None) — read-only Redis key check (DEDUP_WINDOW_MINUTES window). mark_seen(...) records the event and is called last, after the ingest succeeds, so a failed ingest stays retryable. Pass business_id (work_order_id, ptw_id, alarm_id) wherever the event has one. Keyed on (asset_id, event_type) alone it collapses two different permits on one asset into one and the second technician never gets a brief — routine during a turnaround. Wired on all six event routes.
• mark_seen(asset_id, event_type, business_id=None) — the write half of dedup, called as the last step of each event route (marking at check time turned a connector's retry after a 500 into deduplicated and lost the event, and with it a critical PTW brief). Two identical events racing inside one request's runtime can both run; the insert is idempotent and the brief cool-down absorbs the twin.
• correlate_events(asset_id, event_id, occurred_at, supabase) — assigns shared compound_event_id to same-asset events within the dedup window
• check_governor(user_id, priority, site_id, supabase) — returns True if brief can be sent. PTW (priority="critical") always passes. Checks plant state gate then hourly rolling counter.
• get_plant_state(site_id, supabase) — queries plant_operating_states, checks expires_at
• record_push(user_id) — increments hourly counter with 3600s TTL
• get_governor_state(user_id) — returns {state, push_count_last_hour, ceiling, next_delivery_allowed_at}

Event routes (routers/events.py)

• _store_event(supabase, row) — the one insert into operational_events for every event route. event_id is client supplied, so it is a plain insert, not an upsert (an upsert let any ingest role overwrite another event's payload, asset and site by posting its id). A duplicate key (Postgres 23505) is read back: identical type, asset, site and payload is a retry after a partial failure and carries on; anything else is 409 and the stored row is untouched.
• IngestUserDep / _INGEST_ROLES = (engineer, reliability, admin) — require_role guard on the six ingest routes, mirroring the OPA ingest_event grant.
• _canonical_asset(asset_id, driver, supabase, current_user) — resolves a confirmed alias, 404s an unknown tag, and with current_user also 404s an asset on another site (same response, so existence is not disclosed).
• INSPECTION_EVIDENCE_CONFIDENCE = 0.85 — ceiling on the confidence of an inspection-evidence edge. inspection-complete stores the event first, validates that document_id is an inspection_report already in the vault, then writes the graph edge and any quarantine row.
• Acknowledgements (POST /events/{id}/ack) are built server-side: actor from the token, HMAC-SHA256 signature (_sign_acknowledgment, shared with briefs) over event, user, action and server time, one per user per event (a repeat returns repeat: true).
• The pending-brief Redis slot (kairos:brief_pending:{asset_id}:{work_order_id}) belongs to work orders only; tag-out and PTW take none.

SLAService (services/sla_service.py)

Lazy, inline SLA escalation. Called at the top of GET /governance/conflicts and GET /governance/quarantine. No Celery Beat, no scheduled worker.

• check_and_escalate(supabase) — idempotent (guarded by escalated_at IS NULL). Queries overdue conflicts (sla_deadline < NOW()) and quarantine items (sla_due_at < NOW()). Writes escalated_at and appends audit_log entry with action=sla_escalated for each. Returns {conflicts_escalated, quarantine_escalated, checked_at}.

TopologyVerificationService (services/topology.py)

Layer 3 → Layer 7 gate. Derives per-element verification status from each element's quarantine_items.review_status, rolls it up per drawing, and applies engineer decisions.

• SAFETY_CRITICAL_GROUPS = {isolation_boundaries, instrumentation_loops} — canonical_ready requires all of them confirmed and none disputed.
• Confirming an element promotes the CONTAINS_TOPOLOGY_ELEMENT edge the pipeline already wrote; it does not create edges.
• The manifest row is filtered in Python, not in the query — element rows carry no element_type key, so .neq("session_context->>element_type", …) compares against SQL NULL and silently drops every element row.

OtCoverageService (services/ot_coverage.py)

Layer 5 instrumentation coverage, derived from engineer-verified P&ID topology only: verified instrumentation_loops[].instruments[] are the sensor tags. Supabase-only, so the Celery attribution worker uses it directly without an HTTP hop.

coverage_type: "none" means no verified drawing establishes instrumentation — not "this equipment has no sensors". unverified_topology_present separates review backlog from real absence.

Replaces a Go handler that returned hardcoded {asset}-VIBE / {asset}-TEMP / 75% for every asset. Because attribution.py gated its telemetry check on coverage_percent == 0, that value made the check always run and Layer 10's brownfield downgrade unreachable.

TimestampAlignmentService (services/timestamp_alignment.py)

Layer 4 clock alignment. Compares the same correlated event as reported by different source systems, reusing Layer 8's compound_event_id grouping. Normalises to the best-synchronised clock (the historian is site-canonical).

Not occurred_at vs ingested_at — a historical document legitimately occurs months before ingestion, so that comparison flags the whole corpus. The ingestion pipeline still performs that other comparison. Resolved 2026-08-17: the pipeline now treats that gap as ingest lag — it records ingest_lag_recorded and never touches valid_from, which uses the source timestamp.

Report-only while TIMESTAMP_DRIFT_ENFORCE=False.

CircuitBreakerService (services/circuit_breaker.py)

SPC circuit breaker: halts graph writes for an asset class when its 7-day override-count z-score exceeds 2.0.

• check(asset_class) — returns {halted: bool, z_score: float, reason: str, override_count_7d: int}. reason ∈ z_score_exceeded | within_normal_range | stats_error.
• record_override(asset_class, override_type, document_id) — inserts an extraction_overrides row for SPC tracking.
• get_all_states() — returns one state dict per distinct asset_class that has override records (each carries asset_class + the check() fields). GET /governance/circuit-breaker wraps these as {states[], halted_count}.

NERService (services/ner.py)

Named entity recognition for the extraction pipeline. Cloud-first: NIM primary, then a regex last resort. (A local Ollama tier exists in code but is disabled — OLLAMA_BASE_URL is empty; see §6 LLMService.)

• NERService(model=…) — overrides NVIDIA_NIM_NER_MODEL for this instance. The Layer-0 model gate scores a candidate model, so it must be able to pick one; without this the gate always called the env-var model and merely labelled the result with the requested name.
• NIM is called at NVIDIA_NIM_BASE_URL (the same setting synthesis honours; it used to be hard-coded to NVIDIA's public endpoint, so a private gateway still received the key and the document text). An entity returned with a missing, non-numeric or out-of-range confidence gets _DEFAULT_ENTITY_CONFIDENCE = 0.5, below the 0.7 quarantine line, so it goes to review rather than into the graph; one garbled score no longer discards the whole document's entities.
• extract_entities(text, document_type) — tries NIM meta/llama-3.2-11b-vision-instruct; falls back to regex ASSET_TAG pattern matching (Ollama tier skipped while OLLAMA_BASE_URL is empty). Returns [(entity, entity_type, confidence)]
• Character offsets are recovered (_with_spans): the model returns entity text with no positions, so start/end used to come back None — leaving the annotation UI unable to highlight an entity in its source document. Each entity is located in the original text, with a per-value cursor so a repeated mention gets successive positions instead of collapsing onto the first. Entities the model paraphrased are left unlocated rather than guessed. This does not affect the Layer-0 F1 metric: workers/model_validation.py matches on surface-form overlap (_span_match), never on offsets.

OCRService (services/ocr.py)

OCR for the extraction pipeline. Cloud-first for scanned documents; zero-API-cost fast path for native digital PDFs.

• extract_text(file_bytes, mime_type) — Fast paths (no API cost):

  Input	mime	How
  Plain text / markdown / CSV	text/plain, text/markdown, text/csv	UTF-8 decode
  Spreadsheets	…spreadsheetml.sheet (.xlsx), vnd.ms-excel, …opendocument.spreadsheet (.ods)	openpyxl read-only streaming → every sheet flattened to tab-separated rows, prefixed # Sheet: <name>; blank rows dropped
  Email archives	message/rfc822 (.eml), application/mbox, text/rfc822-headers	stdlib email — Date/From/To/Cc/Subject headers + text/plain bodies; attachments listed by filename, never decoded inline (their bytes belong in the vault as their own documents). An mbox is split on its own From delimiter, so every message is extracted
  Digital PDFs	application/pdf	PyMuPDF native text

  Cloud path: nvidia/nemotron-ocr-v2 for scanned documents and images.

  Resource ceilings (uploader-controlled files are rasterised or flattened in the one ingestion worker): MAX_PDF_PAGES = 100, DPI capped at 150 and by MAX_RASTER_PIXELS = 16,000,000 per page (capped_dpi; a page too large to stay legible at 30 DPI is rejected), MAX_RASTER_TOTAL_BYTES = 150 MB of PNG across all pages, spreadsheets capped at 50,000 rows, 256 columns and 200 MB declared uncompressed (checked from the zip directory before openpyxl reads it). A file over a ceiling is treated as unreadable and routes to human review, never silently truncated.

  Spreadsheets and email archives are two of the six source types named in the problem statement. openpyxl is used rather than hand-rolled zip+XML because xlsx cell typing (dates stored as serial numbers, shared vs inline strings) is exactly the silent corruption that must not reach the graph. Covered by tests/test_ingestion_formats.py.

Important: The OCR model uses the NVIDIA CV API (https://ai.api.nvidia.com/v1/cv/<model>), NOT the chat completions endpoint (integrate.api.nvidia.com). Request format uses "input": [{"type": "image_url", "url": "data:..."}].

Response format is {"data": [{"text_detections": [{"bounding_box": ..., "text_prediction": {"text": ..., "confidence": ...}}]}]}. This line previously documented a label key that the API has never returned — _detection_text read it, every line resolved to "", and the caller reported nim_returned_no_text, which was then read for weeks as the "no handwriting model" Layer-3 limitation. Fixed 2026-08-23; label/text are retained as fallbacks. If you change this parser, ocr.detections_unparsed is the log line that tells you the schema moved — it is deliberately distinct from ocr.no_detections (the model ran and saw nothing).

Confidence is the model's own, not a constant. overall_confidence is the per-span confidence weighted by span length, reported with min_span_confidence, low_confidence_spans and span_count. It was hardcoded to 0.95 until 2026-08-23, which made a garbled scan indistinguishable from a clean one to every downstream gate — including the < 0.7 → quarantine rule, applied to a number that could never be below 0.7.

Inline image limit: 180 KB base64. Pages are rasterized at 96 DPI to stay under it, and an image that still exceeds it is re-encoded by _shrink_for_inline (JPEG first, then progressive downscale) rather than skipped. It used to return "" silently, so the corpus's two degraded scans (11x and 13x over) never reached the model at all. Near-duplicate of PIDService._fit_b64; _NIM_IMAGE_SIZE_LIMIT is defined in both modules and they must move together.

PIDService (services/pid.py)

Layer 3, Path B — P&ID engineering-drawing topology extraction via a cloud vision model. Vision-understanding, not OCR (OCR destroys the drawing's connections).

• extract_topology(file_bytes, mime_type) -> dict | None — rasterizes the drawing (150 DPI for PDFs, lowered by ocr.capped_dpi so one page never becomes a multi-gigabyte bitmap; an A0 sheet at 150 DPI already exceeds the pixel cap), downscales to fit the inline size cap (Pillow), sends it to NIM meta/llama-3.2-11b-vision-instruct (chat completions, OpenAI-style image_url content) with a schema-locked prompt, and parses the returned topology JSON (equipment_nodes, isolation_valves, instrumentation_loops, isolation_boundaries). Returns None on any failure so the pipeline falls back to the demo fixture.
• Wired into run_ocr for document_type='pid_drawing'. The result carries topology_source (vision_model | demo_fixture) through the manifest and GET /documents/{id}/topology so a fixture never masquerades as a real extraction. Every element still routes to element-by-element engineer verification (Layer 7).
• Path A (custom YOLOv9 + LayoutLMv3 on GPU) is the documented future upgrade — see ARCHITECTURE.md Layer 3.

PIIService (services/pii.py)

PII detection and masking for the DPDP Act 2023 export boundary.

• detect(text, person_names) -> [span] — non-overlapping spans, earliest-start wins with longest-match tiebreak. Structured identifiers by regex (EMAIL, PAN, AADHAAR, EMPLOYEE_ID, SHIFT_ID, Indian PHONE: +91 / (+91) / 91 / 0 prefixes and a 5+5 split; Aadhaar with spaces, dashes or none); PERSON names supplied by the caller from NERService, plus context patterns (_NAME_CONTEXT) for names the model never saw because NER reads only the first part of a document: after an honorific (Mr, Dr, Shri, Smt, ...) or a sign-off label (Prepared by, Approved by, Operator:, Name:, ...). No second name model.
• redact(text, person_names) -> {redacted_text, spans, counts, pii_found} — masks each distinct value with a stable pseudonym ([PERSON_1]), so cross-references in the text survive redaction where a blanket [REDACTED] would destroy them. Offsets in spans refer to the original text.

Runs at export, never at ingestion. Operational knowledge legitimately contains personnel names — "which technician signed off the EQ-101 seal repair" is a real maintenance question (benchmark Q15) — so redacting on the way in would destroy retrieval. Equipment tags and part numbers are deliberately never matched; tests/test_pii.py pins that.

Exposed via GET /documents/{id}/redacted, which also writes a pii_redacted_export row to audit_log (type counts only, never matched values).

Scope: ARCHITECTURE.md describes this pipeline as gating cross-site knowledge promotion. No cross-site promotion endpoint exists in the codebase, so that wiring is not built; the redaction pipeline itself is.

corpus (services/corpus.py)

Tells real vault documents apart from test artifacts — rows the suite and hand-run sweeps write into the vault. They carry ordinary random DOC- ids, so only the file name identifies them, and on 2026-08-23 they were 87 of the 108 active documents. Anything that counts or renders documents without excluding them reports test hygiene rather than the plant.

• is_test_artifact(file_name) — the single predicate. Anchored prefixes: ann_test_, dbtest_, test_, e2e_, kairos_, probe, tmp, # Kairos. e2e_ and kairos_ were added by decision D8: both name a file after this system or its harness rather than after plant equipment. A _test\.ext stem rule was rejected — it would also match a plausible real hydro_test.pdf, and hiding plant evidence is the one failure this predicate must not have.
• document_rows(supabase, document_ids) — resolves ids to documents rows (document_id, file_name) in one bulk lookup (chunked at 200). Read-only. Fails open: on a lookup error it returns [], so a Supabase blip shows extra noise rather than blanking a real graph.
• partition_test_artifacts(rows) — pure; given those rows, returns the test-artifact ids.
• TEST_ASSET_PREFIXES / REAL_ASSET_CYPHER — the same idea for assets. Prefixes: QA-TEST- (a manual QA sweep's QA-TEST-155635, matched 1 of 11 live assets and 0 of the 10 golden ones) plus the integration suite's ASSET-TEST-, ASSET-DEDUP-, ASSET-EV-, ASSET-ACK-, ASSET-FRESH-. The guard is a Cypher fragment over a, placed inside the asset-list, compliance (gaps, dashboard, audit pack) and coverage queries, because filtering a paginated or aggregated result in Python breaks total. GET /assets/ reports what it hid as excluded_test_assets. The list is deliberately separate from scripts/purge_test_data.py's: that script deletes, this one only hides, and sharing a list would let a display tweak arm a cloud delete.

Two rules that matter:

• An id absent from documents is never an artifact. PROMOTED-<uuid> ids are minted by quarantine promotion for field knowledge that never had a vault document, so "cannot classify" must mean "keep".
• Callers report what they excluded. GET /assets/{id}/knowledge returns excluded_test_documents; run_kg_completeness.py prints its count. A filter that hides its own effect is how the linkage figure stayed wrong for as long as it did.

Consumed by routers/assets.py and benchmark/run_kg_completeness.py — the harness imports it rather than keeping a second copy, since benchmark/ already depends on api/. Covered by tests/test_corpus_filter.py, which pins every real corpus file name against over-matching.

shared_client (services/http.py)

Pooled outbound HTTP client for model-provider calls, replacing a per-call httpx.AsyncClient in llm.py, ner.py, ocr.py and pid.py — each of which meant a fresh TCP + TLS handshake per request, with no keep-alive and socket churn under concurrency.

• shared_client(default_timeout) -> AsyncClient — cached per event loop, not globally. An AsyncClient binds to the loop that created it, and this code runs under several: the FastAPI loop and a fresh asyncio.run() loop per Celery task. A single global client would be handed to a dead loop and raise. Limits: 32 connections, 16 keep-alive, 30 s expiry.
• close_shared_client() — awaited from the FastAPI lifespan shutdown hook.

Always pass an explicit timeout= per request. The cached client keeps the first caller's default, so a 30 s embedding call would otherwise cap a long NIM synthesis.

Covered by tests/test_http_pool.py, including the loop-rebinding case.

metrics (services/metrics.py)

OTEL custom metric instruments. All no-ops when MeterProvider is not configured — which means every process that records one must call setup_telemetry(), not just the API.

briefs.delivered and governor.suppressed are recorded inside Celery tasks (brief_engine, event_bus). The worker never configured telemetry, so both were permanent no-ops and could not reach Grafana under any amount of real traffic (confirmed 2026-08-15 by delivering a real brief and finding kairos_briefs_delivered_total absent). workers/celery_app.py now calls setup_telemetry() on worker_process_init — per forked child, since an exporter's background thread does not survive a fork.

Exported names are the OTLP → Prometheus normalisation of the dotted instrument names, e.g. kairos.briefs.delivered → kairos_briefs_delivered_total, kairos.ingestion.duration → kairos_ingestion_duration_seconds_{bucket,count,sum}.

7. Temporal Workflows and Activities

DocumentIngestionWorkflow (workflows/document_pipeline.py)

Task queue: kairos-ingestion. Triggered by POST /documents/ingest.

Seven sequential activities. store_in_vault raises a non-retryable ApplicationError(type="BadFile") for a missing registry row or a SHA-256 mismatch (non_retryable_error_types=["BadFile"]: a wrong file stays wrong, and retrying it five times only re-crashes the single ingestion worker). link_to_graph resolves aliases from confirmed rows only, and its own alias proposals are written ignore_duplicates=True.

index_vectors and index_text run in parallel.

Canonical valid_from + ingest lag (in mark_complete): KNOWLEDGE_EDGE.valid_from uses the document's occurred_at whenever it is parseable and not future-dated, falling back to ingested_at. A gap beyond TIMESTAMP_DRIFT_TOLERANCE_MINUTES is recorded as audit_log action=ingest_lag_recorded — an observation only.

Until 2026-08-17 this block called the gap "drift" and overwrote valid_from with ingested_at past the tolerance, which would have rewritten every historical document's true date to its upload time and corrupted the validity windows Layer-4 time-travel depends on. It never fired only because occurred_at is NULL across the corpus. Real drift is the same event seen by two source systems — services/timestamp_alignment.py, surfaced at GET /governance/timestamp-drift — and extraction_jobs.timestamp_drift_detected now means only that.

vault_path safety: workflow gracefully exits with {"status": "failed", "reason": "missing vault_path"} if vault_path is absent from params (guards against stale workflow history replays).

MicroInterviewWorkflow (workflows/elicitation_workflow.py)

Task queue: kairos-elicitation. Triggered by POST /elicitation/trigger.

1. Queries Neo4j for gap-filling questions (failure code + asset context)
2. Calls LLM to generate 3–7 targeted questions
3. Saves session to elicitation_sessions with status=questions_ready
4. Waits for StoreElicitationResponseWorkflow signal

StoreElicitationResponseWorkflow

Triggered by POST /elicitation/{work_order_id}/responses. Stores Q&A pairs as a quarantine_items row for human review before graph promotion.

8. Celery Workers

App defined in workers/celery_app.py. Broker and result backend: redis://kairos-redis:6379/1.

Six queues: ingestion, extraction, attribution, transcription, elicitation, validation.

All workers use lazy imports inside the task body (never at module level) to avoid Celery fork-time conflicts.

Brief Assembly Worker (workers/brief_assembly.py)

Task: workers.brief_assembly.assemble_brief(event_type, event_dict) on the ingestion queue.

Supported event_type values and their brief assemblers:

• work_order_created → assemble_work_order_brief
• shift_handover_created → assemble_shift_handover_brief
• ptw_created → assemble_ptw_brief
• equipment_tag_out → assemble_tag_out_brief
• inspection_complete → assemble_inspection_brief
• recurring_failure_detected → assemble_recurring_failure_brief

Triggered with apply_async(countdown=LATE_ARRIVAL_WINDOW_MINUTES*60) for WO, tag-out, and shift-handover events. PTW and inspection (failed/findings) briefs are immediate.

Redis pending key: kairos:brief_pending:{asset_id}:{work_order_id} — stores the Celery task ID so a re-report of the same work order can revoke and re-enqueue its pending brief. A PTW or another work order on the asset never touches it.

Attribution Worker (workers/attribution.py)

Task: workers.attribution.evaluate_outcome(event_id, asset_id)

Triggered from POST /events/work-order when count(WO for same asset in last 30 days) > 1. Its Go historian call carries X-Connector-Secret (CONNECTOR_SHARED_SECRET).

Three independent checks — all must pass for genuine_failure=True:

Failure families in api/utils/failure_families.py: mechanical, seal, electrical, process. If genuine_failure=True: writes attribution_flag to audit_log. No automatic confidence adjustment.

Voice Transcription Worker (workers/voice_transcription.py)

Task: workers.voice_transcription.transcribe_voice_note(work_order_id, storage_path, sha256, submitted_by, filename) on the transcription queue.

1. Downloads audio from Supabase Storage
2. Sends to Groq whisper-large-v3 for transcription
3. Runs NER on transcript
4. Inserts into quarantine_items with input_type=voice_note and session_context={sha256, vault_path, ...}

Off-Boarding Worker (workers/offboarding.py)

Task: workers.offboarding.generate_offboarding_questions(item_id) on the elicitation queue.

Triggered by POST /elicitation/offboarding — one task per equipment family session item, scheduled at the item's scheduled_for time.

1. Fetches session item from offboarding_session_items
2. Queries Neo4j for equipment family knowledge gaps
3. Calls LLM with offboarding-specific prompt to generate 5–8 questions
4. Falls back to 5 generic questions if LLM is unavailable
5. Updates item status=questions_ready and saves questions to offboarding_session_items.questions

Model Validation Worker (workers/model_validation.py)

Task: workers.model_validation.run_model_gate(model_name) on the validation queue.

Triggered by POST /governance/model-gate/run (admin only). model_name is optional — the endpoint defaults it to NVIDIA_NIM_NER_MODEL so the UI can trigger without knowing the model name. The endpoint only enqueues; the task itself runs ~12 min against a 52-row corpus, so the UI shows a "queued" banner and polls history until the run lands.

The "~2.5 min" this section claimed until 2026-08-23 was measured on a run where almost every NER call failed fast on a 429. Once the calls actually reach the model each costs tens of seconds. time_limit/soft_time_limit are 1860/1800 for that reason — at 600/540 two consecutive runs were killed mid-flight and wrote no history entry at all, which is worse than recording a degraded one. The limit tracks corpus_size × per-call latency; revisit it if the corpus grows.

1. Loads validation corpus from validation_corpus
2. Runs NER on each unique document via FallbackCountingNER, which tallies which path produced each extraction (nim / ollama / regex)
3. Computes precision, recall, F1 per entity type, per asset class and per document type — all three partitions share one run-scoped extraction cache, so the extra cuts cost zero model calls. Ground-truth labels outside the prompt's taxonomy are excluded and reported as unscoreable_by_type rather than scored as failures
4. Retrieves the incumbent baseline: the most recent run with validity: "VALID", filtered in Python — a PostgREST .neq() on a missing JSONB key matches nothing, and legacy rows have no validity key at all
5. passed = True when no entity type regressed against that baseline (or there is no eligible baseline). Orthogonal to validity, which reports whether the run reached the model
6. Writes result to audit_log with action=model_gate_result

Graph query policy (ARCHITECTURE.md §7)

• graph.MAX_TRAVERSAL_DEPTH is the single traversal bound, interpolated into the one variable-length pattern rather than written inline per query. Cypher cannot parameterise a variable-length bound (*1..$n is a syntax error), so it is an f-string over an int constant — never over user input. tests/test_graph_query_policy.py fails if an unbounded * ever ships.
• Authority pre-filtering "before traversal" has nothing to apply to here. The Layer 4 hot path is a 1-hop expand, where filter-after-expand is the plan: PROFILE shows NodeUniqueIndexSeek → Expand(All) → Filter. The requirement bites on multi-hop queries, and none exists.
• make graph-perf (scripts/verify_graph_perf.py) is the §7 regression check. It asserts plan shape, not dbHits: thresholds move with the corpus, so they would either be loosened until meaningless or go red on ordinary growth. The regression it exists to catch already happened once — asset_id_unique went missing and the hot path silently became a NodeByLabelScan, returning correct rows and failing nothing.
• Hot-asset Redis precompute is deliberately not built. A precomputed view that goes stale after a KNOWLEDGE_EDGE write is the silent-staleness failure mode the architecture calls its most dangerous. Build it when a PROFILE on a real corpus justifies it, with explicit invalidation on every edge write.

9. Go OT Connector

Service: kairos-backend-go at http://kairos-backend-go:8090.

Source: backend/connectors/cmd/connector/main.go; auth and boot validation in auth.go.

Authentication. Every route except GET /health is behind requireSecret: the caller must send X-Connector-Secret equal to CONNECTOR_SHARED_SECRET (both sides SHA-256 hashed, then constant-time compared; otherwise 401). validateConfig runs at start and the process exits when CONNECTOR_SHARED_SECRET is unset, and, unless APP_ENV is exactly development, when it or INTERNAL_API_KEY is still the dev default. Callers: the FastAPI /health/connectors proxy and workers/attribution.py read CONNECTOR_SHARED_SECRET from their environment (docker-compose passes the same value to all three). Hardening: request bodies capped at 1 MiB (413), upstream replies read through a 4 MiB limit, a shared http.Client with a 15 s timeout, ReadHeaderTimeout 10 s, upstream and fixture failures logged and answered with generic text, and the PI Web API URL never logged (it may carry credentials). internal/ot/client.go URL-escapes the tag, web id and times, and strips the request URL from transport errors. In the dev override the port binds 127.0.0.1 only.

Endpoints

External systems — mock by design. The OT historian (PI Web API) and EAM sync (SAP/Maximo) integrate with plant enterprise systems Kairos does not own. With no live plant to connect to, they run on mock/fixture data by design — the intended state, not a gap. The real paths exist (PIWebAPIClient built; SAP ODS + OPC-UA are stubs) and would activate via PI_WEBAPI_BASE_URL / EAM_ODS_ENDPOINT if a plant were ever connected. Completion status for every layer: docs/implementation/status.md.

PI Web API Client (internal/ot/client.go)

When PI_WEBAPI_BASE_URL is configured:

1. GET {baseURL}/search?q={tag} → resolves WebID (tag URL-escaped)
2. GET {baseURL}/streams/{webId}/recorded?startTime={from}&endTime={to} → time series (each value escaped; response bodies capped at 4 MiB)

Basic Auth via PI_WEBAPI_USERNAME / PI_WEBAPI_PASSWORD.

10. Database Schemas

Supabase PostgreSQL (single source of truth: db/schema.sql)

All 16 historical migrations (001–016) are folded into db/schema.sql — apply that one file to a fresh database. The live applied history is tracked by Supabase itself in supabase_migrations.schema_migrations. The tables:

quarantine_items.input_type CHECK constraint allowed values (migration 013): field_observation, voice_note, elicitation_response, deviation_flag, offboarding_response

audit_log uses timestamp column, not created_at.

RLS policies (migration 004):

• briefs — service-role key bypasses; user JWTs see only own rows
• quarantine_items — read-only for field_worker role

Neo4j Graph Schema (db/neo4j/init_schema.cypher)

Node labels: Asset, Document, Event, Person, Concept, Organisation

Relationship types:

• KNOWLEDGE_EDGE — primary knowledge relationship, carries all 6 temporal properties
• PARENT_OF — asset hierarchy (parent → child)
• LINKED_TO — document↔asset link
• MENTIONS — event references asset
• RESOLVED_BY — conflict resolution link
• INSPECTION_RECORD — inspection result linked to asset

Required KNOWLEDGE_EDGE properties (all 6, every write):

11. Configuration Reference

All settings in api/config.py via pydantic-settings. Source: .env file.

App

Boot guard (_no_insecure_defaults_in_prod). Outside development the API refuses to start while INTERNAL_API_KEY or APP_SECRET_KEY is a dev default, the Supabase service-role key or JWT secret is empty, MOC_WEBHOOK_SECRET is unset, or APP_DEBUG is true. The Go connector applies the same idea to CONNECTOR_SHARED_SECRET.

Supabase

Critical: Never use the service-role key for auth.sign_in_with_password() or auth.get_user(). Always use a fresh create_client(url, ANON_KEY) for auth operations.

Neo4j

Qdrant

Elasticsearch

Redis

Docker note: Inside containers, localhost is not the Redis host. docker-compose.yml explicitly sets REDIS_URL=redis://kairos-redis:6379/0 and CELERY_BROKER_URL=redis://kairos-redis:6379/1.

LLM

Temporal

EEMUA 191 Governor

Go Connector

OpenTelemetry

12. Auth and Authorization

JWT Flow (Supabase Auth)

1. Client calls POST /auth/login → gets access_token (ES256 JWT)
2. Client includes Authorization: Bearer <access_token> on all requests
3. get_current_user calls supabase.auth.get_user(token) using a fresh anon client
4. Role extracted from user.app_metadata.role (not user_metadata, which the user can edit, and not the top-level role field)
5. POST /auth/logout revokes the session server side (scope="local"); the frontend calls it on sign-out and on forced expiry, best effort

app_metadata is writable only with the service-role key. Users created before this rule keep their role in user_metadata and would all resolve as field_worker with no site, so run scripts/migrate_roles_to_app_metadata.py (dry run, then --apply) before deploying the code that reads app_metadata only. It never overwrites an existing app_metadata key and does not touch user_metadata. services/identity.display_name reads the display name from app_metadata for the same reason (a name on a PTW sign-off must not be user-editable).

Test Users

Created by docker exec kairos-backend-api python scripts/seed_users.py.

All five roles in the OPA policy are loginable. reliability and compliance are the two personas that actually demonstrate governance — quarantine promotion and read-only audit — so a demo that skips them skips the point.

Role Permissions (OPA Policy policies/kairos.rego)

Engineers deliberately cannot promote_quarantine or countersign_brief. That is what makes the one-way quarantine gate and the PTW dual signature real: the second signature can never come from the issuing role. _sensitive_actions in the policy lists both, plus resolve_admin_conflict, write_assets, ingest_document, ingest_event and all six read_* actions — without that listing the catch-all rule would grant them to every authenticated role.

read_nonconformance is narrower than read_governance on purpose. The compliance auditor's /compliance/nonconformance page reads /governance/conflicts + /governance/quarantine, so it needs those two children without reaching the model gate, MoC approvals or the circuit breaker. These grants mirror the frontend route table in components/use-role.ts; keep the two in step.

Where a role comes from. dependencies.resolve_token reads role, site_id and name from the token user's app_metadata, which only the service role can write. It never trusts user_metadata (users can edit that). The role migration (scripts/migrate_roles_to_app_metadata.py) has been applied to every account.

OPA Middleware

OPAMiddleware in api/middleware/opa.py intercepts all POST/PUT/PATCH/DELETE requests and GET/HEAD on the sensitive read prefixes (/audit-log, /compliance, /governance, /documents, /events), except /health, /auth, /docs, and the one signed webhook POST /governance/moc/webhook (_SIGNED_WEBHOOKS: authenticated by its own HMAC, and OPA needs a user the plant's MoC system does not have). Maps route prefix to OPA action name; the six event-ingest routes are listed by exact route as ingest_event (not by the /events prefix, so deviation flags, acks and plant-state stay on the generic write action), calls http://kairos-opa:8181/v1/data/kairos/authz/allow. 403 if denied.

Five things about it are load-bearing, each of which was a live defect:

• It reads request.scope["path"], never request.url.path, which is rebuilt from the Host header (a crafted Host made a gated route look like /health/..., which the skip list exempts).

• It fails closed. _ask_opa returns self.debug when OPA is unreachable, never a bare True.

• It does not verify tokens itself — it calls dependencies.resolve_token, the app's single verifier. Supabase issues ES256 here, so the old hand-rolled HS256 decode rejected every real token and the middleware degraded to the dev pass-through.

• OPA_URL must be the compose service name. localhost:8181 inside the container is the API, not OPA; combined with fail-open that made the policy decorative.

• OPTIONS is never gated. This middleware is outermost, so it sees the CORS preflight — which carries no Authorization header — before CORSMiddleware does.

/events/plant-state is exempt from read_events: the app shell renders plant state for every persona, and a field worker who cannot see that the plant is in shutdown is a safety regression.

Verify with tools/verify_authz_policy.sh (39-case decision matrix against a throwaway OPA) — and separately confirm the layer is reached, by probing the live API with a restricted persona's token and checking for a 403.

Rate-limit Middleware

RateLimitMiddleware in api/middleware/ratelimit.py (outermost) caps requests per client IP using a Redis fixed-window counter (RATE_LIMIT_PER_MINUTE, default 120). Enforced whenever APP_ENV is not development (0 = off in development so bursts never trip it). Fails open if Redis is unreachable; /health* exempt. Trusts the first X-Forwarded-For hop (behind Caddy). Pairs with the MAX_UPLOAD_MB (25) cap on /documents/ingest — both are public-exposure abuse guards.

Request hardening (2026-10-01 security pass)

Behaviour that is easy to regress, each pinned by a tests/test_sec_*.py case. Detail and status per finding: implementation/status.md § Accepted risks and deploy checklist.

• Documents (routers/documents.py): DOCUMENT_TYPES (7 values), ALLOWED_MIME_TYPES (anything else is stored as application/octet-stream), safe_filename for the storage path, parse_occurred_at (not future, not older than 30 years), and AUTHORITY_ASSERT_ROLES = {admin, reliability}: any other role asking for authority 1 to 3 is capped to 4 and the response reports authority_capped. Superseding a document of authority 1 to 3 is limited to SUPERSEDE_GATED_ROLES and needs an approved MoC (deterministic id supersede_moc_id); the first request returns 202 pending_moc_approval and a repeat after approval applies it. Signed artifact URLs carry download= so a stored file is never rendered in the storage origin.
• MoC webhook (routers/governance.py): verify_moc_webhook checks HMAC-SHA256 over "{X-Webhook-Timestamp}." + raw body in X-Webhook-Signature, with a 5-minute skew window (MOC_WEBHOOK_MAX_SKEW_SECONDS), fail-closed when no secret is configured outside development, and a replay of an already approved MoC is 409.
• Briefs: an acknowledgement is final (409 on a second one); a PTW brief is readable by any staff role; the inbox filter uses .in_() over _brief_recipients, never a string built from the site_id claim.
• Annotations: only reliability and admin feed validation_corpus; 60 corrections per user per hour; repeat corrections on one document count once toward the circuit breaker.
• Elicitation: the submitter is always the token's user; off-boarding programmes are visible to staff and the person concerned only (_require_staff_or_subject); voice-note storage paths are sanitised (_safe_segment) and read in 1 MB chunks against MAX_UPLOAD_MB.
• Site scope: asset routes use scoped_asset (one 404 for missing or other-site), events, the RCA pack, the audit pack and the coverage and pending-alias lists filter by site_scope.
• Dependencies (requirements.txt): fastapi==0.135.1 with starlette>=1.3.1 (fastapi 0.111.1 capped starlette at 0.37.x, which carries a BadHost advisory and multipart DoS advisories; 0.135.1 is the newest that still accepts the pinned pydantic 2.7.4) and anyio==4.14.2. Removed as unused: python-jose (and its ecdsa), passlib, flower, rich, tenacity, python-dateutil, pytz. numpy stays pinned only because qdrant-client 1.9.1 needs it. The CI pip-audit step now ignores nothing.

Internal Service Auth

Go connector and service-to-service calls use Authorization: Bearer <INTERNAL_API_KEY>, compared in constant time (hmac.compare_digest). The connector's own inbound routes are separately protected by X-Connector-Secret (§9). resolve_token recognizes the key and returns a service admin account without calling Supabase — which is also what keeps the fail-closed OPA middleware from 401ing every connector and Celery write, since the key is not a JWT.

13. Observability

Pipeline, metrics & dashboards

Moved to docs/INFRA.md §6–7 — the OTEL trace/metric pipeline, the custom Prometheus metrics table (kairos_briefs_delivered_total, kairos_governor_suppressed_total, kairos_ingestion_duration_seconds, kairos_conflicts_open), and the Grafana dashboards/credentials.

App-level instrumentation: FastAPIInstrumentor, RedisInstrumentor, HTTPXClientInstrumentor are set up in api/middleware/telemetry.py (setup_telemetry(app)), called from create_app().

Structured Logging

All logs via structlog. Never use print() or stdlib logging. Notable log events:

• governor.suppressed — user_id, count, ceiling
• governor.plant_state_suppression — user_id, site_id, plant_state
• brief_engine.delivered — brief_id, recipient, priority
• attribution.complete — event_id, asset_id, genuine_failure, action
• ingest.complete — document_id, sha256, job_id
• event_bus.compound_event_linked — compound_event_id, event_ids
• activity.ingest_lag_recorded — document_id, lag_minutes (observation; valid_from unaffected)
• activity.occurred_at_in_future / activity.occurred_at_unparseable — fall back to ingest time
• rca_pack.generated — asset_id, failure_code, timeline_count
• document_pipeline.missing_vault_path — document_id (graceful early exit)
• offboarding.programme_created — session_id, personnel_id, sessions

14. Non-Negotiable Rules

These apply to every code change in this codebase. Violations are bugs.

1. All 6 KNOWLEDGE_EDGE properties on every write: valid_from, valid_to, authority_level, document_id, confidence, verification_status

2. Vault is permanent. Never delete. Never overwrite. Supersede by closing valid_to. Supabase Storage objects are immutable.

3. Quarantine is a one-way gate. confidence < 0.7 or unresolved entity → quarantine_items. Human action only to promote. No auto-promotion ever.

4. Asset nodes: MERGE (a:Asset {asset_id: $id}) SET a += $props — never CREATE.

5. Authority pre-filter before graph traversal. Always: WHERE r.authority_level <= $max_level AND r.valid_from <= $as_of

6. Safety-critical parameter queries — explicit refusal when confidence < 0.7. Never hedge. Return sources directly.

7. Phase discipline. Phase 2 LLM synthesis lives only in POST /search/synthesize (and its stream). Never auto-triggered from routers or workers. The evidence and the safety category are produced server-side; a request's context and query_category are ignored.

8. EEMUA 191 Governor. Call EventBusService.check_governor(user_id) before every brief delivery. PTW (priority="critical") always exempt.

9. Secrets via env vars. All config through api/config.py Settings. Never hardcode.

10. OT data is ephemeral. Query historian data in memory, reason with it, discard. Never store time-series in Kairos.

11. structlog only. Never print(), never import logging. workflow.logger in Temporal is a stdlib logger — use f-strings for message formatting, not keyword args.

12. Supabase client hygiene. Never call auth.sign_in_with_password() or auth.get_user() on the global service-role client. Use a fresh create_client(url, ANON_KEY).

13. Docker is the only runtime. No python or pip install outside containers. Hot-reload is active — edits apply immediately. Rebuild only when new pip deps are added.

14. Lazy imports in Celery workers. Import api.services.* inside the task body, never at module level. Add sys.path.insert(0, "/app") at the top of each worker file.

Dev commands and service URLs moved to docs/INFRA.md — Section 9.