Structured events

RelPrim describes every execution twice: the execution report summarizes it afterwards, and structured events describe each step while it happens. Events are opt-in, typed and JSON-compatible, and they are part of RelPrim’s compatibility contract.

Emitting events#

from relprim import EventEmitter, InMemoryEventSink, resilient

sink = InMemoryEventSink()
events = EventEmitter(sinks=(sink,))


@resilient(retries=2, timeout=10, events=events)
async def call_provider(prompt: str) -> str:
    return await provider.generate(prompt)


await call_provider("Summarize this ticket")

for event in await sink.events():
    print(event.event_type, event.payload)

In the builder API, use .with_events(events). One emitter can serve many operations; each event carries its operation_name.

Name	Role
EventType	String enum of stable event identifiers, such as EventType.ATTEMPT_FAILED ("attempt.failed").
StructuredEvent	Immutable event: event_type, operation_name, timestamp, payload, and to_dict().
EventSink	Protocol: an object with async def emit(self, event: StructuredEvent) -> None.
EventEmitter(sinks=..., timestamp_factory=...)	Timestamps events and delivers them to its sinks.
InMemoryEventSink	Collects events; await sink.events() and await sink.clear(). For tests and debugging.
NoopEventSink	Discards events; the emitter’s default sink.
SQLiteEventStore	Persists events to a local SQLite database and queries them.
OpenTelemetryEventSink	Adds events to the active OpenTelemetry span.

Delivery#

EventEmitter.emit() awaits every sink in order, inline, inside the operation. There are no background tasks, queues or buffers, so when an operation returns, all of its events have been delivered.

Within one execution, events arrive in execution order. Concurrent executions sharing a sink interleave, but the events of each execution stay in order.

When a sink fails#

If a sink raises, the exception propagates to the caller unchanged and later sinks do not receive that event. The operation stops at that point:

• the sink error is not retried, does not trigger fallbacks and is not wrapped in OperationExecutionError;
• side effects of the operation may already have happened;
• with idempotency, the execution counts as failed: nothing is cached and a later call executes again.

This is deliberate: RelPrim never silently drops storage or observability failures. To isolate a sink instead, wrap it:

import logging

from relprim import EventSink, StructuredEvent

logger = logging.getLogger(__name__)


class IsolatedSink:
    def __init__(self, sink: EventSink) -> None:
        self._sink = sink

    async def emit(self, event: StructuredEvent) -> None:
        try:
            await self._sink.emit(event)
        except Exception:
            logger.exception("RelPrim event delivery failed")

Cancellation while a sink is awaited propagates as asyncio.CancelledError.

Event contract#

• Payloads are read-only mappings of JSON scalars: str, int, float, bool or None. Floats are always finite. RelPrim validates this when an event is created, so every event can be serialized, stored and exported.
• Timestamps are timezone-aware; a tzinfo whose utcoffset() returns None counts as naive and is rejected. RelPrim produces UTC.
• Durations are finite, non-negative floats in seconds, in keys ending with _seconds.
• Attempt numbers are 1-based; the fallback attempt is numbered after the last primary attempt.
• Status values are lower-case strings: delay_source is "provider" or "backoff", idempotency_status is "executed", "joined" or "replayed", and state is a circuit-breaker state value.
• Idempotency keys never appear in events or report metadata.
• Joined and replayed idempotent calls emit no events; the events of the original execution describe it.

Event identifiers and the keys listed on this page are never renamed or removed outside a major release, and their value types do not change. New event types and new keys can appear in minor releases, so consumers must ignore event types and keys they do not recognize.

Event types#

Event	Emitted when
operation.started	An execution starts.
attempt.started	A primary attempt starts.
validation.succeeded	A result passed validation.
validation.failed	A result failed validation.
circuit_breaker.rejected	The circuit breaker rejected an attempt.
rate_limit.detected	A failure matched the rate-limit policy.
rate_limit.wait_exceeded	The selected delay exceeded the maximum wait.
attempt.failed	An attempt failed, including a failed fallback attempt.
retry.scheduled	A retry delay is about to be awaited.
attempt.succeeded	An attempt succeeded, including a successful fallback attempt.
fallback.started	The fallback chain starts.
fallback.succeeded	A fallback candidate produced a valid result.
fallback.failed	The fallback chain failed.
operation.succeeded	The execution succeeded.
operation.failed	The execution failed.

A run that fails once and succeeds on the retry emits, in order: operation.started, attempt.started, attempt.failed, retry.scheduled, attempt.started, attempt.succeeded, operation.succeeded, with validation.* events before attempt.succeeded or attempt.failed when validation is configured.

Payload keys#

Every event of a type carries the keys listed for it. Keys marked “sometimes” appear only on the path described.

Event	Keys
operation.started	retry_enabled, timeout_enabled, fallback_enabled, circuit_breaker_enabled, validation_enabled, rate_limit_enabled
attempt.started	attempt_number, max_attempts
attempt.succeeded	attempt_number, duration_seconds, attempt metadata
attempt.failed	attempt_number, duration_seconds, error_type, retryable, attempt metadata
retry.scheduled	attempt_number, next_attempt_number, delay_seconds, delay_source, rate_limited, retry_after_seconds
validation.succeeded	attempt_number, validator_name; sometimes fallback_used and fallback_candidate_name, for a fallback candidate’s result
validation.failed	attempt_number, validator_name, reason; sometimes fallback_used and fallback_candidate_name, for a fallback candidate’s result
circuit_breaker.rejected	attempt_number, breaker_name, state, retry_after_seconds
rate_limit.detected	attempt_number, error_type, retry_after_seconds, max_wait_seconds, max_wait_exceeded
rate_limit.wait_exceeded	attempt_number, retry_after_seconds, delay_seconds, max_wait_seconds
fallback.started	attempt_number, after_primary_attempts
fallback.succeeded	attempt_number, duration_seconds, fallback attempt metadata
fallback.failed	attempt_number, duration_seconds, error_type, fallback attempt metadata
operation.succeeded	duration_seconds, attempt_count, retry_count, report metadata
operation.failed	duration_seconds, attempt_count, retry_count, error_type, report metadata

Notes on specific values:

• error_type is the exception class name.
• circuit_breaker.rejected has retry_after_seconds set to None while a half-open probe is in flight.
• rate_limit.detected has retry_after_seconds set to the provider hint or None; in rate_limit.wait_exceeded, delay_seconds is the rejected delay.

Report metadata#

ExecutionReport.metadata always contains:

Key	Meaning
fallback_used	A fallback attempt ran.
circuit_breaker_open	The final attempt was rejected by the circuit breaker.
validation_performed, validation_valid, validation_validator_name, validation_reason	Outcome of the final validation.
rate_limit_encountered	Any attempt was rate-limited.
rate_limit_wait_seconds	Total time waited because of rate limits.
rate_limit_wait_exceeded	A rate-limit delay was rejected because it exceeded the maximum wait.

Depending on the path it also contains the keys of the final attempt (the fallback or rate-limit keys below) and, with idempotency:

Key	Meaning
idempotency_enabled	Always True when present.
idempotency_status	executed, joined or replayed.
idempotency_cache_hit	True for joined and replayed.

Attempt metadata#

Every ExecutionAttempt.metadata contains fallback_used, circuit_breaker_open and the four validation_* keys.

Failed primary attempts also contain:

Key	Meaning
rate_limited	The failure matched the rate-limit policy.
rate_limit_retry_after_seconds	Provider hint, or None.
rate_limit_delay_seconds	Selected rate-limit delay, or None when none was selected (not rate-limited, or no attempt left).
rate_limit_delay_source	provider, backoff or None.
rate_limit_wait_exceeded	The selected delay exceeded the maximum wait.

Successful fallback attempts contain fallback_failed (False), fallback_candidate_name, fallback_candidate_index and fallback_failure_count (the candidates that failed before the successful one). Failed fallback attempts contain fallback_failed (True) and fallback_failure_count. ExecutionAttempt.fallback is True for the fallback attempt.

Writing a sink#

Any object with an async emit method is a sink:

import json

from relprim import StructuredEvent


class JsonLinesSink:
    def __init__(self, path: str) -> None:
        self._path = path

    async def emit(self, event: StructuredEvent) -> None:
        with open(self._path, "a", encoding="utf-8") as file:
            file.write(json.dumps(event.to_dict()) + "\n")

emit runs inline in the operation, so keep it fast, and move blocking I/O to a thread (as SQLiteEventStore does) or to your own queue when volumes are high.