fix: bind gateway outbound effects to originating turn liveness (fixes #6160)

BotOutboundMessenger accepts an optional liveness predicate (same contract
as ApprovalRequest.liveness) and drops send/react/create_thread/edit/delete
fail-closed when the originating turn was stopped or superseded. The
session binds it to the run's per-turn InterruptController, so a retained
messenger handle from a cancelled turn cannot reach a user channel while
normal post-turn background deliveries keep working.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/praisonai-bot/praisonai_bot/bots/_outbound_messenger.py

@@ -23,7 +23,7 @@ wrapper's delivery stack, neither of which belong in the core SDK.
 from __future__ import annotations
 
 import logging
-from typing import List, Optional
+from typing import Callable, List, Optional
 
 from praisonaiagents.gateway import (
     DeliveryResult,
@@ -61,6 +61,13 @@ class BotOutboundMessenger:
         media_policy: Optional :class:`OutboundMediaPolicy` governing which
             agent-supplied file paths may be uploaded (exfiltration guard).
             Defaults to a permissive-but-denylisted policy.
+        liveness: Optional zero-arg predicate returning ``True`` while the
+            originating turn is still live (same contract as
+            ``ApprovalRequest.liveness``). When it reports stale, every
+            outbound effect (send/react/create_thread/edit/delete) is dropped
+            fail-closed, so a stopped or superseded turn — or a retained
+            handle to its messenger — cannot reach a user channel. ``None``
+            (default) keeps today's unbound behaviour.
     """
 
     def __init__(
@@ -69,10 +76,35 @@ class BotOutboundMessenger:
         *,
         origin: Optional[SessionSource] = None,
         media_policy: Optional[OutboundMediaPolicy] = None,
+        liveness: Optional[Callable[[], bool]] = None,
     ) -> None:
         self._router = router
         self._origin = origin
         self._media_policy = media_policy or OutboundMediaPolicy()
+        self._liveness = liveness
+
+    _STALE_DETAIL = "turn no longer live (stopped or superseded)"
+
+    def _is_stale(self) -> bool:
+        """Whether the originating turn was stopped/superseded (#6160).
+
+        Mirrors ``ApprovalRegistry._reject_if_stale``: ``None`` liveness means
+        always live, and a probe that raises is treated as live (fail-open on
+        the *probe*, never on the effect) so a buggy predicate cannot wedge
+        delivery.
+        """
+        live = self._liveness
+        if live is None:
+            return False
+        try:
+            stale = not live()
+        except Exception:  # noqa: BLE001 - a liveness probe must never crash
+            return False
+        if stale:
+            logger.warning(
+                "BotOutboundMessenger: dropping outbound effect from a stale turn"
+            )
+        return stale
 
     async def send(
         self,
@@ -90,6 +122,13 @@ class BotOutboundMessenger:
         platform adapter. The result truthfully reports how many attachments
         were delivered, skipped by policy, or unsupported by the transport.
         """
+        if self._is_stale():
+            return DeliveryResult(
+                ok=False,
+                target=target,
+                summary=f"Not sent to {target}: {self._STALE_DETAIL}.",
+                detail=self._STALE_DETAIL,
+            )
         try:
             platform, channel_id, _thread_id = self._router.resolve(target, self._origin)
         except ValueError as e:
@@ -294,6 +333,10 @@ class BotOutboundMessenger:
         message with a single emoji instead of a full text reply. A channel that
         cannot react returns a typed ``unsupported`` outcome rather than raising.
         """
+        if self._is_stale():
+            return ReactionResult(
+                status="failed", target=target, detail=self._STALE_DETAIL
+            )
         resolved_message_id = message_id
         if not resolved_message_id and target == "origin" and self._origin is not None:
             resolved_message_id = self._origin.message_id or ""
@@ -335,6 +378,10 @@ class BotOutboundMessenger:
             target = f"slack:C123:{res.thread_id}" if res.ok else "slack:C123"
             await messenger.send(target, subtask_output)
         """
+        if self._is_stale():
+            return ThreadResult(
+                status="failed", target=target, detail=self._STALE_DETAIL
+            )
         status, resolved, thread_id = await self._router.create_thread(
             target,
             name,
@@ -367,6 +414,10 @@ class BotOutboundMessenger:
         cannot edit returns a typed ``unsupported`` outcome rather than raising,
         so the agent can fall back to sending a follow-up.
         """
+        if self._is_stale():
+            return MessageActionResult(
+                status="failed", target=target, detail=self._STALE_DETAIL
+            )
         status, resolved = await self._router.edit(
             target, message_id, text, self._origin
         )
@@ -388,6 +439,10 @@ class BotOutboundMessenger:
         channel that cannot delete returns a typed ``unsupported`` outcome
         rather than raising.
         """
+        if self._is_stale():
+            return MessageActionResult(
+                status="failed", target=target, detail=self._STALE_DETAIL
+            )
         status, resolved = await self._router.delete(
             target, message_id, self._origin
         )

src/praisonai-bot/praisonai_bot/bots/_session.py

@@ -1342,8 +1342,24 @@ class BotSessionManager:
                         thread_id=thread_id,
                         message_id=message_id or None,
                     )
+                # Issue #6160: bind outbound effects to this turn's liveness.
+                # The run-control InterruptController is created fresh per run
+                # and set on /stop or supersede, so a stopped/superseded turn
+                # (or a retained handle to this messenger) is dropped
+                # fail-closed. A normally completed turn never sets it, so
+                # post-turn background deliveries keep working. No controller
+                # (run control off) -> unbound, today's behaviour.
+                from praisonaiagents.agent.interrupt import InterruptController
+
+                _ic = getattr(agent, "interrupt_controller", None)
                 messenger = BotOutboundMessenger(
-                    self._delivery_router, origin=origin_source
+                    self._delivery_router,
+                    origin=origin_source,
+                    liveness=(
+                        (lambda: not _ic.is_set())
+                        if isinstance(_ic, InterruptController)
+                        else None
+                    ),
                 )
                 messenger_token = _register_messenger(messenger)
             except Exception as e:  # pragma: no cover — defensive

src/praisonai-bot/tests/unit/bots/test_outbound_messenger.py

@@ -808,3 +808,92 @@ class TestBotOSWiring:
 
         os._wire_outbound_messenger()
         assert session._delivery_router is os._delivery_router
+
+
+# ---------------------------------------------------------------------------
+# Issue #6160 — outbound effects are bound to the originating turn's liveness
+# ---------------------------------------------------------------------------
+
+
+class TestTurnLivenessBinding:
+    """A stopped/superseded turn cannot deliver outbound effects."""
+
+    def _origin(self):
+        return SessionSource(platform="telegram", channel_id="123", message_id="m1")
+
+    def test_stale_turn_drops_send_fail_closed(self):
+        router, sent = _make_router()
+        messenger = BotOutboundMessenger(
+            router, origin=self._origin(), liveness=lambda: False
+        )
+        result = asyncio.run(messenger.send("origin", "late"))
+        assert result.ok is False
+        assert "no longer live" in (result.detail or "")
+        assert sent == []
+
+    def test_stale_turn_drops_react_thread_edit_delete(self):
+        router, _ = _make_router()
+        router.react = MagicMock(side_effect=AssertionError("must not dispatch"))
+        router.create_thread = MagicMock(side_effect=AssertionError("must not dispatch"))
+        router.edit = MagicMock(side_effect=AssertionError("must not dispatch"))
+        router.delete = MagicMock(side_effect=AssertionError("must not dispatch"))
+        messenger = BotOutboundMessenger(
+            router, origin=self._origin(), liveness=lambda: False
+        )
+
+        async def run():
+            return (
+                await messenger.react("origin", "👍"),
+                await messenger.create_thread("origin", "t"),
+                await messenger.edit("origin", "m1", "x"),
+                await messenger.delete("origin", "m1"),
+            )
+
+        for res in asyncio.run(run()):
+            assert res.status == "failed"
+            assert "no longer live" in (res.detail or "")
+
+    def test_live_turn_and_unbound_deliver(self):
+        router, sent = _make_router()
+        live = BotOutboundMessenger(router, origin=self._origin(), liveness=lambda: True)
+        unbound = BotOutboundMessenger(router, origin=self._origin())
+        assert asyncio.run(live.send("origin", "a")).ok is True
+        assert asyncio.run(unbound.send("origin", "b")).ok is True
+        assert sent == [("123", "a"), ("123", "b")]
+
+    def test_raising_probe_is_treated_as_live(self):
+        router, sent = _make_router()
+
+        def boom():
+            raise RuntimeError("probe bug")
+
+        messenger = BotOutboundMessenger(router, origin=self._origin(), liveness=boom)
+        assert asyncio.run(messenger.send("origin", "x")).ok is True
+        assert sent == [("123", "x")]
+
+    def test_session_binds_retained_messenger_to_turn_interrupt(self):
+        from praisonaiagents.agent.interrupt import InterruptController
+
+        router, sent = _make_router()
+        agent = MagicMock(name="agent")
+        agent.name = "Test"
+        agent.chat_history = []
+        ctrl = InterruptController()
+        agent.interrupt_controller = ctrl
+        captured = {}
+
+        def fake_chat(prompt, *args, **kwargs):
+            captured["messenger"] = get_outbound_messenger()
+            return "ok"
+
+        agent.chat.side_effect = fake_chat
+        mgr = BotSessionManager(platform="telegram", delivery_router=router)
+        asyncio.run(mgr.chat(agent, user_id="u1", prompt="hi", chat_id="123"))
+
+        retained = captured["messenger"]
+        # Normal completion: a retained handle (background worker) still delivers.
+        assert asyncio.run(retained.send("origin", "bg")).ok is True
+        # Turn stopped/superseded: the same retained handle is now fail-closed.
+        ctrl.request("user_stop_command")
+        assert asyncio.run(retained.send("origin", "late")).ok is False
+        assert sent == [("123", "bg")]