diff --git a/roboco/events/stream_bus.py b/roboco/events/stream_bus.py
index 6d5150b4..ae70f6f9 100644
--- a/roboco/events/stream_bus.py
+++ b/roboco/events/stream_bus.py
@@ -46,6 +46,9 @@ class StreamEventBus:
     STREAM_PREFIX = "roboco:stream:"
     DEFAULT_GROUP = "roboco-handlers"
     MAX_STREAM_LENGTH = 10000  # Trim streams to this length
+    # Undecodable messages are parked here before ACK so an operator can
+    # inspect a poison pill instead of losing it to a silent ACK.
+    DEAD_LETTER_STREAM = "roboco:stream:dead-letter"
 
     def __init__(
         self,
@@ -66,6 +69,12 @@ class StreamEventBus:
         self._handlers: dict[EventType, list[EventHandler]] = {}
         self._running = False
         self._listen_task: asyncio.Task | None = None
+        # Periodic reclaim: XREADGROUP '>' delivers only NEW messages, so a
+        # runtime handler failure leaves its message pending and unretried
+        # until a restart. This loop re-runs recover_pending so the
+        # idempotency-guarded replay actually fires.
+        self._reclaim_task: asyncio.Task | None = None
+        self._reclaim_interval = 60
 
     async def connect(self) -> None:
         """Connect to Redis."""
@@ -85,6 +94,11 @@ class StreamEventBus:
             with contextlib.suppress(asyncio.CancelledError):
                 await self._listen_task
 
+        if self._reclaim_task:
+            self._reclaim_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._reclaim_task
+
         if self._redis:
             await self._redis.close()
 
@@ -205,6 +219,7 @@ class StreamEventBus:
 
         self._running = True
         self._listen_task = asyncio.create_task(self._listen_loop())
+        self._reclaim_task = asyncio.create_task(self._reclaim_loop())
         logger.info("StreamEventBus listening", streams=streams)
 
     async def _listen_loop(self) -> None:
@@ -235,6 +250,27 @@ class StreamEventBus:
                 logger.error("Error in stream event loop", error=str(e))
                 await asyncio.sleep(1)
 
+    async def _reclaim_loop(self) -> None:
+        """Periodically reclaim pending messages so a runtime handler failure
+        is retried without waiting for an orchestrator restart.
+
+        ``_listen_loop``'s XREADGROUP uses ``>`` (new messages only), so a
+        message left pending by a transient handler failure is never
+        re-delivered at runtime. This loop re-runs :meth:`recover_pending`
+        every ``_reclaim_interval`` seconds; the (event.id, handler)
+        idempotency guard makes the replay safe (already-succeeded handlers
+        are skipped, failed ones re-run).
+        """
+        while self._running:
+            try:
+                await asyncio.sleep(self._reclaim_interval)
+                await self.recover_pending(idle_time_ms=self._reclaim_interval * 1000)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.warning("Reclaim loop error; will retry", error=str(e))
+                await asyncio.sleep(self._reclaim_interval)
+
     @staticmethod
     def _to_str(value: object) -> str:
         """Decode a Redis stream/key value (bytes or str) to str."""
@@ -353,12 +389,48 @@ class StreamEventBus:
             return
         try:
             await handler(event)
-        except Exception:
-            # Clear the marker so a replay re-runs this handler — not a phantom success.
+        except BaseException:
+            # Clear the marker so a replay re-runs this handler — not a phantom
+            # success. BaseException catches asyncio.CancelledError (3.8+ is
+            # BaseException-derived) so a handler cancelled mid-flight (shutdown
+            # or sibling gather cancellation) still clears its marker and is
+            # re-run on reclaim; otherwise the guard would suppress the very
+            # redelivery that completes the work.
             with contextlib.suppress(Exception):
                 await self._redis.delete(key)
             raise
 
+    async def _dead_letter(
+        self, stream: str, message_id: str, data: dict, reason: str
+    ) -> None:
+        """Park an undecodable message on the dead-letter stream for inspection.
+
+        Best-effort: a dead-letter publish failure is logged but never blocks
+        the ACK — we will not strand a poison pill in the pending set because
+        the salvage stream was unwritable.
+        """
+        if self._redis is None:
+            return
+        try:
+            await self._redis.xadd(
+                self.DEAD_LETTER_STREAM,
+                {
+                    "source_stream": stream,
+                    "message_id": message_id,
+                    "data": self._decode_event_data(data) or "",
+                    "reason": reason,
+                },
+                maxlen=self.MAX_STREAM_LENGTH,
+                approximate=True,
+            )
+        except Exception as e:
+            logger.warning(
+                "Dead-letter publish failed",
+                error=str(e),
+                stream=stream,
+                message_id=message_id,
+            )
+
     async def _handle_message(
         self,
         stream: str,
@@ -369,14 +441,32 @@ class StreamEventBus:
         if not self._redis:
             return
 
-        try:
-            event_data = self._decode_event_data(data)
-            if event_data is None:
-                logger.error("Invalid event data", message_id=message_id)
-                await self._redis.xack(stream, self.group_name, message_id)
-                return
+        event_data = self._decode_event_data(data)
+        if event_data is None:
+            logger.error("Invalid event data", message_id=message_id)
+            await self._redis.xack(stream, self.group_name, message_id)
+            return
 
+        try:
             event = Event.from_json(event_data)
+        except Exception as decode_err:
+            # A message whose payload fails to decode (malformed JSON, an
+            # EventType removed in a version bump, bad UUID/timestamp) is a
+            # poison pill: no handler could ever process it, so retrying is
+            # pointless. Dead-letter it for inspection, then ACK so the stream
+            # doesn't wedge on an unkillable pending message re-failing on
+            # every reclaim cycle.
+            logger.error(
+                "Undecodable stream message; dead-lettering and ACKing",
+                message_id=message_id,
+                stream=stream,
+                error=str(decode_err),
+            )
+            await self._dead_letter(stream, message_id, data, str(decode_err))
+            await self._redis.xack(stream, self.group_name, message_id)
+            return
+
+        try:
             all_succeeded = await self._dispatch_event(event)
 
             # ACK the message if all handlers succeeded
diff --git a/tests/unit/events/test_bus.py b/tests/unit/events/test_bus.py
index d5cbfd70..bbcc4a45 100644
--- a/tests/unit/events/test_bus.py
+++ b/tests/unit/events/test_bus.py
@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
+import json
 from typing import TYPE_CHECKING, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -153,3 +156,213 @@ async def test_dispatch_runs_handler_when_redis_guard_unavailable() -> None:
     ok = await bus._dispatch_event(event)
     assert ok is True
     assert calls == ["ran"]
+
+
+# --- poison-pill: an undecodable message must be ACKed, not retried forever ---
+
+
+class _FakeRedisStream:
+    """In-memory redis surface for the xack/xadd/group path.
+
+    Tracks ACKs and dead-letter xadds so the poison-pill test can assert a
+    malformed message is acknowledged (not left pending) and parked on the
+    dead-letter stream. ``xreadgroup`` yields nothing so a listen loop never
+    spins.
+    """
+
+    def __init__(self) -> None:
+        self.xack_calls: list[tuple[str, str, tuple[str, ...]]] = []
+        self.xadd_calls: list[tuple[str, dict]] = []
+
+    async def xack(self, stream: str, group: str, *ids: str) -> int:
+        self.xack_calls.append((stream, group, ids))
+        return len(ids)
+
+    async def xadd(
+        self,
+        stream: str,
+        fields: dict,
+        maxlen: int | None = None,
+        approximate: bool = True,
+    ) -> bytes:
+        del maxlen, approximate
+        self.xadd_calls.append((stream, dict(fields)))
+        return b"1-0"
+
+    async def xreadgroup(self, *args: object, **kwargs: object) -> list:
+        del args, kwargs
+        return []
+
+    async def xgroup_create(self, *args: object, **kwargs: object) -> bool:
+        del args, kwargs
+        return True
+
+    async def xpending(self, *args: object, **kwargs: object) -> dict:
+        del args, kwargs
+        return {"pending": 0}
+
+    async def xpending_range(self, *args: object, **kwargs: object) -> list:
+        del args, kwargs
+        return []
+
+    async def xclaim(self, *args: object, **kwargs: object) -> list:
+        del args, kwargs
+        return []
+
+    async def set(
+        self, key: str, value: str, *, nx: bool = False, ex: int | None = None
+    ) -> bool:
+        del key, value, nx, ex
+        return True
+
+    async def delete(self, key: str) -> int:
+        del key
+        return 1
+
+    async def get(self, key: str) -> None:
+        del key
+
+    async def close(self) -> None:
+        """No-op close for the fake client."""
+
+
+@pytest.mark.asyncio
+async def test_undecodable_message_is_acked_and_dead_lettered() -> None:
+    """A message whose payload fails Event.from_json (unknown EventType value,
+    bad UUID, malformed JSON) is a poison pill: no handler could ever process
+    it, so retrying is pointless. The bus must ACK it (and dead-letter it) so
+    the stream doesn't wedge on an unkillable pending message re-failing on
+    every reclaim."""
+
+    bus = StreamEventBus()
+    fake = _FakeRedisStream()
+    bus._redis = cast("Redis", fake)
+
+    invoked: list[str] = []
+
+    async def _handler(_event: Event) -> None:
+        invoked.append("ran")
+
+    bus.subscribe(EventType.NOTIFICATION_SENT, _handler)
+
+    # type="task.bogus" is not a real EventType → EventType(...) raises ValueError
+    # inside Event.from_json.
+    malformed = json.dumps(
+        {
+            "id": "not-a-uuid",
+            "type": "task.bogus_unknown",
+            "data": {},
+            "timestamp": "2026-06-30T00:00:00+00:00",
+        }
+    )
+    await bus._handle_message("roboco:stream:task", "1234-0", {b"data": malformed})
+
+    # ACKed exactly once — not left pending for reclaim to re-fail forever.
+    assert len(fake.xack_calls) == 1
+    assert fake.xack_calls[0][2] == ("1234-0",)
+    # Dead-lettered for inspection before the ACK.
+    assert len(fake.xadd_calls) == 1
+    assert fake.xadd_calls[0][0] == StreamEventBus.DEAD_LETTER_STREAM
+    # No handler could run — the event never decoded.
+    assert invoked == []
+
+
+# --- periodic reclaim: a runtime handler failure is retried without a restart ---
+
+
+@pytest.mark.asyncio
+async def test_reclaim_loop_periodically_calls_recover_pending() -> None:
+    """XREADGROUP '>' delivers only NEW messages, so a handler that fails at
+    runtime leaves its message pending and unretried until the orchestrator
+    restarts. A periodic reclaim loop must call recover_pending so the
+    idempotency-guarded replay actually fires."""
+
+    bus = StreamEventBus()
+    bus._running = True
+    bus._reclaim_interval = 60
+
+    calls: list[int] = []
+
+    async def _fake_recover(idle_time_ms: int = 60000) -> int:
+        calls.append(idle_time_ms)
+        bus._running = False  # break the loop after the first reclaim
+        return 0
+
+    bus.recover_pending = _fake_recover  # type: ignore[method-assign]
+
+    async def _no_sleep(_seconds: float) -> None:
+        return
+
+    with patch("roboco.events.stream_bus.asyncio.sleep", new=_no_sleep):
+        await bus._reclaim_loop()
+
+    # Reclaim ran once with the interval-aligned idle window, then the loop exited.
+    assert calls == [60000]
+
+
+@pytest.mark.asyncio
+async def test_start_listening_spawns_reclaim_task_alongside_listen() -> None:
+    """start_listening must spawn the reclaim task, not just the listen task —
+    otherwise pending messages are never re-delivered at runtime."""
+
+    bus = StreamEventBus()
+    bus._redis = cast("Redis", _FakeRedisStream())
+
+    async def _noop(self: StreamEventBus) -> None:
+        del self
+
+    async def _handler(_event: Event) -> None: ...
+
+    bus.subscribe(EventType.NOTIFICATION_SENT, _handler)
+
+    with (
+        patch.object(StreamEventBus, "_listen_loop", _noop),
+        patch.object(StreamEventBus, "_reclaim_loop", _noop),
+    ):
+        await bus.start_listening()
+
+    try:
+        assert bus._listen_task is not None
+        assert bus._reclaim_task is not None
+    finally:
+        await bus.disconnect()
+
+
+# --- cancellation mid-handler must clear the idempotency marker ---
+
+
+@pytest.mark.asyncio
+async def test_cancelled_handler_clears_idempotency_marker() -> None:
+    """A handler cancelled mid-flight (shutdown / sibling gather cancellation)
+    is BaseException-cancelled, not Exception-raised, so the old ``except
+    Exception`` left the SET-NX marker set: the message stayed pending but the
+    guard then suppressed the very redelivery that would complete the work.
+    The cleanup must catch BaseException so the marker is cleared and reclaim
+    re-runs the handler."""
+
+    bus = StreamEventBus()
+    fake = _FakeRedis()
+    bus._redis = cast("Redis", fake)
+
+    started = asyncio.Event()
+    proceed = asyncio.Event()
+
+    async def _blocking(_event: Event) -> None:
+        started.set()
+        await proceed.wait()  # block until the dispatch task is cancelled
+
+    bus.subscribe(EventType.NOTIFICATION_SENT, _blocking)
+    event = Event(type=EventType.NOTIFICATION_SENT, data={"task_id": "tc"})
+
+    task = asyncio.create_task(bus._dispatch_event(event))
+    await started.wait()  # handler is now blocked → marker is set
+
+    key = f"bus:processed:{event.id}:_blocking"
+    assert key in fake.keys  # marker acquired before the handler blocked
+
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+    # Marker cleared despite cancellation → a replay re-runs the handler.
+    assert key not in fake.keys