feat(otel): W3C traceparent transport-header propagation — Python SDK (ADR-0028)

Mirrors the Go reference: a dependency-free header seam (publish_with_headers +
headers_from_context surfacing a delivered message's headers) and W3C traceparent
inject/extract in otel.py (consumer span becomes a child of the producer's remote
parent; no header falls back to the v0.1 trace_id behaviour). Per-transport
carriers: Redis __bq_frame (bare-value back-compat), AMQP header table, SQS
MessageAttributes, in-memory reference. Envelope frozen (GR-1), otel stays the
[otel] extra (GR-7), trace_id preserved (GR-4). Also bump to 1.11.0 and reconcile
the drifted __version__ (was 1.6.0) to the package version.

Author: muhammetsafak

CHANGELOG.md

@@ -9,6 +9,31 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.11.0] - 2026-06-21
+
+### Added
+- **W3C `traceparent` span-context propagation** (OpenTelemetry v0.2, ADR-0028) — the optional
+  `babelqueue.otel` module now carries true cross-hop **span** parent-child linkage, not just
+  shared-`trace_id` correlation. On publish, `otel.publish` injects the active span context as a
+  W3C `traceparent` **transport header** (and still stamps `trace_id` for the v0.1 fallback); on
+  consume, `otel.wrap_handler` extracts it and starts the CONSUMER span as a true **child** of the
+  producer span. With no `traceparent` present it falls back to the v0.1 `trace_id`-derived parent,
+  so it is a strict, backward-compatible upgrade — no regression. The header rides **out of band**
+  via a new dependency-free core seam — `BabelQueue.publish_with_headers(urn, data, headers, …)`
+  (produce side) and `babelqueue.headers_from_context()` (consume side, surfaced by the runtime) —
+  so the wire envelope stays **frozen** (`schema_version: 1`, GR-1) and the core stays
+  zero-dependency (OTel remains the optional `[otel]` extra, GR-7). `traceparent` is carried on the
+  **in-memory** (reference), **Redis** (a transport-owned `__bq_frame` JSON frame with bare-value
+  back-compat, so cross-version queues interoperate; degrades to a bare publish in Laravel-compat
+  mode), **RabbitMQ** (native AMQP header table, beside the contract `x-*` headers) and **SQS**
+  (native `MessageAttributes`, beside the contract `bq-*` attributes) transports; where a transport
+  can't carry headers, propagation degrades cleanly to v0.1 `trace_id` correlation with no error.
+  A plain `publish` is byte-identical to before. Unit-tested without a broker (frame round-trip +
+  bare back-compat, header merge/extract per transport, and an in-memory producer→consumer
+  parent-child end-to-end with the OTel SDK's `InMemorySpanExporter`); broker-gated integration
+  tests assert a published `traceparent` arrives on the consumed message's headers beside the
+  unchanged body. The envelope is unchanged; this is purely additive. Ships as a MINOR.
+
 ## [1.6.0] - 2026-06-14
 
 ### Added

README.md

@@ -182,6 +182,42 @@ def on_created(data, meta): ...
 python manage.py babelqueue_worker --queue orders          # run the consumer
 ```
 
+## OpenTelemetry tracing (optional)
+
+`pip install "babelqueue[otel]"` adds the optional `babelqueue.otel` module — the core never
+imports OpenTelemetry, so it stays zero-dependency. It emits a PRODUCER span per publish and a
+CONSUMER span per handled message, correlated across every hop and SDK, at two layered levels:
+
+- **`trace_id` correlation** (v0.1): the envelope's `trace_id` maps 1:1 to an OTel trace id, so
+  every hop that shares a `trace_id` shares one trace — with **zero** wire/transport change.
+- **W3C `traceparent` span linkage** (v0.2): the producer also injects its active span context as
+  a `traceparent` **transport header** (beside the frozen envelope, never in it), so the consumer
+  starts its span as a true **child** of the producer span — real cross-hop parent-child linkage.
+  With no `traceparent` present it falls back to the v0.1 `trace_id` behaviour, so enabling it is a
+  strict, backward-compatible upgrade.
+
+```python
+from opentelemetry import trace
+from babelqueue import BabelQueue, otel
+
+tracer = trace.get_tracer("orders")
+app = BabelQueue("redis://localhost:6379/0", queue="orders")
+
+# consumer: wrap_handler starts a CONSUMER span (child of the producer span when a
+# traceparent rode along; else in the trace_id-derived trace)
+app.register("urn:babel:orders:created", otel.wrap_handler(tracer, on_order_created))
+
+# producer: otel.publish starts a PRODUCER span and carries traceparent + trace_id
+otel.publish(tracer, app, "urn:babel:orders:created", {"order_id": 1042})
+```
+
+The `traceparent` rides the out-of-band transport-header seam (`publish_with_headers` /
+`headers_from_context`) — the same seam the replay-bypass marker uses — so the envelope stays
+frozen (`schema_version: 1`). It is carried on the in-memory, Redis (a transport-owned JSON frame,
+with bare-value back-compat), RabbitMQ (AMQP header table) and SQS (`MessageAttributes`)
+transports; where a transport can't carry it, propagation degrades cleanly to v0.1 `trace_id`
+correlation with no error.
+
 ## What's here
 
 The codec/contracts/dead-letter (zero-dep core), the `BabelQueue` runtime

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "babelqueue"
-version = "1.10.0"
+version = "1.11.0"
 description = "Polyglot Queues, Simplified — the Python core: the canonical BabelQueue wire-envelope codec, contracts and dead-letter helpers."
 readme = "README.md"
 requires-python = ">=3.9"

src/babelqueue/__init__.py

@@ -11,17 +11,18 @@
 
 from __future__ import annotations
 
-from . import dead_letter, idempotency, redrive, replay
+from . import dead_letter, headers, idempotency, redrive, replay
 from .app import BabelQueue
 from .codec import SCHEMA_VERSION, SOURCE_LANG, EnvelopeCodec
 from .contracts import HasTraceId, PolyglotMessage
+from .headers import headers_from_context
 from .idempotency import IdempotencyStore, InMemoryStore
 from .exceptions import BabelQueueError, UnknownUrnError
 from .replay import HEADER_REPLAY_BYPASS, bypass_external_effects, is_replay
 from .routing import UnknownUrnStrategy
 from .transport import HeaderPublisher, InMemoryTransport, ReceivedMessage, Transport
 
-__version__ = "1.6.0"
+__version__ = "1.11.0"
 
 __all__ = [
     "BabelQueue",
@@ -38,11 +39,13 @@
     "BabelQueueError",
     "UnknownUrnError",
     "dead_letter",
+    "headers",
     "idempotency",
     "redrive",
     "replay",
     "is_replay",
     "bypass_external_effects",
+    "headers_from_context",
     "HEADER_REPLAY_BYPASS",
     "IdempotencyStore",
     "InMemoryStore",

src/babelqueue/app.py

@@ -25,9 +25,10 @@ def on_order_created(data, meta):
 from . import dead_letter
 from .codec import EnvelopeCodec
 from .exceptions import UnknownUrnError
+from .headers import _headers_scope
 from .replay import HEADER_REPLAY_BYPASS, _replay_scope
 from .routing import UnknownUrnStrategy
-from .transport import ReceivedMessage, Transport, make_transport
+from .transport import HeaderPublisher, ReceivedMessage, Transport, make_transport
 
 Handler = Callable[..., None]
 
@@ -70,6 +71,37 @@ def publish(
         self.transport.publish(target, EnvelopeCodec.encode(envelope))
         return envelope["meta"]["id"]
 
+    def publish_with_headers(
+        self,
+        urn: str,
+        data: Mapping[str, Any],
+        headers: Mapping[str, str],
+        *,
+        queue: Optional[str] = None,
+        trace_id: Optional[str] = None,
+    ) -> str:
+        """Publish a message together with out-of-band transport ``headers``; returns its id.
+
+        The headers ride **beside** the frozen envelope (GR-1) on the transport's per-message
+        metadata channel — e.g. a W3C ``traceparent`` for cross-hop span linkage (ADR-0028) —
+        never inside it. It is the produce-side counterpart of the headers the runtime surfaces
+        to a handler via :func:`~babelqueue.headers.headers_from_context`.
+
+        When the transport implements :class:`~babelqueue.transport.HeaderPublisher` and
+        ``headers`` is non-empty, the headers are propagated; otherwise it transparently falls
+        back to a plain :meth:`publish` (the headers are dropped — no error, no regression),
+        exactly as :func:`~babelqueue.redrive.redrive` degrades. Passing empty headers is
+        equivalent to :meth:`publish`, so callers need not branch on transport capability.
+        """
+        target = queue or self.queue
+        envelope = EnvelopeCodec.make(urn, data, queue=target, trace_id=trace_id)
+        body = EnvelopeCodec.encode(envelope)
+        if headers and isinstance(self.transport, HeaderPublisher):
+            self.transport.publish_with_headers(target, body, dict(headers))
+        else:
+            self.transport.publish(target, body)
+        return envelope["meta"]["id"]
+
     # -- Register handlers --------------------------------------------------
 
     def handler(self, urn: str) -> Callable[[Handler], Handler]:
@@ -116,8 +148,16 @@ def consume(
     run = consume
 
     def dispatch(self, received: ReceivedMessage) -> None:
-        """Route one reserved message to its handler and acknowledge it."""
-        with _replay_scope(bool(received.headers.get(HEADER_REPLAY_BYPASS))):
+        """Route one reserved message to its handler and acknowledge it.
+
+        The delivered message's out-of-band transport headers are surfaced onto the context for
+        the span of this dispatch (:func:`~babelqueue.headers.headers_from_context`), so a handler
+        or an optional wrapper (e.g. the ``otel`` module reading a W3C ``traceparent``, ADR-0028)
+        can read metadata that travels beside the frozen envelope (GR-1).
+        """
+        with _headers_scope(received.headers), _replay_scope(
+            bool(received.headers.get(HEADER_REPLAY_BYPASS))
+        ):
             envelope = EnvelopeCodec.decode(received.body)
             urn = str(envelope.get("job") or envelope.get("urn") or "")
             handler = self._handlers.get(urn) if urn else None

src/babelqueue/headers.py

@@ -0,0 +1,70 @@
+"""Consume-side out-of-band transport headers (the Python mirror of Go's ``headers.go``).
+
+The runtime surfaces a delivered message's :attr:`~babelqueue.transport.ReceivedMessage.headers`
+onto a :class:`contextvars.ContextVar` for the span of one dispatch, so a handler — or an
+optional wrapper such as the ``otel`` module — can read per-message metadata that travels
+**beside** the frozen envelope (GR-1), never in it. It is the consume-side counterpart of
+:class:`~babelqueue.transport.HeaderPublisher`.
+
+This is the same out-of-band seam the replay-bypass marker rides (ADR-0027); ADR-0028's W3C
+``traceparent`` (for cross-hop span parent-child linkage) is the second rider on it. The header
+map is read-only — treat it as immutable. Adds only :mod:`contextvars` + a plain ``dict`` (no
+dependency), exactly like :mod:`babelqueue.replay`.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import contextvars
+from typing import Dict, Iterator, Mapping, Optional
+
+#: The delivered message's out-of-band transport headers, for the span of one dispatch.
+#: Defaults to an empty mapping so :func:`headers_from_context` is always nil-safe.
+_headers_var: contextvars.ContextVar[Mapping[str, str]] = contextvars.ContextVar(
+    "babelqueue_headers", default={}
+)
+
+
+def headers_from_context() -> Mapping[str, str]:
+    """Return the out-of-band transport headers that arrived with the message currently being
+    handled, or an empty mapping when none were carried (or the transport surfaces none).
+
+    The returned mapping is read-only — do not mutate it. It is the consume-side counterpart of
+    :class:`~babelqueue.transport.HeaderPublisher`: a handler or an optional wrapper (e.g. the
+    ``otel`` module's :func:`~babelqueue.otel.wrap_handler`) reads per-message metadata that
+    travels beside the frozen envelope, never in it (GR-1).
+    """
+    return _headers_var.get()
+
+
+@contextlib.contextmanager
+def _headers_scope(headers: Optional[Mapping[str, str]]) -> Iterator[None]:
+    """Internal: surface ``headers`` on the context for the span of one dispatch, then reset.
+
+    A nil/empty map is fine; reads stay nil-safe. The runtime calls this in
+    :meth:`~babelqueue.app.BabelQueue.dispatch` so wrappers can read the delivered headers.
+    """
+    scoped: Mapping[str, str] = headers or {}
+    token = _headers_var.set(scoped)
+    try:
+        yield
+    finally:
+        _headers_var.reset(token)
+
+
+def merge_headers(*sources: Optional[Mapping[str, str]]) -> Dict[str, str]:
+    """Combine header maps into a single ``dict[str, str]``, dropping blank keys and blank values.
+
+    Later sources win a key collision. Returns a fresh dict (callers may mutate it freely). Used
+    to merge an injected ``traceparent`` onto a transport's contract headers without clobbering
+    them — the contract keys are passed *last* so they win (mirrors the Go merge-not-clobber).
+    """
+    out: Dict[str, str] = {}
+    for source in sources:
+        if not source:
+            continue
+        for key, value in source.items():
+            if not key or value is None or value == "":
+                continue
+            out[str(key)] = str(value)
+    return out