> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognee.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# cognify()
> Transform raw data into a structured knowledge graph
# cognee.cognify()
```python theme={null}
async def cognify(
datasets: Union[str, list[str], list[UUID]] = None,
user: User = None,
graph_model: BaseModel = KnowledgeGraph,
chunker = TextChunker,
chunk_size: int = None,
chunks_per_batch: int = None,
config: Config = None,
vector_db_config: dict = None,
graph_db_config: dict = None,
run_in_background: bool = False,
incremental_loading: bool = True,
custom_prompt: Optional[str] = None,
temporal_cognify: bool = False,
data_per_batch: int = 20,
llm_config: Optional[LLMConfig] = None,
embedding_config: Optional[EmbeddingConfig] = None,
dry_run: bool = False,
)
```
## Description
Transform ingested data into a structured knowledge graph.
This is the core processing step in Cognee that converts raw text and documents
into an intelligent knowledge graph. It analyzes content, extracts entities and
relationships, and creates semantic connections for enhanced search and reasoning.
Prerequisites:
* **LLM\_API\_KEY**: Must be configured (required for entity extraction and graph generation)
* **Data Added**: Must have data previously added via `cognee.add()`
* **Vector Database**: Must be accessible for embeddings storage
* **Graph Database**: Must be accessible for relationship storage
Input Requirements:
* **Datasets**: Must contain data previously added via `cognee.add()`
* **Content Types**: Works with any text-extractable content including:
* Natural language documents
* Structured data (CSV, JSON)
* Code repositories
* Academic papers and technical documentation
* Mixed multimedia content (with text extraction)
Processing Pipeline:
1. **Document Classification**: Identifies document types and structures
2. **Text Chunking**: Breaks content into semantically meaningful segments
3. **Entity Extraction**: Identifies key concepts, people, places, organizations
4. **Relationship Detection**: Discovers connections between entities
5. **Graph Construction**: Builds semantic knowledge graph with embeddings
6. **Content Summarization**: Creates hierarchical summaries for navigation
Graph Model Customization:
The `graph_model` parameter allows custom knowledge structures:
* **Default**: General-purpose KnowledgeGraph for any domain
* **Custom Models**: Domain-specific schemas (e.g., scientific papers, code analysis)
* **Ontology Integration**: Pass an ontology resolver via `config` (or set the `ONTOLOGY_FILE_PATH` environment variable) for predefined vocabularies
Args:
datasets: Dataset name(s) or dataset uuid to process. Processes all available data if None.
* Single dataset: "my\_dataset"
* Multiple datasets: \["docs", "research", "reports"]
* None: Process all datasets for the user
user: User context for authentication and data access. Uses default if None.
graph\_model: Pydantic model defining the knowledge graph structure.
Defaults to KnowledgeGraph for general-purpose processing.
chunker: Text chunking strategy (TextChunker, LangchainChunker).
* TextChunker: Paragraph-based chunking (default, most reliable)
* LangchainChunker: Recursive character splitting with overlap
Determines how documents are segmented for processing.
chunk\_size: Maximum tokens per chunk. Auto-calculated based on LLM if None.
Formula: min(embedding\_max\_completion\_tokens, llm\_max\_completion\_tokens // 2)
Default limits: \~512-8192 tokens depending on models.
Smaller chunks = more granular but potentially fragmented knowledge.
chunks\_per\_batch: Number of chunks to be processed in a single batch in Cognify tasks.
vector\_db\_config: Custom vector database configuration for embeddings storage.
graph\_db\_config: Custom graph database configuration for relationship storage.
run\_in\_background: If True, starts processing asynchronously and returns immediately.
If False, waits for completion before returning.
Background mode recommended for large datasets (>100MB).
Use pipeline\_run\_id from return value to monitor progress.
custom\_prompt: Optional custom prompt string to use for entity extraction and graph generation.
If provided, this prompt will be used instead of the default prompts for
knowledge graph extraction. The prompt should guide the LLM on how to
extract entities and relationships from the text content.
dry\_run: If True, return a stage-level estimate of LLM token usage and rough cost
without making LLM calls or writing graph results. The estimate covers all
data in the selected dataset(s); an incremental run may process fewer items.
Returns:
Union\[dict, list\[PipelineRunInfo], DryRunEstimate]:
* **Blocking mode**: Dictionary mapping dataset\_id -> PipelineRunInfo with:
* Processing status (completed/failed/in\_progress)
* Extracted entity and relationship counts
* Processing duration and resource usage
* Error details if any failures occurred
* **Background mode**: List of PipelineRunInfo objects for tracking progress
* Use pipeline\_run\_id to monitor status
* Check completion via pipeline monitoring APIs
Next Steps:
After successful cognify processing, use search functions to query the knowledge:
```python theme={null}
import cognee
from cognee import SearchType
# Process your data into knowledge graph
await cognee.cognify()
# Query for insights using different search types:
# 1. Natural language completion with graph context
insights = await cognee.search(
"What are the main themes?",
query_type=SearchType.GRAPH_COMPLETION
)
# 2. Get entity relationships and connections
relationships = await cognee.search(
"connections between concepts",
query_type=SearchType.GRAPH_COMPLETION
)
# 3. Find relevant document chunks
chunks = await cognee.search(
"specific topic",
query_type=SearchType.CHUNKS
)
```
Advanced Usage:
```python theme={null}
# Custom domain model for scientific papers
class ScientificPaper(DataPoint):
title: str
authors: List[str]
methodology: str
findings: List[str]
await cognee.cognify(
datasets=["research_papers"],
graph_model=ScientificPaper,
)
# Ground extraction in an ontology (there is no `ontology_file_path` argument;
# pass a resolver through `config` instead).
from cognee.modules.ontology.rdf_xml.RDFLibOntologyResolver import RDFLibOntologyResolver
from cognee.modules.ontology.ontology_config import Config
config: Config = {
"ontology_config": {
"ontology_resolver": RDFLibOntologyResolver(ontology_file="scientific_ontology.owl")
}
}
await cognee.cognify(datasets=["research_papers"], config=config)
# Background processing for large datasets
run_info = await cognee.cognify(
datasets=["large_corpus"],
run_in_background=True
)
# Check status later with run_info.pipeline_run_id
```
Environment Variables:
Required:
* LLM\_API\_KEY: API key for your LLM provider
Optional (same as add function):
* LLM\_PROVIDER, LLM\_MODEL, VECTOR\_DB\_PROVIDER, GRAPH\_DATABASE\_PROVIDER
* AUTO\_RATE\_LIMIT: Turn the rate limiter on automatically when the provider shows overload evidence (default: True)
* LLM\_RATE\_LIMIT\_ENABLED: Enable rate limiting from the first request (default: False)
* LLM\_RATE\_LIMIT\_REQUESTS: Max requests per interval (default: 60; 10 for local inference servers)
Optional (contradiction detection — see [Contradiction detection](#contradiction-detection)):
* CONTRADICTION\_DETECTION: Append the opt-in contradiction check to the pipeline (default: False)
* CONTRADICTION\_CONFIDENCE\_THRESHOLD: Minimum LLM confidence for a pair to be flagged (default: 0.5)
* CONTRADICTION\_MAX\_FACTS: Cap on the facts sent to the LLM in a single check (default: 500)
Optional (provenance ledger — see [Provenance ledger](#provenance-ledger)):
* PROVENANCE\_TRACKING: Append the opt-in provenance-ledger task to the pipeline (default: False)
## Parameters
Dataset name(s) or UUID(s) to process. Processes all datasets if not specified.User performing the operation.Pydantic model defining the knowledge graph schema. Defaults to KnowledgeGraph.Text chunking strategy class.Maximum size of text chunks in tokens.Number of chunks to process per LLM batch.Override the full Cognee config for this run.Override vector database configuration.Override graph database configuration.If true, return immediately and process in background.If true, skip already-processed data. The skip runs whenever this **or** `data_cache` is true, so a full reprocess requires both to be `False`. See [Incremental loading and deduplication](/core-concepts/main-operations/legacy-operations/cognify#examples-and-details).Companion flag to `incremental_loading` — either one being true enables the already-processed skip for a data item.Custom system prompt for entity/relationship extraction.Enable temporal-aware processing.Number of data items per processing batch.LLM settings to install into the current async context for this graph-building operation. When omitted, Cognee uses the active context config or global LLM config. Import `LLMConfig` from `cognee.infrastructure.llm.config`.Embedding settings to install into the current async context for this graph-building operation. When omitted, Cognee uses the active context config or global embedding config. Import `EmbeddingConfig` from `cognee.infrastructure.databases.vector.embeddings.config`.If true, return a `DryRunEstimate` of LLM token usage and rough cost instead of running the pipeline. No LLM calls are made and no graph results are written. See [Dry-run cost estimation](#dry-run-cost-estimation).
## Dry-run cost estimation
Pass `dry_run=True` to preview the LLM token usage and rough USD cost of a `cognify()` run **without making any LLM calls or writing graph results**. This is useful for budgeting a large dataset before committing to the run.
```python theme={null}
import cognee
estimate = await cognee.cognify(datasets=["my_dataset"], dry_run=True)
print(estimate) # human-readable summary table
print(estimate.estimated_cost_usd) # e.g. 0.012345
print(estimate.total_tokens) # input + output tokens across stages
print(estimate.to_dict()) # JSON-serializable dict
```
The estimate covers the two LLM-heavy stages of the default pipeline — `structured_graph_extraction` and `chunk_summarization` — reusing the real document classifier, chunker, and prompt templates so chunk and call counts track an actual run.
The returned `DryRunEstimate` exposes:
| Field | Type | Description |
| ------------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------------------- |
| `operation` | `str` | `"cognify"` for this call. |
| `model` | `str` | The configured LLM model the estimate is priced against. |
| `chunks` | `int` | Number of chunks that would trigger LLM calls. |
| `chunk_tokens` | `int` | Total input tokens across those chunks. |
| `input_tokens` / `output_tokens` / `total_tokens` | `int` | Aggregate token counts across all stages. |
| `estimated_cost_usd` | `float` | Rough total cost across all stages. |
| `skipped_items` | `int` | Items excluded from estimation (e.g. audio/image items, DLT row chunks, code files). |
| `warnings` | `list[str]` | Notes about approximations (reasoning-model output allowance, skipped items, or missing pricing entries). |
| `stages` | `list` | Per-stage breakdown (`name`, `calls`, `input_tokens`, `output_tokens`, `total_tokens`, `estimated_cost_usd`). |
Behavior notes:
* **Datasets are resolved read-only.** Unlike a normal run, a dry run never creates a missing dataset, so estimating a typo'd dataset name fails loudly instead of silently creating one.
* **Estimates are upper bounds for re-runs.** With `incremental_loading=True`, a real run skips already-processed documents, so a dry run may over-estimate a re-run.
* **Not supported with `temporal_cognify=True`** (only the default pipeline is estimated) or while connected to a remote Cognee instance via `serve()` — both raise a `ValueError`.
* **Unknown models emit a warning** rather than reporting a `$0` cost when no pricing entry exists for the configured model.
* **Code files are counted as skipped, at zero cost.** Items that `cognify()` would route down the code graph pipeline are separated out before any document is read, so they are never chunked and contribute no tokens or cost. They are folded into `skipped_items` and reported in `warnings` as `Skipped N code file(s) because they run the deterministic code graph pipeline — no LLM calls.` See [Loaders](/core-concepts/further-concepts/loaders) for which extensions take that route.
## Processing Pipeline
When you call `cognify()`, data goes through these stages:
1. **Document classification** — identify content type
2. **Text chunking** — split into manageable segments
3. **Entity extraction** — identify entities using the LLM
4. **Relationship detection** — find connections between entities
5. **Graph construction** — build the knowledge graph
6. **Summarization** — generate summaries of content
7. **Provenance recording** *(opt-in, off by default)* — append an audit-ledger entry for every document, chunk, entity, and relationship this run produced. Enabled with `PROVENANCE_TRACKING=true`; when off, the task list is identical to the pipeline above. See [Provenance ledger](#provenance-ledger).
8. **Contradiction detection** *(opt-in, off by default)* — compare the facts this run touched against the facts already stored around them and record each conflict as a `contradicts` edge. Enabled with `CONTRADICTION_DETECTION=true`; when off, the task list is identical to the pipeline above. See [Contradiction detection](#contradiction-detection).
## Provenance ledger
When `PROVENANCE_TRACKING` is enabled, `cognify()` splices one extra task (`record_provenance`) into the default pipeline. It runs immediately after the graph and embeddings have been written — so node ids are persisted and stable — and before the contradiction check, so the ledger never depends on contradiction edges. For each item the run produced it appends document → chunk → entity → relationship lineage entries to the append-only `provenance_entries` table in the relational database, committing all entries for one task invocation as a single chained transaction.
There is **no `cognify()` argument** for this feature; like contradiction detection it is configured entirely through `CognifyConfig`:
```bash theme={null}
PROVENANCE_TRACKING=true # default: false
```
The cognify config is cached for the lifetime of the process, so set this in your `.env` or environment **before** the first `cognify()` call. Changing it mid-process has no effect.
Behavior notes:
* **Failure is never fatal.** The task returns its input unchanged and swallows all of its own errors, logging a warning (`Provenance recording failed; ingestion unaffected: ...`) instead of failing the run. Missing pipeline context or ids degrade to entries with a `source_ref_key` of `None` rather than raising. A swallowed failure rolls the whole batch back, so those entries are simply absent — they never claimed sequence numbers, the hash chain stays intact, and `verify_chain()` still reports `valid`. Verification proves the stored entries were not tampered with, not that everything an ingestion produced was recorded, so watch that warning if you depend on the ledger being complete.
* **Ledger keys are dataset-scoped.** Cognee entity ids are deterministic and the ledger lives in the shared relational database, so every node-derived ledger id is prefixed with the dataset id (`"{dataset_id}:{raw_id}"`), and relationship ids are built from the prefixed endpoints. Two datasets mentioning the same entity name keep separate version chains.
* **There is one chain, and one writer at a time.** Dataset scoping applies to ledger keys and version chains, not to the hash chain itself: every entry in the table shares a single `sequence_id` sequence, and each commit takes a ledger-wide write lock (a Postgres advisory lock, so it serializes across processes too). Concurrent `cognify()` runs — several datasets, several workers — therefore queue behind each other for their ledger commits, which is why entries are batched one transaction per task invocation.
* **Custom `graph_model` schemas are covered.** DataPoints that do not follow the default `made_from` / `is_part_of` / `contains` shape are walked with the same traversal `add_data_points` uses, so every node and edge is still recorded.
* **The table must exist.** It ships as Alembic revision `b8c1d3e5f7a9`; run migrations (`cognee.run_migrations()`, or `alembic upgrade head`) before enabling the flag on an existing deployment. The migration is idempotent — it is a no-op if the table is already present.
Entries are read back programmatically through `ProvenanceManager`, which exposes `track_entity`, `track_chunk`, `track_relationship`, `get_provenance`, `get_lineage`, `trace_lineage`, `revision_history`, `invalidate`, `verify_chain`, `check`, and `get_statistics`:
```python theme={null}
from cognee.modules.provenance import get_provenance_manager
manager = get_provenance_manager()
report = await manager.verify_chain()
print(report["valid"], report["total_entries"], report["broken_links"])
```
Each entry carries a SHA-256 checksum over its canonical JSON plus the previous entry's checksum, linked by a unique `sequence_id`, so `verify_chain()` detects deletion, reordering, and single-field edits. It streams the ledger in sequence order by keyset pagination rather than materializing it client-side, so it is safe to run against a large table.
## Contradiction detection
When `CONTRADICTION_DETECTION` is enabled, `cognify()` appends one extra task to the end of the default pipeline. It runs after the graph has been written, so both the new facts and the pre-existing ones are persisted and comparable. For each pair of facts the LLM judges to be in conflict, Cognee logs a warning and writes a `contradicts` edge into the graph — nothing is rewritten or deleted.
There is **no `cognify()` argument** for this feature; it is configured entirely through `CognifyConfig`, which reads these environment variables:
```bash theme={null}
CONTRADICTION_DETECTION=true # default: false
CONTRADICTION_CONFIDENCE_THRESHOLD=0.5 # minimum LLM confidence to flag a pair
CONTRADICTION_MAX_FACTS=500 # cap on facts sent to the LLM in one check
```
The cognify config is cached for the lifetime of the process, so set these in your `.env` or environment **before** the first `cognify()` call. Changing them mid-process has no effect.
Because entity node ids are deterministic (`Entity:`), a re-mentioned entity keeps the id it was first stored under — so a new fact and the stored fact it contradicts share a subject and land in the same neighbourhood:
```python theme={null}
import cognee
await cognee.add("Alice was born in 1985.")
await cognee.cognify()
# Later ingestion, with CONTRADICTION_DETECTION=true
await cognee.add("Alice was born in 1990.")
await cognee.cognify()
```
The second run emits a `WARNING` of the form:
```text theme={null}
Contradiction detected (confidence 0.95): 'alice born in 1985' contradicts 'alice born in 1990' —
```
followed by an `INFO` line reporting how many contradictions were flagged in the graph.
### The `contradicts` edge
Each flagged pair is written as a single edge with `relationship_name` `"contradicts"` and these properties:
| Property | Type | Description |
| ----------------------------------- | ------- | ---------------------------------------------------------------------------------------------------- |
| `relationship_name` | `str` | Always `"contradicts"`. |
| `source_node_id` / `target_node_id` | `str` | The two nodes the edge links. |
| `first_fact` | `str` | Rendered text of the first conflicting fact, e.g. `"alice born in 1985"`. |
| `second_fact` | `str` | Rendered text of the second conflicting fact. |
| `reason` | `str` | Short LLM explanation of why the two facts are incompatible. |
| `confidence` | `float` | The LLM's confidence, in `[0.0, 1.0]`. Pairs below `CONTRADICTION_CONFIDENCE_THRESHOLD` are dropped. |
`first_fact` and `second_fact` are rendered by Cognee from the graph itself (``, with underscores in the relationship name replaced by spaces), not taken from the model output, so the stored text always matches the graph. Node names are stored normalized (lowercase), so the rendered facts are lowercase too.
The edge connects the two nodes that actually differ: the two subjects when the facts have different subjects, otherwise the two objects. If both facts reference exactly the same pair of nodes, no edge is written.
Since it is an ordinary edge, you can read it back with the graph engine:
```python theme={null}
from cognee.infrastructure.databases.graph import get_graph_engine
graph_engine = await get_graph_engine()
_, edges = await graph_engine.get_graph_data()
for source, target, relationship_name, properties in edges:
if relationship_name == "contradicts":
print(properties["first_fact"], "<>", properties["second_fact"])
print(properties["reason"], properties["confidence"])
```
### Scope, cost, and limits
* **Scoped to what you just ingested.** Only the 1-hop neighbourhood of the entities the current run touched is inspected — not the whole graph.
* **Structural edges are ignored.** `contains`, `is_part_of`, `made_from`, `exists_in`, and `contradicts` itself are skipped when building the candidate fact list, as are edges whose endpoints are unnamed (chunks, documents).
* **Fact cap.** At most `CONTRADICTION_MAX_FACTS` facts are compared per check; when the cap is hit, the remainder are skipped and an `INFO` line is logged. Very large neighbourhoods may therefore be only partially compared.
* **Cost.** One additional LLM call per chunk batch, and only when at least two candidate facts were found.
* **Fail-safe and non-destructive.** The task only adds edges, returns its input unchanged, and swallows its own errors (logging a warning), so it can never break ingestion.
Contradiction detection needs a graph backend that supports neighbourhood reads. The default provider (and Neo4j, Neptune, the Postgres graph adapter **(demo)**, and Turso) support it; with `GRAPH_DATABASE_PROVIDER=kuzu` the check currently logs a warning and skips silently, so no `contradicts` edges are written.
## Examples
```python theme={null}
import cognee
# Process all datasets
await cognee.cognify()
# Process a specific dataset
await cognee.cognify(datasets=["my_dataset"])
# Process in background
await cognee.cognify(datasets=["large_dataset"], run_in_background=True)
# Use a custom graph model
from pydantic import BaseModel
class MyGraph(BaseModel):
nodes: list
edges: list
await cognee.cognify(graph_model=MyGraph)
# Custom extraction prompt
await cognee.cognify(
custom_prompt="Extract all technical concepts and their relationships."
)
# Pin specific entity and relationship types
custom_prompt = """
Extract only people and cities as entities.
Connect people to cities with the relationship "lives_in".
Ignore all other entities.
"""
await cognee.cognify(custom_prompt=custom_prompt)
```
When `custom_prompt` is set, it fully **replaces** the default graph extraction prompt (see [`GRAPH_PROMPT_PATH`](/core-concepts/main-operations/legacy-operations/cognify#default-extraction-prompts)) for the entity/relationship extraction step, so you can constrain exactly which entity types and relationship labels the LLM produces. For a step-by-step walkthrough, see the [Custom Prompts guide](/guides/custom-prompts).
`custom_prompt` is ignored when `temporal_cognify=True`.
## Further details
When `run_in_background=True`, `cognify()` starts the processing pipeline as an async background task and **returns immediately**. The return shape is the same as blocking mode — a dict mapping `dataset_id` → `PipelineRunInfo` — but each entry has status `PipelineRunStarted` instead of `PipelineRunCompleted`, and the knowledge graph construction continues in the background.
```python theme={null}
import cognee
# Start processing without waiting for completion
run_info = await cognee.cognify(
datasets=["large_corpus"],
run_in_background=True
)
# run_info is a dict of {dataset_id: PipelineRunInfo}
for dataset_id, info in run_info.items():
print(info.pipeline_run_id) # UUID to track this run
print(info.dataset_id) # Dataset being processed
print(info.status) # Initial status (e.g. "PipelineRunStarted")
```
The returned `PipelineRunInfo` fields relevant for monitoring:
| Field | Type | Description |
| ----------------- | ------ | --------------------------------------- |
| `pipeline_run_id` | `UUID` | Unique identifier for this pipeline run |
| `dataset_id` | `UUID` | The dataset being processed |
| `dataset_name` | `str` | Name of the dataset |
| `status` | `str` | Current status of the run |
Possible status values: `PipelineRunStarted`, `PipelineRunYield`, `PipelineRunCompleted`, `PipelineRunAlreadyCompleted`, `PipelineRunErrored`.
A sixth status, `PipelineRunProgress`, exists only on the WebSocket channel described below. `cognify()` never returns or yields it, so SDK callers that want in-flight progress should poll `GET /api/v1/datasets/status/progress` instead.
When using the REST API, subscribe to real-time pipeline updates with the WebSocket endpoint:
```
WebSocket: /cognify/subscribe/{pipeline_run_id}
```
**Authentication**: The handshake accepts exactly the credentials the HTTP API accepts — an API key header (`X-Api-Key`), a bearer `Authorization` header, or the auth cookie. The configured authentication backends are tried in registration order and the first one that resolves an active user wins. When `REQUIRE_AUTHENTICATION` is off, an unauthenticated handshake falls back to the default user, so single-user deployments can connect without credentials just as their HTTP routes do.
Browsers cannot set custom headers when opening a WebSocket, so both the bearer and API key schemes also accept the credential as a `?token=` query parameter. The fallback applies only to WebSocket connections — plain HTTP requests still require the header.
A WebSocket handshake is itself an HTTP request, so a `?token=` query string can end up in access logs. Uvicorn's own access/error logs redact it automatically, but a reverse proxy or load balancer in front of Cognee (nginx, AWS ALB) logs the full request path by default — redact the `token` query parameter there if you terminate WebSocket traffic through one. Prefer a header wherever your client can set one.
**Usage example (JavaScript)**:
```javascript theme={null}
const pipelineRunId = "your-pipeline-run-id-uuid";
// Same-origin browser client with an auth cookie — nothing extra needed:
const ws = new WebSocket(`ws://your-server/cognify/subscribe/${pipelineRunId}`);
// Otherwise pass the bearer token or API key as a query parameter:
// const ws = new WebSocket(
// `ws://your-server/cognify/subscribe/${pipelineRunId}?token=${token}`
// );
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log("Status:", data.status);
console.log("Run ID:", data.pipeline_run_id);
// data.payload contains the current graph data for the dataset
};
ws.onclose = () => {
// Server closes the connection when processing completes (status: PipelineRunCompleted)
console.log("Pipeline run finished");
};
```
Most WebSocket messages have this shape:
```json theme={null}
{
"pipeline_run_id": "uuid-string",
"status": "PipelineRunYield",
"payload": { /* current graph data for the dataset */ }
}
```
**In-flight progress messages.** While a run is executing, the connection also carries `PipelineRunProgress` messages — one each time a processed result exits the run's task chain — so a backgrounded run signals it is alive and moving instead of going quiet until the terminal event:
```json theme={null}
{
"pipeline_run_id": "uuid-string",
"status": "PipelineRunProgress",
"completed_items": null,
"total_items": null,
"current_stage": "add_data_points",
"stage_index": 4,
"stage_total": 4
}
```
| Field | Type | Description |
| ----------------- | ---------------- | ------------------------------------------------------------------------ |
| `current_stage` | `string \| null` | Name of the task most recently entered in the emitting item's task chain |
| `stage_index` | `int \| null` | 1-based position of that task in the chain |
| `stage_total` | `int \| null` | Total number of tasks in the chain |
| `completed_items` | `null` | Always `null` on this channel — see below |
| `total_items` | `null` | Always `null` on this channel — see below |
Read the stage fields for what they are, not as a stage-by-stage tracker: tasks stream results through the chain, so a message only fires once a result has passed through **every** stage. By then the whole chain has been entered, which means `current_stage` in practice always names the chain's *final* task (`add_data_points` on the default cognify pipeline) and `stage_index` equals `stage_total`. Treat these messages as a liveness heartbeat that also tells you the chain's length, and get progress numbers from the polling endpoint below.
Two more things to handle in a client:
* Progress messages carry **no `payload` key**. Ticks are frequent, and the graph snapshot that the other statuses attach is deliberately not recomputed for each one. Read `data.payload` only when `data.status` is not `PipelineRunProgress`.
* `completed_items` and `total_items` arrive as **`null`** here. The N-of-M file counts are the polling endpoint's signal. For a "3 of 10 files" bar, read `GET /api/v1/datasets/status/progress`, which returns `{status, progress}` per dataset with `completed_items`, `total_items`, and `current_stage`.
The server closes the WebSocket with one of these codes:
| Code | Meaning |
| ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `1000` | The run reached `PipelineRunCompleted` and its final payload was sent |
| `1008` | Rejected: not authenticated, `pipeline_run_id` is not a valid UUID, no such run, or no read permission on the run's dataset. Also sent mid-stream if the dataset is deleted or read access is revoked while streaming |
| `1011` | The server hit an internal failure while streaming the run |
Each `1008` close carries a `reason` naming which of those it was. A retry replays the same rejection, so clients should stop rather than reconnect.
The run's update queue is consumed, not observed: there is one subscriber per run. A second client subscribing to the same `pipeline_run_id` steals events from the first. Authorization is checked before the queue is touched at all, so a rejected caller cannot disturb the real subscriber.
* **Large datasets** (>100 MB) where blocking would time out HTTP connections
* **API integrations** where you want to return a job ID to the caller immediately
* **Parallel processing** of multiple datasets without waiting for each
For small datasets or scripts, the default blocking mode (`run_in_background=False`) is simpler and returns the final result directly.