feat(redis): lease renewal for in-flight stream entries

.github/workflows/ci.yaml

@@ -1,4 +1,4 @@
 name: SDK CI
 
 on:
   push:
@@ -195,6 +195,64 @@
           name: python-${{ matrix.python-version }}
           fail_ci_if_error: false
 
+  # The lock file pins one FastStream release, so the `test` job never sees the
+  # bottom of the range pyproject.toml allows. This job runs the suite against
+  # the lowest supported FastStream, on the lowest supported Python. Keep the
+  # version here in step with the lower bound in sdk/pyproject.toml.
+  test-min-faststream:
+    runs-on: ubuntu-latest
+    name: Test SDK (lowest FastStream, Python 3.10)
+    env:
+      MIN_FASTSTREAM: '0.6.4'
+
+    services:
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Set up Python 3.10
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: '3.10'
+
+      - name: Cache Poetry dependencies
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0
+        with:
+          path: |
+            ~/.cache/pypoetry
+            ~/.cache/pip
+          # Own prefix: this job's venv has FastStream 0.6.4 installed over the
+          # lock, which must not be restored into the regular test jobs.
+          key: poetry-min-faststream-${{ runner.os }}-py3.10-${{ hashFiles('sdk/poetry.lock') }}
+          restore-keys: |
+            poetry-min-faststream-${{ runner.os }}-py3.10-
+
+      - name: Install Poetry
+        run: pipx install poetry==2.4.3
+
+      - name: Copy README to SDK
+        run: cp README.md sdk/
+
+      - name: Install dependencies
+        working-directory: sdk
+        run: |
+          poetry install --all-extras
+          poetry run pip install "faststream[kafka,redis]==${MIN_FASTSTREAM}"
+          poetry run python -c "from importlib.metadata import version; v = version('faststream'); assert v == '${MIN_FASTSTREAM}', v"
+
+      - name: Run tests
+        working-directory: sdk
+        run: poetry run pytest -v
+
   build:
     runs-on: ubuntu-latest
     name: Build and verify SDK package
@@ -263,7 +321,7 @@
   # Summary job to check if all required jobs passed
   all-checks-passed:
     runs-on: ubuntu-latest
-    needs: [lint, changelog, test, build]
+    needs: [lint, changelog, test, test-min-faststream, build]
     if: always()
 
     steps:
@@ -272,6 +330,7 @@
           CHANGELOG_RESULT: ${{ needs.changelog.result }}
           LINT_RESULT: ${{ needs.lint.result }}
           TEST_RESULT: ${{ needs.test.result }}
+          TEST_MIN_FASTSTREAM_RESULT: ${{ needs.test-min-faststream.result }}
           BUILD_RESULT: ${{ needs.build.result }}
         run: |
           # Changelog is optional (skipped on push to main or when no code changes)
@@ -282,6 +341,7 @@
 
           if [[ "$LINT_RESULT" != "success" || \
                 "$TEST_RESULT" != "success" || \
+                "$TEST_MIN_FASTSTREAM_RESULT" != "success" || \
                 "$BUILD_RESULT" != "success" ]]; then
             echo "One or more required checks failed"
             exit 1

docs/docs/sdk/redis-transport.md

@@ -323,10 +323,49 @@ async def handle_order(message):
 
 Guidelines:
 
-- `retry_on_idle_ms` should be comfortably longer than your handler's expected worst-case execution time to avoid false positives.
+- `retry_on_idle_ms` should be comfortably longer than your handler's expected worst-case execution time to avoid false positives — or enable [`renew_lease`](#long-running-handlers-lease-renewal), which decouples the two.
 - `retry_reclaim_interval_s` controls how often the background reclaimer wakes up. Lower values increase Redis load; 15 s is a sensible default for most workloads.
 - `max_retries` prevents poison messages from looping forever. Set to `None` for unlimited retries (no DLQ).
 
+### Long-Running Handlers: Lease Renewal
+
+Redis resets a PEL entry's idle time only on (re)delivery, **not while a handler works on it**. A handler that runs longer than `retry_on_idle_ms` (a multi-minute LLM call, say) is therefore reclaimed and redelivered *while it is still running*: processed twice, in parallel. A larger `retry_on_idle_ms` avoids that, but delays recovery from a crashed worker as much.
+
+`renew_lease=True` removes the trade-off: while a handler runs, its entry is renewed so the reclaimer skips it. A crashed worker stops renewing, and its entries are reclaimed `retry_on_idle_ms` after the last renewal.
+
+```python
+@agent.subscribe(
+    channel=requests,
+    retry_on_idle_ms=60_000,     # crash recovery after ~1 min of silence
+    renew_lease=True,            # ...however long a healthy handler runs
+    max_processing_ms=900_000,   # optional: give up (and retry) after 15 min
+)
+async def handle_request(message):
+    await call_llm(message)      # may take many minutes
+```
+
+| Option | Default | Meaning |
+|---|---|---|
+| `renew_lease` | `False` | Renew in-flight entries. Requires `retry_on_idle_ms`. |
+| `renew_lease_interval_ms` | `retry_on_idle_ms // 3` | How often to renew; must be `< retry_on_idle_ms`. The default tolerates two failed renewals in a row. Requires `renew_lease=True`. |
+| `cancel_on_lease_lost` | `True` | Cancel a handler whose entry was reclaimed anyway (`LeaseLostError`); `False` logs and lets it finish. Requires `renew_lease=True`. |
+| `max_processing_ms` | `None` | Handler deadline: cancel, raise `ProcessingTimeoutError`, NACK → normal retry. Requires `retry_on_idle_ms`. |
+
+**How it works.** Every interval, one Lua script checks per entry that it is still pending *for this consumer* and still in the stream, then runs `XCLAIM <stream> <group> <this consumer> 0 <id> JUSTID`: idle time reset, owner and delivery count unchanged. The ownership check means a renewal never takes an entry back from the reclaimer, whose own `XCLAIM` (`min-idle-time = retry_on_idle_ms`) skips renewed entries. The main and the `.retry` stream are renewed under their own group and consumer. Each call is bounded by half the interval; failures are logged (stream, group, consumer, ids) and retried, never stopping the consumer. The server must allow Lua scripting (`SCRIPT LOAD` / `EVALSHA`); if it refuses, an error is logged at `connect()` and handlers run unleased.
+
+**Lost leases.** An in-flight entry no longer pending for this consumer was normally reclaimed after renewals failed for longer than `retry_on_idle_ms` (a Redis outage, a blocked event loop), and is redelivered. By default the handler is cancelled and `eggai.transport.LeaseLostError` is raised instead of its result, so it can't overlap the redelivery; the entry is not acked. A `batch=True` batch is one unit: if any entry is lost, none is acked (the ones still owned are retried); `lost_ids` names the lost entries, `message_ids` the batch. An entry still pending for this consumer but deleted from the stream (`MAXLEN`, `XTRIM`, `XDEL`) is not lost: nothing can redeliver it, so the handler finishes (with a warning). An entry that left the PEL is lost even if also deleted, since with `delete_on_ack` the reclaimer deletes the original when moving it to `.retry`.
+
+**Prefetch and `max_records`.** FastStream reads up to `max_records` entries per `XREADGROUP` (all available ones when `None`), so entries queued behind a slow handler sit idle in this consumer's PEL; they are renewed too, also with `max_workers > 1`. Entries older than anything this process read (left by an earlier process with the same consumer name, e.g. a restarted container) are not renewed, so the reclaimer recovers them. As a leased entry stays with the worker that read it, `renew_lease=True` defaults `max_records` to **1** unless you set a number or use `batch=True` (an explicit `None` also becomes 1).
+
+**Handler deadline.** A hung handler would keep its lease forever. `max_processing_ms` cancels a handler still running at the deadline with `eggai.transport.ProcessingTimeoutError` (a `TimeoutError`); the entry is NACKed and retried after `retry_on_idle_ms`, counting towards `max_retries`. It also works without `renew_lease`.
+
+**Constraints.**
+
+- Handlers should be `async`. A sync handler's worker thread can't be interrupted, so it is never cancelled: a lost lease or missed deadline takes effect when the thread returns (result discarded, error raised). Past the deadline the entry stays leased until then, so its retry can't overlap the thread, but a sync handler that never returns keeps its lease; after a lost lease the thread may overlap the redelivery.
+- A handler that blocks the event loop blocks renewal too, including a sync handler with `data_type` / `filter_by_message` (the filter wrapper calls it on the event loop).
+- Requires `retry_on_idle_ms` and a consumer group; rejected with `no_ack=True` and `ack_policy=AckPolicy.MANUAL` (an entry acked mid-run would look lost).
+- Delivery stays at-least-once: the lease removes the *parallel* duplicate, not the redelivery after a crash. Handlers must still be idempotent.
+
 ### Automatic Recovery from Redis Stream Loss (NOGROUP)
 
 If Redis loses streams mid-life — due to a restart without persistence, a failover, or memory eviction — FastStream's consume loop retries `XREADGROUP` but never re-runs `XGROUP CREATE`, resulting in an infinite `NOGROUP` error loop.
@@ -372,8 +411,14 @@ async def recovery_handler(message):
 
 ## Backward Compatibility
 
-`retry_on_idle_ms` is fully opt-in. Existing subscriptions without it behave exactly as before — no extra streams, no background tasks, no changed semantics.
+`retry_on_idle_ms` is fully opt-in. Existing subscriptions without it behave exactly as before — no extra streams, no background tasks, no changed semantics. The same holds for `renew_lease` and `max_processing_ms`: without them no lease client or renewal task is created and handlers are not wrapped.
 
 ## API Reference
 
 ::: eggai.transport.RedisTransport
+
+### Lease exceptions
+
+::: eggai.transport.LeaseLostError
+
+::: eggai.transport.ProcessingTimeoutError

sdk/CHANGELOG.md

@@ -7,6 +7,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Redis `subscribe(..., renew_lease=True)`: in-flight lease renewal. Redis
+  resets a PEL entry's idle time only on delivery, so a handler running longer
+  than `retry_on_idle_ms` was reclaimed and redelivered while still running.
+  - Every `renew_lease_interval_ms` (default `retry_on_idle_ms // 3`) a Lua
+    script renews this consumer's in-flight entries with `XCLAIM ... 0 <id>
+    JUSTID` (owner and delivery count unchanged) on the main and `.retry`
+    stream, including prefetched ones; a crashed worker's entries are still
+    reclaimed. `renew_lease` defaults `max_records` to 1.
+  - An entry reclaimed anyway cancels its handler with
+    `eggai.transport.LeaseLostError` (`cancel_on_lease_lost=False` lets it
+    finish); sync handlers are never interrupted, the error is raised when
+    their thread returns.
+  - Requires `retry_on_idle_ms` and Lua scripting; rejected with `no_ack=True`
+    and `AckPolicy.MANUAL`. Opt-in, default behaviour unchanged.
+- Redis `subscribe(..., max_processing_ms=...)`: handler deadline. A handler
+  still running at it is cancelled with `eggai.transport.ProcessingTimeoutError`
+  and its entry NACKed for the normal retry. Requires `retry_on_idle_ms`.
+
+### Changed
+
+- Minimum FastStream is now 0.6.4 (`faststream>=0.6.4,<0.8`, was `>=0.6`).
+  The SDK never worked on the older 0.6 releases: on 0.6.0–0.6.2 every Redis
+  stream subscription failed with a `TypeError` (`StreamSub` has no
+  `min_idle_time` before 0.6.3), and on 0.6.3 a `min_idle_time` subscription
+  never claimed pending entries. Nothing that works today stops working; the
+  failure moves from runtime to install time. CI now also runs the test suite
+  against FastStream 0.6.4.
+- `anyio` (`>=4.0,<5`, the range FastStream 0.6.4 already requires) is now a
+  declared dependency; the SDK imports it directly.
+
+### Fixed
+
+- Redis: consuming resumes after a lost consumer group (stream deleted,
+  Redis flushed or failed over) with FastStream >= 0.7.6. Those versions stop
+  a subscriber for good on `NOGROUP` instead of retrying the read, so the
+  group monitor recreated the group but nothing read from it any more. The
+  monitor now also restarts the stream subscribers `connect()` started, once
+  their group exists again (a partial loss still redelivers from id `0`). No
+  change on FastStream <= 0.7.5, where the read loop never stopped. The lock
+  file moves to FastStream 0.7.7 so CI covers this path.
+
 ## [0.6.0] - 2026-09-25
 
 ### Added

sdk/README.md

@@ -253,6 +253,19 @@ if one joins later, the group monitor turns deletion off for that stream and log
 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.
 
+**Long-running handlers with `renew_lease`:** Redis resets an entry's idle time only on
+delivery, so a handler running longer than `retry_on_idle_ms` is reclaimed and redelivered
+while still running. `renew_lease=True` keeps renewing in-flight entries (a crashed worker's
+are still reclaimed), and `max_processing_ms` adds an optional handler deadline. See
+[Lease Renewal](https://docs.egg-ai.com/sdk/redis-transport/#long-running-handlers-lease-renewal).
+
+```python
+@agent.subscribe(channel=requests, retry_on_idle_ms=60_000, renew_lease=True,
+                 max_processing_ms=900_000)
+async def handle_request(message):
+    await call_llm(message)   # minutes; not redelivered while it runs
+```
+
 **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/agent.py

@@ -8,6 +8,7 @@
 from .hooks import eggai_register_stop
 from .transport import get_default_transport
 from .transport.base import Transport
+from .transport.lease import LEASE_OPTION_KEYS, resolve_lease_options
 
 HANDLERS_IDS: defaultdict[str, int] = defaultdict(int)
 
@@ -127,6 +128,11 @@
                         "idle threshold); a cap below the base would disable backoff "
                         "entirely."
                     )
+            # renew_lease / max_processing_ms: same checks as the transport,
+            # but at decoration time. Validation only: kwargs is read, not
+            # changed, and the transport resolves the options again.
+            if any(k in kwargs for k in LEASE_OPTION_KEYS):
+                resolve_lease_options(kwargs, kwargs.get("retry_on_idle_ms"))
             # Plugins see the kwargs as the caller wrote them (before the DLQ
             # key below is resolved to its namespaced form).
             original_kwargs = kwargs.copy()

sdk/eggai/transport/__init__.py

@@ -7,4 +7,6 @@
 )
 from .inmemory import InMemoryTransport as InMemoryTransport
 from .kafka import KafkaTransport as KafkaTransport
+from .lease import LeaseLostError as LeaseLostError
+from .lease import ProcessingTimeoutError as ProcessingTimeoutError
 from .redis import RedisTransport as RedisTransport