Repository navigation
Proposal: typed event context components and context enrichers #115
Description
Motivation
Context currently stores correlation and causation IDs together with arbitrary additional fields:
Context(
correlation_id=...,
causation_id=...,
custom_field=...,
)This is flexible, but integrations cannot add strongly typed context data without knowing the complete application-specific Context type.
OpenTelemetry is the initial use case. An integration should be able to add its propagation carrier to an event without replacing or understanding context data provided by the application. The same mechanism could support request metadata, tenant information, authentication context, or other optional integrations.
Proposal
Treat Context as a container of independently typed context components.
class ContextComponent(BaseModel):
passExample components:
class EventCorrelation(ContextComponent):
correlation_id: UUID | None = None
causation_id: UUID | None = None
class OpenTelemetryCarrier(ContextComponent):
__context_name__ = "opentelemetry"
values: dict[str, str]
class RequestContext(ContextComponent):
__context_name__ = "application.request"
user_id: str
request_id: UUIDA typed getter would retrieve components by type:
otel = context.get(OpenTelemetryCarrier)
# OpenTelemetryCarrier | None
request = context.get(RequestContext)
# RequestContext | NoneThis avoids trying to express an unsupported intersection type such as:
ApplicationContext & OpenTelemetryContext & RequestContext
Each integration owns and types only its component.
Context enrichers
Introduce an explicit enricher contract:
class EventContextEnricher(Protocol):
def enrich(
self,
context: Context,
*,
event: WrappedEvent,
stream_id: StreamId,
) -> ContextComponent | None:
...Enrichers would run in EventStore._append() after the event already has a Context, but before Serde.serialize_many():
existing Context
-> registered enrichers
-> Context with additional typed components
-> Serde
-> StorageStrategy
An enricher returns only the component it owns. It does not replace or mutate the complete context.
Explicitly supplied context data should take precedence. Enrichers should fill missing components rather than silently overwrite them.
Serialization
Add a registry similar to EventRegistry:
class ContextRegistry:
def add(
self,
component: type[ContextComponent],
) -> type[ContextComponent]:
...
def name_for_type(
self,
component: type[ContextComponent],
) -> str:
...
def type_for_name(
self,
name: str,
) -> type[ContextComponent]:
...Serde would use this registry to serialize and deserialize concrete component types.
Example persisted context:
{
"correlation_id": "...",
"causation_id": "...",
"opentelemetry": {
"values": {
"traceparent": "00-...",
"tracestate": "..."
}
},
"application.request": {
"user_id": "123",
"request_id": "..."
}
}The registry key is the stable serialized type name. The Python class is the key used by application code.
ContextRegistry should remain separate from EventRegistry, because EventRegistry also contains event-specific encryption and data-subject logic. They may share a small generic type-to-name mapping internally if that proves useful.
Backward compatibility
The existing context format is already persisted by all backends, so the proposal should preserve it.
Built-in correlation context
correlation_id and causation_id remain at the context root. They are exposed through the new API as a special built-in component:
correlation = context.get(EventCorrelation)EventCorrelation acts as a typed view over the existing root fields rather than being serialized under another nested key.
If an enricher returns EventCorrelation, its values are written to the existing root fields.
Existing context API
The following should remain valid:
Context(
correlation_id=...,
causation_id=...,
custom_field=...,
)Existing serialized events should remain readable without migration.
Existing whole-context validation should also continue to work:
wrapped_event.get_context(ApplicationContext)Registered component names are deserialized into typed ContextComponent instances. Unregistered legacy fields remain available as ordinary extra context data, as they are today.
This provides two compatible views:
Legacy view:
validate the complete Context as an application-defined model
New view:
retrieve an independently typed component by its class
OpenTelemetry use case
The optional OpenTelemetry integration would register an OpenTelemetryCarrier component and an enricher.
The enricher would use the configured global OpenTelemetry propagator:
current OpenTelemetry Context
-> propagator.inject()
-> OpenTelemetryCarrier
-> persisted event Context
During asynchronous processing:
persisted OpenTelemetryCarrier
-> propagator.extract()
-> new processing span
The core package would not depend on OpenTelemetry.
Scope
This proposal covers:
- typed context components;
- component registration and serialization;
- context enrichers;
- backward-compatible access to the existing context format.
It does not yet define:
- automatic correlation and causation propagation;
- OpenTelemetry span creation;
- outbox or subscription instrumentation;
- mapping pyES context values to OpenTelemetry Baggage.
Those can be built on top of the context component and enricher mechanism.
Questions for review
- Is treating
Contextas a container of typed components a useful direction? - Should component names be stored at the context root, as shown above, or under a dedicated
extensionskey? - Is exposing the existing correlation fields as a special
EventCorrelationcomponent a reasonable compatibility layer? - Should existing values always win over enricher output, or should duplicate components raise an error?
- Should unknown/unregistered context fields continue to pass through unchanged?
- Should
ContextRegistrybe independent, or should a generic registry primitive be extracted fromEventRegistry?
Activity
Enforcer commentedon Sep 26, 2026
It's a significant flaw of the current design :(
I would say so. This reminds me a certain design pattern I cannot recall the name of right now.
I'd use
extensionsor similar, e.g. justcomponents?Works for me plus accessors to maintain compatiblity
WDYM? That once some values are in context enrichment logic runs again? IMO that should be an error, indicating there is an issue with context management and potential cross-contamination
For backwards compatibility only, but undocumented. This should be discouraged as much as possible.
Do we even need it at all? Registry, I have an impression, was created partially when Pydantic was imagined as a swappable dependency. It is not really. If it's still needed though, I'd keep it separate
lzukowski commentedon Oct 1, 2026
Thanks, this is helpful. I agree with the direction.
Typed components — agreed. The current design prevents integrations from adding typed context without knowing the application-owned complete context model.
Storage shape — agreed on a dedicated
componentscontainer. I originally mentionedextensions, butcomponentsis more precise. I think this should be a mapping rather than a list:{ "correlation_id": "...", "causation_id": "...", "components": { "opentelemetry": { "values": { "traceparent": "00-...", "tracestate": "..." } }, "application.request": { "user_id": "123", "request_id": "..." } } }The mapping key is the stable wire name of the component, and its value is the component payload. This makes the serialized type explicit, prevents duplicate components by construction, and avoids repeating a type discriminator inside every value.
Correlation compatibility — agreed.
EventCorrelationcan be the typed view of the existing root-levelcorrelation_idandcausation_id; existing accessors and the persisted shape must remain compatible.Enricher collisions — my intention was specifically to permit a caller to supply a component before the event reaches the Event Store. For example, an event may already carry an explicit
OpenTelemetryCarrier; in that case an OpenTelemetry enricher should not overwrite it.I agree that silently resolving collisions between enrichers would hide a context-management error. The contract I have in mind is: components present on the incoming context are explicit and win; if two enrichers attempt to add the same missing component during one append, that is an error. Enrichers must return a new context/component and never mutate the incoming context, so enrichment cannot leak from one append into another.
I am still considering whether an enricher belongs directly in the Event Store. For OpenTelemetry it would only snapshot the currently active OTel context through
propagator.inject()before serialization. It would not create spans, activate contexts, or otherwise mutate OTel state. Span lifecycle and consumer instrumentation would remain outside the core Event Store. Does that separation address your cross-contamination concern?Unknown fields — agreed. Passing unknown root-level fields through should exist only to read historical contexts and retain backward compatibility. It should not be documented or supported as a way for new code to add context data. New extensions should be registered typed components under
components.Registry — I agree that this must not be motivated by making Pydantic swappable, and it should not share the event-specific responsibilities of
EventRegistry. I still think a minimal separate registry is needed for deserialization: the persisted key ("opentelemetry") must map to the concrete Python type (OpenTelemetryCarrier) that validates its payload.It would only provide stable
wire name <-> component typemapping. On read, Serde can look up each component key and validate the payload with the registered type. Unknown component keys should remain opaque for backward-compatible reading/round-tripping, but they would not be exposed as typed components or become an API for writing arbitrary data.