@@ -9,6 +9,40 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+### Added
+- **`ReceivedMessage.route_urn` routing seam** (MINOR). An optional field a transport can set
+  when its broker carries the URN out of band. When non-empty `BabelQueue.dispatch()` looks the
+  handler up by it instead of the decoded body's URN; empty (the default for every existing
+  transport) keeps routing on the body. The body is never rewritten, so the handler, the retry
+  re-publish and the dead-letter record carry the producer's original body (including a `urn`
+  alias). No wire change.
+- **`RoutedPublisher` optional transport capability** (MINOR). `publish_routed(queue, body,
+  route_urn)`. When a delivery has a `route_urn` and the transport implements it, the retry, release and dead-letter
+  writes, and `redrive`, use it so the broker's routing property keeps the original
+  value across attempts (no flip to the body URN on the second delivery). `PulsarTransport`
+  implements it (`bq-job` pinned; blank derives from the body). Others keep plain `publish`.
+  No wire change.
+
+### Changed
+- `BabelQueue.dispatch()` applies `route_urn` only to a body that decodes; an undecodable body
+  stays on the poison / unknown-URN path instead of being routed to a handler.
+
+### Fixed
+- **Pulsar consumer now follows broker-bindings §5.2 / §5.7.** `PulsarTransport.pop()` routes on
+  the `bq-job` message property (set on `ReceivedMessage.route_urn`; the body's URN is used only
+  when the property is absent or blank) and takes `attempts` from the authoritative
+  `bq-attempts` property (the body's value is used only when the property is absent or not a
+  plain ASCII-decimal integer that fits a 32-bit int), then applies `max(.., redelivery_count)`.
+  Previously only the body was read. The body's `job` / `urn` is **not** rewritten, so a
+  property/body mismatch no longer leaks into retries or the DLQ; `attempts` is written into the
+  body only when the body already holds an integer `attempts` and the value changes (a missing
+  or non-integer `attempts` is still rejected by `EnvelopeCodec.accepts`). When the property and
+  the body name different URNs the property wins and the new optional
+  `on_routing_mismatch(property_urn, body_urn, envelope)` constructor hook is called for an
+  accepted envelope, after attempts are reconciled (its exceptions are swallowed; it is separate
+  from error handling). A body without a URN is still rejected. No wire change
+  (`schema_version: 1`).
+
 ## [1.14.0] - 2026-10-03
 
 ### Added
 
@@ -29,7 +29,14 @@
 from .exceptions import BabelQueueError, DecryptError, UnknownUrnError
 from .replay import HEADER_REPLAY_BYPASS, bypass_external_effects, is_replay
 from .routing import UnknownUrnStrategy
-from .transport import HeaderPublisher, InMemoryTransport, ReceivedMessage, Redeliverer, Transport
+from .transport import (
+    HeaderPublisher,
+    InMemoryTransport,
+    ReceivedMessage,
+    Redeliverer,
+    RoutedPublisher,
+    Transport,
+)
 
 __version__ = "1.14.0"
 
@@ -46,6 +53,7 @@
     "ReceivedMessage",
     "HeaderPublisher",
     "Redeliverer",
+    "RoutedPublisher",
     "BabelQueueError",
     "UnknownUrnError",
     "DecryptError",
 
@@ -45,6 +45,7 @@ def on_order_created(data, meta):
     Redeliverer,
     Transport,
     make_transport,
+    publish_keeping_route,
 )
 
 Handler = Callable[..., None]
@@ -278,6 +279,10 @@ def dispatch(self, received: ReceivedMessage) -> None:
         ):
             envelope = EnvelopeCodec.decode(received.body)
             urn = str(envelope.get("job") or envelope.get("urn") or "")
+            if received.route_urn and envelope:
+                # Transport-supplied routing override; body untouched. Only for a body that
+                # decodes: an undecodable one (decode() yields {}) stays on the poison path.
+                urn = received.route_urn
             handler = self._handlers.get(urn) if urn else None
 
             try:
@@ -381,9 +386,16 @@ def _release(self, received: ReceivedMessage, body: str, delay: float) -> None:
         if isinstance(self.transport, Redeliverer):
             self.transport.redeliver(received, body, delay)
             return
-        self.transport.publish(received.queue, _without_forbidden_keys(body))
+        self._publish_for(received, received.queue, _without_forbidden_keys(body))
         self._ack_settled(received)
 
+    def _publish_for(self, received: ReceivedMessage, queue: str, body: str) -> None:
+        """Publish ``body`` to ``queue`` on behalf of ``received`` (a retry / release re-publish
+        or its dead-letter record). Keeps the routing URN the broker property carried on this
+        delivery when the transport can pin it (:class:`RoutedPublisher`), so a property/body
+        mismatch does not flip to the body URN on the next attempt or in the DLQ."""
+        publish_keeping_route(self.transport, queue, body, received.route_urn)
+
     def _ack_settled(self, received: ReceivedMessage) -> None:
         """Acknowledge a message whose outcome is already final (handled, dead-lettered,
         re-published, dropped or deleted by the unknown-URN strategy).
@@ -474,7 +486,7 @@ def _dead_letter(
             exception=(type(exc).__name__ if exc is not None else None),
         )
         target = self.dead_letter_queue or (received.queue + self.dead_letter_suffix)
-        self.transport.publish(target, EnvelopeCodec.encode(annotated))
+        self._publish_for(received, target, EnvelopeCodec.encode(annotated))
         self._ack_settled(received)
 
 
 
@@ -7,11 +7,17 @@
 ``bq-trace-id`` = trace_id, ``bq-message-id`` = meta.id, plus ``bq-schema-version`` /
 ``bq-source-lang`` / ``bq-attempts`` — so a Java/.NET/... peer can route on ``bq-job``
 without parsing the body. Consuming receives one message at a time
-(``receive`` -> process -> ``acknowledge``); the authoritative attempt count is the
-envelope's ``bq-attempts`` (the body), cross-checked against the broker's native
-``redelivery_count`` (0-based) and reconciled onto the envelope as
-``attempts = max(body, redelivery_count)`` — no -1, because Pulsar's redelivery count is
-0-based (0 on first delivery). A message left un-acked is redelivered (at-least-once).
+(``receive`` -> process -> ``acknowledge``). Per §5.2 / §5.7 the ``bq-job`` property is the
+routing key (the body's URN is used only when ``bq-job`` is absent or blank) and the
+``bq-attempts`` property is the authoritative attempt count (the body's ``attempts`` is used
+only when the property is absent or not a plain non-negative ASCII-decimal integer that fits
+a 32-bit int). The result is then raised to the broker's native ``redelivery_count``
+(0-based): ``attempts = max(bq-attempts, redelivery_count)`` — no -1, because Pulsar's
+redelivery count is 0-based (0 on first delivery). The routing URN is handed to the app through
+``ReceivedMessage.route_urn`` — the body's ``job`` / ``urn`` is never rewritten, so retries and
+the DLQ keep the producer's original body. When ``bq-job`` and the body name different URNs
+the property wins and the optional ``on_routing_mismatch(property_urn, body_urn, envelope)``
+hook is called (for an accepted envelope, after attempts are reconciled). A message left un-acked is redelivered (at-least-once).
 
 This implements §5 of the broker-bindings contract. The envelope is unchanged
 (``schema_version`` stays 1); Apache Pulsar is purely additive.
@@ -24,11 +30,17 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Optional
+import re
+from typing import Any, Callable, Dict, Optional, Tuple
 
 from .codec import EnvelopeCodec, parse_envelope
 from .transport import ReceivedMessage, Transport
 
+#: ``bq-attempts`` is valid only as a run of ASCII digits (no sign, no whitespace) ...
+_ATTEMPTS_PATTERN = re.compile(r"[0-9]+")
+#: ... that fits a signed 32-bit int, matching the Java and .NET consumers (GR-5).
+_MAX_ATTEMPTS = 2**31 - 1
+
 
 class PulsarTransport(Transport):
     def __init__(
@@ -40,6 +52,7 @@ def __init__(
         consumer_type: Any = None,
         topic_prefix: str = "",
         receive_timeout_millis: int = 1000,
+        on_routing_mismatch: Optional[Callable[[str, str, Dict[str, Any]], None]] = None,
         **subscribe_options: Any,
     ) -> None:
         self._url = url or "pulsar://localhost:6650"
@@ -48,6 +61,7 @@ def __init__(
         self._topic_prefix = topic_prefix
         self._receive_timeout_millis = receive_timeout_millis
         self._subscribe_options = subscribe_options
+        self._on_routing_mismatch = on_routing_mismatch
         self._producers: Dict[str, Any] = {}
         self._consumers: Dict[str, Any] = {}
 
@@ -105,9 +119,12 @@ def _default_shared_type() -> Any:
         return pulsar.ConsumerType.Shared  # pragma: no cover - needs Pulsar
 
     @staticmethod
-    def _projection(body: str) -> Dict[str, str]:
+    def _projection(body: str, route_urn: str = "") -> Dict[str, str]:
         """Native Pulsar message properties (string->string) — a redundant, routable view of
-        the body: bq-job/bq-trace-id/bq-message-id + bq-schema-version/lang/attempts. §5.2."""
+        the body: bq-job/bq-trace-id/bq-message-id + bq-schema-version/lang/attempts. §5.2.
+
+        A non-empty ``route_urn`` pins ``bq-job`` (a retry / release re-publish keeps the original
+        delivery's routing URN); blank derives it from the body."""
         env = parse_envelope(body)
         if not env:
             return {}
@@ -125,25 +142,80 @@ def _projection(body: str) -> Dict[str, str]:
         if meta.get("lang"):
             props["bq-source-lang"] = str(meta["lang"])
         props["bq-attempts"] = str(int(env.get("attempts", 0) or 0))
+        if route_urn:
+            props["bq-job"] = route_urn
         return props
 
     @staticmethod
-    def _reconcile(body: str, redelivery_count: Any) -> str:
-        """Set attempts to max(current, redelivery_count). Pulsar's redelivery count is
-        0-based (0 on first delivery), so it maps directly to attempts with no -1. The
-        runtime retries by republishing with attempts+1 in the body (redelivery count back to
-        0), so a republished message must not have its higher body count lowered."""
+    def _parse_attempts(raw: Any) -> Optional[int]:
+        """The ``bq-attempts`` value as an int, or ``None`` when absent or invalid (§5.5)."""
+        if not isinstance(raw, str) or _ATTEMPTS_PATTERN.fullmatch(raw) is None:
+            return None
+        value = int(raw)
+        return value if value <= _MAX_ATTEMPTS else None
+
+    @staticmethod
+    def _reconcile(
+        body: str,
+        redelivery_count: Any,
+        properties: Optional[Dict[str, str]] = None,
+        on_routing_mismatch: Optional[Callable[[str, str, Dict[str, Any]], None]] = None,
+    ) -> Tuple[str, str]:
+        """Apply §5.2 / §5.7 to a received body; returns ``(body, route_urn)``.
+
+        * Routing: a non-blank ``bq-job`` property is the URN the app routes on. It is returned
+          as ``route_urn`` (set on :attr:`ReceivedMessage.route_urn`); the body's ``job`` /
+          ``urn`` is never rewritten, so the handler, retries and the dead-letter record keep the
+          producer's original body. A body with no URN gets no ``route_urn`` — it stays
+          non-conformant and is rejected whatever ``bq-job`` says.
+        * Attempts: a valid ``bq-attempts`` property replaces the body's value (the body's is
+          kept when it is absent or invalid), then the result is raised to
+          ``redelivery_count``. Pulsar's count is 0-based (0 on first delivery), so it maps
+          directly with no -1; the max never lowers a runtime-incremented counter (the runtime
+          retries by republishing with attempts+1, which resets the redelivery count to 0). The
+          value is written into the body only when the body already holds an integer
+          ``attempts`` and the value really changes, so a missing / non-integer ``attempts`` is
+          still rejected by ``EnvelopeCodec.accepts``.
+        * ``on_routing_mismatch(property_urn, body_urn, envelope)`` is called when the property
+          and the body name different URNs — only for an envelope the consumer accepts and after
+          attempts are reconciled (as in the Java and .NET consumers), so it sees what the
+          handler will get. The hook is optional, separate from any error handling, and any
+          exception it raises is swallowed — it never affects the message flow.
+        """
+        properties = properties or {}
+        env = parse_envelope(body)
+        if not env:
+            return body, ""
+
+        routed = str(properties.get("bq-job") or "").strip()
+        body_urn = EnvelopeCodec.urn(env)
+        route_urn = routed if routed and body_urn else ""
+
+        current = env.get("attempts")
+        has_int_attempts = isinstance(current, int) and not isinstance(current, bool)
+        attempts = current if has_int_attempts else 0
+        from_property = PulsarTransport._parse_attempts(properties.get("bq-attempts"))
+        if from_property is not None:
+            attempts = from_property
         try:
             rc = int(redelivery_count)
-        except (ValueError, TypeError):
-            return body
-        if rc <= 0:
-            return body
-        env = EnvelopeCodec.decode(body)
-        if not env or rc <= int(env.get("attempts", 0) or 0):
-            return body
-        env["attempts"] = rc
-        return EnvelopeCodec.encode(env)
+        except (TypeError, ValueError):
+            rc = 0
+        if rc > attempts:
+            attempts = rc
+
+        out = body
+        if has_int_attempts and current != attempts:
+            env["attempts"] = attempts
+            out = EnvelopeCodec.encode(env)
+
+        if route_urn and route_urn != body_urn and on_routing_mismatch is not None:
+            if EnvelopeCodec.accepts(env):
+                try:
+                    on_routing_mismatch(route_urn, body_urn, dict(env))
+                except Exception:  # noqa: BLE001 - a diagnostic hook must never break consumption
+                    pass
+        return out, route_urn
 
     @staticmethod
     def _redelivery_count(message: Any) -> int:
@@ -156,6 +228,12 @@ def _redelivery_count(message: Any) -> int:
         except (TypeError, ValueError):
             return 0
 
+    @staticmethod
+    def _properties(message: Any) -> Dict[str, str]:
+        getter = getattr(message, "properties", None)
+        value = getter() if callable(getter) else getter
+        return dict(value) if isinstance(value, dict) else {}
+
     @staticmethod
     def _payload(message: Any) -> str:
         getter = getattr(message, "data", None)
@@ -167,7 +245,14 @@ def _payload(message: Any) -> str:
     # -- Transport ----------------------------------------------------------
 
     def publish(self, queue: str, body: str) -> None:
-        self._producer(queue).send(body.encode("utf-8"), properties=self._projection(body))
+        self.publish_routed(queue, body, "")
+
+    def publish_routed(self, queue: str, body: str, route_urn: str) -> None:
+        """:meth:`publish` with ``bq-job`` pinned to ``route_urn`` (:class:`RoutedPublisher`);
+        a blank ``route_urn`` derives it from the body."""
+        self._producer(queue).send(
+            body.encode("utf-8"), properties=self._projection(body, route_urn)
+        )
 
     def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]:
         wait = self._receive_timeout_millis
@@ -182,8 +267,13 @@ def pop(self, queue: str, timeout: float = 1.0) -> Optional[ReceivedMessage]:
             raise
         if message is None:  # pragma: no cover - defensive (real client raises instead)
             return None
-        body = self._reconcile(self._payload(message), self._redelivery_count(message))
-        return ReceivedMessage(body=body, queue=queue, handle=message)
+        body, route_urn = self._reconcile(
+            self._payload(message),
+            self._redelivery_count(message),
+            self._properties(message),
+            self._on_routing_mismatch,
+        )
+        return ReceivedMessage(body=body, queue=queue, handle=message, route_urn=route_urn)
 
     def ack(self, message: ReceivedMessage) -> None:
         if message.handle is None:
 
@@ -36,7 +36,7 @@
 
 from .codec import EnvelopeCodec
 from .replay import HEADER_REPLAY_BYPASS
-from .transport import HeaderPublisher, ReceivedMessage, Transport
+from .transport import HeaderPublisher, ReceivedMessage, Transport, publish_keeping_route
 
 Envelope = Mapping[str, Any]
 Select = Callable[[Envelope], bool]
@@ -88,7 +88,7 @@ def redrive(
     result = RedriveResult()
     for message, envelope in batch:
         if envelope is None:
-            transport.publish(dlq, message.body)  # restore the poison body; never drop it
+            publish_keeping_route(transport, dlq, message.body, message.route_urn)  # restore the poison body; never drop it
             transport.ack(message)
             result.skipped += 1
             result.items.append(RedriveItem("", "", "", "", dlq, "", False))
@@ -109,7 +109,7 @@ def redrive(
         )
 
         if select is not None and not select(envelope):
-            transport.publish(dlq, message.body)  # not selected: restore unchanged
+            publish_keeping_route(transport, dlq, message.body, message.route_urn)  # not selected: restore unchanged
             transport.ack(message)
             result.skipped += 1
             result.items.append(item)
@@ -119,7 +119,7 @@ def redrive(
         item.to = target
 
         if dry_run:
-            transport.publish(dlq, message.body)  # report the plan; restore unchanged
+            publish_keeping_route(transport, dlq, message.body, message.route_urn)  # report the plan; restore unchanged
             transport.ack(message)
             result.skipped += 1
             result.items.append(item)
@@ -130,10 +130,10 @@ def redrive(
         reset["attempts"] = 0
         try:
             item.bypassed = _publish_redriven(
-                transport, target, EnvelopeCodec.encode(reset), bypass
+                transport, target, EnvelopeCodec.encode(reset), bypass, message.route_urn
             )
         except Exception:
-            transport.publish(dlq, message.body)  # restore on a publish failure, then surface
+            publish_keeping_route(transport, dlq, message.body, message.route_urn)  # restore on a publish failure, then surface
             transport.ack(message)
             raise
         transport.ack(message)
@@ -156,14 +156,18 @@ def _decoded(body: str) -> Optional[Dict[str, Any]]:
     return envelope
 
 
-def _publish_redriven(transport: Transport, queue: str, body: str, bypass: bool) -> bool:
+def _publish_redriven(
+    transport: Transport, queue: str, body: str, bypass: bool, route_urn: str = ""
+) -> bool:
     """Re-publish a reset message to ``queue``. When ``bypass`` is set and the transport is a
     :class:`~babelqueue.transport.HeaderPublisher`, stamp the ``bq-replay-bypass`` header and
-    return True; otherwise publish plainly and return False."""
+    return True; otherwise publish plainly and return False. Either way a non-empty ``route_urn``
+    (the dead-lettered message's routing property) is carried back through
+    :class:`~babelqueue.transport.RoutedPublisher` when the transport implements it."""
     if bypass and isinstance(transport, HeaderPublisher):
         transport.publish_with_headers(queue, body, {HEADER_REPLAY_BYPASS: "1"})
         return True
-    transport.publish(queue, body)
+    publish_keeping_route(transport, queue, body, route_urn)
     return False