Correlation and causation IDs

CQRS ES

Every command and event in Protean carries two tracing identifiers (correlation_id and causation_id) that let you reconstruct the full causal history of any business operation. This guide covers how the IDs are generated, how they propagate, how to supply them from external callers, and how every observability layer (logging, OTEL, Observatory) uses them.

The out-of-the-box guarantee

Protean generates and propagates correlation and causation IDs automatically. No opt-in configuration, no manual wiring. The moment you call domain.process(), every command and event in the resulting chain carries both IDs:

correlation_id: A constant identifier shared by every message in the chain. It answers: "Which business operation does this message belong to?"
causation_id: The headers.id of the immediate parent message. It answers: "What directly caused this message?"

Take a small order domain. The command handler for PlaceOrder raises OrderPlaced. An event handler reacts to OrderPlaced by dispatching ConfirmOrder:

from protean import Domain, current_domain, handle
from protean.fields import Dict, Identifier, List, String

domain = Domain(name="Shipping")


@domain.aggregate
class Order:
    customer_id: Identifier(required=True)
    items: List(content_type=Dict)
    status: String(default="PLACED")


@domain.command(part_of=Order)
class PlaceOrder:
    customer_id: Identifier(required=True)
    items: List(content_type=Dict)


@domain.event(part_of=Order)
class OrderPlaced:
    order_id: Identifier(required=True)


@domain.command(part_of=Order)
class ConfirmOrder:
    order_id: Identifier(required=True)


@domain.event(part_of=Order)
class OrderConfirmed:
    order_id: Identifier(required=True)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place(self, command: PlaceOrder) -> None:
        order = Order(customer_id=command.customer_id, items=command.items)
        order.raise_(OrderPlaced(order_id=order.id))
        current_domain.repository_for(Order).add(order)

    @handle(ConfirmOrder)
    def confirm(self, command: ConfirmOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        order.status = "CONFIRMED"
        order.raise_(OrderConfirmed(order_id=order.id))
        repo.add(order)


@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderPlaced)
    def confirm_order(self, event: OrderPlaced) -> None:
        current_domain.process(ConfirmOrder(order_id=event.order_id))

Placing an order takes one call:

def place_order(items):
    # This is all you need. Both IDs are set automatically.
    domain.process(PlaceOrder(customer_id="cust-123", items=items))

When no external correlation_id is provided, Protean generates a UUID4 hex string (32 characters, no dashes). Every downstream event and command inherits it. You never need to check for None.

Passing correlation IDs from external callers

In production, the correlation_id typically originates outside Protean, from an API gateway, frontend client, or upstream service. There are two ways to pass it in.

Explicitly via `domain.process()`

Pass the correlation_id parameter directly:

def place_order_from_gateway(items):
    domain.process(
        PlaceOrder(customer_id="cust-123", items=items),
        correlation_id="req-abc-123-from-gateway",
    )

Automatically via HTTP headers

When using the DomainContextMiddleware with FastAPI, the middleware extracts X-Correlation-ID (falling back to X-Request-ID) from the incoming request and stores it in the global context. current_domain.process() reads the ID from the global context automatically:

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

from protean import current_domain
from protean.integrations.fastapi import DomainContextMiddleware

app = FastAPI()
router = APIRouter()


class PlaceOrderRequest(BaseModel):
    customer_id: str
    items: list[dict]


app.add_middleware(
    DomainContextMiddleware,
    route_domain_map={"/orders": domain},
)


@router.post("/orders")
async def place_order(request: PlaceOrderRequest):
    # No need to extract headers: the middleware already did it.
    # The correlation ID flows through automatically.
    current_domain.process(PlaceOrder(**request.model_dump()))


app.include_router(router)

The middleware also injects X-Correlation-ID into the response, reflecting the ID that was actually used (from header, explicit param, or auto-generated).

Error responses include correlation ID

When register_exception_handlers is used alongside DomainContextMiddleware, error responses automatically include correlation_id in the JSON body. API consumers can correlate errors with the originating request without inspecting response headers:

{
  "error": "Order not found",
  "correlation_id": "ext-123"
}

This works for all Protean exception types (ValidationError, ObjectNotFoundError, InvalidStateError, InvalidOperationError, InvalidDataError, and plain ValueError). The body correlation_id matches the X-Correlation-ID response header. Routes outside a domain context do not include the field.

Priority of correlation ID sources

When multiple sources provide a correlation ID, Protean resolves them in this order:

Explicit correlation_id parameter on domain.process(), highest priority
Parent message context (g.message_in_context), used when an event handler dispatches a new command
HTTP request header (g.request_correlation_id), set by DomainContextMiddleware
Auto-generated UUID4 hex: Fallback when none of the above is present

Full propagation flow

The following diagram shows how correlation_id and causation_id flow through a typical command-event chain, starting from an HTTP request:

Sequence diagram

Every message in the chain shares the same correlation_id (ext-123), while each causation_id points to the immediate parent message.

Causation chain semantics

The two IDs together form a causation tree. The correlation_id is the tree's label; the causation_id links define the edges:

PlaceOrder [corr=ext-123, cause=None]                <-- root command
  +-- OrderPlaced [corr=ext-123, cause=PlaceOrder.id]
        +-- ReserveInventory [corr=ext-123, cause=OrderPlaced.id]
        |     +-- InventoryReserved [corr=ext-123, cause=ReserveInventory.id]
        +-- NotifyCustomer [corr=ext-123, cause=OrderPlaced.id]
              +-- NotificationSent [corr=ext-123, cause=NotifyCustomer.id]

Propagation rules

Entry point	`correlation_id`	`causation_id`
`domain.process(cmd)`	Auto-generated or caller-provided	`None` (root)
`domain.process(cmd, correlation_id="ext-123")`	`"ext-123"`	`None` (root)
Command handler raises event	Inherited from command	Command's `headers.id`
Event handler dispatches new command	Inherited from event	Event's `headers.id`

Where the IDs live

Both IDs are stored in DomainMeta, the domain-specific section of message metadata:

def trace_ids(event, message):
    # On a command or event object
    event_ids = (
        event._metadata.domain.correlation_id,  # "ext-123"
        event._metadata.domain.causation_id,  # the causing message's headers.id
    )

    # On a deserialized Message
    message_ids = (
        message.metadata.domain.correlation_id,
        message.metadata.domain.causation_id,
    )
    return event_ids, message_ids

Traversing the chain programmatically

The event store provides three methods for causation chain traversal:

def traverse(some_message, command_message):
    store = domain.event_store.store

    # Walk UP from a message to the root command
    chain = store.trace_causation(some_message)
    # Returns [root_command, ..., target_message]

    # Walk DOWN from a command to find all its effects
    effects = store.trace_effects(command_message)
    # Returns downstream events/commands in chronological order

    # Build the full tree for a correlation ID
    root = store.build_causation_tree("ext-123")
    # Returns a CausationNode with .children recursively populated

    return chain, effects, root

To get the whole chain in causal order, call domain.correlation_trace(). It wraps build_causation_tree() and flattens the tree into a depth-first, pre-order list (each parent appears before its effects):

def print_chain():
    chain = domain.correlation_trace("ext-123")
    for node in chain:
        print(f"{node.kind}: {node.message_type}")
    # Returns [] when no messages match the correlation ID.

Asserting a chain in tests

protean.testing.assert_chain() checks that a correlation chain contains an exact sequence of message types, in order. Pass type names or the element classes themselves (their __type__ is used):

from protean.testing import assert_chain


def check_order_chain(correlation_id):
    chain = domain.correlation_trace(correlation_id)

    # Compare against element classes...
    assert_chain(chain, [PlaceOrder, OrderPlaced, ConfirmOrder, OrderConfirmed])

    # ...or against fully-qualified type strings:
    assert_chain(
        chain,
        [
            "Shipping.PlaceOrder.v1",
            "Shipping.OrderPlaced.v1",
            "Shipping.ConfirmOrder.v1",
            "Shipping.OrderConfirmed.v1",
        ],
    )

assert_chain() raises AssertionError with the expected and actual sequences when the length or any type does not match. You can check an end-to-end causation flow in a single assertion.

CLI inspection

Follow a causal chain from the terminal:

# Tree view (default) -- shows parent-child causation structure
protean events trace "ext-123" --domain=myapp

# Flat table view -- chronological list with trace columns
protean events trace "ext-123" --flat --domain=myapp

Service boundary handling

When one Protean service consumes events from another via subscribers, the correlation chain must be preserved across the boundary. Protean handles this automatically.

Automatic bridging in subscribers

When the server processes an incoming broker message, it extracts the correlation_id from the message's metadata.domain.correlation_id and sets it on the processing context. If the external message has no correlation ID, the server generates a fresh UUID, so the chain within the consuming service is still fully traced.

@domain.subscriber(stream="payments")
class PaymentSubscriber:
    def __call__(self, message):
        # The correlation_id from the source service is already in context.
        # Any commands dispatched here inherit it automatically.
        domain.process(
            ConfirmPayment(order_id=message["order_id"]),
        )

Nested command preservation

When a subscriber dispatches multiple commands within a single handler, Protean preserves the outer message context across each domain.process() call. This prevents the second command from losing the original correlation chain:

@domain.subscriber(stream="fulfillment")
class FulfillmentSubscriber:
    def __call__(self, message):
        # Both commands inherit the same correlation_id from the broker message.
        domain.process(ReserveInventory(order_id=message["order_id"]))
        domain.process(NotifyWarehouse(order_id=message["order_id"]))

Relationship to OpenTelemetry

Protean maintains two distinct tracing layers that complement each other:

Layer	Location	Purpose	Format
Domain tracing	`DomainMeta.correlation_id` + `causation_id`	Business operation tracking	Flexible strings
Distributed tracing	`MessageHeaders.traceparent`	Infrastructure span tracking (Jaeger, Datadog)	W3C 32-hex / 16-hex

Both layers carry the correlation_id. It identifies the same business operation whether you're looking at domain metadata or a Jaeger trace.

When OTel is enabled, Protean sets protean.correlation_id on all major spans, and protean.causation_id wherever a single parent message is well-defined. This lets you filter and group spans by business operation in your APM dashboard.

OTEL span attribute reference

The following spans carry correlation and causation ID attributes:

Span name	`protean.correlation_id`	`protean.causation_id`
`protean.command.process`	Yes	—
`protean.handler.execute`	Yes	Yes
`protean.uow.commit`	Yes	Yes
`protean.outbox.process`	Yes (when uniform)	Yes (when uniform)
`protean.outbox.publish`	Yes	Yes

"When uniform" means the attribute is set only when all messages in a batch share the same ID.

For the complete span catalog, see the OpenTelemetry integration guide.

Viewing correlation chains in Observatory

The Protean Observatory includes correlation_id and causation_id in every MessageTrace event. This enables real-time grouping and filtering of trace events by business operation.

Timeline view

The Observatory Timeline page supports deep-linking by correlation ID:

http://localhost:9000/timeline?correlation=ext-123

This opens the correlation chain sub-view, showing a vertical causation tree of all messages in the chain.

Trace events that carry correlation context

Most trace event types include correlation_id and causation_id when available. Handler and outbox events always carry these fields; acknowledgment and dead-letter events may have null values when correlation context is not available at the point of emission.

Trace event	Description
`handler.started`	Handler began processing a message
`handler.completed`	Handler finished successfully
`handler.failed`	Handler raised an exception
`pm.transition`	Process manager state transition
`outbox.published`	Message published from outbox
`outbox.failed`	Outbox publish failure
`message.acked`	Message acknowledged
`message.dlq`	Message moved to dead letter queue

For details on the Observatory, see the Observability reference.

Structured logging setup

Domain.init() auto-configures logging and installs the correlation integrations: ProteanCorrelationFilter on the root logger and on each of its handlers, and protean_correlation_processor on the structlog pipeline. Every log record emitted during message processing includes correlation_id and causation_id fields. You do not need to wire anything by hand.

For the full setup, config keys, and the wide event schema, see the Logging guide and the Logging reference.

Manual wiring when auto-configuration is disabled

When PROTEAN_NO_AUTO_LOGGING=1 is set, or when the framework is embedded in an application that manages its own logging, wire the integrations explicitly:

import logging

from protean.integrations.logging import ProteanCorrelationFilter

for handler in logging.getLogger().handlers:
    handler.addFilter(ProteanCorrelationFilter())

Attach the filter to each handler on the root logger. A filter on a logger runs only for records logged on that logger, so a filter on the root logger misses records from child loggers such as logging.getLogger("myapp.orders").

import structlog

from protean.integrations.logging import protean_correlation_processor

structlog.configure(
    processors=[
        protean_correlation_processor,
        structlog.dev.ConsoleRenderer(),
    ]
)

Safe when no context is active

Both integrations check for an active domain context and a g.message_in_context before reading IDs. When no context is available (e.g., during startup or in a background thread), both fields default to "" so formatters never raise KeyError. When no message is in scope, both integrations fall back to g.correlation_id / g.causation_id. g.correlation_id and g.causation_id are the documented extension point for HTTP middleware, CLI commands, and background jobs that need logs tagged before any domain message exists.

Summary

Aspect	Detail
`correlation_id` format	Flexible string (UUID4 hex default, or any caller-provided string)
`causation_id` format	Protean message ID (e.g., `myapp::order:command-abc123-0`)
Auto-generation	Always. Every chain has a `correlation_id` even if none is supplied
HTTP header	`X-Correlation-ID` (fallback: `X-Request-ID`), extracted by `DomainContextMiddleware`
Error responses	`correlation_id` included in JSON body via `register_exception_handlers`
Cross-service	Automatic bridging through subscribers; generates fresh UUID when source has none
OTEL spans	`protean.correlation_id` and `protean.causation_id` on all major spans
Observatory	Every `MessageTrace` event carries both IDs; Timeline supports correlation deep-links
Logging	`ProteanCorrelationFilter` (stdlib) and `protean_correlation_processor` (structlog)
CLI	`protean events trace <correlation_id>` with tree and flat views
Programmatic API	`trace_causation()`, `trace_effects()`, `build_causation_tree()`