feat(redis): opt-in delete_on_ack to XDEL entries when they are acked

docs/docs/sdk/redis-transport.md

@@ -281,6 +281,33 @@ Two rules make a DLQ entry safe to *consume*, not just inspect. `_retry_count` i
 
 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.
 
+### Freeing Memory with `delete_on_ack`
+
+Redis keeps a stream entry after it is acked; only `MAXLEN` trims it, and `MAXLEN` trims by count whether or not an entry was processed. With `delete_on_ack=True` a subscription `XDEL`s every entry at the moment it acks it (`XACK` + `XDEL` in one `MULTI`), so a stream only holds in-flight messages:
+
+```python
+@agent.subscribe(channel=emails, delete_on_ack=True)
+async def handle_email(message):
+    ...
+```
+
+It applies wherever the subscription acks: after the handler returns (including messages a filter skips), on the SDK retry stream, and when the reclaimer moves a stuck entry to `.retry` or the DLQ. Under the default `NACK_ON_ERROR` a failing handler's entry stays in the PEL for retry; under `AckPolicy.ACK` / `ACK_FIRST`, which ack failures too, it is deleted too. DLQ entries are never deleted by this option, and a poison entry dropped with no DLQ configured is only acked so it stays readable with `XRANGE`.
+
+**Only one consumer group per stream.** `XDEL` removes the entry for every group, so a second group (or a group-less subscriber) that has not read an entry yet never will. Competing workers inside one group are fine. For fan-out to several groups, keep `delete_on_ack` off and bound memory with `max_len`. Requires a consumer group; rejected with `no_ack=True` and `AckPolicy.MANUAL`. On Redis >= 8.2 the equivalent command is `XACKDEL`.
+
+How the one-group rule is enforced:
+
+- **At `connect()`**, before any consumer starts: another group on the stream raises `RuntimeError` naming it.
+- **At runtime**, the group monitor (every `group_monitor_interval_s`) checks again. If a group has appeared, it turns deletion **off** for that stream (plain `XACK`, entries kept) and logs an error. Memory grows again, but the other group only misses what was deleted during one monitor interval. Deletion stays off until restart, and the next `connect()` refuses to start until the extra group is gone.
+- **Renaming a handler renames its group.** `Agent` names groups `<agent>-<handler function>-<n>`, so after renaming a handler function the old group is still on the stream and every pod refuses to start. Remove it with `XGROUP DESTROY <stream> <old-group>`, and on the old `.retry` stream `XGROUP DESTROY <stream>.<old-group>.retry <old-group>-retry`.
+
+Two consequences to plan for:
+
+- **Keep `max_len` as a backstop.** A few paths still leave acked entries behind: a delete that fails (it is logged, not retried, so a successful handler isn't run again), deletion turned off by the monitor, and a crash mid-handler under `ACK_FIRST`. A generous `max_len` next to `delete_on_ack` bounds them.
+- **No replay.** Deleted entries are gone. Rewinding a group (`XGROUP SETID <stream> <group> 0`) or adding a group to backfill history finds nothing. Runbooks that rely on rewinds need another source.
+
+Tested against Redis 7.4 (open source). Redis Enterprise Active-Active / geo-replicated Azure Managed Redis has not been tested; replicated consumer-group state may behave differently there.
+
 ### Tuning the Reclaimer
 
 ```python

sdk/CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Redis `subscribe(..., delete_on_ack=True)`: every entry the subscription
+  acks is also `XDEL`'d (same `MULTI`), so consumed messages stop occupying
+  Redis memory instead of waiting for `MAXLEN` trimming. Covers handler
+  success (including filtered-out messages), the retry stream, and the
+  reclaimer's moves to the retry stream / DLQ. A failing handler's entry
+  stays in the PEL under the default `NACK_ON_ERROR`, and is deleted under
+  `AckPolicy.ACK` / `ACK_FIRST` (which ack it anyway). Opt-in because it
+  assumes a single consumer group per stream: `XDEL` removes the entry for
+  every group. `connect()` raises `RuntimeError` if another group already
+  reads the stream; if one joins later, the group monitor turns deletion off
+  for that stream (plain `XACK`) and logs an error.
+
 ### Fixed
 
 - MCP adapter reads tool schemas from fastmcp's own `Tool` instead of the

sdk/README.md

@@ -241,6 +241,18 @@ A poison entry (unparseable envelope) is no longer copied to the DLQ verbatim: i
 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.
 
+**Freeing memory with `delete_on_ack`:** acked stream entries stay in Redis until `MAXLEN`
+trims them, and `MAXLEN` trims by count regardless of whether an entry was processed.
+`subscribe(..., delete_on_ack=True)` `XDEL`s each entry when it is acked (`XACK` + `XDEL`
+in one `MULTI`): after the handler returns, on the retry stream, and when the reclaimer
+moves an entry to `.retry` or the DLQ. A failing handler's entry still stays in the PEL
+for retry (under the default `NACK_ON_ERROR`). **It assumes one consumer group per
+stream:** `XDEL` removes the entry for every group, so don't use it on a stream that
+fans out to several groups. `connect()` raises if another group already reads the stream;
+if one joins later, the group monitor turns deletion off for that stream and logs an error.
+Deleted entries can't be replayed, and a generous `max_len` is still worth keeping as a
+backstop for the few paths that leave acked entries behind.
+
 **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

sdk/eggai/transport/pending_reclaimer.py

@@ -81,6 +81,22 @@
     # stream key no longer does.
     source_stream: str | None = None
     handler: str | None = None
+    # XDEL the original entry together with its XACK once it has been moved to the
+    # retry stream or DLQ (the subscription's delete_on_ack). Assumes this group is
+    # the only consumer of `stream`.
+    delete_on_ack: bool = False
+
+
+async def xack_del(client: Any, stream: str, group: str, *ids: Any) -> None:
+    """XACK and XDEL ``ids`` in one MULTI (the ``delete_on_ack`` primitive).
+
+    Both commands touch the same key, so this is safe on Redis Cluster /
+    Enterprise too. On Redis >= 8.2 this is ``XACKDEL``; swap it here.
+    """
+    async with client.pipeline(transaction=True) as pipe:
+        pipe.xack(stream, group, *ids)
+        pipe.xdel(stream, *ids)
+        await pipe.execute()
 
 
 def _encode_envelope(headers: dict, body_bytes: bytes) -> bytes:
@@ -263,16 +279,27 @@
       1. Pages through XPENDING to find entries idle longer than min_idle_ms.
       2. XCLAIM them under a dedicated reclaimer consumer.
       3. XADD the fields to retry_stream (a separate stream — avoids duplicates).
-      4. XACK the original PEL entry.
+      4. XACK the original PEL entry (plus XDEL with delete_on_ack).
 
     Delivery guarantee: at-least-once. XADD and XACK are not atomic; a crash
     between them will re-deliver the message on the next reclaim cycle.
     Handlers must be idempotent. The injected _original_message_id field (in the
     message body) can be used for application-level deduplication.
     """
 
-    def __init__(self, redis_url: str, connection_kwargs: dict[str, Any] | None = None):
+    def __init__(
+        self,
+        redis_url: str,
+        connection_kwargs: dict[str, Any] | None = None,
+        delete_on_ack_disabled: set[str] | None = None,
+    ):
         self._redis_url = redis_url
+        # Stream keys where delete_on_ack was turned off at runtime (another
+        # consumer group appeared). Shared with, and mutated by, the transport's
+        # group monitor; _ack() falls back to a plain XACK for these.
+        self._delete_on_ack_disabled: set[str] = (
+            delete_on_ack_disabled if delete_on_ack_disabled is not None else set()
+        )
         # Connection-resilience settings (socket_timeout, socket_keepalive,
         # health_check_interval, retry_on_timeout, …) forwarded from the
         # transport so this independent client recovers from a silently dropped
@@ -311,7 +338,7 @@
         # decode_responses is pinned here and must not be overridden by callers.
         self._redis_client = aioredis.from_url(
             self._redis_url,
             **{**self._connection_kwargs, "decode_responses": False},
         )
         self._running = True
         for key, config in self._configs.items():
@@ -500,7 +527,7 @@
             stale_ids: list[Any] = [msg_id for msg_id, _ in candidates]
         else:
             stale_ids = []
             for msg_id, idle in candidates:
                 retry_count = await self._read_retry_count(config.stream, msg_id)
                 if idle >= self._effective_idle_ms(config, retry_count):
                     stale_ids.append(msg_id)
@@ -549,7 +576,7 @@
                     dlq_fields = dict(fields)
                     dlq_fields[data_key] = _encode_body(headers, body)
                     await self._xadd(config.dlq_stream, dlq_fields, config.dlq_max_len)
-                    await self._client.xack(config.stream, config.group, msg_id)
+                    await self._ack(config, msg_id)
                     logger.warning(
                         "Message %s has an unparseable envelope; moved to DLQ %s "
                         "(retry count cannot be tracked)",
@@ -558,7 +585,9 @@
                     )
                     await self._invoke_on_dlq(config, body, msg_id_str, 0)
                 else:
+                    # Plain XACK even with delete_on_ack: with no DLQ this entry
+                    # is the only copy, keep it for XRANGE forensics.
                     await self._client.xack(config.stream, config.group, msg_id)
                     logger.error(
                         "Message %s has an unparseable envelope and no DLQ is "
                         "configured; dropping it to avoid a retry-stream livelock",
@@ -582,7 +611,7 @@
                 dlq_fields = dict(fields)
                 dlq_fields[data_key] = _encode_body(headers, body)
                 await self._xadd(config.dlq_stream, dlq_fields, config.dlq_max_len)
-                await self._client.xack(config.stream, config.group, msg_id)
+                await self._ack(config, msg_id)
                 logger.warning(
                     "Message %s exceeded max_retries=%d; moved to DLQ %s",
                     msg_id_str,
@@ -593,9 +622,15 @@
             else:
                 fields[data_key] = _encode_body(headers, body)
                 await self._xadd(config.retry_stream, fields, config.max_len)
-                await self._client.xack(config.stream, config.group, msg_id)
+                await self._ack(config, msg_id)
                 logger.debug("Reclaimed %s → %s", msg_id_str, config.retry_stream)
 
+    async def _ack(self, config: ReclaimerConfig, msg_id: Any) -> None:
+        if config.delete_on_ack and config.stream not in self._delete_on_ack_disabled:
+            await xack_del(self._client, config.stream, config.group, msg_id)
+        else:
+            await self._client.xack(config.stream, config.group, msg_id)
+
     async def _invoke_on_dlq(
         self,
         config: ReclaimerConfig,