Introduction

What is Qadra?

Qadra is a multitenant workload orchestration platform designed for enterprise AI workflows. Unlike traditional chat applications, Qadra treats AI agents as specialized team members that perform discrete tasks within structured pipelines.

The Problem with Traditional AI Applications

Most LLM applications follow this pattern:

1. User asks a question
2. System retrieves "relevant" documents via RAG
3. Everything gets stuffed into a prompt
4. Model generates a response
5. Repeat

This approach has fundamental problems:

Context Window Abuse: Models have limited context windows. Traditional RAG dumps retrieved documents into the prompt, hoping the model figures out what's relevant. As complexity grows, you hit limits—and even before that, quality degrades because signal gets lost in noise.

Hallucination at Scale: When you ask a model to do everything (retrieve, reason, synthesize, format) in one shot, it hallucinates. The model confidently produces plausible-sounding nonsense because it has no structured way to verify claims against facts.

No Accumulation: Each interaction starts fresh. A research analyst who spent an hour gathering data doesn't pass that structured knowledge to the due diligence specialist—instead, you re-retrieve, re-prompt, and hope for consistency.

Prompt Spaghetti: As applications grow, prompts become unmaintainable. You end up with 10,000-token system prompts trying to handle every edge case, version-controlled in git with no testing strategy.

Core Thesis

Agents are programs that sometimes call models, not prompts that do everything.

This single idea changes everything:

Programs have structure. An agent has an identity (display name, avatar), a defined purpose (description), and behavioral instructions (system prompt). This is inspectable, testable, versionable.

Programs compose. Complex workflows aren't giant prompts—they're pipelines of specialized agents, or visual flow graphs that orchestrate multiple agents with branching, conditions, and human-in-the-loop gates. The research analyst hands off to the due diligence specialist, who hands off to the memo writer. Each does one job well.

Programs are bounded. An agent doesn't see "everything"—it sees exactly what the orchestration layer provides. Context flows forward through pipeline stages, accumulated and structured. This is demand-driven context, not supply-driven context dump.

Qadra's Approach

• Persona-driven Agents: Each agent is a named specialist with a system prompt that defines its behavior, a description of its capabilities, and an identity (display name, avatar). Agents call LLMs with scoped context—they are team members, not catch-all assistants.

• Flow-based Orchestration: Work is coordinated through visual flow graphs (built with a drag-and-drop flow builder) or sequential pipelines. Flow graphs support branching, conditions, loops, HTTP calls, human-in-the-loop gates, knowledge lookups, and sub-flow execution—all without writing code.

• Demand-driven Context: Context flows forward through pipeline stages. Each stage receives the accumulated output of prior stages. No stage sees everything—only what was produced upstream. This prevents context window abuse and keeps signal-to-noise high.

• Fact Grounding: Responses are verified against a knowledge graph (SPO triples). Claims that can't be grounded are flagged. This catches hallucinations before they reach users.

• Knowledge Core Ingestion: Documents are analyzed, chunked, embedded, and routed into the appropriate store—structured facts become SPO triples, narrative content becomes vector chunks. The ingestion pipeline understands the difference and routes accordingly.

• Conversations with Epistemic Grounding: Users can interact with agents directly through conversations. When an agent responds, the system automatically injects relevant knowledge graph context (facts, contradiction detections, knowledge gaps) so the agent's response is grounded in what the organization actually knows.

First Use Case: Investment Banking

Qadra's initial deployment focuses on investment banking workflows:

Deal → Research → Due Diligence → Investment Memo → Review

Why investment banking? These workflows are:

• High-stakes: A bad investment memo can lose millions. Accuracy matters.
• Structured: Deals follow predictable stages with clear handoffs.
• Knowledge-intensive: Analysts synthesize data from dozens of sources.
• Compliance-heavy: Every decision needs an audit trail.

This is the opposite of casual chat—it's exactly where "prompts that do everything" fail hardest.

Each stage has specialized agents that produce artifacts:

Each agent is an expert at one thing. The research analyst doesn't worry about IC formatting—that's the memo writer's job. This separation of concerns is why agents are specialized personas with focused system prompts, not monolithic catch-all assistants.

This is NOT a Chat App

Understanding what Qadra is NOT is as important as understanding what it is:

Why This Distinction Matters

Chat apps optimize for engagement. They want users talking to the AI as much as possible. Success is measured in messages sent, sessions started, time spent.

Qadra optimizes for outcomes. A workload has a goal (produce an investment memo). Success is measured in workloads completed, accuracy of outputs, time-to-decision.

This changes everything about the architecture:

• No conversation history bloat: Workloads have structured context, not message logs
• No "let me think about that": Agents either have the capability or they don't
• No hallucination tolerance: Ungrounded claims are flagged, not glossed over
• Conversations are supplementary: Users can chat with agents directly, but the core value is in orchestrated workflows that produce auditable artifacts

Key Differentiators

Persona-based Agents

Agents in Qadra are not prompt templates. They are named specialists—team members with defined identities, behavioral instructions, and scoped responsibilities.

What makes a Qadra agent different from a prompt?

An agent has four components:

• Identity: Display name, avatar, and description. This is the agent's face—how it appears in the UI, how other parts of the system discover and reference it. Example: "Research Analyst" with a description of "Gathers market data and competitive analysis."

• System Prompt: Behavioral instructions that define how the agent thinks, responds, and operates. This is not a one-shot prompt—it's persistent guidance that shapes every interaction the agent has within a workload or conversation.

• Orchestration Controls: How the agent participates in larger workflows. control_type determines what happens after the agent completes (retain control, hand off to parent, or return to pipeline start). output_visibility controls whether output is user-facing or kept internal. max_calls_per_parent_agent prevents runaway loops.

• Versioning: Agents support version numbers. New versions can be deployed alongside old ones, with is_active controlling which version is live. This enables safe iteration without breaking running workloads.

Visual Flow Builder

Qadra includes a visual flow builder for designing agent workflows as directed graphs. Flows are stored as ReactFlow JSONB and executed by an event-driven graph executor.

Flow nodes provide the building blocks:

Flows support human-in-the-loop patterns: when a HumanInput node is reached, execution pauses and waits for user response via the API. This enables approval gates, review steps, and interactive decision points within automated workflows.

Knowledge Graph Integration

Traditional RAG has a fundamental problem: it treats all knowledge as bags of text chunks. "Apple was founded by Steve Jobs in 1976" becomes a chunk that might or might not be retrieved when you ask about Apple's founding.

Qadra uses a dual-storage approach that separates facts from narrative:

SPO Index (Subject-Predicate-Object): Structured facts stored as triples.

(Apple, founded_by, Steve Jobs)
(Apple, founding_year, 1976)
(Steve Jobs, co-founded, Apple)

Facts are queryable, linkable, and verifiable. When an agent claims "Apple was founded in 1976," the system can check: does the triple (Apple, founding_year, 1976) exist? This is grounding—the antidote to hallucination.

Vector Store: Embeddings for semantic similarity search.

Narrative content that doesn't decompose into facts—analysis, opinions, context—is chunked, embedded, and stored in pgvector with HNSW indexes. This enables "find me content similar to X" queries.

Why both? They serve different purposes:

The ingestion pipeline analyzes content and routes intelligently:

This isn't just optimization—it's architectural. Facts go where they can be verified. Narrative goes where it can be searched. The system knows the difference.

Conversations with Epistemic Grounding

While Qadra's core is workload orchestration, users can also interact with agents directly through conversations. These are ad-hoc chat threads—not pipeline executions—but they benefit from the same knowledge infrastructure.

When a user sends a message in a conversation, the system automatically queries the knowledge graph for relevant SPO triples, detects contradictions, and identifies knowledge gaps. This epistemic context is injected into the agent's prompt as structured facts, so the agent's response is grounded in the organization's actual knowledge base rather than the LLM's training data.

Conversations also serve as the interface for flow execution: when the LLM determines that a user's intent maps to an existing flow, it can trigger flow execution inline and narrate the real results back to the user. Progress streams in real time via Server-Sent Events.

Multitenancy

Enterprise AI isn't one company—it's many. Each customer needs complete isolation: their agents, their knowledge, their workflows, their data.

Qadra is multitenant by design, not by afterthought:

Organization (Tenant)
├── Users              ← Multi-tenant via junction table
├── API Keys           ← Tenant-scoped, revocable
├── Agents             ← Custom specialists for this org
├── Pipelines          ← Workflows specific to their processes
├── Workloads          ← Their active and completed work
│   ├── Tasks          ← Agent assignments
│   └── Artifacts      ← Produced documents
├── Agent Flows        ← Visual workflow definitions
│   └── Flow Executions
├── Conversations      ← Direct agent interactions
├── SPO Index          ← Their private knowledge graph
└── Vector Store       ← Their private embeddings

Why multitenancy matters:

Row-level security is enforced at the repository layer, not application code:

#![allow(unused)]
fn main() {
// Every repository method requires tenant_id
async fn get_workload(&self, tenant_id: Uuid, id: Uuid) -> Result<Workload> {
    sqlx::query!("SELECT * FROM workloads WHERE tenant_id = $1 AND id = $2",
                 tenant_id, id)
        .fetch_one(&self.pool)
        .await
}
}

There's no way to accidentally query across tenants because tenant_id is a required parameter on every operation. This is defense in depth—even if application code has a bug, the database layer enforces isolation.

Technology Stack

Every technology choice has a reason. We optimized for correctness, observability, and operational simplicity.

Why These Specific Choices?

Rust for the data layer: Investment banking workflows are long-running and high-stakes. Rust's memory safety eliminates entire classes of bugs (use-after-free, null pointers). No garbage collector means predictable latency—no surprise pauses during critical operations. All data operations (CRUD, auth, flows, knowledge) go through Rust Core via NATS.

PostgreSQL over specialized databases: One database for relational data AND vector search (pgvector). This eliminates sync complexity between separate vector databases. Transactions work across both—you can atomically insert a document and its embeddings.

NATS for service communication: The gateway, Rust Core, and Python Agent communicate via NATS request-reply. This decouples the HTTP layer from the data layer and allows horizontal scaling of any component independently.

Redis for events, not Kafka: Our event volumes don't justify Kafka's operational complexity. Redis Streams provide durable, acknowledged delivery with consumer groups. If we outgrow Redis, the abstraction layer allows swapping implementations.

MongoDB for audit only: Audit logs have different access patterns (append-heavy, rarely queried, schema-evolving). Keeping them separate from PostgreSQL prevents table bloat and allows independent scaling.

Development Guide

This guide covers local development setup, common tasks, and patterns for contributing to Qadra.

Prerequisites

Quick version check:

rustc --version    # rustc 1.75.0 or higher
docker --version   # Docker 24.x or higher
node --version     # v18.x or higher
yarn --version     # 1.x or higher
python --version   # 3.11 or higher

Local Setup

Clone and Configure

git clone https://github.com/HGP-Technologies/qadra-ai.git
cd qadra-ai

# Copy environment template
cp .env.example .env

# Edit configuration (set API keys, passwords)
vim .env

Start Infrastructure

# Start all infrastructure services
docker compose up -d postgres mongodb nats minio imaginary

# Wait for PostgreSQL to be ready
docker compose exec postgres pg_isready -U qadra

Run Migrations

DATABASE_URL=postgres://qadra:qadra_dev_password@localhost:5432/qadra \
  sqlx migrate run

Start Development Servers

Three application processes run independently during development:

# Terminal 1: Rust Core (NATS handlers, truth layer)
cargo run

# Terminal 2: Node.js Gateway (HTTP API)
cd gateway && yarn install && yarn dev

# Terminal 3: React SPA (frontend)
cd web && yarn install && yarn dev

The React SPA runs at http://localhost:5173, the Gateway at http://localhost:4000, and Rust Core connects to NATS internally.

Project Structure

qadra-ai/
├── Cargo.toml                 # Workspace manifest (10 crates)
├── crates/                    # Library crates
│   ├── qadra-traits/          # ALL interfaces and shared types (tiny, stable)
│   ├── qadra-container/       # IoC container, mock implementations
│   ├── qadra-db-postgres/     # Repository: PostgreSQL
│   ├── qadra-llm-openrouter/  # LLM: OpenRouter (OpenAI-compatible)
│   ├── qadra-cache-redis/     # Cache: Redis
│   ├── qadra-queue-redis/     # Queue: Redis Streams
│   ├── qadra-vector-pgvector/ # Vector: pgvector
│   ├── qadra-semantic-cache-redis/ # Semantic cache for verification results
│   ├── qadra-epistemic-core/  # Epistemic operations (KG queries, attribution)
│   └── qadra-nats/            # NATS client wrapper
├── src/                       # Main Rust application
│   ├── main.rs                # Entry point (8 NATS handler spawns)
│   ├── lib.rs                 # Library exports
│   ├── config.rs              # Configuration loading
│   ├── nats_service.rs        # NATS handler dispatch
│   ├── nats_observer.rs       # Catch-all NATS audit observer
│   ├── flow_executor.rs       # Flow graph execution engine
│   ├── flow_generator.rs      # Flow graph generation
│   ├── email_service.rs       # SMTP2GO email delivery
│   ├── email_renderer.rs      # Handlebars template rendering
│   ├── email_logging.rs       # LoggingEmailService decorator
│   ├── error.rs               # Error types
│   ├── ingestion/             # Document processing pipeline
│   │   ├── mod.rs
│   │   ├── analyzer.rs        # Content routing (SPO/Vector/Both/Discard)
│   │   ├── chunker.rs         # Text chunking
│   │   ├── embedder.rs        # Embedding generation
│   │   ├── extractor.rs       # LLM-based triple extraction
│   │   └── service.rs         # Orchestrates the ingestion pipeline
│   ├── tracing/               # Audit trace to MongoDB
│   └── telemetry/             # OpenTelemetry integration
├── gateway/                   # Node.js HTTP gateway
│   ├── src/
│   │   ├── index.ts           # Express server entry point
│   │   ├── nats.ts            # NATS client and request helper
│   │   ├── config.ts          # Configuration
│   │   ├── storage.ts         # MinIO S3 client
│   │   ├── images.ts          # Imaginary image processing client
│   │   ├── auth/              # JWT middleware
│   │   ├── routes/            # HTTP route handlers (15 route files)
│   │   │   ├── auth.ts        # Login, register, refresh, verify
│   │   │   ├── users.ts       # Profile, avatar
│   │   │   ├── teams.ts       # Members, invites, roles
│   │   │   ├── tenants.ts     # Tenant settings, branding
│   │   │   ├── agents.ts      # Agent CRUD
│   │   │   ├── flows.ts       # Flow CRUD and execution
│   │   │   ├── conversations.ts # Chat conversations and messages
│   │   │   ├── knowledge.ts   # Document ingestion and proposals
│   │   │   ├── artifactTemplates.ts # Artifact template CRUD
│   │   │   ├── files.ts       # File upload/download
│   │   │   ├── email.ts       # Email send
│   │   │   ├── notifications.ts # Email templates, notification logs
│   │   │   ├── audit.ts       # NATS audit trail
│   │   │   ├── admin.ts       # Super admin operations
│   │   │   └── health.ts      # Health checks
│   │   └── types/             # TypeScript type definitions
│   ├── package.json
│   └── tsconfig.json
├── web/                       # React SPA (Vite + Tailwind + zustand)
│   ├── src/
│   │   ├── App.tsx            # Router and layout
│   │   ├── main.tsx           # Entry point
│   │   ├── api/               # API client functions
│   │   ├── stores/            # zustand state stores
│   │   ├── pages/             # Page components
│   │   │   ├── Dashboard.tsx
│   │   │   ├── Conversations.tsx
│   │   │   ├── ConversationChat.tsx
│   │   │   ├── Agents.tsx
│   │   │   ├── AgentEditor.tsx
│   │   │   ├── Flows.tsx
│   │   │   ├── FlowBuilder.tsx
│   │   │   ├── ArtifactTemplates.tsx
│   │   │   ├── ArtifactTemplateEditor.tsx
│   │   │   ├── Settings.tsx
│   │   │   ├── Admin.tsx
│   │   │   ├── Login.tsx
│   │   │   ├── Register.tsx
│   │   │   └── ...
│   │   └── components/        # Reusable components
│   │       └── flow/nodes/    # ReactFlow node type components
│   ├── package.json
│   ├── vite.config.ts
│   └── tailwind.config.ts
├── agent/                     # Python orchestration agent
│   ├── src/qadra_agent/
│   │   └── orchestrator.py    # WorkloadOrchestrator (LLM-driven pipelines)
│   ├── tests/
│   ├── pyproject.toml
│   └── Dockerfile
├── migrations/                # SQL migrations (001-030)
├── tests/                     # Rust integration tests
│   ├── common/                # TestClient, TestDb, fixtures
│   ├── workload_lifecycle.rs
│   ├── task_lifecycle.rs
│   ├── stage_forwarding.rs
│   ├── multi_stage_forwarding.rs
│   └── artifact_pipeline.rs
├── deploy/otel/               # Observability configuration
├── kong/                      # Kong API gateway config
├── docs/                      # Documentation
└── docker-compose.yml         # Full development stack

Crate Workspace

The Rust code is organized into 10 workspace crates plus the root binary:

Design principle: Everything is swappable. Core business logic depends only on traits, not implementations. Switch PostgreSQL for SQLite, swap OpenRouter for Ollama, replace Redis with in-memory---all via configuration.

Common Tasks

Running Tests

# Unit tests (all workspace crates, ~120 tests)
cargo test --workspace --lib

# Integration tests (requires running PostgreSQL with pgvector)
DATABASE_URL="postgres://qadra:qadra_dev_password@localhost:5432/qadra" \
  cargo test --test workload_lifecycle -- --test-threads=1

# All integration test files
DATABASE_URL="postgres://qadra:qadra_dev_password@localhost:5432/qadra" \
  cargo test -- --test-threads=1

# Specific test by name
cargo test test_name

# With output
cargo test -- --nocapture

# Web tests (if test files exist)
cd web && yarn test

# Gateway type checking
cd gateway && yarn build

Linting

# Rust formatting
cargo fmt
cargo fmt --check

# Rust linting
cargo clippy -- -D warnings

# Web linting (Biome)
cd web && npx @biomejs/biome check src/

# Web type checking
cd web && npx tsc --noEmit

Building

# Debug build
cargo build

# Release build
cargo build --release

# Check without building (faster)
cargo check

# Gateway build
cd gateway && yarn build

# Web build
cd web && yarn build

Database Operations

# Connect to database
psql postgres://qadra:qadra_dev_password@localhost:5432/qadra

# Run migrations
DATABASE_URL=postgres://... sqlx migrate run

# Create new migration
sqlx migrate add migration_name

# Revert last migration
sqlx migrate revert

# Prepare offline query data (for CI)
cargo sqlx prepare

Adding New Features

Standard Feature Flow

The typical flow for adding a new feature:

1. Migration -- Create SQL migration for any new tables or columns
2. Traits types -- Add types and trait methods in qadra-traits
3. PostgreSQL impl -- Implement trait methods in qadra-db-postgres
4. Container mock stubs -- Add default return values in qadra-container
5. NATS handler -- Add handler case in src/nats_service.rs
6. Gateway route -- Add HTTP route in gateway/src/routes/
7. Frontend -- Add page/store in web/src/
8. Rule docs -- Update .claude/rules/ files

Adding a New NATS Handler

1. Define the operation in src/nats_service.rs:

#![allow(unused)]
fn main() {
"your_operation" => {
    // Parse request, call repository, return response
}
}

2. Add gateway route in gateway/src/routes/yourFeature.ts:

router.get("/your-endpoint", requireAuth, async (req, res) => {
    const result = await natsRequest(subject, payload);
    res.json(result);
});

3. Register route in gateway/src/index.ts

Adding a New Database Table

1. Create migration:

sqlx migrate add my_table

2. Write SQL in migrations/XXX_my_table.sql:

CREATE TABLE my_table (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    -- columns
    created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX idx_my_table_tenant ON my_table(tenant_id);

3. Add trait in crates/qadra-traits/src/lib.rs
4. Implement in crates/qadra-db-postgres/src/lib.rs
5. Add mock stub in crates/qadra-container/src/lib.rs

Code Style

Naming Conventions

Error Handling

Use the project's error types:

#![allow(unused)]
fn main() {
use crate::{Error, Result};

fn my_function() -> Result<()> {
    something_that_might_fail()?;
    Ok(())
}
}

Logging

Use tracing macros with structured fields:

#![allow(unused)]
fn main() {
use tracing::{info, warn, error, debug, instrument};

#[instrument(skip(self))]
async fn my_function(&self, id: Uuid) -> Result<()> {
    info!(id = %id, "Processing item");
    Ok(())
}
}

Testing Patterns

Unit Tests

Unit tests live at the bottom of source files (Rust convention):

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_my_function() {
        let result = my_function(input);
        assert_eq!(result, expected);
    }

    #[tokio::test]
    async fn test_async_function() {
        let result = async_function().await;
        assert!(result.is_ok());
    }
}
}

Integration Tests

Integration tests are in tests/ and require a running PostgreSQL instance:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_workload_lifecycle() {
    let container = setup_test_container().await;
    let workload = container.workload_repository()
        .create_workload(params)
        .await
        .unwrap();
    assert_eq!(workload.status, "pending");
}
}

Test Database

The TestDb helper resolves the test database URL:

1. TEST_DATABASE_URL -- Explicit test database URL (if set)
2. DATABASE_URL -- Derives test URL by replacing database name with qadra_test
3. Default -- postgres://qadra:qadra_dev_password@localhost:5433/qadra_test

The helper auto-creates the qadra_test database if it does not exist.

Mocking

Use mock implementations from qadra-container:

#![allow(unused)]
fn main() {
use qadra_container::ContainerBuilder;

let container = ContainerBuilder::new()
    .with_mock_repository()
    .with_mock_llm()
    .build();
}

Mock implementations return empty/default values.

Git Workflow

Commit Configuration

All commits must be signed as:

• Name: Mark Scott
• Email: mark.scott@hgp-technologies.com

Use Co-Authored-By: Mark Scott <mark.scott@hgp-technologies.com> for co-authored commits.

Branch Strategy

• main -- Production-ready code
• develop -- Integration branch
• feat/* -- Feature branches
• fix/* -- Bug fix branches

CI Pipeline

GitHub Actions at .github/workflows/ci.yml runs on push to main/develop and PRs targeting main. Key jobs:

Common Pitfalls

Forgetting tenant_id

Every repository method takes tenant_id. Every query includes it. No exceptions.

#![allow(unused)]
fn main() {
// WRONG
sqlx::query!("SELECT * FROM workloads WHERE id = $1", id)

// RIGHT
sqlx::query!("SELECT * FROM workloads WHERE tenant_id = $1 AND id = $2",
             tenant_id, id)
}

Blocking the async runtime

Never call blocking functions directly. Use tokio::spawn_blocking for unavoidable blocking work.

#![allow(unused)]
fn main() {
// WRONG
let result = std::fs::read_to_string("file.txt")?;

// RIGHT
let result = tokio::fs::read_to_string("file.txt").await?;
}

DbWorkload/DbTask/DbArtifact do not derive Clone

Access fields directly instead of .clone().into().

Container repository access patterns

#![allow(unused)]
fn main() {
// Repository (agents, graphs, SPO)
container.repository()

// Auth-specific operations
container.auth()

// Workload operations
container.workload_repository()

// Agent flow operations
container.agent_flow_repository()
}

Debugging

Enabling Debug Logs

# Rust Core logs
RUST_LOG=qadra=debug cargo run

# Specific module
RUST_LOG=qadra::flow_executor=debug cargo run

# SQL queries
RUST_LOG=sqlx=debug cargo run

# Gateway logs
cd gateway && DEBUG=* yarn dev

Inspecting Database State

psql postgres://qadra:qadra_dev_password@localhost:5432/qadra

\dt                           # List tables
SELECT * FROM agent_flows;
SELECT * FROM flow_executions WHERE status = 'running';
SELECT * FROM conversations ORDER BY created_at DESC LIMIT 5;
SELECT * FROM documents WHERE ingestion_status = 'pending_approval';

Inspecting NATS

# Monitor all NATS subjects
nats sub "qadra.>"

# Monitor specific subject
nats sub "qadra.*.flow.do.*"

# NATS server info
curl http://localhost:8222/varz

Inspecting MongoDB Audit

docker exec qadra-mongodb mongosh \
  "mongodb://qadra:qadra_traces_dev@localhost:27017/qadra_traces?authSource=admin" \
  --quiet --eval "db.nats_audit.find().sort({timestamp: -1}).limit(5).toArray()"

Performance Tips

Compile Times

# Use cargo check instead of build when possible
cargo check

# Incremental builds (default)
export CARGO_INCREMENTAL=1

# Full workspace build ~3 min, incremental ~30s

Runtime Performance

#![allow(unused)]
fn main() {
// Pre-allocate vectors when size is known
let mut results = Vec::with_capacity(expected_count);

// Use references instead of clones
fn process(data: &Data) -> Result<()>  // Preferred

// Batch database operations
repo.insert_many(&items).await?;       // 1 query, not N
}

Tutorials

These walkthroughs take you end-to-end through the three workflows that most onboarding teams need first: getting knowledge into the graph and querying it, wiring an SPO persona pipeline as a flow, and tracing a memo claim back to its sources for sign-off.

Every step is a real HTTP call against the Node.js Gateway (http://localhost:4000). If you have not started the stack yet, follow the Development Guide first --- you need Rust Core, the Gateway, PostgreSQL, and NATS running.

A few conventions used throughout:

• All authenticated calls send Authorization: Bearer $TOKEN. The $TOKEN is the access_token from login (15-minute lifetime --- re-login or refresh if it expires).
• You never pass tenant_id in a request body. It is derived from your JWT (see API Reference -> Authentication).
• Responses are abbreviated to the fields each step needs. The examples are illustrative --- UUIDs and tokens are placeholders.

Tutorial 1: Ingest knowledge & query it

Goal: register an account, propose a document, approve it through the human gate so it ingests, then ask a question and see the answer grounded in the freshly ingested facts.

This tutorial exercises the human-gated ingestion lifecycle described in Knowledge Graph -> Document Ingestion Pipeline. Nothing enters the SPO graph or vector store until a human approves it.

1. Register a user and tenant. Registration creates the user, a default workspace tenant, and returns a token pair in one call.

   curl -X POST http://localhost:4000/auth/register \
     -H 'Content-Type: application/json' \
     -d '{
       "email": "analyst@acme.test",
       "password": "supersecret",
       "name": "Ada Analyst"
     }'
   

   {
     "access_token": "eyJ0eXAiOiJKV1Q...",
     "refresh_token": "eyJ0eXAiOiJKV1Q...",
     "active_tenant": { "tenant_id": "tnt-1111", "tenant_name": "Ada Analyst's Workspace", "role": "owner" }
   }
   

   If the account already exists, log in instead --- the response shape is identical:

   curl -X POST http://localhost:4000/auth/login \
     -H 'Content-Type: application/json' \
     -d '{ "email": "analyst@acme.test", "password": "supersecret" }'
   

   Capture the access token for the rest of the tutorial:

   export TOKEN="eyJ0eXAiOiJKV1Q..."
   

2. Propose a document for review. Proposing does not ingest --- it places the document in the review queue with status pending_approval.

   curl -X POST http://localhost:4000/knowledge/propose \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{
       "title": "Acme Corp Overview",
       "content": "Acme Corporation was founded in 1990 by John Smith. The company is headquartered in San Francisco and reported 2024 revenue of $391B. Acme acquired DataCo in 2015 for $50 million.",
       "content_type": "text/plain",
       "source_url": "https://example.com/acme",
       "source_type": "web"
     }'
   

   { "document_id": "doc-2222" }
   

3. Approve the proposal (the human gate). Approval stamps approved_by/approved_at and kicks off ingestion asynchronously on the Rust side. It returns immediately --- ingestion (analyze -> extract triples -> chunk -> embed) runs in the background.

   curl -X POST http://localhost:4000/knowledge/documents/doc-2222/approve \
     -H "Authorization: Bearer $TOKEN"
   

   {
     "document_id": "doc-2222",
     "status": "approved",
     "message": "Ingestion started -- poll document status for progress"
   }
   

4. Poll until ingestion completes. Status moves approved -> processing -> completed. The ingestion_result carries the routing decision and counts.

   curl http://localhost:4000/knowledge/documents/doc-2222 \
     -H "Authorization: Bearer $TOKEN"
   

   {
     "document_id": "doc-2222",
     "title": "Acme Corp Overview",
     "ingestion_status": "completed",
     "ingestion_result": { "routing": "both", "triples_extracted": 4, "chunks_created": 2 }
   }
   

   Four triples are now in the SPO index --- e.g. (Acme Corporation, founded_by, John Smith) and (Acme Corporation, revenue_is, $391B) --- each linked back to doc-2222 for provenance.

5. Create an agent to talk to. Conversations route through an agent persona (see API Reference -> Agents).

   curl -X POST http://localhost:4000/agents \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{
       "name": "research-analyst",
       "display_name": "Research Analyst",
       "description": "Answers questions grounded in the knowledge graph",
       "system_prompt": "You are a research analyst. Answer using only grounded facts."
     }'
   

   { "id": "agt-3333", "name": "research-analyst" }
   

6. Open a conversation.

   curl -X POST http://localhost:4000/conversations \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{ "title": "Acme research" }'
   

   { "id": "cnv-4444", "title": "Acme research" }
   

7. Ask a question and see epistemic grounding. Sending a message runs the epistemic query pipeline on your message before synthesis: it pulls matching SPO triples, runs the competence gate, and injects the grounding context into the LLM call. After synthesis, the verification gate records whether the answer stayed grounded --- both results land in the message metadata.

   curl -X POST http://localhost:4000/conversations/cnv-4444/messages \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{
       "agent_id": "agt-3333",
       "content": "Who founded Acme Corporation and what was its 2024 revenue?"
     }'
   

   {
     "id": "msg-5555",
     "role": "assistant",
     "content": "Acme Corporation was founded by John Smith and reported 2024 revenue of $391B.",
     "metadata": {
       "verification": {
         "passed": true,
         "faithfulness": 1.0,
         "attributions": [
           { "subject": "Acme Corporation", "predicate": "founded_by", "object": "John Smith", "score": 0.62 },
           { "subject": "Acme Corporation", "predicate": "revenue_is", "object": "$391B", "score": 0.41 }
         ],
         "ungrounded_claims": []
       }
     }
   }
   

   The answer is drawn from the triples you just ingested. The verification block confirms the response stayed grounded --- if you ask about something not in the graph, the competence gate coverage drops and the verification gate surfaces the unsupported claims under ungrounded_claims instead of hiding them.

   To watch the gates and node steps in real time, POST to /conversations/cnv-4444/messages/stream instead and read the text/event-stream. See API Reference -> Stream Message.

Tutorial 2: Build & run an SPO persona pipeline flow

Goal: wire a four-persona research pipeline as a flow graph, run it, and inspect the execution results --- including the claim-validation checkpoint that blocks unverified claims from flowing downstream.

The personas map to flow node behaviors described in API Reference -> Flow Node Types:

The ClaimValidation node is the hard checkpoint: it labels every prior-stage claim SUPPORTED, CONTRADICTED, or NOT_MENTIONED before Eliza builds on them. See Knowledge Graph -> Claim Validation Node.

1. Create the flow definition. The graph is a ReactFlow {nodes, edges, viewport} object. Each node's data carries its type-specific config. Reuse $TOKEN from Tutorial 1.

   curl -X POST http://localhost:4000/flows \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{
       "name": "spo-persona-dd-pipeline",
       "description": "Scarlet writes triples, Ayana filters, ClaimValidation checks, Eliza drafts DD, Reagan reviews",
       "flow_data": {
         "nodes": [
           { "id": "start",   "type": "Start", "data": {}, "position": { "x": 0, "y": 0 } },
   
           { "id": "scarlet", "type": "Agent",
             "data": { "label": "Scarlet (Research)", "agent_name": "scarlet",
                       "prompt": "Research {{target_company}} and state concrete facts.",
                       "output_key": "scarlet_output" },
             "position": { "x": 200, "y": 0 } },
           { "id": "scarlet-write", "type": "SpoWrite",
             "data": { "label": "Persist Scarlet triples", "content_key": "scarlet_output",
                       "asserter": "scarlet", "source": "spo-persona-dd-pipeline",
                       "output_key": "scarlet_triples" },
             "position": { "x": 400, "y": 0 } },
   
           { "id": "ayana",   "type": "Agent",
             "data": { "label": "Ayana (Analysis)", "agent_name": "ayana",
                       "prompt": "Analyze the research for {{target_company}}.",
                       "output_key": "ayana_output" },
             "position": { "x": 600, "y": 0 } },
           { "id": "ayana-filter", "type": "SpoFilter",
             "data": { "label": "Filter Scarlet triples", "asserter": "scarlet",
                       "output_key": "scarlet_facts" },
             "position": { "x": 800, "y": 0 } },
   
           { "id": "validate", "type": "ClaimValidation",
             "data": { "label": "Ground prior claims",
                       "input_keys": "scarlet_output,ayana_output",
                       "output_key": "claim_validation" },
             "position": { "x": 1000, "y": 0 } },
   
           { "id": "eliza",   "type": "Agent",
             "data": { "label": "Eliza (Due Diligence)", "agent_name": "eliza",
                       "prompt": "Write a due-diligence summary using only SUPPORTED claims from {{claim_validation}}.",
                       "output_key": "eliza_output" },
             "position": { "x": 1200, "y": 0 } },
   
           { "id": "reagan",  "type": "Agent",
             "data": { "label": "Reagan (Review)", "agent_name": "reagan",
                       "prompt": "Review Eliza's DD for sign-off readiness.",
                       "output_key": "reagan_output" },
             "position": { "x": 1400, "y": 0 } },
   
           { "id": "end",     "type": "End", "data": {}, "position": { "x": 1600, "y": 0 } }
         ],
         "edges": [
           { "id": "e1", "source": "start",        "target": "scarlet" },
           { "id": "e2", "source": "scarlet",      "target": "scarlet-write" },
           { "id": "e3", "source": "scarlet-write","target": "ayana" },
           { "id": "e4", "source": "ayana",        "target": "ayana-filter" },
           { "id": "e5", "source": "ayana-filter", "target": "validate" },
           { "id": "e6", "source": "validate",     "target": "eliza" },
           { "id": "e7", "source": "eliza",        "target": "reagan" },
           { "id": "e8", "source": "reagan",       "target": "end" }
         ],
         "viewport": { "x": 0, "y": 0, "zoom": 1 }
       }
     }'
   

   { "id": "flw-6666", "name": "spo-persona-dd-pipeline", "node_count": 9 }
   

   Don't want to hand-author the graph? POST /flows/generate builds one from a natural-language prompt instead (see API Reference -> AI Flow Generation).

2. Execute the flow. The input object seeds the flow state --- here, the company under review. The executor walks the graph from Start, running each node and threading outputs through shared state.

   curl -X POST http://localhost:4000/flows/flw-6666/execute \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{ "input": { "target_company": "Acme Corporation" } }'
   

   { "execution_id": "exe-7777", "status": "running" }
   

3. Poll the execution until it finishes. Status follows pending -> running -> completed (or failed/cancelled). If you had inserted a HumanInput node, it would pause at paused here and wait for POST /flows/executions/exe-7777/input.

   curl http://localhost:4000/flows/executions/exe-7777 \
     -H "Authorization: Bearer $TOKEN"
   

4. Inspect the results. A completed execution exposes the per-node node_states (each with status, input, output) and the shared state blackboard. The ClaimValidation node's output is the checkpoint you care about for sign-off.

   {
     "execution_id": "exe-7777",
     "status": "completed",
     "state": {
       "target_company": "Acme Corporation",
       "scarlet_output": "Acme was founded by John Smith; 2024 revenue $391B.",
       "scarlet_triples": [
         { "subject": "Acme Corporation", "predicate": "founded_by", "object": "John Smith" }
       ],
       "claim_validation": {
         "summary": { "supported": 2, "contradicted": 0, "not_mentioned": 1 },
         "claims": [
           { "claim": "Acme was founded by John Smith", "label": "SUPPORTED" },
           { "claim": "2024 revenue was $391B",          "label": "SUPPORTED" },
           { "claim": "Acme operates in 40 countries",   "label": "NOT_MENTIONED" }
         ]
       },
       "eliza_output": "Due-diligence summary (grounded claims only)...",
       "reagan_output": "Approved for sign-off."
     },
     "node_states": {
       "scarlet":      { "status": "completed" },
       "scarlet-write":{ "status": "completed", "output": { "triples_written": 4 } },
       "ayana":        { "status": "completed" },
       "ayana-filter": { "status": "completed" },
       "validate":     { "status": "completed" },
       "eliza":        { "status": "completed" },
       "reagan":       { "status": "completed" }
     }
   }
   

   The NOT_MENTIONED label on "Acme operates in 40 countries" is the system telling you Scarlet asserted something the graph cannot back up --- Eliza is instructed to drop it, and Reagan signs off only on the grounded remainder.

   List every run of a flow with GET /flows/flw-6666/executions.

Tutorial 3: Trace a memo claim for sign-off

Goal: an investment-committee reviewer reads a claim in a memo and needs to verify it against the underlying knowledge in one round trip --- which triples support it, who asserted them, and from which source documents.

This uses the claim-trace path documented in Knowledge Graph -> Claim Trace. It runs the inverse of ingestion: free-text claim in, supporting triples + source documents out, under a 500 ms budget (a 450 ms internal timeout returns partial results rather than failing the review).

1. Trace the claim. Paste the memo statement verbatim. max_results caps how many supporting matches come back (1--20). Reuse $TOKEN.

   curl -X POST http://localhost:4000/knowledge/trace \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{
       "claim": "Acme Corporation was founded by John Smith",
       "max_results": 5
     }'
   

2. Read the trace. Each match groups a supporting triple (with its asserter and assertion_type) under the claim and links it to the source documents that ground it --- each carrying the excerpt and the full source content.

   {
     "claim": "Acme Corporation was founded by John Smith",
     "matches": [
       {
         "triple": {
           "subject": "Acme Corporation",
           "predicate": "founded_by",
           "object": "John Smith",
           "confidence": 0.95,
           "source": "Acme Corp Overview",
           "asserter": "scarlet",
           "assertion_type": "fact"
         },
         "documents": [
           {
             "document_id": "doc-2222",
             "title": "Acme Corp Overview",
             "source_url": "https://example.com/acme",
             "source_type": "web",
             "excerpt": "Acme Corporation was founded in 1990 by John Smith",
             "content": "Acme Corporation was founded in 1990 by John Smith. The company is headquartered in San Francisco...",
             "confidence": 0.95
           }
         ]
       }
     ],
     "duration_ms": 312
   }
   

   The reviewer now has everything to sign off: the claim resolves to a fact-type triple asserted by scarlet, grounded in doc-2222 with the exact excerpt. duration_ms confirms it came in under the sub-500 ms IC budget.

3. Handle an unsupported claim. If the memo asserts something the graph cannot back up, matches comes back empty --- a clear signal the statement is not grounded and should not be signed off:

   curl -X POST http://localhost:4000/knowledge/trace \
     -H "Authorization: Bearer $TOKEN" \
     -H 'Content-Type: application/json' \
     -d '{ "claim": "Acme Corporation operates in 40 countries", "max_results": 5 }'
   

   { "claim": "Acme Corporation operates in 40 countries", "matches": [], "duration_ms": 88 }
   

   An empty trace and a NOT_MENTIONED label from Tutorial 2's ClaimValidation node are two views of the same gap --- the pipeline catches it before the memo is written, and the trace endpoint catches it again at sign-off. Together they close the loop from ingestion to investment-committee review.

Architecture

System Overview

Qadra uses a NATS-centric, multi-container architecture. There is no monolithic HTTP API in Rust. Instead, each concern lives in a dedicated container, and all inter-service communication flows through NATS request-reply messaging.

The Four Containers

Request Flow

Browser (:5173)
    |
    v
Kong (:8000) ── JWT validation, rate limiting, routing
    |
    v
Node.js Gateway (:4000) ── JWT create/validate, CORS, multipart uploads
    |
    v  NATS request-reply
    |
Rust Core (no HTTP port) ── PostgreSQL, MongoDB, Redis
    |
    v  NATS request-reply (for LLM workloads)
    |
Python Agent ── LLM providers (OpenRouter, Cerebras)

Why NATS Instead of Direct HTTP?

• Decoupling: Gateway knows nothing about database schemas. Rust Core knows nothing about HTTP routing.
• Horizontal scaling: Spin up N Rust Core instances; NATS distributes requests automatically.
• Uniform messaging: Auth, workloads, flows, email, admin, audit, epistemic operations all use the same transport.
• Observability: A single catch-all NATS observer captures every message for audit without modifying handlers.

Trait-Based Dependency Injection

The Rust application layer uses a ServiceContainer with trait-based DI. All external dependencies (database, cache, LLM providers, vector index) are abstracted behind traits defined in qadra-traits:

#![allow(unused)]
fn main() {
pub struct ServiceContainer {
    pub repository: Arc<dyn Repository>,
    pub llm: Arc<dyn LlmProvider>,
    pub cache: Arc<dyn Cache>,
    pub docs: Arc<dyn DocumentStore>,
    pub queue: Arc<dyn MessageQueue>,
    pub runtime: Arc<dyn RuntimeExecutor>,
    pub vector_index: Arc<dyn VectorIndex>,
    pub embeddings: Arc<dyn EmbeddingProvider>,
    // ... more trait objects
}
}

Why traits?

• Testability: Mock implementations in qadra-container for unit tests without a real database
• Swappability: Change PostgreSQL to SQLite by implementing Repository
• Clarity: Business logic expresses intent, not infrastructure concerns

Persistence Layer

Three databases serve distinct purposes:

┌──────────────────┐   ┌──────────────────┐   ┌──────────────────┐
│   PostgreSQL     │   │   Redis Stack    │   │    MongoDB       │
│                  │   │                  │   │                  │
│ - Source of truth│   │ - Cache          │   │ - Audit traces   │
│ - pgvector HNSW  │   │ - Streams        │   │ - NATS audit log │
│ - All domain     │   │ - Pub/Sub        │   │ - Decision trees │
│   tables         │   │ - Session state  │   │                  │
└──────────────────┘   └──────────────────┘   └──────────────────┘

Why not just PostgreSQL? Redis provides 10-100x faster reads for hot data (agent definitions, pipeline configs). MongoDB handles append-heavy, schema-evolving audit logs without bloating the primary database.

What Lives Where

Data Flow Patterns

Write Operations

All writes go to PostgreSQL first, then publish events:

Gateway → NATS → Rust Core → PostgreSQL → Redis Stream (event)

Example: Creating a workload

1. Gateway receives HTTP request, validates JWT, publishes to qadra.{tenant}.workload.do.create
2. Rust Core handler validates payload, inserts into workloads table within a transaction
3. Snapshots pipeline stages into pipeline_snapshot JSONB
4. Publishes workload.created to Redis Stream
5. Returns response via NATS reply

Why PostgreSQL first? PostgreSQL is the source of truth. If the Redis publish fails, the workload still exists. Event delivery can be retried.

Read Operations

Reads use the cache-aside pattern:

Gateway → NATS → Rust Core → Redis Cache → miss? → PostgreSQL → populate cache

What gets cached: Agent definitions, pipeline configurations, tenant settings (read frequently, change rarely).

What doesn't get cached: Workloads, tasks (frequently updated, stale data is dangerous), artifacts (too large, accessed infrequently).

NATS Request-Reply

All service communication uses NATS request-reply. The gateway publishes a request to a NATS subject and awaits a reply:

Gateway: NATS.request("qadra.auth.login", payload) → response
Rust Core: subscribes to "qadra.auth.>" → processes → replies

NATS Subject Hierarchy

NATS is the message bus connecting all services. Subjects follow a hierarchical naming convention.

Subject Pattern Summary

Auth Operations (qadra.auth.*)

Auth subjects are tenant-less (pre-authentication context):

Admin Operations (qadra.admin.*)

Admin subjects are tenant-less. Every operation checks is_super_admin before executing.

Workload Operations (qadra.{tenant}.workload.do.*)

Flow Operations (qadra.{tenant}.flow.do.*)

Email Operations (qadra.email.*)

Epistemic Operations (qadra.{tenant}.epistemic.*)

Tenant-scoped knowledge-graph operations handled by Rust Core (and qadra-epistemic-core):

NATS Observer (qadra.>)

A catch-all subscriber that records every NATS message to the nats_audit MongoDB collection. Fire-and-forget via tokio::spawn. Skips qadra.audit.> subjects to avoid recursive logging. Graceful no-op if MongoDB is unavailable. Metadata only -- no payload content stored.

Orchestration Flow (Python Agent)

Client
    |
    v  NATS: qadra.{tenant}.workload.execute
    |
Python Agent (WorkloadOrchestrator)
    |
    +-- For each pipeline stage:
    |     +-- get_forwarded_context  (Rust Core via NATS)
    |     +-- LLM selects best agent for stage
    |     +-- create_task            (Rust Core via NATS)
    |     +-- LLM executes with agent persona + context
    |     +-- complete_task          (Rust Core via NATS)
    |     +-- create_artifact        (Rust Core via NATS)
    |     +-- record_stage_result    (Rust Core via NATS)
    |     +-- advance_stage          (Rust Core via NATS)
    |
    +-- Return WorkloadResult (success, stages, artifacts)

Flow Execution Engine

Event-driven graph executor in src/flow_executor.rs. Processes ReactFlow graph definitions:

1. Start: Parse flow graph, find Start node, seed execution queue
2. Loop: Pop node from queue, check predecessors complete, execute, store output, enqueue successors, persist state
3. Pause: HumanInput nodes pause execution, wait for user response
4. Resume: Human input submitted, merge into state, re-enqueue successors, continue loop
5. End: End node reached, execution status completed

Node types: Start, End, Agent, Condition, HumanInput, Loop, Http, ExecuteFlow, Transform, Delay, KnowledgeLookup, Research, Artifact, Extract, SpoFilter, SpoWrite, ClaimValidation.

Safety rails: MAX_TOTAL_STEPS=500, max Loop iterations=100, max Delay=300s.

SPO Pipeline Node Types

Three node types integrate the SPO knowledge store directly into flow execution:

SSE Streaming (Conversation Progress)

Real-time flow execution progress streamed to the frontend via Server-Sent Events:

React (SSE consumer) <-- SSE -- Gateway (SSE endpoint) <-- NATS pub/sub -- Rust Core
                                                                            |
POST /messages/stream -> subscribes to _PROGRESS.{uuid}       FlowExecutor emits
returns text/event-stream  forwards events as SSE              node_started/completed
                           closes on "done" event              to progress_subject

Source Code Organization

Rust Core (src/)

Rust Core has no HTTP server. It subscribes to NATS subjects and handles all data operations:

src/
├── main.rs               # NATS handler spawning (all wildcard subscribers)
├── nats_service.rs        # NATS message routing and handler dispatch
├── nats_observer.rs       # Catch-all NATS observer -> MongoDB audit
├── config.rs              # Configuration loading
├── error.rs               # Error types
├── lib.rs                 # Library root
│
├── ingestion/             # Knowledge Core ingestion pipeline
│   ├── analyzer.rs        # LLM content classification (Spo/Both/Discard)
│   ├── chunker.rs         # Text chunking (pure Rust, paragraph boundaries)
│   ├── extractor.rs       # LLM triple extraction with epistemic metadata
│   ├── embedder.rs        # Batch embedding wrapper (chunks + SPO nodes)
│   └── service.rs         # Pipeline orchestrator (IngestionService impl)
│
├── flow_executor.rs       # Event-driven graph executor for agent flows
├── flow_generator.rs      # AI-powered flow graph generation from prompts
│
├── email_service.rs       # SMTP2GO email delivery
├── email_renderer.rs      # Handlebars template rendering
├── email_logging.rs       # LoggingEmailService decorator (notification audit)
│
├── tracing/               # Audit trace system (MongoDB)
│   ├── context.rs         # Trace context management
│   ├── decision.rs        # Decision tree types
│   ├── repository.rs      # MongoDB trace repository
│   └── types.rs           # Trace, Decision, Metrics structs
│
├── telemetry/             # OpenTelemetry integration
│   └── mod.rs             # Init, config
│
└── db/                    # Database utilities
    └── mod.rs             # Connection pool helpers

Crate Workspace (crates/)

crates/
├── qadra-traits/              # ALL interfaces (tiny, stable, no deps)
├── qadra-container/           # IoC container, service wiring, mock stubs
├── qadra-db-postgres/         # Repository: PostgreSQL (sqlx)
├── qadra-vector-pgvector/     # Vector: pgvector HNSW index
├── qadra-llm-openrouter/      # LLM: OpenRouter (OpenAI-compatible)
├── qadra-cache-redis/         # Cache: Redis
├── qadra-queue-redis/         # Queue: Redis Streams
├── qadra-semantic-cache-redis/ # Semantic cache: Redis with embedding similarity
├── qadra-nats/                # NATS client utilities
└── qadra-epistemic-core/      # Epistemic operations (SPO, curvature, attribution)

Key Traits (qadra-traits)

#![allow(unused)]
fn main() {
/// Repository - Database operations (PostgreSQL)
#[async_trait]
pub trait Repository: Send + Sync {
    async fn get_agent(&self, id: Uuid) -> Result<Option<Agent>>;
    async fn list_active_agents(&self, tenant_id: Uuid) -> Result<Vec<Agent>>;
    // ... more methods
}

/// AuthRepository - User/tenant/auth operations
#[async_trait]
pub trait AuthRepository: Send + Sync {
    async fn login(&self, email: &str, password: &str) -> Result<User>;
    async fn register_user(&self, params: RegisterParams) -> Result<(User, Tenant)>;
    // ... more methods
}

/// WorkloadRepository - Workload/task/artifact CRUD
#[async_trait]
pub trait WorkloadRepository: Send + Sync {
    async fn create_workload(&self, params: CreateWorkloadParams) -> Result<Workload>;
    async fn create_task(&self, params: CreateTaskParams) -> Result<Task>;
    // ... more methods
}

/// AgentFlowRepository - Flow/execution CRUD
#[async_trait]
pub trait AgentFlowRepository: Send + Sync {
    async fn create_flow(&self, params: CreateFlowParams) -> Result<AgentFlow>;
    async fn create_execution(&self, params: CreateExecutionParams) -> Result<FlowExecution>;
    // ... more methods
}

/// LlmProvider - Model completions
#[async_trait]
pub trait LlmProvider: Send + Sync {
    async fn complete(&self, request: CompletionRequest) -> Result<CompletionResponse>;
    async fn stream(&self, request: CompletionRequest) -> Result<CompletionStream>;
}

/// VectorIndex - Embedding storage and search
#[async_trait]
pub trait VectorIndex: Send + Sync {
    async fn upsert(&self, tenant_id: Uuid, records: Vec<EmbeddingRecord>) -> Result<()>;
    async fn search(&self, tenant_id: Uuid, query: Embedding, k: usize) -> Result<Vec<SearchHit>>;
}
}

Node.js Gateway (gateway/)

The gateway is an Express application that handles all HTTP concerns:

gateway/src/
├── index.ts               # Express app, middleware setup
├── config.ts              # Environment configuration
├── nats.ts                # NATS client connection
├── storage.ts             # S3/MinIO file operations
├── images.ts              # Imaginary image processing client
├── auth/                  # JWT creation, validation, middleware
├── routes/
│   ├── auth.ts            # Login, register, refresh, verify email
│   ├── users.ts           # User profile, avatar
│   ├── teams.ts           # Team members, invites
│   ├── tenants.ts         # Tenant settings, branding
│   ├── admin.ts           # Super admin operations
│   ├── agents.ts          # Agent CRUD (-> NATS)
│   ├── flows.ts           # Flow CRUD and execution (-> NATS)
│   ├── conversations.ts   # Chat/conversation operations (-> NATS)
│   ├── knowledge.ts       # Knowledge ingestion (-> NATS)
│   ├── artifactTemplates.ts # Artifact template CRUD (-> NATS)
│   ├── files.ts           # File upload/download (S3 + NATS)
│   ├── email.ts           # Email send (-> NATS)
│   ├── notifications.ts   # Email templates, logs (-> NATS)
│   ├── audit.ts           # NATS audit log (-> NATS)
│   └── health.ts          # Liveness and readiness checks
└── types/                 # TypeScript type definitions

Gateway responsibilities:

• JWT creation (login/register) and validation (all authenticated routes)
• CORS configuration
• Multipart file uploads to MinIO (S3-compatible)
• Image processing via imaginary (avatar resize)
• HTTP-to-NATS bridging (every route publishes a NATS request and returns the reply)
• SSE streaming for real-time conversation progress

Gateway does NOT:

• Touch PostgreSQL directly
• Execute business logic
• Manage workload state

React SPA (web/)

web/src/
├── App.tsx                # Router, layout, auth guard
├── main.tsx               # Entry point
├── api/                   # HTTP client (Axios, talks to Gateway)
├── stores/                # Zustand state management
├── pages/
│   ├── Login.tsx
│   ├── Register.tsx
│   ├── Dashboard.tsx
│   ├── Conversations.tsx / ConversationChat.tsx
│   ├── Agents.tsx / AgentEditor.tsx
│   ├── Flows.tsx / FlowBuilder.tsx
│   ├── ArtifactTemplates.tsx / ArtifactTemplateEditor.tsx
│   ├── Admin.tsx
│   ├── Settings.tsx
│   ├── TenantPicker.tsx
│   └── ...
└── components/
    └── flow/nodes/        # ReactFlow custom node components

Python Agent (agent/)

The Python Agent is the intelligence layer. It handles everything that requires LLM calls: workload orchestration, synthesis, research, gap detection, and verification. It communicates exclusively with Rust Core via NATS -- it never touches PostgreSQL, MongoDB, or Redis directly.

Entry point: qadra-agent CLI command starts a FastAPI/uvicorn HTTP server and subscribes to NATS subjects for workload execution, agent queries, and document verification.

agent/src/qadra_agent/
├── main.py                # Entry point (uvicorn server)
├── api.py                 # FastAPI HTTP endpoints
├── nats_client.py         # NATS connection and request helpers (CoreClient)
├── orchestrator.py        # WorkloadOrchestrator (pipeline execution)
├── llm.py                 # LLM provider client (OpenRouter, Anthropic, OpenAI)
├── loop.py                # Agent loop (synthesis, research, gap detection)
├── research.py            # Autonomous research via Exa, Tavily, Perplexity
├── ingestion.py           # Document ingestion coordination
├── attribution.py         # KG-SMILE attribution
├── verification.py        # Document verification
├── conflict_detection.py  # Contradiction detection in KG
└── config.py              # Pydantic settings (QADRA_ env prefix)

WorkloadOrchestrator

The WorkloadOrchestrator class drives workload execution through pipeline stages. It subscribes to qadra.{tenant}.workload.execute via NATS and processes each workload sequentially through its pipeline.

Execution flow for each workload:

1. Load workload and agents: Fetch the workload (including pipeline_snapshot) and list all active agents from Rust Core via NATS
2. Sort stages: Order pipeline stages by their order field
3. For each stage (sequential):
   • get_forwarded_context -- retrieve merged outputs and artifact references from all prior completed stages (Rust Core via NATS)
   • _select_agent -- LLM picks the best agent for the stage based on agent structure, description, and forwarded context. If only one agent exists, it is selected directly. Temperature is set to 0.0 for deterministic selection.
   • create_task -- record the agent assignment in PostgreSQL (Rust Core via NATS)
   • _execute_task -- build a system prompt from the agent persona (display_name, structure, stage name) and a user prompt incorporating forwarded context from prior stages, then call the LLM
   • complete_task -- store the LLM output on the task (Rust Core via NATS)
   • If output exceeds 200 characters, create_artifact -- persist the output as a named document artifact (Rust Core via NATS)
   • record_stage_result + advance_stage -- persist the stage result and move the workload to the next stage (Rust Core via NATS)
4. Return WorkloadResult with success status, stages completed count, and artifact list

Error handling: If any stage fails, the task is marked as failed via fail_task, and the orchestrator returns a partial result with the error and any artifacts produced by earlier stages.

Key design choice: The orchestrator never writes to the database directly. Every data operation is a NATS request-reply to Rust Core. This keeps the Python Agent stateless and horizontally scalable.

NATS Subjects Handled by Python Agent

NATS Callbacks to Rust Core

The CoreClient wraps NATS request-reply calls to Rust Core for all data operations:

LLM Provider Configuration

The Python Agent uses an OpenAI-compatible client (via httpx) that routes through OpenRouter by default. Provider selection is via environment variables:

All OpenRouter HTTP requests are rate-limited through a process-global semaphore (max 8 concurrent) with exponential backoff and jitter on 429 responses.

Dependencies

Python 3.10+, key packages: nats-py (NATS client), fastapi/uvicorn (HTTP API), openai (OpenAI-compatible LLM calls), anthropic (Claude API), httpx (async HTTP), pydantic-settings (configuration), tenacity (retry logic), structlog (structured logging).

Infrastructure

Docker Compose Services

Observability Stack

Resource Limits

Kong Configuration

Kong runs in DB-less mode with declarative config at kong/kong.yml:

• JWT validation for Qadra-issued tokens
• Injects headers: X-User-Email, X-Tenant-Id, Authorization
• Routes: /api/* to Gateway, /* to React SPA

Network

All services run on the qadra-network Docker network. Internal service discovery uses container names (e.g., postgres:5432, nats:4222, loki:3100).

Event Streams

Redis Streams for asynchronous coordination:

Why Redis Streams over simple Pub/Sub?

• Durability: Messages persist until acknowledged
• Consumer Groups: Multiple workers share a stream without duplicate processing
• Replay: New consumers can read historical messages

Security Model

Authentication

Two authentication methods:

JWT Tokens (Human Users):

• Access token (15 min): { sub, email, name, tid, tname, role, sa? }
• Refresh token (7 days): { sub, type: "refresh" }, SHA256-hashed for DB storage
• sa claim present only for super admins
• Gateway creates and validates JWTs; Rust Core never sees raw HTTP auth headers

API Keys (Machine-to-Machine):

• Tenant-scoped, permission-restricted, rate-limited, revocable

Authorization Flow

Browser -> Kong (validates JWT) -> Gateway (extracts claims) -> NATS (claims in payload) -> Rust Core (enforces tenant_id)

Authorization happens at the boundary:

1. Kong: Validates JWT signature and expiry
2. Gateway: Extracts user/tenant claims, injects into NATS payload
3. Rust Core: Enforces tenant_id scoping on every database query

Row-Level Security

Every query is scoped to a tenant at the repository layer:

#![allow(unused)]
fn main() {
impl Repository for PostgresRepository {
    async fn get_workload(&self, tenant_id: Uuid, id: Uuid) -> Result<Workload> {
        sqlx::query_as!(Workload,
            "SELECT * FROM workloads WHERE tenant_id = $1 AND id = $2",
            tenant_id, id
        ).fetch_one(&self.pool).await
    }
}
}

Defense in Depth

Super Admin

Platform-level is_super_admin boolean on the users table. NOT a per-tenant role. Capabilities: create tenants, drop into any tenant (bypasses membership with synthetic JWT), reset passwords, promote/demote. NATS handler: qadra.admin.>.

File Storage

MinIO provides S3-compatible object storage:

• Gateway-proxied downloads: URLs are always /api/files/{id}/download, never direct S3 links
• Image processing: Avatar uploads auto-resized via imaginary (HTTP service) to 256x256 WebP
• Purpose-based validation: avatar/logo have 2MB limit + image-only types; others 10MB default
• Object key format: tenants/{tenant_id}/{purpose}/{entity_id}{ext}
• Multi-backend: Tenants can configure their own S3 endpoint via tenants.settings.storage

Observability

Services (Rust, Node, etc.)
    |
    v  OpenTelemetry Protocol (OTLP)
    |
OTel Collector (receives, batches)
    |
    +-- Loki (logs)
    +-- Prometheus (metrics)
    |
    v
Grafana (visualization + alerting)

• Grafana dashboards: Qadra Observability (/d/qadra-overview), Container Monitoring (/d/qadra-containers)
• 8 provisioned alerts: High CPU, high memory, critical memory, scrape target down, OTel export errors, container restart, high error rate, low disk space
• Structured JSON logs: All services emit to stdout with trace_id, tenant_id, service fields

Qadra Architecture

C4 Level 1 — System Context

Who uses Qadra and what does it talk to.

flowchart TB
    user["fa:fa-user Tenant User<br/><i>Owner, admin, or member</i>"]
    admin["fa:fa-user-shield Super Admin<br/><i>Cross-tenant platform operator</i>"]

    subgraph platform["Qadra Platform"]
        qadra["Multitenant Workload Orchestration<br/><i>Agents perform tasks · produce artifacts · ingest knowledge</i>"]
    end

    subgraph external["External Services"]
        direction LR
        openrouter["OpenRouter<br/><i>LLM inference</i><br/>Claude · GPT · Llama"]
        smtp["SMTP2GO<br/><i>Email delivery</i>"]
        minio["S3 / MinIO<br/><i>Object storage</i>"]
    end

    user -->|"Workloads, agents,<br/>flows, conversations"| qadra
    admin -->|"Create tenants,<br/>manage users"| qadra
    qadra -->|"LLM completions<br/>+ embeddings"| openrouter
    qadra -->|"Verification, invite,<br/>notification emails"| smtp
    qadra -->|"File upload<br/>/ download"| minio

    style platform fill:#0d1117,stroke:#58a6ff,stroke-width:2px
    style external fill:#1a1a2e,stroke:#30363d

C4 Level 2 — Container Diagram

What's inside the Qadra platform.

flowchart TB
    user["fa:fa-user Tenant User<br/><i>Browser</i>"]

    subgraph platform["Qadra Platform"]
        direction TB

        subgraph frontend["Frontend"]
            spa["Web SPA<br/><i>React · Vite · Tailwind</i><br/>:5173"]
        end

        subgraph gateway_layer["Gateway"]
            gateway["API Gateway<br/><i>Node.js · Express</i><br/>:4000<br/>JWT auth · CORS · HTTP→NATS"]
        end

        nats{{"fa:fa-arrows-alt NATS<br/><i>Request-Reply Bus</i><br/>:4222"}}

        subgraph services["Services"]
            direction LR
            core["Rust Core<br/><i>Truth layer</i><br/>SPO index · ingestion<br/>flow executor · auth"]
            agent["Python Agent<br/><i>Intelligence layer</i><br/>LLM orchestration<br/>workload execution"]
        end

        subgraph data["Data Stores"]
            direction LR
            postgres[("PostgreSQL<br/><i>pgvector</i><br/>Source of truth")]
            mongodb[("MongoDB<br/>Audit traces")]
            redis[("Redis<br/>Cache · streams")]
        end
    end

    subgraph external["External Services"]
        direction LR
        openrouter["OpenRouter<br/><i>LLM API</i>"]
        minio["MinIO<br/><i>S3 storage</i>"]
        smtp["SMTP2GO<br/><i>Email</i>"]
    end

    user -->|"HTTPS"| spa
    spa -->|"HTTPS"| gateway
    gateway <-->|"publish / request"| nats
    nats <-->|"subscribe / reply"| core
    nats <-->|"subscribe / reply"| agent

    core --> postgres
    core -.-> mongodb
    core --> redis
    gateway --> minio

    core -->|"HTTPS"| openrouter
    agent -->|"HTTPS"| openrouter
    core -->|"HTTPS"| smtp

    style platform fill:#0d1117,stroke:#58a6ff,stroke-width:2px
    style frontend fill:#161b22,stroke:#30363d
    style gateway_layer fill:#161b22,stroke:#30363d
    style services fill:#161b22,stroke:#30363d
    style data fill:#161b22,stroke:#30363d
    style external fill:#1a1a2e,stroke:#30363d

C4 Level 3 — Rust Core Components

What's inside the Rust Core container.

flowchart TB
    NATS["fa:fa-envelope NATS Bus"]

    subgraph core["Rust Core"]
        direction TB
        nats_svc["NATS Service<br/><i>8 wildcard handlers</i>"]

        subgraph handlers["Request Handlers"]
            direction LR
            h_epist["epistemic.*<br/>KG queries, ingestion"]
            h_work["workload.do.*<br/>CRUD + stage ops"]
            h_auth["auth.*<br/>Login, register, tokens"]
            h_flow["flow.do.*<br/>CRUD + execution"]
            h_email["email.*<br/>Send + templates"]
            h_admin["admin.*<br/>Super admin ops"]
        end

        subgraph engines["Engines"]
            direction LR
            ingestion["Ingestion Pipeline<br/><i>analyzer → extractor → embedder</i>"]
            flow_exec["Flow Executor<br/><i>graph runner, 13 node types</i>"]
            epistemic["Epistemic Core<br/><i>competence, verification, curvature</i>"]
        end

        subgraph infra["Infrastructure"]
            direction LR
            container["Service Container<br/><i>IoC, trait wiring</i>"]
            observer["NATS Observer<br/><i>catch-all audit trail</i>"]
            email_svc["Email Service<br/><i>SMTP2GO + Handlebars</i>"]
        end

        nats_svc --> handlers
        h_epist --> ingestion
        h_epist --> epistemic
        h_flow --> flow_exec
        h_email --> email_svc
    end

    NATS <--> nats_svc
    observer -.->|"fire & forget"| MongoDB[(MongoDB)]
    container --> PostgreSQL[(PostgreSQL)]
    ingestion --> PostgreSQL

    style core fill:#0d1117,stroke:#58a6ff,stroke-width:2px
    style handlers fill:#161b22,stroke:#30363d
    style engines fill:#161b22,stroke:#30363d
    style infra fill:#161b22,stroke:#30363d

Data Flow — Document Ingestion

flowchart LR
    A[Agent proposes document] -->|NATS: propose_document| B[documents table<br/>status: pending_approval]
    B --> C{Human reviews}
    C -->|Approve| D[stamps approved_by + approved_at]
    C -->|Reject| E[stamps rejected_by + reason]
    D -->|async tokio::spawn| F[Ingestion Pipeline]
    F --> G[LLM Analyzer<br/>Spo / Both / Discard]
    G -->|Spo| H[LLM Extractor<br/>SPO triples]
    G -->|Both| H
    G -->|Both| I[Chunker + Embedder<br/>vector chunks]
    H --> J[(spo_nodes + spo_triples)]
    I --> K[(pgvector embeddings)]
    H --> L[document_triples<br/>provenance links]
    I --> M[document_chunks<br/>provenance links]

Data Flow — Workload Execution

flowchart LR
    A[Client] -->|NATS: workload.execute| B[Python Agent]
    B --> C[For each pipeline stage]
    C --> D[Get forwarded context<br/>from prior stages]
    D --> E[LLM selects best agent]
    E --> F[Create task<br/>via NATS to Rust Core]
    F --> G[LLM executes with<br/>agent persona + context]
    G --> H[Complete task + create artifact<br/>via NATS to Rust Core]
    H --> I[Record stage result<br/>advance to next stage]
    I --> C

NATS Subject Map

flowchart TB
    subgraph "Tenant-Scoped"
        E["qadra.{tid}.epistemic.*<br/>KG queries, ingestion, attribution"]
        W["qadra.{tid}.workload.do.*<br/>Workload/task/artifact CRUD"]
        F["qadra.{tid}.flow.do.*<br/>Flow CRUD + execution"]
    end

    subgraph "Tenant-Less"
        AU["qadra.auth.*<br/>Login, register, tokens, invites"]
        EM["qadra.email.*<br/>Send, templates, logs"]
        AD["qadra.admin.*<br/>Super admin operations"]
        AUD["qadra.audit.*<br/>Audit log queries"]
    end

    subgraph "Observer"
        OB["qadra.><br/>Catch-all → MongoDB"]
    end

Design Decisions

This page captures the "why X over Y" behind Qadra's architecture -- the load-bearing choices that shape the codebase, and the alternatives we considered and rejected. It is a distillation of the full decision log (.claude/rules/decision-index.md), grouped by theme and trimmed to the decisions a future reader is most likely to question.

Decisions are not permanent. If the context that justified one changes, we re-evaluate and update the log. The goal here is to make the reasoning legible so nobody "fixes" a deliberate choice without understanding it first.

Context Retrieval

The single most foundational choice in the system.

• Chose vector similarity (pgvector HNSW) + SPO triples over BALLS -- the original design used a 3D graph with GPU backends and Ricci-curvature traversal. We replaced it with pgvector HNSW (fast approximate nearest-neighbour search) plus an SPO triple store for fact grounding. It reuses existing infrastructure, is far simpler to operate, and dropped a large amount of complexity (3D geometry, GPU dependencies, curvature math) that was not earning its keep.
• Chose demand-driven context over supply-driven -- context is retrieved against requirements (vector similarity + SPO paths) rather than dumping everything into the prompt. Standard RAG searches everything; SPO-RAG searches where the triple graph points.

Data & Storage

Three databases, each doing the job it is best at.

• Chose PostgreSQL as source of truth over MongoDB / CockroachDB -- ACID transactions, native pgvector for embeddings, mature tooling, and sqlx compile-time query verification. All authoritative state lives here.
• Chose Redis as the hot layer over Memcached / in-process cache -- one tool, three jobs: Streams for events, Pub/Sub for WebSocket fan-out, and read-through caching. Cache-aside for agent/pipeline/tenant config (read often, change rarely); workloads and tasks are never cached (stale data is dangerous).
• Chose MongoDB for audit storage over PostgreSQL JSONB / Elasticsearch -- decision trees and NATS audit entries have a flexible, append-heavy, schema-evolving shape that fits a document store naturally, without bloating the primary database or requiring joins.
• Chose Redis Streams over plain Pub/Sub -- durability (messages persist until acked), consumer groups (workers share a stream without duplicates), and replay for new consumers.

Service Topology

How the containers talk to each other.

• Chose a Node.js gateway + NATS split over a Rust HTTP (axum) core -- JWT creation/validation, CORS, and HTTP routing are a thin layer that doesn't need Rust's performance. The gateway owns the HTTP boundary and file uploads; Rust Core stays NATS-only and never touches PostgreSQL via HTTP. This keeps the truth layer decoupled from the web framework and lets each scale independently.
• Chose NATS request-reply over direct HTTP between services -- the gateway knows nothing about database schemas; Rust Core knows nothing about HTTP routing. NATS gives uniform messaging for every concern (auth, workloads, flows, email, admin, epistemic) and lets a single catch-all observer audit all traffic without modifying handlers.
• Chose a catch-all NATS observer (qadra.>) -> MongoDB over per-handler logging / a PostgreSQL audit table -- one subscriber captures all traffic in one place, metadata only, fire-and-forget. Skips qadra.audit.> to avoid recursive logging; degrades to a no-op if MongoDB is down.
• Chose Kong as the API gateway over Traefik / nginx / custom -- battle-tested JWT validation, rate limiting, and routing with declarative config.
• Chose Axum for any Rust HTTP needs over Actix / Rocket / Go -- Tower middleware ecosystem, async-first, type-safe extractors.

Agent Model

Agents are team members, not prompts.

• Chose "agents are programs that call models" over "agents are prompts that do everything" -- deterministic orchestration with models used for intelligence, not control flow.
• Chose pure-persona agents (system prompt) over atomics (JavaScript + structure/requirements/sequencing) -- the atomics model was never used in practice. Agents are LLM personas orchestrated by flow graphs; the system prompt defines behaviour directly. Model selection became a platform concern, not a per-agent one. (Migration 028 removed the atomics-era columns.)
• Chose per-message agent_id over per-conversation -- users can switch agents mid-thread (e.g. start with Research Analyst, pivot to Memo Writer) without opening a new conversation.

Knowledge Graph & SPO Personas

The epistemic layer -- grounding, attribution, and specialist sub-agents over the SPO store.

• Chose SPO as the always-on index over a vector-only path -- ingestion routes content to Spo (triples only), Both (triples + chunks), or Discard. There is no vector-only route; SPO navigates structure, pgvector retrieves similar content.
• Chose the KG-SMILE Two Gates model for explainability -- a Competence Gate (pre-synthesis: "does my knowledge cover this?") and a Verification Gate (post-synthesis: "did I stay grounded?"). Production runs two LLM-free verifiers on every chat response: a rule-based grounding gate (sentence parsing + word overlap) and a Belnap four-valued gate (pure-Rust UD parse + polarity-tagged graph, distinguishing "false" from "no evidence", with an optional ONNX parser behind the onnx-ort feature). Perturbation-based attribution is the staged higher-precision path.
• Chose epistemic metadata on every triple -- each triple carries an asserter and an assertion_type (fact / claim / opinion). A fact from NASA is weighted differently from an opinion in a blog post; synthesis differentiates verified facts from analyst claims.
• Chose SPO-aware flow nodes (SpoFilter, SpoWrite, ClaimValidation) over an external grounding service -- the knowledge store integrates directly into flow execution: query triples into state, persist agent output as provenanced triples, and ground prior-stage claims against the KG.
• Chose a financial-data-extraction sub-agent persona over inline extraction -- a specialised internal-visibility persona queries the SPO index for an entity's financial metrics, with RelinquishToParent control and a single call budget.

Runtime & Plugins

Lightweight, sandboxed, swappable.

• Chose QuickJS over V8 / Deno / Node.js -- atomics and plugins need isolation and instant-on (<1ms cold start, ~500KB), not raw throughput. Without instant-on the pattern collapses back into batched compound prompts.
• Chose a unified plugins table with a plugin_type discriminator over separate tables per type -- extractors, validators, and evaluators share the same storage (code, bundled code, schemas), the same QuickJS execution model, and the same CRUD/execute API. One schema makes adding a new plugin type trivial.
• Chose JSONB graph/state storage over normalized tables -- flow definitions (ReactFlow {nodes, edges, viewport}), flow node-states, and pipeline snapshots are always read/written as a whole batch. JSONB stores them verbatim with no impedance mismatch or N+1 joins. The flow executor is a Rust in-process event-driven queue (fast DB access, low latency) rather than a separate microservice.

Workloads & Orchestration

• Chose a JSONB pipeline snapshot at workload-creation time over a version FK -- the snapshot is self-contained and immutable; running workloads are unaffected by later pipeline edits, with no extra table or runtime join.
• Chose a separate usage_records table over JSONB inside task metadata -- enables aggregate queries (sum tokens per tenant/workload), indexed time-range filtering, and batch inserts. JSONB would force a full scan to report usage.
• Chose application-side context forwarding over database triggers -- easier to test; stage results carry artifact references (metadata only), keeping full content in the artifacts table.

Identity & Multi-Tenancy

• Chose a user_tenants junction table over a single tenant_id on users -- users belong to multiple organizations, each with a per-tenant role and an is_default flag. The legacy users.tenant_id column is kept for backward compatibility (tracked as tech debt).
• Chose argon2 (new) + bcrypt (legacy) over argon2-only -- detect $2b$/$2a$ prefixes to verify existing bcrypt hashes; all new passwords use argon2.
• Chose a boolean is_super_admin on users over a separate admins table / RBAC roles -- a binary platform privilege needs one column and a partial index, not joins on every check. Super-admin drop-in uses a synthetic JWT (role: "super_admin") rather than polluting user_tenants with fake membership.

Conversations & Tool-Use

• Chose LLM tool-use (run_flow, generate_chart) over UI buttons / regex detection -- the model autonomously decides when intent maps to a flow or a visualization. Flow IDs are constrained via a JSON-Schema enum; flows run inline (45s) so the agent narrates actual outputs, not just "started". Chart configs travel as artifacts in existing metadata JSONB -- no new endpoints or columns.
• Chose inline epistemic grounding in send_message over RAG-only / bare LLM / a separate service -- reuses the existing SPO + curvature + contradiction pipeline. Grounding context is injected as a System message; it degrades gracefully to an ungrounded call if no triples match.
• Chose SSE over a NATS progress subject over WebSocket / polling -- NATS request-reply allows only one response, so progress streams over an ephemeral _PROGRESS.{uuid} subject that the gateway forwards as Server-Sent Events. SSE is HTTP-native, works with existing auth, and auto-reconnects.

Email & Files

• Chose PostgreSQL per-tenant email templates with global defaults over SMTP2GO-hosted templates -- per-tenant customization without external coupling; Handlebars rendering lives in Rust Core, SMTP2GO is delivery-only. Post-registration email is fire-and-forget so it never adds latency.
• Chose a transparent LoggingEmailService decorator over per-call-site logging -- wrapping send() means every caller gets notification_logs auditing for free. A multi-channel notification_logs table (with a channel column) is future-proofed for SMS/push without a migration.
• Chose MinIO (S3-compatible) with gateway-proxied downloads over direct/pre-signed S3 URLs -- proxied /api/files/{id}/download URLs enforce tenant isolation at the gateway and avoid expiry management. A polymorphic files table (purpose + entity_type/entity_id) handles avatars, logos, and documents with one schema. Image processing uses the standalone imaginary HTTP service (no native deps), falling back to the original on failure.

Frontend

• Chose Apache ECharts over Chart.js / D3 / Recharts / Plotly -- JSON option config is LLM-friendly (the model just emits a JSON object), and it has native financial chart types (candlestick, heatmap, treemap, sankey). D3 needs imperative code; Chart.js lacks financial charts; Plotly is large and opinionated.
• Chose react-markdown + remark-gfm over dangerouslySetInnerHTML -- safe AST-based rendering with a shared MarkdownContent component as the single source of markdown styling.
• Chose extracted ConfirmDialog / Toast components over native window.confirm() -- native dialogs block the thread and can't be themed.
• Chose two-key chord keyboard shortcuts (G then a letter) over single keys / a command palette -- GitHub-style chords are discoverable, don't conflict with typing, and need no extra UI.
• Chose decomposed settings tab components over a monolithic page -- the original 1,759-line file was unmaintainable; each tab now owns its own state and data fetching behind a thin shell.

CI/CD & Deploy

• Chose path-gated, dependency-graphed CI jobs over a single monolithic pipeline -- clippy and unit tests share one Rust compilation (rust-check-test), rust-format runs instantly in parallel, the web job does install/lint/typecheck/test/build once, and Cypress component/E2E jobs run concurrently after it. A single aggregate ci-pass job gates branch protection.
• Chose deterministic, fixture/mock-based agent and E2E evaluation over live-backend tests -- Cypress E2E stubs every /api/* route via cy.intercept() (no backend, deterministic and fast); Vitest store tests mock the API client and SSE streams. UI components are covered by co-located *.cy.tsx component tests, not Vitest.
• Chose routing NATS traffic to Loki via the Rust observer's tracing emission over an OTel NATS receiver / Vector -- the contrib OTel collector has no NATS receiver, so the existing observer emits a structured tracing::info event (target nats.traffic) that flows through the standard OTLP -> collector -> Loki pipeline. Zero new services.
• Chose Loki's native OTLP endpoint (otlp_http -> /otlp) over the removed loki exporter, and baking the collector config into the image via Dockerfile COPY over a bind mount -- bind mounts from a Coolify-managed devcontainer don't resolve on the Docker host, so the config is built in.

Domain Model

Core Concepts

Entity Relationships

Understanding the entity hierarchy is crucial for working with Qadra. Every entity belongs to a tenant, and relationships enforce data integrity.

Tenant
├── Users (via user_tenants junction)
│   ├── is_super_admin (platform-level flag)
│   ├── email_verified
│   └── avatar_url
├── API Keys
├── Team Invites
├── Files (S3/MinIO metadata)
├── Email Templates (override global defaults)
├── Notification Logs (multi-channel audit)
├── Agents[]
├── Pipelines[]
│   ├── stages (live JSONB)
│   └── draft_stages (in-progress edits)
├── Workloads[]
│   ├── pipeline_id -> Pipeline
│   ├── pipeline_snapshot (immutable copy at creation)
│   ├── stage_results (accumulated JSONB)
│   ├── Tasks[]
│   │   └── agent_id -> Agent
│   ├── Artifacts[]
│   │   ├── task_id -> Task
│   │   └── template_id -> ArtifactTemplate
│   └── Usage Records[]
├── Artifact Templates[]
│   └── renderer_id -> Plugin
├── Agent Flows[]
│   └── Flow Executions[]
├── Conversations[]
│   └── Messages[]
│       ├── agent_id -> Agent
│       └── flow_execution_id -> FlowExecution
├── Documents[] (human-gated ingestion)
│   ├── proposed_by -> User
│   ├── approved_by -> User
│   ├── DocumentTriples[] -> SPO Triples
│   └── DocumentChunks[] -> Embeddings
├── SPO Index
│   ├── Nodes[]
│   ├── Triples[]
│   └── Entity Aliases[]
├── Plugins[]
├── Renderers[]
├── BALLS Graphs[]
└── Memory Store[]

Key Relationships Explained

Tenant -> Everything: The tenant is the root of all data. Every table has a tenant_id foreign key. You cannot query across tenants, period.

Users -> Tenants (many-to-many): Users belong to tenants through the user_tenants junction table. Each membership carries a per-tenant role (owner, admin, user) and an is_default flag indicating which tenant loads on login. A user's email is globally unique.

Super Admin: The is_super_admin boolean on the users table is a platform-level flag, not a per-tenant role. Super admins can create tenants, drop into any tenant (bypasses membership), reset passwords, and manage other super admins. The first super admin is promoted via direct SQL.

Pipeline -> Stages: Pipelines define the workflow structure. Stages are stored as JSONB within the pipeline. Pipelines support draft_stages for editing without affecting running workloads, and a pipeline_snapshot is captured on each workload at creation time.

Workload -> Pipeline: A workload references a pipeline and snapshots its stages at creation. This ensures running workloads are unaffected by subsequent pipeline edits.

Workload -> Stage Results: Completed stages record structured results in workloads.stage_results JSONB, keyed by stage name. Downstream stages consume this as forwarded context containing task summaries, merged outputs, and artifact references.

Task -> Agent: Each task is assigned to exactly one agent. The agent performs the work. Multiple tasks in a stage might use the same or different agents.

Task -> Artifacts: Artifacts are outputs. One task might produce multiple artifacts. Artifacts can optionally reference an artifact_template for structured output formatting.

Flow -> Executions: Agent flows are visual graph definitions stored as ReactFlow JSONB. Each execution is a state machine (pending -> running -> paused -> completed/failed/cancelled) with a blackboard state dictionary, per-node states, and an execution queue.

Conversation -> Messages: Conversations are direct agent chat sessions. Messages reference the agent that produced them (switchable per message) and optionally link to a flow execution.

Document -> Triples + Chunks: Documents go through a human-gated approval flow before ingestion. Once approved, the ingestion pipeline extracts SPO triples and/or vector chunks, linking them back to the source document for provenance.

Agents

Agents are the workers in Qadra. Unlike chatbots that respond to prompts, agents are specialists with defined personas and system prompts that perform tasks within workflows and conversations.

Agent Model

Agents have been refactored from the atomics model (JavaScript programs in QuickJS) to a pure persona model. Agents are LLM personas with system prompts, not code artifacts.

Persona fields -- what defines the agent:

• name: Internal identifier ("research-analyst")
• display_name: Human-readable name ("Research Analyst")
• description: What this agent does in plain language
• avatar_url: Profile image
• system_prompt: The LLM system prompt that defines behavior and expertise

Orchestration fields -- how the agent participates in workflows:

• control_type: What happens after the agent completes (retain, relinquish_to_parent, relinquish_to_start)
• output_visibility: Whether output is shown to users (user_facing) or kept internal (internal)
• max_calls_per_parent_agent: Safety rail limiting recursive invocations (default: 3)

Agent Schema

CREATE TABLE agents (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    name TEXT NOT NULL,
    version INTEGER DEFAULT 1,

    -- Persona
    display_name TEXT,
    avatar_url TEXT,
    description TEXT,
    system_prompt TEXT NOT NULL DEFAULT '',

    -- Orchestration
    control_type control_type NOT NULL DEFAULT 'relinquish_to_parent',
    output_visibility output_visibility NOT NULL DEFAULT 'user_facing',
    max_calls_per_parent_agent INTEGER NOT NULL DEFAULT 3,

    -- Metadata
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),

    UNIQUE(tenant_id, name, version)
);

Control Types

Control types define what happens to execution flow after an agent completes its work:

Output Visibility

Example Agent

{
  "name": "research-analyst",
  "display_name": "Research Analyst",
  "description": "Gathers market data and competitive analysis for deal evaluation",
  "avatar_url": "/avatars/analyst.png",
  "system_prompt": "You are a senior research analyst at an investment bank. Your role is to gather and synthesize market data, competitive intelligence, and industry trends. Always cite sources. Structure your output with clear sections: Market Overview, Competitive Landscape, Key Trends, and Risk Factors.",
  "control_type": "relinquish_to_parent",
  "output_visibility": "user_facing",
  "max_calls_per_parent_agent": 3
}

Why system prompts instead of JavaScript code? The original atomics model required agents to be JavaScript programs executed in QuickJS sandboxes. In practice, agent behavior is better expressed as natural language instructions to an LLM. System prompts are easier to write, test, and iterate on. The orchestration layer handles execution coordination; agents focus on expertise.

Versioning still works the same way. You might have research-analyst v1 and v2 with different system prompts but the same persona. Users see "Research Analyst" throughout -- the upgrade is invisible.

Canonical SPO Pipeline Personas

After the SPO-architecture decision (Story 98), four seeded agent presets form the canonical investment pipeline. Each writes to or reads from the SPO knowledge graph, so claims stay grounded and traceable as a deal moves through the stages.

These are the canonical personas following the SPO-architecture decision -- Scarlet establishes grounded facts, Ayana and Eliza consume and extend them, and Reagan produces the human-facing recommendation.

Agent CRUD

Agents are managed through the gateway HTTP API, which routes to Rust Core via NATS.

NATS subjects: qadra.{tenant_id}.agent.do.{operation} where operation is create, get, list, update, or delete.

Create request fields (all except name are optional with defaults):

Update semantics: The gateway uses a sparse update pattern. Only fields explicitly included in the request body are sent to Rust Core. Missing fields are not updated. Fields set to null clear the value. This means you can update just the system prompt without touching other fields.

Agent Editor (UI)

The agent editor (web/src/pages/AgentEditor.tsx) provides a full editing interface organized into three panels:

1. Persona panel: Display name, description. These define the agent's identity as seen by users.
2. System Prompt panel: A monospace textarea for the LLM system prompt. This is the core of the agent -- the instructions that define behavior when invoked by a flow or pipeline. Supports Ctrl/Cmd+S to save.
3. Behavior panel: Control type (dropdown), output visibility (dropdown), and max calls per parent (number input). These govern how the agent participates in orchestration.

The top toolbar shows the agent name (click to rename inline) and a save button with status indicator (saved/error). The editor loads the agent by ID from the URL param and populates all fields from the store.

Pipelines

Pipeline Schema

CREATE TABLE pipelines (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    name TEXT NOT NULL,
    description TEXT,
    stages JSONB NOT NULL DEFAULT '[]',       -- Live stage definitions
    draft_stages JSONB,                       -- In-progress edits (NULL = no draft)
    max_handoffs_per_turn INTEGER NOT NULL DEFAULT 25,  -- Loop prevention
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),

    UNIQUE(tenant_id, name)
);

Draft/Live Versioning

Pipelines support editing without affecting running workloads:

• stages: The live configuration. Running workloads use this (via snapshot).
• draft_stages: An in-progress edit. NULL means no active draft.
• Publishing: Overwrites stages with draft_stages, then clears the draft.
• Discarding: Sets draft_stages to NULL, leaving stages untouched.

When a workload is created, the pipeline's current stages are copied into workloads.pipeline_snapshot. This makes the workload immune to subsequent pipeline edits.

Safety Rails

• max_handoffs_per_turn (pipeline-level, default 25): Maximum agent-to-agent handoffs per orchestration turn. Prevents infinite loops.
• max_calls_per_parent_agent (agent-level, default 3): Maximum times a parent agent can invoke a specific child agent per turn. Prevents recursive abuse.

Stage Definition

{
  "stages": [
    {
      "name": "Research",
      "order": 1,
      "description": "Gather market data and competitive analysis"
    },
    {
      "name": "Due Diligence",
      "order": 2,
      "description": "Validate claims and assess risks"
    },
    {
      "name": "Investment Memo",
      "order": 3,
      "description": "Synthesize findings into recommendation"
    },
    {
      "name": "Review",
      "order": 4,
      "description": "Final quality check"
    }
  ]
}

Workloads

Workloads are the central concept in Qadra -- they represent units of work that progress through pipeline stages. In investment banking terms, a workload is a deal. In IT terms, it might be a ticket or project.

Unlike chat threads that accumulate messages, workloads accumulate structured context as agents complete tasks. This context flows forward, enabling downstream agents to build on previous work.

Workload Schema

CREATE TABLE workloads (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    pipeline_id UUID REFERENCES pipelines(id),
    title TEXT NOT NULL,
    description TEXT,
    status workload_status DEFAULT 'pending',   -- pending, active, completed, failed, cancelled
    current_stage TEXT,
    current_stage_order INTEGER,
    metadata JSONB DEFAULT '{}',
    context JSONB DEFAULT '{}',                 -- Accumulated context from completed tasks
    pipeline_snapshot JSONB,                    -- Immutable copy of pipeline stages at creation
    stage_results JSONB NOT NULL DEFAULT '{}',  -- Per-stage structured results
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

Workload Lifecycle

Workloads follow a strict state machine. Understanding these states is essential for building UIs and debugging issues.

          +-------------+     +-------------+     +-------------+
  [*] --> |   pending   | --> |   active    | --> |  completed  | --> [*]
          +-------------+     +------+------+     +-------------+
                                     |
                                     v
                              +-------------+
                              |   failed    | --> [*]
                              +-------------+

State Descriptions:

State Transitions:

Stage Results and Context Forwarding

Completed pipeline stages produce structured results that flow to downstream stages. This replaces the simple context accumulation model with a richer structure.

Stage Result structure:

{
  "Research": {
    "stage_name": "Research",
    "stage_order": 1,
    "tasks": [
      {
        "task_id": "uuid",
        "agent_id": "uuid",
        "stage": "Research",
        "output": { "market_size": "$50B", "competitors": ["CompA", "CompB"] },
        "artifacts": [
          { "id": "uuid", "name": "Market Analysis", "artifact_type": "document", "content_type": "text/markdown", "size_bytes": 4200 }
        ],
        "completed_at": "2026-01-24T10:00:00Z"
      }
    ],
    "merged_output": { "market_size": "$50B", "competitors": ["CompA", "CompB"] },
    "completed_at": "2026-01-24T10:00:00Z"
  }
}

Key types:

Forwarded context is built by build_forwarded_context() and contains {prior_stages, merged_outputs, artifact_refs} for downstream stage consumption.

Why references only? Artifact references store metadata, not content. Full content stays in the artifacts table. This keeps stage_results lightweight while providing enough information for downstream agents to know what's available.

Tasks

Task Schema

CREATE TABLE tasks (
    id UUID PRIMARY KEY,
    workload_id UUID NOT NULL REFERENCES workloads(id),
    agent_id UUID NOT NULL REFERENCES agents(id),
    stage TEXT NOT NULL,
    stage_order INTEGER NOT NULL,
    status task_status DEFAULT 'pending',     -- pending, queued, running, completed, failed, cancelled
    progress INTEGER DEFAULT 0,               -- 0-100
    input JSONB DEFAULT '{}',
    output JSONB,
    error TEXT,
    queued_at TIMESTAMPTZ,
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

Task Execution Flow

The Python Agent orchestrator (WorkloadOrchestrator) drives task execution. Data operations go to Rust Core via NATS.

1. Client sends qadra.{tenant}.workload.execute (NATS)
2. Python Agent creates workload -> Rust Core (workload.do.create)
3. For each pipeline stage (sequential):
   a. get_forwarded_context -> Rust Core (prior stage outputs, artifact refs)
   b. LLM selects best agent for stage (based on agent description/persona)
   c. create_task -> Rust Core (agent assigned to stage)
   d. LLM executes with agent system prompt + forwarded context
   e. complete_task -> Rust Core (output stored)
   f. create_artifact -> Rust Core (if output > 200 chars)
   g. record_stage_result -> Rust Core (persist to stage_results JSONB)
   h. advance_stage -> Rust Core (move to next stage)
4. Return WorkloadResult (success, stages_completed, artifacts)

Progress updates flow through Redis pub/sub to WebSocket connections, giving users real-time visibility into long-running tasks.

Artifacts

Artifact Schema

CREATE TABLE artifacts (
    id UUID PRIMARY KEY,
    workload_id UUID NOT NULL REFERENCES workloads(id),
    task_id UUID REFERENCES tasks(id),
    agent_id UUID NOT NULL REFERENCES agents(id),
    template_id UUID REFERENCES artifact_templates(id),  -- Optional template
    name TEXT NOT NULL,
    artifact_type TEXT NOT NULL,           -- 'document', 'analysis', 'memo', 'report', etc.
    content_type TEXT,                     -- MIME type
    content TEXT,                          -- Inline for small artifacts
    file_path TEXT,                        -- Storage path for large artifacts
    size_bytes BIGINT,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW()
);

Artifact Types

Artifact Templates

Artifact templates define the expected structure and output format for artifacts. They give agents a blueprint: what sections to produce, what schema to follow, and optionally which renderer plugin transforms the output into a final format (PDF, HTML, etc.).

Template Schema

CREATE TABLE artifact_templates (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    name TEXT NOT NULL,
    display_name TEXT,
    description TEXT,
    version INTEGER NOT NULL DEFAULT 1,

    -- Template structure
    artifact_type TEXT NOT NULL DEFAULT 'document',   -- 'document', 'analysis', 'memo', 'report', etc.
    expected_sections JSONB NOT NULL DEFAULT '[]',    -- Section definitions (see below)
    output_schema JSONB,                              -- JSON Schema for structured output validation
    output_format TEXT NOT NULL DEFAULT 'markdown',   -- 'markdown', 'html', 'pdf', 'json'

    -- Renderer binding (optional)
    renderer_id UUID REFERENCES plugins(id) ON DELETE SET NULL,

    -- Metadata
    is_active BOOLEAN NOT NULL DEFAULT true,
    is_global BOOLEAN NOT NULL DEFAULT false,

    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    UNIQUE(tenant_id, name, version)
);

-- Artifacts can optionally reference a template
ALTER TABLE artifacts ADD COLUMN template_id UUID REFERENCES artifact_templates(id) ON DELETE SET NULL;

Indexes: Tenant-scoped indexes on artifact_type and active templates (partial index on is_active = true).

Creating a Template

A template requires a name and defaults to artifact type document with markdown output. All other fields are optional.

Renderer binding: When renderer_id is set, it references a plugin (from the plugins table) that transforms the agent's raw output into the target output_format. For example, a renderer plugin might convert markdown content into a styled PDF. If no renderer is bound, the raw output is stored as-is. Deleting the referenced plugin sets renderer_id to NULL (ON DELETE SET NULL).

Expected Sections

The expected_sections JSONB array defines the structure an agent should follow when producing an artifact from this template. Each section is an object:

[
  {
    "name": "Executive Summary",
    "description": "High-level overview of findings and recommendation",
    "required": true
  },
  {
    "name": "Market Analysis",
    "description": "Market size, growth trends, competitive landscape",
    "required": true
  },
  {
    "name": "Risk Factors",
    "description": "Key risks and mitigation strategies",
    "required": true
  },
  {
    "name": "Appendix",
    "description": "Supporting data tables and references",
    "required": false
  }
]

Sections are passed to the agent as part of the prompt context. The agent's output is expected to follow this structure, though enforcement depends on the orchestration layer.

Using Templates

When an artifact references a template_id, the template's expected_sections and output_schema guide the agent's output:

1. During task execution: The orchestrator includes the template's section definitions in the agent's prompt context, instructing it to produce output matching the template structure.
2. After generation: If output_schema is set, the output can be validated against the JSON Schema.
3. Rendering: If renderer_id is set, the renderer plugin transforms the raw output into the target output_format (e.g., markdown to PDF).
4. Storage: The resulting artifact is stored in the artifacts table with template_id set, linking it back to the template for provenance.

In the investment banking example, the "Investment Memo" artifact at Stage 3 uses a template that ensures every memo has an Executive Summary, Market Analysis, Risk Factors, and Recommendation section -- consistent formatting across all deals.

Artifact Template API

Templates are managed through the gateway HTTP API, routed to Rust Core via NATS.

NATS subjects: qadra.{tenant_id}.artifact_template.do.{operation} where operation is create, get, list, update, or delete.

List query params: limit (1-100, default 20), offset (default 0), artifact_type (optional filter).

Artifact Template Editor (UI)

The template editor (web/src/pages/ArtifactTemplateEditor.tsx) provides four panels:

1. Template Info: Display name, description, artifact type (dropdown), and output format (dropdown).
2. Expected Sections: A dynamic list where you add, remove, and reorder sections. Each section has a name field, optional description, and a required checkbox.
3. Output Schema: A monospace JSON textarea for defining a JSON Schema. Validated on save -- invalid JSON prevents saving.

The editor supports Ctrl/Cmd+S to save and shows inline save status (saved/error). The name is editable inline from the top toolbar.

Usage Records

Every LLM call, embedding generation, tool invocation, plugin execution, and renderer execution is tracked per-operation:

CREATE TABLE usage_records (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    workload_id UUID REFERENCES workloads(id),
    task_id UUID REFERENCES tasks(id),
    agent_id UUID REFERENCES agents(id),
    model TEXT,
    tokens_in INTEGER NOT NULL DEFAULT 0,
    tokens_out INTEGER NOT NULL DEFAULT 0,
    operation_type usage_operation_type NOT NULL DEFAULT 'llm_call',
    metadata JSONB NOT NULL DEFAULT '{}',
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Operation types: llm_call, embedding, tool_call, plugin_exec, renderer_exec.

Usage records enable aggregate queries (SUM tokens per tenant/workload), time-range filtering via indexed created_at, and batch inserts via UNNEST. Stored in a separate table rather than JSONB inside tasks because aggregate queries across workloads would require scanning all tasks and extracting nested arrays.

Agent Flows

Agent flows provide visual, graph-based workflows with branching, conditions, loops, and human-in-the-loop gates.

Flow Definition Schema

CREATE TABLE agent_flows (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    name TEXT NOT NULL,
    description TEXT,
    flow_data JSONB NOT NULL DEFAULT '{"nodes":[],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}',
    node_count INTEGER NOT NULL DEFAULT 0,
    is_active BOOLEAN NOT NULL DEFAULT true,
    created_by UUID REFERENCES users(id),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    UNIQUE(tenant_id, name)
);

flow_data stores the ReactFlow graph definition as JSONB: {nodes, edges, viewport}. The executor always loads the full graph, never queries individual nodes. JSONB stores it verbatim with no impedance mismatch.

Flow Execution Schema

CREATE TABLE flow_executions (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    flow_id UUID NOT NULL REFERENCES agent_flows(id),
    flow_snapshot JSONB NOT NULL,              -- Immutable copy at execution start
    status flow_execution_status NOT NULL DEFAULT 'pending',
    state JSONB NOT NULL DEFAULT '{}',         -- Blackboard dictionary
    node_states JSONB NOT NULL DEFAULT '{}',   -- Per-node execution state
    execution_queue JSONB NOT NULL DEFAULT '[]',
    current_node_id TEXT,
    error TEXT,
    triggered_by UUID REFERENCES users(id),
    input JSONB NOT NULL DEFAULT '{}',
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Flow Execution Lifecycle

stateDiagram-v2
    [*] --> pending
    pending --> running : execute
    running --> completed : End node reached
    running --> failed : node error / safety limit
    running --> cancelled : user cancels
    running --> paused : HumanInput node
    paused --> running : user submits input
    paused --> cancelled : user cancels

Node Types

The flow executor supports 14 node types. Each node stores its configuration in the data JSONB field of the ReactFlow node object. All nodes that produce output can specify an output_key in their data; after execution, the output is merged into the flow state dictionary under that key.

Start

Entry point for the flow. Exactly one per flow.

Behavior: Pass-through. The execution input (provided when starting the flow) becomes the initial state. All successors are enqueued immediately. No output_key -- state is seeded from execution input directly.

End

Terminal node. At least one per flow. When reached, the execution status becomes completed.

Behavior: Collects final output. If output_keys is specified, filters the state to only those keys. Otherwise returns the full state. No successors are enqueued.

Agent

Invokes an LLM with the agent's system prompt and a rendered prompt template. This is the primary intelligence node.

Behavior: Renders prompt_template by replacing {{key}} placeholders with values from the flow state. Loads the agent's system_prompt by agent_id. Injects upstream state data as a second system message so the agent has context from prior nodes. Calls the LLM with temperature: 0.7, max_tokens: 4096. Output is a JSON object {response, model, tokens}. All successors are enqueued.

Condition

Boolean branching node. Evaluates an expression against the flow state and routes to the matching branch.

Expression evaluation rules:

• key == value: Compares state[key] against value. Strings, numbers, and booleans are compared by string representation. Returns route "true" or "false".
• key != value: Inverse of ==. Returns "true" or "false".
• key (bare): Truthy check. true for non-empty strings, any number, true booleans, non-empty arrays/objects. false for empty strings, null, missing keys, false booleans.

Edge behavior: Outgoing edges MUST use sourceHandle to indicate which route they belong to. An edge with sourceHandle: "true" is followed when the expression evaluates to true; sourceHandle: "false" for the false branch. Only the matching branch is enqueued. Nodes on the non-matching branch (and their entire downstream subtree) are marked Skipped, unless they are also reachable from the taken branch.

HumanInput

Pauses the entire flow execution and waits for a user to submit input via the API.

Behavior: Immediately returns NodeResult::Paused. The node status is set to WaitingForInput and the execution status is set to Paused. State and queue are persisted to the database. When the user submits input via POST /flows/executions/:execId/input, the NATS handler stores the response in the node's output, marks the node as Completed, merges the response into state under output_key, re-enqueues successors, and resumes the execution loop.

Loop

Iterates over a collection in the flow state. Currently a simplified batch processor -- it reads the array, caps it to max_iterations, and outputs the items. It does NOT re-execute successor nodes per item.

Behavior: Reads state[collection_key] as an array. If missing or not an array, outputs {iterations: 0, results: []}. Otherwise caps at max_iterations and outputs {iterations, results}. All successors are enqueued once (not per item).

Http

Makes an outbound HTTP request. Supports GET, POST, PUT, PATCH, DELETE.

Behavior: Templates the URL by replacing {{key}} with state values. Sends the request with a 30-second timeout. Output is {status, body} where body is parsed as JSON if possible, otherwise stored as a string. All successors are enqueued.

ExecuteFlow

Triggers another flow as a sub-flow. Currently a stub -- returns a placeholder message.

Behavior: Returns {message: "Nested flow execution not yet implemented"}. All successors are enqueued. Future implementation will create a child FlowExecution, run it to completion, and merge its output back into the parent state.

Transform

Applies a data transformation by mapping state fields to new output fields.

Behavior: For each entry in mappings: if the value is a string, it is treated as a state key reference and the corresponding state value is copied. If the value is any other JSON type, it is used as a literal. Output is a JSON object with the mapped fields. All successors are enqueued.

Example: {"mappings": {"company": "company_name", "ready": true}} copies state.company_name into output.company and sets output.ready to true.

Delay

Waits for a specified duration before continuing. Capped at 300 seconds.

Behavior: Sleeps for min(duration_seconds, 300) seconds using tokio::time::sleep. Output is {delayed_seconds}. All successors are enqueued after the delay.

KnowledgeLookup

Queries the tenant's SPO knowledge graph for relevant triples.

Behavior: Renders the query template with state variables. Searches SPO nodes matching the query text via search_spo_nodes, then fetches triples connected to those nodes via get_triples_for_nodes. Deduplicates by triple ID. Output is {query, source, limit, results} where results is an array of {subject, predicate, object, confidence, source} objects. Gracefully returns empty results on error. All successors are enqueued.

Research

Performs external research via LLM. Designed for Exa neural search integration (not yet wired -- currently uses LLM knowledge as a stand-in).

Behavior: Renders the query template. Calls the LLM with a research analyst system prompt and the rendered query plus workflow context (truncated to 4000 chars). Output is {query, findings, model, source: "llm_knowledge"}. All successors are enqueued.

Artifact

Produces a structured artifact, optionally formatted against an artifact template.

Behavior: Four modes depending on available inputs:

1. No upstream content, no template: Calls LLM to generate content from the full state context.
2. No upstream content, with template: Calls LLM with template section guidance to generate structured content.
3. Upstream content, with template: Calls LLM to reformat the upstream content into the template structure.
4. Upstream content, no template: Uses the upstream content as-is (no LLM call).

Output is {artifact_name, template_id, artifact_type, output_format, content}. All successors are enqueued.

Extract

Extracts structured parameters from text using an LLM. Typically placed immediately after the Start node to parse user intent into typed variables that downstream nodes reference via {{variable_name}}.

Behavior: Reads state[input_key] as the raw text. If empty, sets all output variables to null. Otherwise, calls the LLM (temperature: 0.0, max_tokens: 1024) with a prompt asking it to extract the named variables as a JSON object. Strips markdown fences from the response if present. Parses the JSON and maps each requested variable into the output. If the LLM returns non-JSON, gracefully sets all variables to null. The output is a flat object -- each extracted variable becomes a top-level key. Unlike other nodes, Extract merges its output variables directly into state (each variable becomes a state key), making them available to all downstream nodes via {{variable_name}}.

Note: Extract nodes do NOT use output_key. Instead, each extracted variable is merged individually into the flow state. This is because the node returns a flat object (not nested under a key), and the executor's output-merge logic writes each top-level key into state.

AI Flow Generation

Flows can be generated from natural language descriptions using an LLM. The flow generator (src/flow_generator.rs) translates user prompts into valid ReactFlow graph definitions.

How It Works

1. User submits a prompt via POST /flows/generate with a natural language description of the desired workflow.
2. Gateway forwards the request to Rust Core via NATS subject qadra.{tid}.flow.do.generate.
3. Rust Core loads context: active agents and artifact templates for the tenant.
4. LLM generates the graph: A system prompt describes all 14 node types with their data field schemas, edge format, layout rules, and structural constraints. The user message includes the available agents (with IDs, names, descriptions) and artifact templates, followed by the user's request.
5. Validation: The generated JSON is validated for structural correctness.
6. Position assignment: If nodes lack positions, a top-to-bottom grid layout is applied automatically.
7. Result returned: The validated flow_data JSONB is returned to the gateway, ready to be saved as a flow.

Node Catalog

The node catalog (NODE_CATALOG in flow_generator.rs) is the single source of truth for node types. It defines the type_name, description, data_example, and notes for each node type. The LLM system prompt, validation logic, and frontend palette all derive from this catalog. Adding a new node type means adding one NodeSpec entry.

LLM Prompt Structure

The system prompt instructs the LLM to return raw JSON (no markdown fences) with this structure:

{
  "nodes": [
    {
      "id": "{type}_{unique_number}",
      "type": "{node_type}",
      "position": { "x": number, "y": number },
      "data": { "label": "Display Name", ...type_specific_fields }
    }
  ],
  "edges": [
    {
      "id": "e_{source_id}-{target_id}",
      "source": "{source_node_id}",
      "target": "{target_node_id}",
      "animated": true,
      "sourceHandle": "true|false"
    }
  ],
  "viewport": { "x": 0, "y": 0, "zoom": 1 }
}

Layout rules: Start node at (400, 50), 200px vertical spacing, 300px horizontal spacing for parallel branches. Condition edges use sourceHandle of "true" or "false".

LLM parameters: temperature: 0.3, max_tokens: 4096.

Validation

validate_flow_data() checks:

• nodes and edges arrays exist and are non-empty
• Exactly 1 start node
• At least 1 end node
• Every node type is in the catalog (rejects unknown types)
• Every edge references valid source and target node IDs

If the LLM wraps its response in markdown code fences despite instructions, they are stripped before parsing.

Position Assignment

ensure_positions() checks whether any node is missing a position or has a position without x/y. If so, all nodes receive a simple top-to-bottom grid layout starting at (400, 50) with 200px vertical spacing.

Flow Execution Engine

The event-driven graph executor in src/flow_executor.rs processes flow graphs. It is a Rust in-process executor that shares the PostgreSQL connection pool and NATS handlers with the rest of Rust Core.

Execution Loop

1. Load execution record (must be status=Pending)
2. Parse graph from flow_snapshot (immutable copy)
3. Initialize: all nodes -> Pending, seed queue with Start node
4. Set execution status -> Running
5. LOOP:
   a. Pop node_id from front of queue
   b. Check all predecessors are Completed or Skipped (join gate)
      - If not all done: re-enqueue and continue
   c. Set node status -> Running, record started_at, snapshot state as input
   d. Persist state to database (crash recovery point)
   e. Emit "node_started" progress event via SSE channel
   f. Execute the node (dispatch by type)
   g. Handle result:
      - Completed: mark Completed, merge output into state via output_key,
        enqueue successors (respecting condition routing), skip non-taken branches
      - Paused: mark WaitingForInput, persist state, set execution -> Paused, RETURN
      - Failed: mark Failed, set execution -> Failed with error, RETURN
   h. Emit "node_completed" or "node_failed" progress event
6. Queue empty: set execution status -> Completed, persist final state

State Merging

The flow state is a JSONB blackboard dictionary shared by all nodes. State flows through the graph as follows:

• Execution input seeds the initial state (e.g., {"user_message": "Analyze NVIDIA"})
• After each node completes, if the node has an output_key in its data, the node's output is stored as state[output_key] = output
• Extract nodes are special: they return a flat object of extracted variables, and each variable becomes an individual state key (no nesting under output_key)
• Each node receives the full state as its input. Nodes read from state via template substitution ({{key}}) or direct key access
• State is never pruned -- it accumulates throughout the execution. All prior node outputs remain available to all downstream nodes

Predecessor Join Gate

Before executing a node, the executor checks that ALL predecessor nodes (based on incoming edges) have status Completed or Skipped. This implements an implicit AND-join for nodes with multiple incoming edges. If predecessors are not yet done, the node is re-enqueued at the back of the queue.

Condition Routing and Branch Skipping

When a Condition node completes, it returns a route value ("true" or "false"). The executor then:

1. For each outgoing edge, checks if the edge's sourceHandle matches the route
2. Matching edges: target nodes are enqueued
3. Non-matching edges: the skip_subtree() function marks the target node and its entire downstream subtree as Skipped (via depth-first traversal), unless a node is also reachable from the taken branch

This ensures that only the relevant branch executes, and nodes on dead branches are marked Skipped rather than left Pending.

HumanInput Pause and Resume

Pause:

1. HumanInput node immediately returns NodeResult::Paused
2. Node status set to WaitingForInput
3. Full execution state (state dict, node_states, queue) persisted to database
4. Execution status set to Paused
5. Executor returns -- the execution is now dormant

Resume (triggered by POST /flows/executions/:execId/input):

1. NATS handler stores user input in the waiting node's output field
2. FlowExecutor::resume() is called
3. Loads persisted state from database
4. Finds the WaitingForInput node, marks it Completed
5. Merges the node's output into state under its output_key
6. Re-enqueues the node's successors
7. Sets execution status to Running
8. Re-enters the same execution loop as run()

Nested ExecuteFlow

The ExecuteFlow node type is currently a stub. It returns {message: "Nested flow execution not yet implemented"} and enqueues successors normally. The intended future behavior: load the target flow by target_flow_id, create a child FlowExecution, run it synchronously, and merge its final state back into the parent flow's state.

Safety Rails

MAX_TOTAL_STEPS is checked on every iteration of the execution loop (both initial run() and resume()). It counts total node executions, not unique nodes -- a loop that re-enqueues nodes can exhaust this limit.

State Persistence and Crash Recovery

The executor persists the full execution state to PostgreSQL after every node execution via update_execution_state(). This writes the current state, node_states, execution_queue, and current_node_id to the flow_executions row. If the Rust process crashes mid-execution, the persisted state contains enough information to determine which nodes completed and where the queue was, though automatic crash recovery is not currently implemented.

Progress Events

The executor optionally emits real-time progress events via an mpsc::UnboundedSender<FlowProgressEvent> channel (attached via with_progress()). Events are fire-and-forget and never block execution. Three event types:

These events are streamed to the frontend via SSE for live flow execution visualization.

Conversations

Conversations provide a direct agent chat interface, separate from pipeline-based workloads. While workloads move structured data through pipeline stages, conversations are freeform -- a user talks to agents, switches between them, and can even trigger flow executions inline.

Conversation Model

A conversation belongs to a single user within a tenant. It holds an ordered sequence of messages, each of which records the agent that responded (if any), the LLM model used, performance metrics, and optional links to flow executions.

CREATE TABLE conversations (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id   UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    user_id     UUID NOT NULL REFERENCES users(id) ON DELETE SET NULL,
    title       TEXT,                            -- Auto-generated or user-set
    is_archived BOOLEAN NOT NULL DEFAULT false,
    deleted_at  TIMESTAMPTZ,                     -- Soft delete (filtered by default)
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE conversation_messages (
    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id     UUID NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    agent_id            UUID REFERENCES agents(id) ON DELETE SET NULL,
    role                TEXT NOT NULL CHECK (role IN ('user', 'assistant', 'system')),
    content             TEXT NOT NULL,
    model               TEXT,                    -- LLM model identifier
    token_count         INTEGER,                 -- Total tokens consumed
    latency_ms          INTEGER,                 -- End-to-end response time
    finish_reason       TEXT,                    -- 'stop', 'length', 'tool_calls', etc.
    rating              SMALLINT CHECK (rating IN (-1, 1)),  -- Thumbs up/down
    flow_execution_id   UUID REFERENCES flow_executions(id) ON DELETE SET NULL,
    metadata            JSONB NOT NULL DEFAULT '{}',
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Indexes:

• idx_conversations_tenant -- tenant-scoped listing, excludes soft-deleted (WHERE deleted_at IS NULL)
• idx_conversations_tenant_user -- per-user listing within a tenant, excludes soft-deleted
• idx_conversation_messages_conv_created -- ordered message retrieval by conversation
• idx_conversation_messages_agent -- find all messages by a specific agent
• idx_conversation_messages_rating -- query rated messages for feedback analysis

Relationship to other entities:

• Each conversation_messages.agent_id points to the agent that produced the assistant response. User messages have agent_id = NULL.
• conversation_messages.flow_execution_id links to a flow_executions row when the message was produced by running a flow.
• conversation_messages.metadata is a JSONB bag that carries artifacts (inline content blocks for rendering) and suggested_actions (follow-up prompts the UI can present).

Agent Switching

Unlike most chat interfaces where the "assistant" is a single model, Qadra conversations let users switch agents on every message. The agent_id lives on conversation_messages, not on conversations.

How it works:

1. The user selects an agent in the UI (agent picker dropdown).
2. When sending a message, the selected agent_id is included in the request body.
3. The backend invokes the chosen agent's system prompt + conversation history to generate a response.
4. Both the user message and assistant message are persisted. The assistant message records the agent_id that produced it.
5. The UI auto-selects the last responding agent for the next message, but the user can switch at any time.

This means a single conversation can contain messages from multiple agents. For example, a user might ask the Research Analyst about market data, then switch to the Due Diligence Specialist to probe risk factors -- all within the same conversation thread.

Flow Tool-Use

Agents can trigger flow executions from within a conversation. When an agent determines that a user's request maps to an existing flow, the backend executes the flow and streams progress events back to the user in real time.

How flow integration works:

1. The send_message handler on the Rust Core evaluates whether the agent should invoke a flow (based on the agent's tool configuration and the conversation context).
2. If a flow is triggered, the executor runs the flow graph and publishes progress events to a NATS progress subject.
3. The gateway forwards these NATS events as SSE events to the client (see SSE Streaming below).
4. When the flow completes, the agent receives the flow's output state and synthesizes a final response.
5. The assistant message is persisted with flow_execution_id set, linking the conversational response to the full flow execution record.

The frontend renders flow progress inline in the chat: a live visualization of node execution with status indicators, output previews, and duration metrics.

SSE Streaming

The POST /conversations/:id/messages/stream endpoint uses Server-Sent Events to deliver real-time progress during message processing. This is a POST endpoint (not GET) because it sends the user's message in the request body and streams the response.

Connection setup:

• The gateway sets SSE headers (Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no).
• A unique NATS progress subject (_PROGRESS.{uuid}) is created for this request.
• The gateway subscribes to the progress subject, then sends the NATS request to Rust Core with the progress_subject field set.
• Rust Core publishes FlowProgressEvent messages ({event, data, seq}) to the progress subject during processing.
• The gateway forwards each event to the SSE stream.

Event types and their data payloads:

Wire format (standard SSE):

event: thinking
data: {"status":"Thinking..."}

event: flow_started
data: {"flow_name":"Intake Triage","execution_id":"abc-123"}

event: node_started
data: {"node_id":"node-1","node_type":"Agent","label":"Classify Request"}

event: node_completed
data: {"node_id":"node-1","output_preview":"Category: Research","duration_ms":1200}

event: assistant_message
data: {"user_message":{...},"assistant_message":{...},"conversation":{...}}

event: done
data: {}

Timeouts and fallbacks:

• The NATS request has a 240-second timeout (flow executor 180s + follow-up LLM ~30s + buffer).
• The gateway has a 250-second overall safety timeout on the SSE stream.
• If the progress events do not deliver an assistant_message before the NATS reply arrives, the gateway uses the NATS reply as a fallback and writes it as an assistant_message SSE event.
• If the SSE stream fails entirely, the frontend falls back to the non-streaming POST /conversations/:id/messages endpoint.

Message Rating

Users can rate individual assistant messages with thumbs up (+1) or thumbs down (-1). Ratings are stored directly on the conversation_messages row.

• Endpoint: POST /conversations/:id/messages/:messageId/rate
• Body: { "rating": 1 } or { "rating": -1 }
• NATS subject: qadra.{tenant_id}.chat.do.rate_message
• Constraint: rating column is SMALLINT CHECK (rating IN (-1, 1)). Only -1 and 1 are valid.
• Idempotent: Rating the same message again overwrites the previous rating.
• Index: idx_conversation_messages_rating enables efficient queries on rated messages for feedback analysis and model evaluation.

Conversation Lifecycle

Create -- POST /conversations

Creates an empty conversation. Title is optional (can be auto-generated later from the first message).

Send message -- POST /conversations/:id/messages (non-streaming) or POST /conversations/:id/messages/stream (SSE)

Sends a user message and receives the agent's response. The request body includes agent_id (required), content (required, max 100,000 characters), and optionally flow_execution_id. Both endpoints persist the user message and assistant message, returning them along with the updated conversation record.

List conversations -- GET /conversations?limit=20&offset=0

Returns ConversationSummary objects with message_count, last_message_preview, and last_agent_id for sidebar display. Excludes soft-deleted conversations.

Get conversation -- GET /conversations/:id

Returns the full conversation record.

List messages -- GET /conversations/:id/messages?limit=100&offset=0

Returns messages ordered by created_at. Used on page load and as a recovery mechanism when SSE streaming fails.

Update title -- PATCH /conversations/:id

Updates the conversation title.

Delete -- DELETE /conversations/:id

Soft-deletes the conversation by setting deleted_at. The conversation and its messages are preserved for audit but excluded from all listing queries.

NATS Subjects

All conversation operations go through tenant-scoped NATS subjects:

Key Design Decisions

• agent_id on messages, not conversations: Users can switch agents mid-conversation. One message might come from the Research Analyst, the next from the Due Diligence Specialist. The conversation itself is agent-agnostic.
• flow_execution_id on messages: Messages can be produced by flow executions, linking conversational output to the full flow execution record with its node states and blackboard data.
• Soft delete with deleted_at: Conversations are never hard-deleted. Setting deleted_at excludes them from listing queries while preserving the audit trail.
• metadata JSONB bag: Carries artifacts (inline content blocks like charts or agent-produced documents) and suggested_actions (follow-up prompts). Extensible without schema migration.
• SSE over WebSocket for streaming: SSE is simpler to implement (one-way server-to-client), works through proxies/CDNs, and naturally maps to the request-response pattern of sending a message and streaming the response. WebSocket would be overkill for this use case.

Document Ingestion

Documents go through a human-gated approval flow before knowledge ingestion. Agents can propose documents, but humans must approve before ingestion runs.

Document Schema

CREATE TABLE documents (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    title TEXT,
    content TEXT NOT NULL,
    content_type TEXT DEFAULT 'text/plain',
    source_url TEXT,
    source_type TEXT,                          -- 'upload', 'web', 'api', 'research'

    -- Ingestion lifecycle
    ingestion_status TEXT DEFAULT 'pending_approval',
    ingestion_result JSONB,

    -- Proposal gate
    proposed_by UUID REFERENCES users(id),     -- Who submitted for review
    approved_by UUID REFERENCES users(id),     -- Human who opened the gate
    approved_at TIMESTAMPTZ,
    rejected_by UUID REFERENCES users(id),     -- Who rejected (if rejected)
    rejected_at TIMESTAMPTZ,
    rejection_reason TEXT,

    created_at TIMESTAMPTZ DEFAULT NOW(),
    processed_at TIMESTAMPTZ
);

Document Lifecycle

pending_approval --> approved --> processing --> completed
                                            --> failed
pending_approval --> rejected

Why human-gated? Agents can autonomously discover and propose documents for ingestion, but a human must approve before content enters the knowledge graph. This prevents polluting the KG with low-quality or irrelevant content.

Ingestion Routing

Once approved, the ingestion pipeline routes content:

Results link back to the document:

• document_triples: Which SPO triples came from this document
• document_chunks: Which vector chunks came from this document

Users and Authentication

User Schema

CREATE TABLE users (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),  -- Legacy, kept for backward compat
    email TEXT NOT NULL,
    email_verified BOOLEAN NOT NULL DEFAULT false,
    password_hash TEXT NOT NULL,
    name TEXT NOT NULL,
    role TEXT NOT NULL DEFAULT 'user',
    avatar_url TEXT,
    is_super_admin BOOLEAN NOT NULL DEFAULT false,
    settings JSONB DEFAULT '{}',
    last_login_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

Multi-Tenant Memberships

CREATE TABLE user_tenants (
    user_id UUID NOT NULL REFERENCES users(id),
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    role TEXT NOT NULL DEFAULT 'user',       -- 'owner', 'admin', 'user'
    is_default BOOLEAN NOT NULL DEFAULT false,
    joined_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (user_id, tenant_id)
);

• Globally unique email: One account per email, belonging to multiple tenants
• Per-tenant roles: owner (immutable), admin, user
• is_default: Which tenant loads on login
• Registration: Atomic transaction creates tenant + user + user_tenants in one operation

Email Verification

CREATE TABLE email_verification_tokens (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL REFERENCES users(id),
    token TEXT NOT NULL UNIQUE,            -- 32 random bytes, hex-encoded (64 chars)
    expires_at TIMESTAMPTZ NOT NULL,       -- 24h expiry
    used_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

• Token validated + not used + not expired -> mark used_at, set email_verified = true
• Post-registration: fire-and-forget async task (doesn't block registration response)
• Resend: invalidates existing tokens, creates new, sends new email

Team Invites

CREATE TABLE team_invites (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    email TEXT NOT NULL,
    role TEXT NOT NULL DEFAULT 'user',
    invited_by UUID NOT NULL REFERENCES users(id),
    token TEXT NOT NULL UNIQUE,            -- 32 random bytes, hex-encoded, 7-day expiry
    expires_at TIMESTAMPTZ NOT NULL,
    accepted_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Smart invite logic: If a user with the email already exists globally, they are added to the tenant directly. If not, an invite is created and an email is sent. Only owner/admin can invite, change roles, or remove members.

Email Templates

CREATE TABLE email_templates (
    id UUID PRIMARY KEY,
    tenant_id UUID REFERENCES tenants(id),  -- NULL = global default
    slug TEXT NOT NULL,                      -- 'verification', 'team_invite', etc.
    name TEXT NOT NULL,
    subject TEXT NOT NULL,                   -- Handlebars: 'Verify your email, {{user_name}}'
    html_body TEXT NOT NULL,                 -- Handlebars HTML template
    text_body TEXT,                          -- Handlebars plain text fallback
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE(tenant_id, slug)
);

• Template resolution: Tenant-specific wins over global (ORDER BY tenant_id IS NULL ASC LIMIT 1)
• Handlebars rendering: {{var}} syntax for subject, html_body, text_body
• Tenant overrides: Tenants can customize any global template by slug

Super Admin

The is_super_admin boolean on the users table is a platform-level privilege:

The first super admin is promoted via direct SQL. JWT carries sa: true claim when present.

Files

File metadata is stored in PostgreSQL. Actual blobs live in S3-compatible object storage (MinIO by default, tenant-configurable).

CREATE TABLE files (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    uploaded_by UUID NOT NULL REFERENCES users(id),
    storage_provider TEXT NOT NULL DEFAULT 'internal',  -- 'internal' (MinIO) or 's3'
    bucket TEXT NOT NULL DEFAULT 'qadra-files',
    object_key TEXT NOT NULL,
    filename TEXT NOT NULL,
    content_type TEXT NOT NULL,
    size_bytes BIGINT NOT NULL,
    purpose TEXT NOT NULL,                     -- 'avatar', 'artifact', 'logo', 'attachment'
    entity_type TEXT,                          -- Polymorphic: 'user', 'tenant', 'workload'
    entity_id UUID,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

• Gateway-proxied downloads: URLs are always /api/files/{id}/download, never direct S3 links
• Image processing: Avatar uploads auto-resized via imaginary (HTTP service) to 256x256 WebP
• Purpose-based validation: avatar/logo have 2MB limit + image-only types; others 10MB default
• Object key format: tenants/{tenant_id}/{purpose}/{entity_id}{ext}

Notification Logs

Every notification sent by the platform is automatically logged. Multi-channel from day one.

CREATE TABLE notification_logs (
    id UUID PRIMARY KEY,
    tenant_id UUID REFERENCES tenants(id),
    channel TEXT NOT NULL DEFAULT 'email',      -- 'email', future: 'sms', 'push'
    recipient TEXT NOT NULL,
    template_slug TEXT,
    subject TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'sent',        -- 'sent', 'failed'
    error TEXT,
    provider_id TEXT,                           -- SMTP2GO email_id, future: Twilio SID
    triggered_by TEXT,                          -- 'registration', 'verification', 'team_invite', etc.
    user_id UUID REFERENCES users(id),
    metadata JSONB NOT NULL DEFAULT '{}',
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Logging is transparent via the LoggingEmailService decorator pattern. All callers of EmailService::send() get automatic audit logging without code changes. Fire-and-forget via tokio::spawn to avoid blocking delivery.

Multitenancy Model

Each tenant is fully isolated with its own resources:

Example Tenant: Acme Capital

Row-level security enforced at the database layer. The user_tenants junction table determines which tenants a user can access.

Example: Investment Banking Pipeline

Putting it all together with a concrete example:

Pipeline: "Deal Pipeline" with 4 stages

Agents:

• Research Analyst: Gathers market data, competitive analysis (system_prompt focused on sourcing and synthesis)
• Due Diligence Specialist: Validates claims, checks risks (system_prompt focused on verification and risk assessment)
• Memo Writer: Synthesizes findings into investment memo (system_prompt focused on clear financial writing)
• Reviewer: Final quality check (system_prompt focused on completeness and accuracy)

Workload: "Acme Corp Acquisition Analysis"

Stage 1: Research
  -> Research Analyst executes with forwarded context: {}
  -> Output: { market_size: "$50B", competitors: [...], key_trends: [...] }
  -> Artifact: "Acme Corp Market Analysis" (markdown)
  -> stage_results["Research"] recorded

Stage 2: Due Diligence
  -> Due Diligence Specialist executes with forwarded context from Research
  -> Output: { risk_score: 0.3, red_flags: [], verified_claims: 12 }
  -> Artifact: "Acme Corp Risk Assessment" (markdown)
  -> stage_results["Due Diligence"] recorded

Stage 3: Investment Memo
  -> Memo Writer executes with forwarded context from Research + DD
  -> Output: { recommendation: "proceed", confidence: 0.85 }
  -> Artifact: "Acme Corp Investment Memo" (pdf, via artifact template + renderer)
  -> stage_results["Investment Memo"] recorded

Stage 4: Review
  -> Reviewer executes with forwarded context from all prior stages
  -> Output: { approved: true, comments: "Strong analysis, proceed to IC" }
  -> Artifact: "Review Sign-off" (markdown)
  -> Workload status -> completed

Each stage builds on the structured output of previous stages. The artifact template for the investment memo ensures consistent formatting across all deals.

Qadra Database ERD

42 tables across 8 domains. Each diagram renders independently.

Core: Tenants & Users

erDiagram
    tenants {
        UUID id PK
        TEXT name
        TEXT slug
        JSONB settings
    }
    users {
        UUID id PK
        UUID tenant_id FK
        TEXT email
        TEXT name
        TEXT role
        BOOLEAN is_super_admin
        BOOLEAN email_verified
    }
    user_tenants {
        UUID user_id PK
        UUID tenant_id PK
        TEXT role
        BOOLEAN is_default
    }
    api_keys {
        UUID id PK
        UUID tenant_id FK
        TEXT key_hash
        TEXT name
    }
    refresh_tokens {
        UUID id PK
        UUID user_id FK
        TEXT token_hash
    }
    email_verification_tokens {
        UUID id PK
        UUID user_id FK
        TEXT token
    }
    team_invites {
        UUID id PK
        UUID tenant_id FK
        UUID invited_by FK
        TEXT email
        TEXT token
    }
    files {
        UUID id PK
        UUID tenant_id FK
        UUID uploaded_by FK
        TEXT object_key
        TEXT purpose
    }
    email_templates {
        UUID id PK
        UUID tenant_id FK
        TEXT slug
        TEXT subject
    }
    notification_logs {
        UUID id PK
        UUID tenant_id FK
        TEXT channel
        TEXT recipient
        TEXT status
    }

    tenants ||--o{ users : has
    tenants ||--o{ user_tenants : members
    users ||--o{ user_tenants : belongs_to
    tenants ||--o{ api_keys : has
    users ||--o{ refresh_tokens : has
    users ||--o{ email_verification_tokens : has
    tenants ||--o{ team_invites : has
    users ||--o{ team_invites : invited_by
    tenants ||--o{ files : has
    users ||--o{ files : uploaded_by
    tenants ||--o{ email_templates : has
    tenants ||--o{ notification_logs : has

Agents, Pipelines & Workloads

erDiagram
    agents {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        TEXT display_name
        TEXT system_prompt
        TEXT control_type
        BOOLEAN is_active
    }
    pipelines {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        JSONB stages
        JSONB draft_stages
    }
    workloads {
        UUID id PK
        UUID tenant_id FK
        UUID pipeline_id FK
        TEXT title
        TEXT status
        TEXT current_stage
        JSONB stage_results
        JSONB pipeline_snapshot
    }
    tasks {
        UUID id PK
        UUID workload_id FK
        UUID agent_id FK
        TEXT stage
        TEXT status
        INTEGER progress
        JSONB output
    }
    artifacts {
        UUID id PK
        UUID workload_id FK
        UUID task_id FK
        UUID agent_id FK
        UUID template_id FK
        TEXT name
        TEXT artifact_type
    }
    usage_records {
        UUID id PK
        UUID tenant_id FK
        UUID workload_id FK
        UUID task_id FK
        UUID agent_id FK
        TEXT operation_type
        INTEGER tokens_in
        INTEGER tokens_out
    }
    artifact_templates {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        TEXT output_format
        UUID renderer_id FK
    }

    pipelines ||--o{ workloads : uses
    workloads ||--o{ tasks : has
    agents ||--o{ tasks : assigned_to
    workloads ||--o{ artifacts : produces
    tasks ||--o{ artifacts : produces
    agents ||--o{ artifacts : created_by
    artifact_templates ||--o{ artifacts : template_for
    workloads ||--o{ usage_records : tracks

Knowledge Graph (SPO)

erDiagram
    spo_nodes {
        UUID id PK
        UUID tenant_id FK
        TEXT value
        TEXT node_type
        VECTOR embedding
        JSONB content
        FLOAT confidence
    }
    spo_triples {
        UUID id PK
        UUID tenant_id FK
        UUID subject_id FK
        UUID predicate_id FK
        UUID object_id FK
        FLOAT confidence
        TEXT asserter
        TEXT assertion_type
    }
    entity_aliases {
        UUID id PK
        UUID tenant_id FK
        TEXT surface_form
        UUID canonical_node_id FK
        TEXT entity_type
    }
    documents {
        UUID id PK
        UUID tenant_id FK
        TEXT title
        TEXT content
        TEXT ingestion_status
        JSONB ingestion_result
        UUID proposed_by FK
        UUID approved_by FK
        TIMESTAMPTZ approved_at
        UUID rejected_by FK
    }
    document_triples {
        UUID document_id PK
        UUID triple_id PK
        FLOAT confidence
        TEXT excerpt
    }
    document_chunks {
        UUID document_id PK
        UUID chunk_id PK
        INTEGER chunk_index
        INTEGER start_offset
    }

    spo_nodes ||--o{ spo_triples : subject
    spo_nodes ||--o{ spo_triples : predicate
    spo_nodes ||--o{ spo_triples : object
    spo_nodes ||--o{ entity_aliases : canonical
    documents ||--o{ document_triples : extracted
    spo_triples ||--o{ document_triples : from
    documents ||--o{ document_chunks : chunked

Attribution (KG-SMILE)

erDiagram
    attribution_sessions {
        UUID id PK
        UUID tenant_id FK
        UUID workload_id FK
        TEXT query
        BOOLEAN competence_passed
        FLOAT spo_coverage
        BOOLEAN verification_passed
        FLOAT fidelity
        FLOAT faithfulness
        FLOAT stability
    }
    triple_attributions {
        UUID id PK
        UUID session_id FK
        UUID triple_id FK
        FLOAT attribution_score
        FLOAT weight
    }
    perturbation_results {
        UUID id PK
        UUID session_id FK
        FLOAT graph_similarity
        FLOAT response_similarity
    }
    spo_component_groups {
        UUID id PK
        UUID session_id FK
        TEXT group_name
    }
    conflict_events {
        UUID id PK
        UUID tenant_id FK
        TEXT new_subject
        TEXT new_predicate
        TEXT new_object
        UUID existing_triple_id FK
        TEXT severity
        TEXT reason
    }

    attribution_sessions ||--o{ triple_attributions : has
    attribution_sessions ||--o{ perturbation_results : has
    attribution_sessions ||--o{ spo_component_groups : has

Flows & Conversations

erDiagram
    agent_flows {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        JSONB flow_data
        INTEGER node_count
        UUID created_by FK
        BOOLEAN is_active
    }
    flow_executions {
        UUID id PK
        UUID tenant_id FK
        UUID flow_id FK
        JSONB flow_snapshot
        TEXT status
        JSONB state
        JSONB node_states
        UUID triggered_by FK
    }
    conversations {
        UUID id PK
        UUID tenant_id FK
        UUID user_id FK
        TEXT title
    }
    conversation_messages {
        UUID id PK
        UUID conversation_id FK
        UUID agent_id FK
        UUID flow_execution_id FK
        TEXT role
        TEXT content
    }

    agent_flows ||--o{ flow_executions : runs
    conversations ||--o{ conversation_messages : has
    flow_executions ||--o{ conversation_messages : produced_by

BALLS Graphs

erDiagram
    graphs {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        JSONB metadata
    }
    vertices {
        UUID id PK
        UUID graph_id FK
        FLOAT x
        FLOAT y
        FLOAT z
        TEXT label
        FLOAT curvature
    }
    edges {
        UUID id PK
        UUID graph_id FK
        UUID source_id FK
        UUID target_id FK
        FLOAT weight
        FLOAT curvature
    }

    graphs ||--o{ vertices : has
    graphs ||--o{ edges : has
    vertices ||--o{ edges : source
    vertices ||--o{ edges : target

Plugins & Renderers

erDiagram
    plugins {
        UUID id PK
        UUID tenant_id FK
        TEXT plugin_type
        TEXT name
        INTEGER version
        TEXT code
        BOOLEAN is_active
        BOOLEAN is_global
    }
    renderers {
        UUID id PK
        UUID tenant_id FK
        TEXT name
        INTEGER version
        TEXT output_format
        TEXT code
        BOOLEAN is_active
        BOOLEAN is_global
    }
    memory {
        UUID id PK
        UUID tenant_id FK
        TEXT key
        JSONB value
        VECTOR embedding
    }
    sessions {
        UUID id PK
        UUID tenant_id FK
        TEXT session_type
    }
    usage {
        UUID id PK
        UUID tenant_id FK
        UUID agent_id FK
        UUID session_id FK
        INTEGER tokens_in
        INTEGER tokens_out
    }

    sessions ||--o{ usage : during

Knowledge Graph

Overview

Qadra uses a dual-storage approach for knowledge retrieval, combining the precision of graph databases with the flexibility of semantic search.

The core insight: Different types of knowledge need different retrieval methods.

• Facts ("Apple was founded in 1976") --- Exact lookup via SPO triples
• Concepts ("What are the risks of this acquisition?") --- Semantic similarity via vectors

Why not just vectors? Vector search is powerful but imprecise. Ask "When was Apple founded?" and you might get chunks discussing Apple's founding, but not necessarily the exact year. SPO gives you (Apple, founded_in, 1976) directly.

Why not just SPO? Triple stores require structured data. For narrative content, exploratory queries, or fuzzy concepts, vectors find relevant context that wouldn't match any predefined pattern.

SPO Index

What is SPO?

SPO stands for Subject-Predicate-Object --- the fundamental unit of knowledge representation used in semantic web and knowledge graph systems.

Example Triple: (Apple, founded_by, Steve Jobs)

Why SPO over property graphs?

Property graphs (Neo4j, etc.) allow arbitrary properties on nodes and edges. SPO is more constrained --- everything is a triple. This constraint is a feature:

• Uniformity: All facts have the same shape, simplifying queries and storage
• Composability: Complex facts decompose into simple triples
• Interoperability: RDF/SPARQL compatibility if needed later

Example: Representing a complex fact

"Apple was founded by Steve Jobs in Cupertino in 1976"

Becomes multiple triples:

(Apple, founded_by, Steve Jobs)
(Apple, founded_in_location, Cupertino)
(Apple, founded_in_year, 1976)
(Steve Jobs, co_founded, Apple)

Each fact is independently queryable, updateable, and traceable to its source document.

Schema

-- Nodes store entities and relationships
CREATE TABLE spo_nodes (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    value TEXT NOT NULL,
    node_type TEXT NOT NULL,               -- 'subject', 'predicate', 'object'
    embedding vector(1536),                -- For semantic search
    content JSONB,                         -- Rich metadata
    source_document TEXT,                  -- Provenance
    confidence REAL DEFAULT 1.0,
    created_at TIMESTAMPTZ DEFAULT NOW(),

    UNIQUE(tenant_id, value, node_type)
);

-- Triples connect nodes
CREATE TABLE spo_triples (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL REFERENCES tenants(id),
    subject_id UUID NOT NULL REFERENCES spo_nodes(id),
    predicate_id UUID NOT NULL REFERENCES spo_nodes(id),
    object_id UUID NOT NULL REFERENCES spo_nodes(id),
    confidence REAL DEFAULT 1.0,
    source TEXT,                           -- "document:uuid", "user:uuid"
    asserter TEXT,                         -- Who made this claim
    assertion_type TEXT,                   -- 'fact', 'claim', 'opinion'
    created_at TIMESTAMPTZ DEFAULT NOW(),

    UNIQUE(tenant_id, subject_id, predicate_id, object_id)
);

-- All 6 SPO permutation indexes for O(1) lookups
CREATE INDEX idx_spo ON spo_triples(tenant_id, subject_id, predicate_id, object_id);
CREATE INDEX idx_sop ON spo_triples(tenant_id, subject_id, object_id, predicate_id);
CREATE INDEX idx_pso ON spo_triples(tenant_id, predicate_id, subject_id, object_id);
CREATE INDEX idx_pos ON spo_triples(tenant_id, predicate_id, object_id, subject_id);
CREATE INDEX idx_osp ON spo_triples(tenant_id, object_id, subject_id, predicate_id);
CREATE INDEX idx_ops ON spo_triples(tenant_id, object_id, predicate_id, subject_id);

-- HNSW index for embedding similarity on SPO nodes
CREATE INDEX idx_spo_nodes_embedding ON spo_nodes
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

Query Patterns

SPO supports six fundamental query patterns, all scoped by tenant_id:

Examples:

-- SP?: What is Apple's founding year?
-- (Apple, founded_in_year, ?) -> ["1976"]
SELECT o.value FROM spo_nodes s
JOIN spo_triples t ON t.subject_id = s.id
JOIN spo_nodes p ON t.predicate_id = p.id
JOIN spo_nodes o ON t.object_id = o.id
WHERE t.tenant_id = $1 AND s.value = 'Apple' AND p.value = 'founded_in_year';

-- ?PO: Who founded Apple?
-- (?, founded_by, Apple) -> ["Steve Jobs"]
SELECT s.value FROM spo_nodes s
JOIN spo_triples t ON t.subject_id = s.id
JOIN spo_nodes p ON t.predicate_id = p.id
JOIN spo_nodes o ON t.object_id = o.id
WHERE t.tenant_id = $1 AND p.value = 'founded_by' AND o.value = 'Apple';

-- S?O: What is the relationship between Apple and Steve Jobs?
-- (Apple, ?, Steve Jobs) -> ["founded_by", "co_founded_by"]
SELECT p.value FROM spo_nodes s
JOIN spo_triples t ON t.subject_id = s.id
JOIN spo_nodes p ON t.predicate_id = p.id
JOIN spo_nodes o ON t.object_id = o.id
WHERE t.tenant_id = $1 AND s.value = 'Apple' AND o.value = 'Steve Jobs';

Predicate Design

Good predicates are meaningful relationships:

Rules:

• Predicates should be verbs or verb phrases
• Use snake_case: is_located_in, not isLocatedIn
• Be specific: founded_by not associated_with
• Avoid meta-predicates that describe the data, not relationships

The extraction pipeline enforces this with a junk-predicate filter that rejects is_about, key_point, research_finding, relates_to, has_aspect, mentioned_in, discussed, and noted.

Vector Store

pgvector with HNSW

All vector storage uses PostgreSQL pgvector with HNSW (Hierarchical Navigable Small World) indexes for fast approximate nearest-neighbor search.

CREATE TABLE embeddings (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    chunk_id UUID NOT NULL,
    document_id UUID NOT NULL,
    embedding vector(1536) NOT NULL,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}',
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_embeddings_embedding ON embeddings
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

Two things get embedded:

1. Document chunks: Paragraph-boundary text chunks (~500 chars) for narrative retrieval
2. SPO node values: Subject/predicate/object values for semantic search on the knowledge graph itself

Semantic Search

#![allow(unused)]
fn main() {
let results = vector_index.search(
    tenant_id,
    query_embedding,
    k: 10,
    filter: None,
).await?;

for hit in results {
    println!("{}: {:.3}", hit.content, hit.score);
}
}

SPO navigates structure; pgvector retrieves similar content. Standard RAG searches everything; SPO-RAG searches where SPO points.

Document Ingestion Pipeline

Human-Gated Design

Ingestion into the Knowledge Core is explicitly human-gated. Agents can surface matching documents and propose them for ingestion, but a human must confirm before any data is written to the SPO graph or vector store.

Why: Prevents accidental data co-mingling across tenants or knowledge domains. We never want unintended data polluting the Knowledge Core.

How it works:

1. An agent discovers relevant content (via search, connector, or user upload)
2. The agent proposes the document for ingestion --- it appears in a review queue
3. A human reviews and approves (or rejects) the proposal
4. Only after explicit confirmation does the ingestion pipeline run

No automatic or background ingestion of discovered content. The UI presents proposed documents for review, and only after explicit confirmation does data enter the system.

Pipeline Overview

Once a document is approved, the pipeline transforms raw content into queryable knowledge:

Approved Document
    |
    v
1. Record document (status='processing')
    |
    v
2. Content Analyzer (LLM routing: Spo / Both / Discard)
    |
    +--- Discard ---> Mark discarded, stop
    |
    v
3a. Triple Extractor (LLM, epistemic metadata, windowed for large docs)
    |                --> batch-create SPO nodes + triples
    |                --> embed SPO node values into HNSW index
    |
3b. Text Chunker (if routing = Both)
    |                --> pure Rust, paragraph boundaries, ~500 chars
    |                --> batch embed chunks into pgvector
    |
    v
4. Provenance linking (document_triples, document_chunks)
    |
    v
5. Update document status='completed'

Key principle: SPO is always the index. Every non-discard document gets SPO triples extracted. There is no "vector-only" category. The Both routing adds vector chunks on top of SPO extraction for documents that also benefit from narrative retrieval.

Implementation: src/ingestion/ module, orchestrated by IngestionPipeline (implements the IngestionService trait).

Content Analyzer

The analyzer determines how to route content. It uses an LLM classifier with a heuristic fallback.

Routing categories:

There is no "vector-only" routing. If the LLM returns "vector", the analyzer upgrades it to "both" --- SPO extraction always runs on non-junk content.

LLM classification: The analyzer sends a 2000-character sample to a haiku-tier model with a routing prompt. The LLM returns JSON with routing, has_extractable_facts, estimated_chunks, confidence, and reasoning.

Heuristic fallback: When the LLM is unavailable or returns unparseable output, the analyzer falls back to length-based heuristics:

• Content < 500 chars --> Spo (SPO extraction only)
• Content >= 500 chars --> Both (SPO + vector chunks)

Implementation: src/ingestion/analyzer.rs

Triple Extraction

The extractor uses an LLM to pull Subject-Predicate-Object triples from text, including epistemic metadata on every triple.

Extraction prompt rules:

• Use meaningful predicates: founded_by, is_located_in, acquired, revenue_is
• Reject junk predicates: is_about, key_point, research_finding, relates_to
• Each triple must represent a concrete, verifiable relationship
• Include assertion type: fact, claim, or opinion
• Include the source excerpt for provenance

Windowed extraction for large documents: When content exceeds 2000 characters, the pipeline chunks it into extraction windows (~2000 chars with 100-char overlap) and runs extraction on each window independently. Results are deduplicated by (subject, predicate, object), keeping the highest-confidence version.

Post-extraction filtering:

• Triples below 0.5 confidence are discarded
• Triples with junk predicates are discarded
• Remaining triples are batch-created as SPO nodes and triples in PostgreSQL
• Subject node values are embedded into the HNSW index for semantic search on the KG itself

Example extraction:

Input: "Acme Corporation was founded in 1990 by John Smith.
        The company is headquartered in San Francisco."

Output:
[
  {
    "subject": "Acme Corporation",
    "predicate": "founded_in",
    "object": "1990",
    "confidence": 0.95,
    "assertion_type": "fact",
    "excerpt": "Acme Corporation was founded in 1990"
  },
  {
    "subject": "Acme Corporation",
    "predicate": "founded_by",
    "object": "John Smith",
    "confidence": 0.95,
    "assertion_type": "fact",
    "excerpt": "Acme Corporation was founded in 1990 by John Smith"
  },
  {
    "subject": "Acme Corporation",
    "predicate": "headquartered_in",
    "object": "San Francisco",
    "confidence": 0.90,
    "assertion_type": "fact",
    "excerpt": "The company is headquartered in San Francisco"
  }
]

Implementation: src/ingestion/extractor.rs

Text Chunking

Large content routed as Both gets split into overlapping chunks for vector embedding. The chunker is pure Rust --- no LLM calls.

Algorithm:

• Target chunk size: ~500 characters
• Overlap: ~50 characters (re-includes the tail of the previous chunk)
• Split on paragraph boundaries (double newlines)
• Each chunk tracks index, start_offset, and end_offset for provenance

The chunker accumulates paragraphs until the target size is exceeded, then flushes. Overlap is achieved by carrying the last paragraph(s) from the previous chunk into the next, up to the overlap budget.

Implementation: src/ingestion/chunker.rs

Embedding

The embedder handles two kinds of embedding in batch:

1. Document chunks --> embed_and_store_chunks: Batch-embeds chunk content, upserts EmbeddingRecord entries into the vector index with metadata (chunk_index, start_offset, end_offset)
2. SPO node values --> embed_spo_nodes: Batch-embeds subject node values, updates their embedding column in spo_nodes via VectorSearchRepository

Both operations use the EmbeddingProvider trait, which can be backed by OpenAI, local models, or any provider implementing the trait.

Implementation: src/ingestion/embedder.rs

Provenance Linking

All ingested content maintains full provenance --- every triple and every chunk links back to its source document.

-- Source documents
CREATE TABLE documents (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    title TEXT,
    content TEXT NOT NULL,
    content_type TEXT,
    source_url TEXT,
    source_type TEXT,                      -- 'upload', 'web', 'api', 'research'
    ingestion_status TEXT,                 -- 'pending', 'processing', 'completed', 'discarded'
    ingestion_result JSONB,                -- Routing decision, counts, analysis
    created_at TIMESTAMPTZ,
    processed_at TIMESTAMPTZ
);

-- Links documents to extracted triples
CREATE TABLE document_triples (
    document_id UUID REFERENCES documents(id),
    triple_id UUID REFERENCES spo_triples(id),
    confidence REAL,
    excerpt TEXT,                          -- Source text the triple was extracted from
    PRIMARY KEY (document_id, triple_id)
);

-- Links documents to vector chunks
CREATE TABLE document_chunks (
    document_id UUID REFERENCES documents(id),
    chunk_id UUID,
    chunk_index INTEGER,
    start_offset INTEGER,
    end_offset INTEGER,
    PRIMARY KEY (document_id, chunk_id)
);

This enables:

• Tracing any triple back to its source document and excerpt
• Tracing any vector chunk back to its document and character offsets
• Answering "where did this fact come from?" for audit and compliance

Chain-of-Custody Provenance

Provenance linking answers which document a triple came from. Chain-of-custody extends this to a complete, queryable audit trail: every SPO triple resolves to its originating document along with when the source was published and when Qadra retrieved it.

Migration 035 adds custody columns:

-- On spo_triples: where this triple came from and when it was captured
ALTER TABLE spo_triples ADD COLUMN source_uri TEXT;
ALTER TABLE spo_triples ADD COLUMN retrieval_timestamp TIMESTAMPTZ;

-- On documents: when the source was published vs. when we ingested it
ALTER TABLE documents ADD COLUMN publication_date TIMESTAMPTZ;
ALTER TABLE documents ADD COLUMN retrieval_timestamp TIMESTAMPTZ;

The TripleProvenance type resolves the full chain in a single indexed join --- triple to its document, carrying title, source URL, publication date, retrieval timestamp, and the source excerpt:

#![allow(unused)]
fn main() {
pub struct TripleProvenance {
    pub triple_id: Uuid,
    pub source_uri: Option<String>,
    pub retrieval_timestamp: Option<DateTime<Utc>>,
    pub document_title: Option<String>,
    pub document_source_url: Option<String>,
    pub publication_date: Option<DateTime<Utc>>,
    pub document_retrieval_timestamp: Option<DateTime<Utc>>,
    pub excerpt: Option<String>,
}
}

NATS subject: qadra.{tid}.epistemic.trace_triple --- resolves the custody chain for a single triple. The single-join design keeps resolution under a 100 ms budget.

This answers not just "where did this fact come from?" but "how old is the source, and when did we last see it?" --- essential for time-sensitive financial knowledge where a stale figure is worse than no figure.

Claim Trace

Where chain-of-custody traces a single triple, claim trace runs the inverse direction for investment-committee (IC) sign-off: given a free-text claim from a memo (e.g. "Acme's 2024 revenue was $391B"), trace it back to the supporting SPO triples --- with their asserter and source --- and the source documents, including the excerpt and full content.

Flow: two indexed queries plus a reverse document join.

1. SPO node search --- match the claim text against SPO node values (subject/object) to find candidate nodes
2. Triple lookup --- fetch triples involving those nodes, carrying asserter and assertion_type
3. Reverse document join --- walk document_triples back to documents for each supporting triple, returning excerpt + content

The whole trace is budgeted under 500 ms (a 450 ms internal timeout guards the budget; partial results are returned rather than failing the IC review).

The response groups supporting triples (with asserter + source) under the claim and links each to its source documents (with excerpt + content), so a reviewer can verify a memo statement against the underlying knowledge in one round trip.

Epistemic Metadata

Not All Knowledge Is Equal

A fact from a regulatory filing differs from an opinion in a blog post. Qadra tracks who made a claim and its epistemic status on every SPO triple.

Assertion Types

The default assertion type is Claim (conservative). If the extractor is uncertain, it defaults to claim rather than fact.

Schema

Two columns on spo_triples:

ALTER TABLE spo_triples ADD COLUMN asserter TEXT;
ALTER TABLE spo_triples ADD COLUMN assertion_type TEXT
    CHECK (assertion_type IN ('fact', 'claim', 'opinion'));

Extraction

The triple extraction prompt explicitly asks the LLM to classify each triple's assertion type. The ExtractedTriple struct carries this through the pipeline:

#![allow(unused)]
fn main() {
pub struct ExtractedTriple {
    pub subject: String,
    pub predicate: String,
    pub object: String,
    pub confidence: f32,
    pub assertion_type: Option<AssertionType>,  // fact, claim, or opinion
    pub excerpt: Option<String>,
}
}

Query Filtering

Epistemic metadata enables filtering by trust level:

#![allow(unused)]
fn main() {
// Get only verified facts
let facts = repository
    .query_triples(tenant_id)
    .where_assertion_type(AssertionType::Fact)
    .fetch_all()
    .await?;

// Get claims from a specific analyst
let analyst_claims = repository
    .query_triples(tenant_id)
    .where_asserter("Goldman Sachs")
    .where_assertion_type(AssertionType::Claim)
    .fetch_all()
    .await?;
}

Claim Classification

Beyond the three-way assertion type (fact, claim, opinion), the ClaimClassifier in qadra-epistemic-core provides fine-grained classification of assertions into six epistemic categories.

ClaimType enum:

How classification works:

The classifier uses pre-compiled regex indicator patterns (word-boundary matching, inflection-aware) across four categories: projection, opinion, assessment, and fact. For each input text:

1. Score the text against each indicator set (count of matches / total indicators in set)
2. Apply priority ordering: projection > opinion > assessment > fact
3. The first category exceeding a 0.05 score threshold wins
4. Confidence is computed as 0.6 + (score * 0.4), capped at 1.0
5. Multiple fact indicators (score > 0.15) promote StatedFact to VerifiedFact
6. If no indicators match, default to StatedFact at 0.5 confidence

Response-level classification:

classify_response splits a full response into sentences (handling abbreviations like "Dr.", "U.S.", "e.g.") and classifies each sentence independently. This enables per-claim epistemic tracking within a single agent output.

Integration with groundedness validation:

The ClaimType maps to AssertionType for storage via to_assertion_type(). This bridges the fine-grained classification (useful for analysis and debugging) with the storage-level three-way system. During groundedness validation, claim classification helps distinguish which parts of a response need strict fact-checking (verified facts) versus which are inherently subjective (opinions, projections).

Customization:

The classifier supports custom indicator lists via builder methods (with_projection_indicators, with_opinion_indicators) for domain-specific vocabulary. For example, an investment banking pipeline might add financial projection indicators like "DCF implies" or "base case".

Implementation: crates/qadra-epistemic-core/src/claims.rs

Confidence Weighting

Epistemic status feeds into synthesis context weighting:

In synthesis prompts, facts and claims are separated:

VERIFIED FACTS:
(Apple, founded_in_year, 1976) [asserter: SEC filings]

ANALYST CLAIMS (verify independently):
(NVIDIA, will_reach, $5T_market_cap) [asserter: Morgan Stanley]

KG-SMILE Attribution

The Two Gates Model

Every agent response passes through two gates that determine whether the system has the competence to answer and whether the answer stayed grounded in known facts.

Query --> Extract Entities --> Competence Gate --> Synthesis --> Verification Gate --> Response
                                    |                                |
                             "Should I attempt?"            "Did I stay grounded?"

Based on the KG-SMILE paper --- explainable GraphRAG attribution.

Gate 1: Competence (Pre-Synthesis)

Question: "Does my knowledge domain cover this query?"

Before the agent synthesizes a response, the competence gate checks whether the SPO graph has meaningful coverage of the query's entities.

#![allow(unused)]
fn main() {
CompetenceResult {
    passed: bool,
    entities: Vec<String>,      // Extracted from query
    coverage: f32,              // Fraction with SPO triples (0-1)
    connectivity: f32,          // How connected in SPO graph (0-1)
    relevant_triples: Vec<Uuid>,
}
}

Thresholds:

• min_coverage: 0.3 --- At least 30% of query entities must have triples
• min_connectivity: 0.2 --- At least 20% of entity pairs must be connected

Failure modes:

• "Insufficient SPO coverage" --- Query about an unknown domain
• "Insufficient connectivity" --- Entities exist but aren't related in the graph

If the competence gate fails, the system can decline to answer, trigger research (see 5.9), or return a response with a low-confidence warning.

Gate 2: Verification (Post-Synthesis)

Question: "Does my output's semantic structure exist in my knowledge domain?"

After synthesis, the verification gate checks whether the response's implied facts are grounded in the SPO graph. Qadra ships three verifier implementations with different cost/precision trade-offs. Two of them — both LLM-free — run on every chat response in production; the perturbation-based attribution path is implemented but not yet wired into the live pipeline.

Both production gates write their result into the assistant message's metadata (metadata["verification"] and metadata["belnap"]) and always surface ungrounded or contradicted claims rather than silently dropping them. Both are a passing no-op when the query produced no context triples — this prevents false negatives when the knowledge graph has nothing relevant to say.

Implementation: crates/qadra-epistemic-core/src/attribution/verification.rs (rule-based + perturbation) and crates/qadra-epistemic-core/src/belnap.rs (Belnap).

Rule-Based Grounding Gate (production)

#![allow(unused)]
fn main() {
VerificationGateResult {
    passed: bool,
    fidelity: f32,
    faithfulness: f32,                   // fraction of implied triples that are grounded
    attributions: Vec<AttributedEdge>,
    ungrounded_claims: Vec<UngroundedClaim>,
    reason: Option<String>,
}
}

Process:

1. Extract implied triples from the response — split into sentences; detect a predicate keyword (e.g. "reported", "produces", "owns"); subject = noun phrase before the keyword, object = noun phrase after.
2. Classify each triple — compute word overlap with every context triple; SUPPORTED if any overlap exists, otherwise UNGROUNDED.
3. Faithfulness = supported / total implied (1.0 if none were extracted).
4. Gate decision — pass iff faithfulness >= min_faithfulness (default 0.5).
5. Always surface ungrounded_claims.

fidelity and stability are reported as rule-based proxies (fidelity mirrors faithfulness; stability is a determinism flag) — they become true R²/Jaccard only on the perturbation path below.

Belnap Four-Valued Gate (production)

A second, complementary gate from the qadra-belnap crate. It evaluates each claim over Belnap four-valued logic, where the graph can simultaneously support and contradict a claim, and "no evidence" is distinct from "false":

Pipeline (verify_with_belnap):

1. build_belnap_graph(triples) — each SPO triple becomes a polarity-tagged edge. Edge polarity is derived from the predicate string via predicate_polarity (a negation/contrast-marker heuristic) — v1. The research-faithful source is classify_polarity over the asserting excerpt computed at SPO write time; that is a tracked follow-up (see crates/qadra-belnap/MODEL.md).
2. Parse the response into claims (subject entity, predicate polarity) using the UD parser.
3. Ground each claim against the graph and reconcile the claim's polarity against the four-valued result.
4. passed is false iff the graph contradicts a claim. (Ungrounded/contested claims are reported but do not fail the gate.)

The UD parser comes from shared_parser(), a process-wide OnceLock. Under the onnx-ort Cargo feature it loads the real ONNX goeswith parser (KoichiYasuoka/roberta-base-english-ud-goeswith, run via ort) from QADRA_BELNAP_MODEL / QADRA_BELNAP_TOKENIZER / QADRA_BELNAP_LABELS. Without the feature it returns a StubParser whose empty parse makes the gate a passing no-op — so the default build is unchanged, and the Belnap gate only does real work when the feature is on and the model assets are present. BelnapVerification is serialized into metadata["belnap"] only when the response yields at least one claim.

Perturbation Attribution (future, higher-precision)

The verify_response path measures grounding by perturbation — removing subsets of triples, re-synthesizing, and measuring how much the response changes — then trains a weighted surrogate model to produce per-triple attribution scores (fidelity = R², faithfulness = correlation with actual impact, stability = Jaccard robustness). It is the highest-precision verifier but costs one LLM re-synthesis per perturbation, so it is not yet invoked by the live chat pipeline. Its distinguishing optimization — curvature-guided perturbation, which targets structurally important edges instead of 20 blind random removals — is described next.

Curvature-Guided Perturbation

The KG-SMILE paper recommends 20 random perturbations to measure how triple removal affects response quality. Random selection wastes LLM calls on redundant edges that change nothing. Qadra replaces random selection with Forman-Ricci curvature --- a discrete curvature measure computed directly on the knowledge graph --- to target perturbations at structurally important edges.

Implementation: crates/qadra-epistemic-core/src/curvature/

What Is Forman-Ricci Curvature

Forman-Ricci curvature is a discrete analogue of Ricci curvature from differential geometry, adapted for graphs. For an edge $e = (u, v)$, the formula is:

$$\kappa_F(e) = 4 - \deg(u) - \deg(v) + 3 \cdot |\triangle(u, v)|$$

Where:

• $\deg(u)$ is the degree of node $u$ (number of neighbors)
• $\deg(v)$ is the degree of node $v$
• $|\triangle(u, v)|$ is the number of triangles containing edge $e$ (i.e., common neighbors of $u$ and $v$)

Intuition: High-degree nodes pull curvature negative (the edge is a bottleneck funneling many paths). Triangles push curvature positive (the edge has parallel alternatives through shared neighbors).

Complexity: $O(|E| \cdot d_{\max})$ where $d_{\max}$ is the maximum node degree --- linear in graph size for bounded-degree graphs.

Worked examples:

Edge Classification

Curvature values partition edges into three structural roles:

The threshold $-0.1$ is configurable via CurvatureConfig::bridge_threshold.

graph LR
    subgraph "Cluster (kappa = 3)"
        A((A)) --- B((B))
        B --- C((C))
        C --- A
    end

    subgraph "Bridge (kappa = -3)"
        D((D)) --- H((Hub))
        E((E)) --- H
        F((F)) --- H
        G((G)) --- H
    end

    C -. "bridge edge<br/>kappa < -0.1" .-> D

    style H fill:#f96,stroke:#333

Structure analysis: Before choosing a sampling method, the system computes the percentage of bridge edges. If bridges make up at least 15% of all edges (min_bridge_percentage), curvature-guided sampling is recommended. Otherwise, the graph is too uniform and random sampling is used as a fallback.

Weight Computation

Raw curvature values are transformed into sampling probabilities through a shifted-inversion + normalization pipeline:

Step 1: Shifted inversion --- Flip the sign so negative curvature (bridges) maps to high weight:

$$w(e) = \max!\bigl(\varepsilon,; -\kappa_F(e) + \kappa_{\max} + 1\bigr)$$

Where $\kappa_{\max} = \max_e \kappa_F(e)$ and $\varepsilon$ is the weight floor (default $0.01$) ensuring every edge retains a nonzero sampling chance.

Step 2: Normalize to probabilities:

$$P(e) = \frac{w(e)}{\sum_{e'} w(e')}$$

Effect on sampling distribution:

When all edges have identical curvature (e.g., a uniform hub-spoke), all weights are equal and sampling degrades gracefully to uniform random.

Weighted sampling: Edges are drawn without replacement using the computed probabilities. The sampler iteratively selects an edge proportional to its probability, removes it from the pool, renormalizes, and repeats. This ensures each perturbation set contains unique edges.

How It Feeds Into Attribution

The curvature pipeline replaces random perturbation selection in the KG-SMILE verification gate:

flowchart TD
    A[SPO Triples for Query] --> B[Compute Forman-Ricci Curvature]
    B --> C{Bridge % >= 15%?}
    C -->|Yes| D[Curvature-Guided Sampling]
    C -->|No| E[Random Sampling Fallback]
    D --> F[Select Perturbation Sets]
    E --> F
    F --> G[Run Perturbations<br/>Remove selected edges, re-synthesize]
    G --> H[Measure Response Similarity]
    H --> I[Train Surrogate Model<br/>Weighted Linear Regression]
    I --> J[Attribution Scores per Triple]

Why this matters:

• Fewer perturbations needed: Bridge edges produce the strongest attribution signal. By sampling them preferentially, each perturbation is more informative. Where random sampling might need 20 perturbations to achieve stable attributions, curvature-guided sampling can achieve comparable fidelity with fewer runs.
• Each perturbation is more informative: Removing a bridge edge disconnects knowledge clusters, producing a measurable drop in response quality. Removing a cluster edge changes almost nothing --- it wastes an LLM call that produces near-zero signal.
• Cost reduction: Each perturbation requires an LLM re-synthesis call. Targeting structurally important edges reduces the total number of LLM calls needed for reliable attribution.
• Context edge selection: Beyond perturbation sampling, the curvature pipeline also selects which edges to include in the LLM context window. Bridge edges (sorted by curvature ascending, most negative first) are prioritized. If there are at least min_bridges_for_selection bridge edges, only bridges are included. Otherwise, edges below the mean curvature are used as a fallback. The result is capped at max_context_edges.

Configuration

All curvature behavior is controlled by CurvatureConfig:

Full pipeline entry point: compute_curvature(triples, config) runs the complete pipeline --- Forman-Ricci computation, structure analysis, weight transformation, and context edge selection --- returning a CurvatureResult with edges sorted by curvature ascending (bridges first).

Attribution Scoring

Triple attribution via weighted linear regression:

S_i = B_0 + B_1*W_1*P_1 + ... + B_k*W_k*P_k + e

Where:

• S_i = Response similarity for perturbation i
• P_k = Binary indicator (triple k present?)
• W_k = Kernel weight (Gaussian: exp(-(C_i)^2 / sigma^2))
• B_k = Attribution score (higher = more influential)

Configuration

Schema

-- Sessions track both gates
CREATE TABLE attribution_sessions (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    query TEXT NOT NULL,
    competence_passed BOOLEAN,
    spo_coverage REAL,
    spo_connectivity REAL,
    verification_passed BOOLEAN,
    fidelity REAL,
    faithfulness REAL,
    stability REAL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Triple-level attribution scores
CREATE TABLE triple_attributions (
    id UUID PRIMARY KEY,
    session_id UUID REFERENCES attribution_sessions(id),
    triple_id UUID REFERENCES spo_triples(id),
    attribution_score REAL,
    weight REAL,
    component_group TEXT
);

-- Raw perturbation data
CREATE TABLE perturbation_results (
    id UUID PRIMARY KEY,
    session_id UUID REFERENCES attribution_sessions(id),
    removed_triple_ids UUID[],
    graph_similarity REAL,
    response_similarity REAL
);

Attribution Session Persistence

Every gate run is durable, not just in-memory. Each call into handle_query persists a row to attribution_sessions and --- when triples were ranked --- batch-inserts the per-triple curvature/weight into triple_attributions.

The write happens via fire-and-forget (tokio::spawn), so query latency is unaffected and a failed insert never propagates to the caller.

Persisted on every gate run:

• query, tenant_id, agent name
• spo_coverage and spo_connectivity (the curvature analysis bridge percentage)
• competence_passed and competence timing
• per-triple attributions (weight = curvature)

Verification-side columns (fidelity, faithfulness, stability, verification_passed) are left NULL until the verification gate is invoked from a higher-level pipeline.

Querying persisted sessions:

The list endpoint is paginated and filterable by workload_id, agent_name, date range (from/to), and the gate-pass booleans (competence_passed, verification_passed). Getting a single session returns it together with its triple_attributions.

Metrics: each gate run emits a tracing event with the target kg_smile.gate_run, carrying gate, result, coverage, and connectivity. These flow through the standard OTLP --> OTel Collector --> Loki pipeline and feed a KG-SMILE Grafana dashboard that charts pass/fail rates and average coverage/connectivity over time --- no new exporter or metrics endpoint required.

Groundedness Validation

Purpose

Detect hallucinations by checking if claims in an agent's response can be traced back to the provided context. Groundedness validation complements the Two Gates model as a third layer of quality assurance.

Validation Flow

Agent Response
    |
    v
Extract Claims (LLM)
    "NVIDIA has market cap of $3T"
    "Revenue grew 200% YoY"
    |
    v
For each claim:
    --> Find supporting triple or chunk
    --> Check entailment (SUPPORTED / CONTRADICTED / NOT_MENTIONED)
    --> Score confidence
    |
    v
Aggregate: { grounded: bool, score: 0-1, ungrounded_claims: [] }

Scoring

#![allow(unused)]
fn main() {
pub struct ValidationResult {
    pub is_grounded: bool,
    pub score: f32,              // 0-1, fraction of claims grounded
    pub total_claims: usize,
    pub grounded_claims: usize,
    pub ungrounded_claims: Vec<String>,
}
}

Integration

Called after synthesis, before returning a response:

• If grounded: return as-is
• If partially grounded: attach warning listing ungrounded claims
• If poorly grounded: trigger gap detection and autonomous research (see 5.9), or regenerate with stricter prompt

Relationship to KG-SMILE

The three quality layers form a chain:

1. Competence Gate (pre-synthesis): Do we have knowledge about this topic?
2. Verification Gate (post-synthesis): Did we stay grounded in that knowledge?
3. Groundedness Validation (post-synthesis): Are specific claims individually supported?

Claim Validation Node

Groundedness validation runs inside a single agent's synthesis. Claim validation lifts the same idea into a flow graph: a ClaimValidation node grounds every claim produced by prior-stage agents against the SPO triples in the Knowledge Core.

This is the cross-stage check for pipelines where one agent's output feeds another --- for example, validating Scarlet's research claims and Ayana's analysis against known facts before a downstream stage builds on them.

What it does:

1. Read the prior-stage agent outputs from the flow state (the input_keys)
2. Extract individual claims from each output
3. Ground each claim against SPO triples in the Knowledge Core
4. Label each claim SUPPORTED, CONTRADICTED, or NOT_MENTIONED
5. Write a per-claim result list plus a validation summary back to state under output_key

Configuration:

The node surfaces contradictions and unsupported claims explicitly rather than silently passing them downstream, giving the pipeline a hard checkpoint before unverified claims propagate into an investment memo.

Entity Aliases

The Problem

The same entity can appear under different names: "Apple", "Apple Inc.", "Apple Inc", "AAPL". Without alias resolution, SPO queries miss related triples.

Approach

Entity aliases are stored as SPO triples themselves:

(Apple Inc., alias_of, Apple)
(AAPL, alias_of, Apple)
(Apple Computer, alias_of, Apple)

During query processing, the system resolves aliases before SPO lookup:

1. Query entities are checked against alias_of predicates
2. If an alias is found, the canonical entity is used for the SPO query
3. Results from all aliases are merged

This uses the existing SPO infrastructure --- no separate alias table needed.

Gap Detection and Autonomous Research

When to Research

After initial synthesis, the system analyzes the response for signs of insufficient knowledge. Research is triggered (not on every query) when any of these conditions are detected:

1. Uncertainty phrases in the response:

"i don't know", "i'm not sure", "cannot find",
"don't have enough information", "not in the context",
"unable to answer", "no relevant", "insufficient"

2. Weak attribution signal: Average absolute attribution score < 0.001 across all attributed triples. Indicates no triples strongly influenced the response.

3. Missing bridges: Bridge edge ratio < 10% when the graph has 5+ edges. The graph lacks structural connectors between knowledge regions.

Research Flow

Gap Detected
    |
    v
Exa Neural Search (3 results)
    Returns: URLs + full text content
    |
    v
For each source:
    --> Chunk text (500 chars, 50 overlap)
    --> Embed each chunk into pgvector
    --> Extract triples via LLM (with epistemic metadata)
    --> Store triples in SPO index
    --> Link provenance (document_triples, document_chunks)
    |
    v
Re-synthesize with enriched context

Re-Synthesis

After research, context is rebuilt from BOTH the original and new knowledge:

EXISTING KNOWLEDGE:
(Apple, founded_in_year, 1976) [fact]
(Apple, headquartered_in, Cupertino) [fact]

NEW RESEARCH:
(Apple, revenue_2024, $391B) [claim, asserter: Reuters]
(Apple, ceo, Tim Cook) [fact, asserter: SEC filing]

SOURCE EXCERPTS:
[chunk text from newly ingested documents]

Constraints

• Only trigger research when gaps are detected (not every query)
• Limit to 3 Exa results per research cycle to control cost
• All new triples carry epistemic metadata (asserter, assertion_type)
• All new content links to source documents for provenance
• Research triggers are logged for debugging and cost monitoring
• Research results go through the same ingestion pipeline (human-gated for background research, automatic for in-flight gap filling during agent execution)

Conflict Detection

The Problem

When new triples contradict existing ones, the system must detect and handle the conflict rather than silently storing both.

Example:

Existing: (Tesla, founded_by, Elon_Musk)      confidence=0.95  fact
New:      (Tesla, founded_by, Martin_Eberhard)  confidence=0.80  claim
--> Contradiction detected: different founders asserted for the same subject+predicate

How Contradictions Are Detected

Two triples contradict when they share the same subject and predicate but assert different objects. The detect_contradictions function in qadra-epistemic-core implements this:

1. Normalize and group: All triples are grouped by normalized (subject, predicate) pairs. Normalization lowercases and collapses whitespace, so "Tesla" and "tesla" group together.
2. Check for distinct objects: Within each group, object values are normalized and compared. A contradiction exists only when 2+ distinct objects appear for the same subject+predicate.
3. Same object, different confidence: Two triples asserting (Tesla, founded_by, Elon Musk) at different confidence levels are NOT a contradiction --- they reinforce each other.
4. Parallel analysis: Groups are analyzed in parallel using rayon, since each (subject, predicate) group is independent.

Resolution Scoring

Each competing assertion receives a resolution score that determines the winner. The score combines four factors:

resolution_score = confidence * assertion_type_weight * source_bonus * trust_factor

Assertion type weights:

Source bonus: When multiple independent sources assert the same object value, the representative gets a source bonus of 1.0 + (source_count - 1) * 0.1. Two sources asserting the same object = 1.1x multiplier. This rewards corroboration.

Source trust scores: An optional source_trust map (source name to 0.0-1.0 score) can downweight unreliable sources. For example, {"sketchy-blog.com": 0.3, "reuters.com": 1.0} causes a high-confidence fact from a sketchy blog to lose to a lower-confidence claim from Reuters:

sketchy-blog.com: 0.9 * 1.0 (fact) * 0.3 (trust) = 0.27
reuters.com:      0.8 * 0.6 (claim) * 1.0 (trust) = 0.48  --> winner

Unknown sources default to trust factor 1.0 (full trust until proven otherwise).

Winner selection: Assertions are sorted by resolution score descending. The highest-scoring assertion wins. The Contradiction struct records the winner's triple ID and all competing assertions with their scores.

Contradiction Result

#![allow(unused)]
fn main() {
pub struct ContradictionResult {
    pub has_contradictions: bool,
    pub contradictions: Vec<Contradiction>,   // Each with subject, predicate, assertions, winner
    pub triples_analyzed: usize,
    pub conflict_count: usize,                // Number of unique S+P pairs with conflicts
}

pub struct Contradiction {
    pub subject: String,
    pub predicate: String,
    pub assertions: Vec<CompetingAssertion>,  // Sorted by resolution_score descending
    pub winner: Option<Uuid>,                 // Triple ID of the highest-scoring assertion
    pub detection_confidence: f32,            // 0.9 when scores are computable, 0.5 otherwise
}
}

Multi-Way Contradictions

The system handles contradictions with more than two competing assertions. For example, three different CEO assertions for the same company:

(Company_X, ceo, Alice)   [fact,    confidence=0.9]  --> score: 0.9 * 1.0 = 0.9  (winner)
(Company_X, ceo, Bob)     [claim,   confidence=0.7]  --> score: 0.7 * 0.6 = 0.42
(Company_X, ceo, Charlie) [opinion, confidence=0.5]  --> score: 0.5 * 0.3 = 0.15

All three assertions are recorded in the Contradiction with their resolution scores. Alice wins.

Integration with Ingestion Pipeline

During triple extraction, before storing a new triple, the pipeline checks for existing triples with the same subject and predicate but different object. When found:

1. Same assertion type: Flag for human review --- which is current?
2. Higher-confidence new triple: Replace old triple, archive the original
3. Lower-confidence new triple: Store as alternative, mark as contested
4. Temporal context available: Both may be correct at different points in time --- store both with temporal metadata

Contested Triples

When a conflict cannot be automatically resolved, both triples are stored and marked as contested. Synthesis prompts include both with appropriate hedging:

CONTESTED:
(Acme Corp, ceo, John Smith) [fact, asserter: 10-K filing, 2024-01]
(Acme Corp, ceo, Jane Doe) [claim, asserter: Bloomberg, 2024-06]
Note: CEO may have changed. Verify with latest source.

Implementation: crates/qadra-epistemic-core/src/contradictions.rs

Usage Examples

Ingesting a Document

POST /v1/documents
Content-Type: application/json

{
  "title": "Acme Corp Overview",
  "content": "Acme Corporation was founded in 1990 by John Smith. The company
              is headquartered in San Francisco and specializes in enterprise
              software. Acme acquired DataCo in 2015 for $50 million...",
  "source_type": "web",
  "source_url": "https://example.com/acme"
}

Response:

{
  "document_id": "uuid",
  "routing": "both",
  "triples_extracted": 4,
  "chunks_created": 2,
  "analysis": {
    "routing": "both",
    "content_type": "text/plain",
    "has_extractable_facts": true,
    "estimated_chunks": 2,
    "confidence": 0.87,
    "reasoning": "Mixed content with facts and narrative"
  }
}

Querying the Knowledge Graph

#![allow(unused)]
fn main() {
// Find all companies founded by a person
let companies = repository.query_spo_sp(
    tenant_id,
    "John Smith",
    "founded"
).await?;

// Find who founded a specific company
let founders = repository.query_spo_po(
    tenant_id,
    "founded_by",
    "Acme Corporation"
).await?;

// Semantic search for relevant context
let context = vector_index.search(
    tenant_id,
    embed("Acme acquisition history"),
    k: 5
).await?;
}

Combined SPO + Vector Query

#![allow(unused)]
fn main() {
// 1. SPO gives us structured facts about the entity
let facts = repository.query_spo_s(tenant_id, "Acme Corporation").await?;

// 2. Vector search gives us narrative context
let narrative = vector_index.search(
    tenant_id,
    embed("Acme Corporation competitive position"),
    k: 5
).await?;

// 3. Combine for synthesis
let context = format!(
    "STRUCTURED FACTS:\n{}\n\nNARRATIVE CONTEXT:\n{}",
    format_triples(&facts),
    format_chunks(&narrative),
);
}

Financial Data Extraction

A specialised sub-agent that queries the SPO index for financial metrics on a named entity. Instead of free-form synthesis, it pulls structured financial facts directly from the knowledge graph and returns them with confidence and provenance.

Recognised predicates:

revenue, ebitda, market_cap, founded_in_year,
headcount, funding_round, valuation

How it works: given an entity (e.g. "Acme Corporation"), the sub-agent looks up SPO triples whose subject matches the entity and whose predicate is one of the financial predicates above. Each match becomes a metric carrying its object value, confidence, and asserter.

#![allow(unused)]
fn main() {
pub struct FinancialMetric {
    pub metric: String,        // predicate name, e.g. "revenue"
    pub value: String,         // object value, e.g. "$391B"
    pub confidence: f32,       // from the source triple
    pub asserter: Option<String>,
    pub low_confidence: bool,  // true when confidence < 0.6
}
}

Metrics with confidence below 0.6 are flagged low_confidence so a reviewer can treat them with caution rather than excluding them outright.

NATS subject: qadra.{tid}.epistemic.extract_financial_metrics

Request { entity } --> response { entity, metrics: [...], retrieved_at }. Because the extraction is pure SPO lookup over a fixed predicate set, it is fast and deterministic --- no LLM call is needed to surface the figures.

Semantic Verification Cache

When verifying LLM responses against the knowledge graph, we can avoid redundant LLM calls by caching verification results for semantically similar queries.

The Problem

Every LLM response verification requires:

1. LLM call to generate/parse the response (~500ms-2s, ~$0.01-0.10)
2. SPO lookup to ground facts against triples (~5ms)

For similar queries ("When was Apple founded?" vs "What year did Apple start?"), we repeat the expensive LLM call even though the SPO verification would be identical.

The Solution: Perturbation Cache

Cache verification results keyed by semantic similarity, not exact string match.

Key insight: The SPO lookup is fast (~5ms). Even on cache hit, we re-validate against current facts. We only skip the expensive LLM call.

Cache Schema

Redis Stack with RediSearch provides vector similarity search:

FT.CREATE perturbation_cache ON HASH PREFIX 1 pcache:
  SCHEMA
    tenant_id TAG
    embedding VECTOR HNSW 6
      TYPE FLOAT32
      DIM 1536
      DISTANCE_METRIC COSINE
    query_hash TAG
    created_at NUMERIC SORTABLE
    ttl NUMERIC

Similarity Threshold

Start at 0.93, monitor cache hit rate and verification accuracy.

Performance

Connectors & Content Ingestion

Overview

Connectors are how external content enters Qadra. A connector mirrors documents from a source system (Confluence, Google Drive, SharePoint, Slack, S3, or the gateway), feeds them through a processing pipeline, and loads the result into the vector index and SPO knowledge graph.

The design follows a Moveworks-inspired three-stage pipeline:

MIRROR (Connectors)  -->  PROCESS (Chunk + Enrich + Embed)  -->  LOAD (VectorIndex)

Everything is trait-based and swappable. Each connector implements the ContentConnector trait; the platform composes connectors, a DocumentProcessor, an EmbeddingProvider, and a VectorIndex behind abstract interfaces, so the ingestion core knows nothing about any specific source system.

The human-gated approval flow for documents (propose -> review -> approve -> ingest) and the ingestion internals (content analyzer, triple extraction, chunking, provenance) are covered in the Knowledge Graph chapter. This page covers the connectors, sync, and entitlements that sit in front of that pipeline.

The MIRROR -> PROCESS -> LOAD Pipeline

The MIRROR stage produces a SyncDelta (added / modified / deleted document IDs). Only changed documents flow downstream, so an incremental sync never re-processes the entire corpus.

The ContentConnector Model

A connector is a streaming source of documents. The ContentConnector trait exposes two core operations:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait ContentConnector: Send + Sync {
    /// List documents, optionally only those changed since a timestamp (delta sync).
    async fn list_documents(&self, since: Option<DateTime<Utc>>) -> Result<Vec<DocumentState>>;

    /// Fetch full document content (streaming, zero-copy where possible).
    async fn fetch_documents(&self, ids: &[String]) -> Result<DocumentStream>;
}
}

Connectors are registered in the ServiceContainer via a ConnectorRegistry (HashMap<String, Arc<dyn ContentConnector>>), keyed by connector type. The supporting DocumentProcessor (chunk / enrich / embed) and VectorIndex (semantic search) traits complete the pipeline:

Available Connectors

The gateway connector is always available (it backs user uploads). The rest are opt-in per tenant and gated by entitlements (see below).

Sync Strategy

Each connector can be driven by three sync modes:

• Incremental sync is the common path: ask the source for everything changed since the last successful sync, build a SyncDelta, and process only the added/modified documents.
• Full sync runs daily to catch deletions and repair drift. Because it lists the entire corpus, it can detect documents that disappeared from the source (which incremental sync cannot reliably see) and rebuild the sync_state table from scratch.
• Webhook sync handles real-time updates: a single document is processed immediately on notification, without waiting for the next incremental window.

Sync history is recorded in the sync_runs table; per-document state lives in sync_state (primary key tenant_id + connector + source_id).

Change Detection

To decide whether a document actually changed, connectors compare source metadata in a strict priority order --- cheapest and most reliable first:

1. ETag comparison --- fastest, HTTP-native. If the source returns an ETag and it matches the stored one, the document is unchanged. No content fetch needed.
2. Version comparison --- for systems with explicit version numbers (Confluence, SharePoint). A higher version means the document changed.
3. Timestamp comparison --- the fallback. Compare last_modified against the stored synced_at.

Each document's tracked state is captured in DocumentState:

#![allow(unused)]
fn main() {
pub struct DocumentState {
    pub source_id: String,
    pub content_hash: String,        // BLAKE3 of content
    pub last_modified: DateTime<Utc>,
    pub etag: Option<String>,
    pub version: Option<u64>,
    pub synced_at: DateTime<Utc>,
}
}

A BLAKE3 content_hash is the final tiebreaker: even if ETag/version/timestamp suggest a change, an identical hash means the processed content is unchanged and downstream re-embedding can be skipped.

Zero-Copy Processing

Ingestion is engineered to avoid unnecessary allocations as content flows from source to vector store:

• DocumentContent is an enum of Mmap, Bytes, or Borrowed --- documents can be memory-mapped or borrowed with zero allocation on load.
• Chunk borrows from the document content (&'doc str) --- no string copies during chunking.
• EnrichedChunk uses Cow<'doc, str> --- it only allocates if a chunk is actually mutated during enrichment.

Supporting crates: memmap2 (memory-mapped files), bytes (zero-copy byte buffers), bumpalo (arena allocation), blake3 (fast content hashing).

Per-Tenant Connector Configuration

Connector configuration is per-tenant, stored in the tenant_connectors table and managed through the TenantConnectorStore trait:

Each configuration holds:

• Credentials --- one of OAuth2, ApiKey, Basic, or Aws, encrypted at rest.
• Sync config --- cron schedule, incremental interval, webhook settings, rate limits.
• Source filter --- connector-specific scoping: Confluence spaces, Google Drive folders, Slack channels, S3 prefixes.

Because credentials and source filters live per-tenant, two tenants can connect to the same source system (e.g. two Confluence instances) with completely isolated configuration and sync state.

Tenant Entitlements

Connectors and LLM providers are gated by service plans. Every tenant has an entitlement record defining what it can use and how much.

Service Plans

Three tiers --- free, pro, enterprise --- each defines:

• Limits: max_connectors, max_documents, api_requests, llm_tokens, storage, agents, pipelines, concurrent_workloads.
• Available services: which connectors and LLM providers are enabled.

Enforcement

Before any connector or service operation, the entitlement is checked:

#![allow(unused)]
fn main() {
// Before adding a connector / running a service operation:
let ent = entitlements.get(tenant_id).await?;
ent.can_add_connector(connector_type)?;  // checks service availability + limits
}

can_add_connector validates both that the connector type is available on the tenant's plan and that the tenant is under its max_connectors limit.

Override Hierarchy

Entitlements resolve in priority order, most specific first:

explicit service_overrides  >  limit_overrides  >  plan defaults

A tenant on the pro plan can be granted a single extra connector via a limit_override, or have a specific connector enabled via a service_override, without changing its plan.

Entitlement Schema

The TenantEntitlementStore trait exposes get, set_plan, set_limit_override, enable_service, and disable_service. Admin (Qadra employee) endpoints manage entitlements via GET/PUT/PATCH /admin/tenants/{id}/entitlements|plan|limits|services|usage|audit.

Feature-Flag Compilation

Beyond runtime entitlement checks, connectors are also gated at compile time via Cargo feature flags. A connector that is not compiled in cannot be enabled regardless of plan.

[features]
default     = ["postgres", "openrouter", "redis", "mongo", "pgvector", "connector-gateway"]
minimal     = ["sqlite", "ollama", "embed-local", "connector-gateway"]   # zero cloud deps
enterprise  = ["postgres", "openrouter", "pgvector", "redis", "mongo",
               "embed-openai", "connector-confluence", "connector-sharepoint",
               "connector-s3", "connector-gateway"]
full        = ["..."]  # everything

Two gates work together:

1. Build-time --- is_service_compiled(service) checks the active feature flags. The minimal build ships with only the gateway connector and zero cloud dependencies.
2. Runtime --- entitlements.get(tenant_id) checks the tenant's plan and overrides.

A connector must pass both gates to run: it must be compiled into the binary and enabled on the tenant's plan.

Agent Evaluation

This section covers the Agent Output Quality Evaluation framework (Story 94)---a deterministic, fixture-driven way to score what each agent produces, used as a CI release gate.

Purpose

LLM-based agents are non-deterministic, but the shape of their output is not. Scarlet emits SPO triples, Ayana filters items, Eliza grounds claims, Reagan writes review sections. The evaluation framework scores these structured outputs against fixed rubrics so we can detect regressions before they ship.

The framework is:

• Deterministic -- no LLM calls during scoring. Same fixtures in, same scores out.
• LLM-free -- rubrics inspect emitted structure (triples, labels, claims, sections), not prose quality.
• Fixture-driven -- every agent ships golden (should-pass) and negative (should-fail) cases.
• A release gate -- CI fails the release build if any agent's golden average drops below its threshold.

Rubrics

Each agent has one primary rubric. A rubric inspects the agent's emitted output and returns a score in [0.0, 1.0].

Reagan's required sections:

• Recommendation
• Conviction
• Key Positives
• Key Risks
• Reviewer Notes

Scoring

A rubric scores a single case. An agent's score aggregates those case scores.

• Per-agent score = average over GOLDEN cases only (should_pass = true).
• Negative cases (should_pass = false) must score below the threshold---this confirms the rubric actually discriminates.
• Coverage -- at least 10 cases per agent (8 golden + 4 negative is the working baseline).
• Release gate -- release_gate_check fails if any agent's golden average falls below its threshold.

agent_score(agent) = mean(score(case) for case in golden_cases(agent))

release_gate_check:
    for agent in agents:
        assert agent_score(agent) >= threshold(agent)
    # negative cases independently asserted < threshold

Metrics

After an evaluation run, emit_agent_eval_metrics emits a structured tracing event:

• Target: qadra.eval
• Event: agent_eval_complete
• Fields: eval_agent, eval_score, eval_rubric

These flow through the standard pipeline---OTLP -> OTel Collector -> Loki -> the Eval Quality Grafana dashboard (deploy/otel/dashboards/eval-quality.json)---so per-agent scores can be charted against their thresholds over time. No new exporter or /metrics endpoint is required.

Code Locations

Constraints

• Rubrics must stay aligned to the live SPO persona output formats. If a persona changes the structure it emits (triple shape, section headings, label schema), the corresponding rubric and fixtures must be updated in lockstep---otherwise the gate either passes blindly or false-fails.
• Scoring stays deterministic and LLM-free: a rubric may only inspect emitted structure, never call a model.
• Every agent keeps both golden and negative fixtures so the gate verifies discrimination, not just a high average.

AI Fluency & Response Calibration

Overview

Qadra adapts how it responds based on each user's AI fluency --- their skill at interacting with an AI system. The system observes interaction patterns, infers a fluency level, and calibrates every response accordingly: terse bullets for experts, thorough walkthroughs for novices.

The core principle: Fluency is about HOW users interact, not WHAT they know.

• A novice investor asking about DCF is still a novice AI user.
• An expert prompter asking about stocks is still an expert AI user.

Domain expertise and jargon are not fluency indicators. A user who fluently discusses discounted cash flows but accepts the first AI answer without refining it is interacting like a novice. Fluency lives in the interaction loop --- iteration, format specification, chaining --- never in the subject matter.

Fluency Detection

The system scores each query for signals that reveal interaction skill, weights them, and feeds them into a rolling window. Never use jargon or topic complexity as a signal.

Expert Indicators

High-weight signals that the user is steering the AI deliberately:

Novice Indicators

Lower-weight signals that the user is accepting output passively:

Pattern Detection

Signals are detected with compiled regular expressions over the raw query text. Format specification, context provision, and query chaining each have their own pattern set:

#![allow(unused)]
fn main() {
lazy_static! {
    static ref FORMAT_SPEC_PATTERNS: Vec<Regex> = vec![
        Regex::new(r"(?i)\b(in json|as json|json format)\b").unwrap(),
        Regex::new(r"(?i)\b(as a (list|table)|bullet points)\b").unwrap(),
        Regex::new(r"(?i)\b(step.?by.?step|break.?(it )?(down|into))\b").unwrap(),
    ];
}

lazy_static! {
    static ref CONTEXT_PROVISION_PATTERNS: Vec<Regex> = vec![
        Regex::new(r"(?i)\b(for context|some background)\b").unwrap(),
        Regex::new(r"(?i)\b(i'm (working on|trying to|building))\b").unwrap(),
        Regex::new(r"(?i)\b(the goal is|my goal is)\b").unwrap(),
    ];
}

lazy_static! {
    static ref CHAINING_PATTERNS: Vec<Regex> = vec![
        Regex::new(r"(?i)\b(now|next|then).*\b(that|this|the above)\b").unwrap(),
        Regex::new(r"(?i)\b(take that|use that|apply that)\b").unwrap(),
    ];
}
}

Iteration and chaining are also detected relationally --- comparing the current query against the previous one --- not just by single-query keywords.

Fluency Levels

Every assessment resolves to one of four levels:

#![allow(unused)]
fn main() {
pub enum FluencyLevel {
    Novice,
    Intermediate,
    Advanced,
    Expert,
}
}

Level Determination

An expert score is computed as a weighted blend of the expert signals, then normalized and compared against thresholds:

#![allow(unused)]
fn main() {
// Expert: iteration + format specs + examples + chaining
let expert_score =
    counts.iterates_prompt * 0.35 +
    counts.specifies_format * 0.25 +
    counts.uses_examples * 0.2 +
    counts.chains_queries * 0.2;

// Thresholds
if norm_expert > 0.4 { FluencyLevel::Expert }
else if norm_expert > 0.25 || norm_intermediate > 0.5 { FluencyLevel::Advanced }
else if norm_novice > 0.4 { FluencyLevel::Novice }
else { FluencyLevel::Intermediate }
}

Rolling Window & Confidence

Fluency is not judged from a single query. The assessment maintains a rolling window of recent signals:

• Tracks the last 50 signals (configurable window size).
• Uses an exponential moving average for complexity, so recent behavior weighs more.
• Confidence increases with signal count, reaching its maximum at ~15 signals --- early in a session, confidence is low and the system stays cautious.

#![allow(unused)]
fn main() {
pub struct FluencyAssessment {
    signals: VecDeque<FluencySignal>,
    window_size: usize,
    profile: FluencyProfile,
}

impl FluencyAssessment {
    pub fn process_query(
        &mut self,
        query: &str,
        previous_query: Option<&str>,
    ) -> &FluencyProfile {
        let signals = self.analyze_query(query);
        let is_iteration = self.detect_iteration(query, previous_query);
        let is_chaining = self.detect_chaining(query);

        // Add signals to rolling window
        // Update profile
        // Return updated profile
    }

    pub fn needs_more_guidance(&self) -> bool {
        matches!(self.profile.level, FluencyLevel::Novice)
            || (matches!(self.profile.level, FluencyLevel::Intermediate)
                && self.profile.confidence < 0.5)
    }

    pub fn prefers_brevity(&self) -> bool {
        matches!(self.profile.level, FluencyLevel::Expert)
            || (matches!(self.profile.level, FluencyLevel::Advanced)
                && self.profile.confidence > 0.6)
    }
}
}

Both level and confidence are always checked together --- a high-fluency guess with low confidence should not produce terse expert-style output.

Response Calibration

Once a fluency level is established, the response is tuned across several dimensions:

#![allow(unused)]
fn main() {
pub struct ResponseCalibration {
    pub verbosity: Verbosity,
    pub use_headers: bool,
    pub use_bullets: bool,
    pub technical_level: TechnicalLevel,
    pub anticipate_followups: bool,
    pub should_simplify: bool,
    pub should_elaborate: bool,
}

pub enum Verbosity {
    Minimal,
    Concise,
    Detailed,
    Comprehensive,
}

pub enum TechnicalLevel {
    Layman,
    Intermediate,
    Technical,
    Expert,
}
}

Calibration by Level

Expert users get dense, concise delivery --- no headers (they know what they're looking for), no hand-holding, no anticipated follow-ups (they'll ask if needed). Novice users get comprehensive, layman-level explanations with headers, bullets, simplification, and suggested next steps.

Confidence Blending

When confidence is low, the calibration blends toward Intermediate rather than committing to an uncertain extreme:

#![allow(unused)]
fn main() {
impl ResponseCalibrator {
    pub fn calibrate(
        &self,
        fluency_level: FluencyLevel,
        confidence: f32,
    ) -> ResponseCalibration {
        // Low confidence = blend toward intermediate
        if confidence < 0.5 {
            self.blend_calibrations(
                &self.get_calibration(fluency_level),
                &self.get_calibration(FluencyLevel::Intermediate),
                confidence,
            )
        } else {
            self.get_calibration(fluency_level)
        }
    }
}
}

Calibration → Synthesis Prompt

Calibration is not cosmetic post-processing --- it is injected directly into the synthesis prompt so the model generates the right shape of answer from the start:

#![allow(unused)]
fn main() {
fn build_prompt(
    query: &str,
    context: &str,
    calibration: &ResponseCalibration,
) -> String {
    let mut instructions = String::new();

    match calibration.verbosity {
        Verbosity::Minimal => instructions.push_str("Be extremely brief. "),
        Verbosity::Comprehensive => {
            instructions.push_str("Provide thorough explanation with examples. ")
        }
        _ => {}
    }

    if calibration.should_simplify {
        instructions.push_str("Use simple language, avoid jargon. ");
    }

    if calibration.anticipate_followups {
        instructions.push_str("Suggest what the user might want to explore next. ");
    }

    format!("{}\n\nContext:\n{}\n\nQuestion: {}", instructions, context, query)
}
}

Real-Time Adaptation

Because the rolling window updates on every interaction, fluency --- and therefore calibration --- shifts within a single session. As the user demonstrates more sophisticated interaction patterns, the responses tighten accordingly:

Query 1: "What is NVIDIA worth?"
-> Fluency: intermediate (simple question)
-> Response: Detailed with explanation

Query 2: "Break it down as a table"
-> Fluency: advanced (format spec detected!)
-> Response: Concise table format

Query 3: "Now compare to AMD, bullet points only"
-> Fluency: expert (chaining + constraints)
-> Response: Minimal bullet comparison

Note that none of these shifts depend on the topic (NVIDIA, AMD valuation) --- they are driven entirely by the interaction patterns: a bare question, then a format specification, then chaining plus an explicit constraint.

Constraints

• Never use jargon or keywords as fluency indicators --- domain expertise is not AI fluency.
• Update the profile on every interaction.
• Confidence requires sufficient signal history; blend toward Intermediate when confidence is low.
• Always check both level and confidence.
• Respect explicit format requests --- they override calibration defaults.
• Don't over-explain to experts; don't under-explain to novices.

Flows

Overview

A flow is a visual, graph-based definition of agent work. Where a conversation is an ad-hoc chat and a pipeline is a fixed sequence of stages, a flow is an explicit directed graph --- nodes are steps, edges are control flow --- authored in a ReactFlow canvas and executed by an event-driven engine in the Rust Core.

Flows are how Qadra expresses branching, conditions, human-in-the-loop gates, loops, sub-flows, and knowledge operations as a single composable artifact. An agent can trigger a flow from a conversation (via LLM tool-use), or a flow can be executed directly through the API.

The core idea: agents are programs that sometimes call models. A flow makes the program explicit --- you can see the control flow, the branch points, and where a human has to sign off.

The Flow Data Model

A flow is stored in two pieces: the editable definition and an immutable snapshot taken at execution time.

flow_data --- the editable definition

agent_flows.flow_data holds the ReactFlow graph verbatim as JSONB:

{
  "nodes": [ { "id": "start", "type": "Start", "data": {} }, ... ],
  "edges": [ { "id": "e1", "source": "start", "target": "agent-1" }, ... ],
  "viewport": { "x": 0, "y": 0, "zoom": 1 }
}

ReactFlow serializes to a single object, so JSONB stores it with no impedance mismatch. The executor always loads the full graph --- it never queries individual nodes --- so there is no benefit to normalizing nodes and edges into separate tables.

CREATE TABLE agent_flows (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id   UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    name        TEXT NOT NULL,
    description TEXT,
    flow_data   JSONB NOT NULL DEFAULT '{"nodes":[],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}',
    node_count  INTEGER NOT NULL DEFAULT 0,
    is_active   BOOLEAN NOT NULL DEFAULT true,
    created_by  UUID REFERENCES users(id) ON DELETE SET NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE(tenant_id, name)
);

flow_snapshot --- the immutable run copy

When an execution starts, the current flow_data is copied into flow_executions.flow_snapshot. The execution runs entirely against this snapshot, so editing the flow definition mid-run --- or after a run completes --- never changes what an in-flight or historical execution did. This is the same immutability pattern as workloads.pipeline_snapshot.

CREATE TYPE flow_execution_status AS ENUM
    ('pending','running','paused','completed','failed','cancelled');

CREATE TABLE flow_executions (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id       UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    flow_id         UUID NOT NULL REFERENCES agent_flows(id) ON DELETE CASCADE,
    flow_snapshot   JSONB NOT NULL,
    status          flow_execution_status NOT NULL DEFAULT 'pending',
    state           JSONB NOT NULL DEFAULT '{}',
    node_states     JSONB NOT NULL DEFAULT '{}',
    execution_queue JSONB NOT NULL DEFAULT '[]',
    current_node_id TEXT,
    error           TEXT,
    triggered_by    UUID REFERENCES users(id) ON DELETE SET NULL,
    input           JSONB NOT NULL DEFAULT '{}',
    started_at      TIMESTAMPTZ,
    completed_at    TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Execution State Machine

Every execution moves through a fixed set of states:

                  +-----------+
   create  -----> |  pending  |
                  +-----+-----+
                        |
                        v
                  +-----------+        HumanInput node
                  |  running  | <----------------------+
                  +-----+-----+                        |
                        |                              |
        +---------------+---------------+              |
        |               |               |        user submits
        v               v               v          input
  +-----------+   +-----------+   +-----------+        |
  | completed |   |  failed   |   |  paused   | -------+
  +-----------+   +-----------+   +-----------+
                                        |
                        cancel ---------+--> +-----------+
                                             | cancelled |
                                             +-----------+

The Blackboard State

The state column is a blackboard --- a shared JSONB dictionary that every node can read from and write to. There is no per-node message passing; instead, nodes communicate through named keys in this dictionary.

• A node reads its inputs from state keys (e.g. an Extract node reads state[input_key]).
• A node writes its outputs back to state under a configured output_key.
• Downstream nodes read those keys as their own inputs.

Most nodes use {{variable}} templating in their config to interpolate state values --- for example a KnowledgeLookup node's query can be "financials for {{company}}", where company was written to state by an earlier Extract node.

The execution input (an initial JSONB object) seeds the blackboard at run start --- this is how a conversation passes user_message into a flow it triggers.

The Executor Loop

The engine lives in src/flow_executor.rs. It is an event-driven queue processor, not a recursive tree-walker.

1. Start --- Parse the flow snapshot graph, find the Start node, seed the execution queue.
2. Loop --- Pop a node from the queue, check that all its predecessors are complete, execute it, store its output to state, enqueue its successors, and persist the full execution row.
3. Pause --- A HumanInput node sets the execution to paused and stops the loop, waiting for user input.
4. Resume --- When the user submits input, it is merged into state, the node's successors are re-enqueued, and the loop continues.
5. End --- Reaching an End node sets the execution status to completed.

State is persisted after each step (the executor reads/writes the full node_states and state as a batch, the same pattern as stage_results on workloads), so an execution is always recoverable from its database row.

Safety Rails

Because flows can branch and loop, the executor enforces hard limits to prevent runaway or recursive executions:

Hitting a rail transitions the execution to failed with an explanatory error rather than spinning indefinitely.

HumanInput: Pause and Resume

A HumanInput node is the human-in-the-loop gate. When the executor reaches one:

• The execution status becomes paused.
• The node status becomes waiting_for_input.
• The executor loop stops and persists state.

The execution stays paused --- indefinitely, durably --- until a user submits input via POST /flows/executions/:execId/input (NATS flow.do.human_input). The submitted input is merged into the blackboard state, the node's successors are re-enqueued, and the loop resumes from where it stopped.

The resume path is guarded against a time-of-check/time-of-use race: the resume SQL only updates when the row is still paused, so two concurrent input submissions cannot both resume the same execution.

Node Types

A flow supports 17 node types. Each has its own data config and execution behavior.

Notes on specific nodes

• Extract is the bridge between unstructured conversation and structured flow inputs. It reads state[input_key], prompts the LLM to return JSON for the requested output_variables, and merges the result into state. If the LLM returns non-JSON, it gracefully sets each variable to null rather than failing the run.
• KnowledgeLookup is the read side of the Knowledge Core. source chooses SPO triples, vector chunks, or both; results land in state under output_key.
• SpoFilter / SpoWrite are the SPO read/write pair --- a flow can pull facts into state, run agents over them, then write new claims back to the graph with provenance (asserter + source).
• ClaimValidation is the cross-stage groundedness checkpoint: it labels each prior-stage claim SUPPORTED, CONTRADICTED, or NOT_MENTIONED before unverified claims propagate downstream (e.g. into an investment memo). See the Knowledge Graph page for the grounding mechanics.
• Research currently runs as a placeholder pending full Exa integration; the node and config exist so flows authored against it remain valid.

SSE Streaming of Execution Progress

Long-running executions stream progress to the frontend over Server-Sent Events, so a user watching a flow run sees nodes start and complete in real time rather than waiting for one final response.

Because a NATS request-reply only allows a single response, progress cannot be streamed over the reply itself. Instead, a dedicated NATS progress subject is used:

React (SSE consumer) <--SSE-- Gateway <--NATS pub/sub-- Rust Core (FlowExecutor)

1. The Gateway creates a unique NATS subject _PROGRESS.{uuid} and subscribes to it.
2. It sends the execution request with that subject in the payload.
3. The Rust Core's executor publishes FlowProgressEvent { event, data, seq } messages to the subject as it runs --- fire-and-forget, with a monotonically increasing sequence counter for ordering.
4. The Gateway forwards each as an SSE event (event: {type}\ndata: {json}\n\n) and closes the stream on done or client disconnect.

Emitted event types include flow_started, node_started, node_completed, node_failed, flow_completed, flow_failed, and done. The same machinery powers streaming flow execution from within a conversation.

NATS Operations and REST Endpoints

Flow subjects are tenant-scoped under qadra.{tid}.flow.do.>, handled by the Rust Core.

These are reached over HTTP through the Gateway at /flows and /flows/executions/... (including POST /flows/:flowId/execute, POST /flows/executions/:execId/input, and the SSE-friendly execution endpoints). See the API Reference for the full request/response shapes --- the flow routes there map one-to-one onto the NATS operations above.

Conversations

Overview

Conversations are ad-hoc chat threads between a user and one or more agents, outside of flow executions and pipeline workloads.

Qadra is a workload orchestration platform, not a chat app --- but conversations exist as a supplementary, informal interaction mode. Where a workload moves through pipeline stages and a flow executes a structured graph, a conversation is a free-form thread. There is no pipeline, no stage progression, no immutable snapshot.

A conversation can still reach into the formal machinery --- an agent can autonomously trigger a flow from inside a conversation (see Flow Tool-Use) --- but the conversation itself stays informal.

Data Model

Two tables (migration 029): conversations (the thread) and conversation_messages (the turns).

CREATE TABLE conversations (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id   UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    user_id     UUID NOT NULL REFERENCES users(id) ON DELETE SET NULL,
    title       TEXT,
    is_archived BOOLEAN NOT NULL DEFAULT false,
    deleted_at  TIMESTAMPTZ,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE conversation_messages (
    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id     UUID NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    agent_id            UUID REFERENCES agents(id) ON DELETE SET NULL,
    role                TEXT NOT NULL CHECK (role IN ('user', 'assistant', 'system')),
    content             TEXT NOT NULL,
    model               TEXT,
    token_count         INTEGER,
    latency_ms          INTEGER,
    finish_reason       TEXT,
    rating              SMALLINT CHECK (rating IN (-1, 1)),
    flow_execution_id   UUID REFERENCES flow_executions(id) ON DELETE SET NULL,
    metadata            JSONB NOT NULL DEFAULT '{}',
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Key design choices:

• Soft delete: Conversations carry a deleted_at timestamp. All queries filter WHERE deleted_at IS NULL --- threads are never hard-deleted.
• Analytics columns: model, token_count, latency_ms, and finish_reason live inline on each message (not in a side table) so training-data export, cost tracking, and latency monitoring are direct SQL queries with no joins.
• Rating: Each assistant message can be rated thumbs up (+1) or thumbs down (-1).
• Flow linkage: flow_execution_id is a real FK linking a message to a flow execution the agent triggered.
• Auto-title: The first exchange auto-generates a conversation title from the user's message (truncated to ~60 characters).