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.