feat(redis): dlq_channel — one shared DLQ stream, provenance on every entry

docs/docs/sdk/redis-transport.md

@@ -1,4 +1,4 @@
 # RedisTransport
 
 Redis Streams transport for EggAI — the recommended choice for most production workloads.
 
@@ -226,6 +226,61 @@
 !!! note "At-least-once applies to DLQ writes too"
     The `XADD` (to DLQ) and `XACK` are not atomic. A crash between them can produce duplicate DLQ entries — and fire `on_dlq` more than once — for the same logical message. Re-drive tooling and `on_dlq` callbacks should deduplicate using `_original_message_id`.
 
+#### Shared DLQ channel (`dlq_channel`)
+
+A DLQ stream is a plain Redis stream with the same envelope as any channel, so an eggai subscriber can consume it. What makes that awkward by default is the naming: one key per handler, derived from the agent name, the function name and a counter. Rename a handler and a listener silently watches an empty stream.
+
+`dlq_channel` replaces the per-handler key with one you choose, so every handler — across services, if they share the value — dead-letters into a single stream that one consumer can subscribe to:
+
+```python
+dlq = Channel("dlq", transport=transport)
+
+@agent.subscribe(channel=orders, retry_on_idle_ms=30_000, max_retries=3, dlq_channel=dlq)
+async def handle_order(message): ...
+
+@agent.subscribe(channel=payments, retry_on_idle_ms=30_000, max_retries=3, dlq_channel="dlq")
+async def handle_payment(message): ...
+```
+
+`dlq_channel` accepts a `Channel` or a topic name; a name is namespaced exactly like `Channel(name)`, so `"dlq"` becomes `<EGGAI_NAMESPACE>.dlq`. Only the terminal sink is shared — each handler keeps its own `.retry` stream (sharing it would re-introduce the cross-handler fan-out fixed in #225).
+
+The consumer is an ordinary subscription. Two things to get right:
+
+```python
+@ops_agent.subscribe(
+    channel=dlq,
+    group_start="0",   # new group starts at the beginning: failures that happened
+)                      # before this service booted are picked up, not skipped
+async def on_dead_letter(message):
+    # No data_type here: the shared stream carries every payload type.
+    # Dispatch on message["type"] or on the provenance below.
+    source  = message["_dlq_source"]     # e.g. "eggai.orders"
+    handler = message["_dlq_handler"]    # e.g. "order-service-handle_order-1"
+    ...
+```
+
+Do **not** give the sink `retry_on_idle_ms` together with the same `dlq_channel`: a handler dead-lettering into its own input is a loop, and `subscribe()` rejects it. If the sink needs retries, let it default to its own per-handler DLQ.
+
+**Provenance on every DLQ entry.** Because a shared stream's key no longer says where an entry came from, the reclaimer stamps it into the JSON body (body, not extra `XADD` fields: FastStream's decoder reads only `__data__`, so body keys are all a subscriber sees). This applies to per-handler DLQs too.
+
+| Field | Value |
+|-------|-------|
+| `_dlq_source` | full key of the channel the handler subscribed to |
+| `_dlq_handler` | handler suffix / consumer group |
+| `_dlq_at` | ISO-8601 UTC timestamp of the DLQ write |
+| `_dlq_reason` | `"max_retries"`, or `"poison"` for an unparseable envelope |
+| `_dlq_retries` | how many retries actually ran before giving up (`max_retries`; `"0"` for poison) |
+| `_retry_count` | **reset to `"0"`** on the DLQ write |
+| `_original_message_id` | as on retry delivery |
+
+Two rules make a DLQ entry safe to *consume*, not just inspect. `_retry_count` is reset so a DLQ consumer that has its own `retry_on_idle_ms` starts with a fresh budget — otherwise it would inherit an already-exceeded count, be dead-lettered again on its first transient failure and, with backoff, wait `base × multiplier^exhausted` for its first reclaim. And when that consumer does give up and dead-letters the entry into its own DLQ, `_dlq_source` / `_dlq_handler` / `_dlq_at` keep the original failure (set-if-absent) while `_dlq_reason` / `_dlq_retries` describe the latest hop, so neither piece of information is lost.
+
+**No `MAXLEN` on a shared DLQ.** `retry_max_len` caps the retry streams and per-handler DLQs, not a shared `dlq_channel`. `XADD MAXLEN` trims the whole stream regardless of which writer appended, so with several services writing to one DLQ the smallest cap among them would silently delete the others' unconsumed dead letters. A shared DLQ is written untrimmed by default; retention is the sink's responsibility (`XTRIM`, or a scheduled trim once entries are processed). `RedisTransport(dlq_max_len=...)` is the opt-in hard ceiling for operators who prefer bounded memory over the risk; use the same value on every writer to that DLQ.
+
+**Poison entries are wrapped.** An envelope the reclaimer cannot parse used to be copied to the DLQ verbatim; FastStream's parser then falls back to raw bytes, which a typed subscription silently skips and an untyped one cannot use. It is now written as a fresh envelope whose body holds the fields above plus the original bytes as `_dlq_raw_b64`, so every DLQ entry decodes to a JSON object. Re-drive tooling that handled raw poison entries should read `_dlq_raw_b64` instead.
+
+Validation: `dlq_channel` requires `retry_on_idle_ms` and a non-`None` `max_retries` (with `max_retries=None` there is no DLQ to redirect), must differ from the subscribed channel, and must not end in `.retry` (every `.retry` stream is auto-consumed by some handler, so dead-lettering into one would feed that handler's retry loop). All of it fails at `subscribe()` time, before anything is registered on the broker.
+
 ### Tuning the Reclaimer
 
 ```python

sdk/CHANGELOG.md

@@ -7,6 +7,89 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+**Migration note for DLQ consumers.** The *shape of DLQ entries* changes in
+this release; the SDK API, retry-stream entries, handler-side
+`_retry_count` / `_original_message_id`, stream key names and per-handler
+trimming do not. If you read DLQ entries yourself (`XRANGE`, re-drive
+scripts, monitoring) or read the dict passed to `on_dlq`:
+
+- read the exhausted retry budget from `_dlq_retries`, not `_retry_count`
+  (which is now `"0"` on every DLQ entry);
+- expect the additional `_dlq_source`, `_dlq_handler`, `_dlq_at`,
+  `_dlq_reason` keys;
+- poison entries are no longer raw bytes: recover the original payload by
+  base64-decoding `_dlq_raw_b64`, and `on_dlq` receives that decoded dict;
+- stream entries with no `__data__` field (only possible from a non-eggai
+  producer) are no longer retried at all; they are dead-lettered as poison on
+  the first reclaim.
+
+The `on_dlq(body, msg_id, retry_count)` signature and its `retry_count`
+argument are unchanged.
+
+### Added
+- `RedisTransport`: new `dlq_channel` subscribe option. Dead-letters into ONE
+  stream you name instead of the per-handler `{channel}.{handler_suffix}.dlq`,
+  so a single consumer can subscribe to every handler's (or every service's)
+  failures with a plain `@agent.subscribe(channel=dlq, group_start="0")`.
+  Accepts a `Channel` or a topic name (namespaced like `Channel(name)`) on
+  `Agent.subscribe` / `Channel.subscribe`; a full key on
+  `RedisTransport.subscribe`. Requires `retry_on_idle_ms` and a non-`None`
+  `max_retries`; rejects the subscribed channel and the handler's retry stream
+  as targets. The `.retry` stream stays per-handler (#225) — only the terminal
+  sink is shared. A shared DLQ is written without `MAXLEN` by default
+  (`retry_max_len` keeps applying to retry streams and per-handler DLQs):
+  stream-wide trimming would let one writer's cap delete other writers'
+  unconsumed dead letters. `RedisTransport(dlq_max_len=...)` is the opt-in
+  hard ceiling for a shared DLQ. Default behaviour without `dlq_channel` is
+  unchanged.
+- Every DLQ entry now carries provenance in its JSON body, alongside the
+  existing `_retry_count` / `_original_message_id`: `_dlq_source` (the channel
+  key the handler subscribed to), `_dlq_handler` (handler suffix / consumer
+  group), `_dlq_at` (ISO-8601 UTC) and `_dlq_reason` (`"max_retries"` or
+  `"poison"`) and `_dlq_retries` (how many retries actually ran). Additive:
+  typed models ignore the extra keys, as they already do for `_retry_count`.
+  On an entry dead-lettered a second time (by a DLQ consumer that gave up)
+  the origin keys `_dlq_source` / `_dlq_handler` / `_dlq_at` keep the first
+  failure, while `_dlq_reason` / `_dlq_retries` describe the latest hop.
+
+### Changed
+- **`_retry_count` is reset to `"0"` on the DLQ write** (the exhausted budget
+  moves to `_dlq_retries`). A DLQ entry is also a *first* delivery to whatever
+  consumes the DLQ; carrying the exceeded count over meant a DLQ consumer with
+  its own `retry_on_idle_ms` got zero retries and, with backoff, an escalated
+  first reclaim. Anything that read `_retry_count` off a DLQ entry should read
+  `_dlq_retries`; the `on_dlq` callback's `retry_count` argument is unchanged.
+- Poison messages (envelopes the reclaimer cannot parse) are no longer copied to
+  the DLQ verbatim. They are wrapped in a well-formed envelope whose body holds
+  the `_dlq_*` fields, `_original_message_id`, and the original bytes as
+  `_dlq_raw_b64`, so a DLQ subscriber always receives a decodable JSON object
+  instead of raw bytes. Envelope headers are preserved when only the body was
+  unusable. Re-drive scripts that handled raw poison entries should read
+  `_dlq_raw_b64`. Consequently `on_dlq` now receives that decoded dict for
+  poison entries too, instead of the raw `{b"__data__": bytes}` fields.
+- `Agent.subscribe` / `Channel.subscribe` reject a `dlq_channel` *string* that
+  already starts with `EGGAI_NAMESPACE` (e.g. a pasted `channel.get_name()`),
+  which would otherwise be namespaced twice and route dead letters to an
+  unwatched `<ns>.<ns>.dlq`. Pass the bare topic name or a `Channel`.
+- `RedisTransport.subscribe` rejects `retry_on_idle_ms` without a consumer
+  group (`handler_id=` / `group=`): the reclaimer works on a group's PEL, so
+  without one every reclaim cycle just errored. `Agent`/`Channel` always set a
+  group; this only affects direct transport callers.
+- `dlq_channel` rejects any key ending in `.retry`, not just the handler's own
+  retry stream: every `.retry` stream is auto-consumed by some handler, so
+  dead-lettering into one would feed that handler's retry loop. The check runs
+  before the broker subscriber is registered.
+
+### Fixed
+- `PendingReclaimer`: a pending entry with a parseable envelope but a non-object
+  JSON body (e.g. a published list), a non-integer `_retry_count`, or no
+  `__data__` field at all (non-eggai producer) escaped the poison handling. The
+  first two raised outside the guarded parse and aborted the whole reclaim
+  cycle after `XCLAIM`, head-of-line blocking every later pending entry; the
+  third ping-ponged between main and retry stream forever with a retry count
+  that could never grow. All three are now dead-lettered as poison; in
+  particular, entries without `__data__` are no longer retried at all.
+
 ## [0.4.1] - 2026-09-07
 
 ### Fixed

sdk/README.md

@@ -188,6 +188,59 @@ the per-handler `{channel}.{handler_suffix}.dlq`. The DLQ is terminal — no aut
 reclaimer. Set `max_retries=None` for unlimited retries. An optional `on_dlq` callback
 fires when a message lands in the DLQ.
 
+**Shared DLQ channel:** A DLQ stream is an ordinary stream, so an eggai subscriber can
+consume it — but with one key per handler there is nothing stable to subscribe to. Pass
+`dlq_channel` to dead-letter every handler (or every service) into **one** stream instead:
+
+```python
+dlq = Channel("dlq", transport=transport)
+
+@agent.subscribe(channel=orders, retry_on_idle_ms=30_000, max_retries=3, dlq_channel=dlq)
+async def handle_order(message): ...
+
+@agent.subscribe(channel=payments, retry_on_idle_ms=30_000, max_retries=3, dlq_channel="dlq")
+async def handle_payment(message): ...          # a topic name is namespaced like Channel("dlq")
+
+# One consumer for all of them — start the group at "0" to pick up failures that
+# happened before this service booted.
+@ops_agent.subscribe(channel=dlq, group_start="0")
+async def on_dead_letter(message):
+    print(message["_dlq_source"], message["_dlq_handler"], message["_dlq_retries"])
+```
+
+Only the terminal sink is shared; each handler keeps its own `.retry` stream (a shared
+retry stream would re-introduce the cross-handler fan-out of #225). `dlq_channel` requires
+`retry_on_idle_ms` and a non-`None` `max_retries`, and must not be the channel the handler
+subscribes to. Don't give the sink itself `retry_on_idle_ms` with the same `dlq_channel`.
+
+Every DLQ entry — shared or per-handler — carries provenance in its body so a shared
+consumer can tell entries apart:
+
+| Field | Value |
+|-------|-------|
+| `_dlq_source` | full key of the channel the handler subscribed to (e.g. `eggai.orders`) |
+| `_dlq_handler` | handler suffix / consumer group (e.g. `order-service-handle_order-1`) |
+| `_dlq_at` | ISO-8601 UTC timestamp of the DLQ write |
+| `_dlq_reason` | `"max_retries"`, or `"poison"` for an envelope the reclaimer could not parse |
+| `_dlq_retries` | how many retries actually ran before giving up (`max_retries`; `"0"` for poison) |
+| `_retry_count` | **reset to `"0"`** on the DLQ write, so a DLQ consumer with its own `retry_on_idle_ms` starts with a fresh budget |
+| `_original_message_id` | as on retry delivery |
+
+If a DLQ consumer itself gives up on an entry and dead-letters it again, `_dlq_source`,
+`_dlq_handler` and `_dlq_at` keep the *original* failure while `_dlq_reason` and
+`_dlq_retries` describe the latest hop.
+
+A shared DLQ is written **without** `MAXLEN` by default — `retry_max_len` applies to the
+retry streams and to per-handler DLQs only. `XADD MAXLEN` trims the whole stream regardless
+of which writer appended, so one service's cap would silently delete other services'
+unconsumed dead letters. Retention of a shared DLQ is the sink's job (`XTRIM`, or ack and
+trim on a schedule). Operators who prefer a hard memory ceiling can opt in with
+`RedisTransport(dlq_max_len=...)`; set the same value on every writer to that DLQ.
+
+A poison entry (unparseable envelope) is no longer copied to the DLQ verbatim: it is
+wrapped in a fresh envelope whose body holds the metadata above plus the original bytes
+as `_dlq_raw_b64`, so a DLQ subscriber always receives a decodable JSON object.
+
 **Retry backoff:** By default retries fire at a constant cadence equal to
 `retry_on_idle_ms`. Set `retry_backoff_multiplier > 1.0` to back off exponentially: a
 message is treated as due for reclaim once it has been idle for
@@ -238,8 +291,9 @@ Retry stream (eggai.orders.order-service-handle_order-1.retry)   ← per-handler
             _retry_count ≤ max_retries → XADD <retry stream> → XACK
             _retry_count > max_retries → XADD <dlq stream>   → XACK
 
-DLQ stream   (eggai.orders.order-service-handle_order-1.dlq)      ← per-handler
-        terminal — no reclaimer, manual re-drive only
+DLQ stream   (eggai.orders.order-service-handle_order-1.dlq)      ← per-handler by default,
+        terminal — no reclaimer                                      or one shared key via dlq_channel=
+        entries carry _dlq_source / _dlq_handler / _dlq_at / _dlq_reason
 ```
 
 ## Observability (OpenTelemetry)

sdk/eggai/agent.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable
 from typing import Any, TypeVar
 
-from .channel import Channel
+from .channel import Channel, resolve_dlq_channel
 from .hooks import eggai_register_stop
 from .transport import get_default_transport
 from .transport.base import Transport
@@ -68,7 +68,7 @@
             self._transport = get_default_transport()
         return self._transport
 
-    def subscribe(self, channel: Channel | None = None, **kwargs):
+    def subscribe(self, channel: Channel | None = None, **subscribe_kwargs):
         """
         Decorator for adding a subscription.
 
@@ -81,6 +81,11 @@
         """
 
         def decorator(handler: Callable[[dict[str, Any]], "asyncio.Future"]):
+            # Work on a copy: the decorator object may be applied to several
+            # handlers, and the resolution/popping below must not leak from one
+            # application into the next (e.g. re-namespacing an already-resolved
+            # dlq_channel into "<ns>.<ns>.dlq").
+            kwargs = dict(subscribe_kwargs)
             channel_name = (channel or Channel()).get_name()
             if "min_idle_time" in kwargs and "retry_on_idle_ms" in kwargs:
                 raise ValueError(
@@ -122,8 +127,41 @@
                         "idle threshold); a cap below the base would disable backoff "
                         "entirely."
                     )
+            # Plugins see the kwargs as the caller wrote them (before the DLQ
+            # key below is resolved to its namespaced form).
             original_kwargs = kwargs.copy()
 
+            # Shared DLQ: resolve Channel / topic name to a full key here (the
+            # transport only knows keys) and fail at decoration time, like the
+            # other retry knobs, so a typo can't silently no-op.
+            if kwargs.get("dlq_channel") is not None:
+                if "retry_on_idle_ms" not in kwargs:
+                    raise ValueError(
+                        "dlq_channel requires retry_on_idle_ms to be set. "
+                        "Set retry_on_idle_ms to enable SDK-managed retries with a DLQ."
+                    )
+                if "max_retries" in kwargs and kwargs["max_retries"] is None:
+                    raise ValueError(
+                        "dlq_channel requires max_retries: max_retries=None disables "
+                        "the DLQ entirely, so there is nothing to route to the shared "
+                        "channel."
+                    )
+                dlq_key = resolve_dlq_channel(kwargs["dlq_channel"])
+                if dlq_key is None:  # unreachable: input checked non-None above
+                    raise ValueError("dlq_channel could not be resolved to a key")
+                if dlq_key == channel_name:
+                    raise ValueError(
+                        f"dlq_channel {dlq_key!r} is the channel this handler "
+                        "subscribes to; dead-lettering into its own input would loop."
+                    )
+                if dlq_key.endswith(".retry"):
+                    raise ValueError(
+                        f"dlq_channel {dlq_key!r} names an SDK-managed retry stream "
+                        "('.retry' suffix); dead-lettering into a retry stream would "
+                        "feed a handler's retry loop. Pick a different name."
+                    )
+                kwargs["dlq_channel"] = dlq_key
+
             # Extract plugin-specific kwargs dynamically and clean them from kwargs
             plugin_found_keys = set()
             for plugin_name in PLUGIN_LIST: