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