diff --git a/src/praisonai-bot/praisonai_bot/bots/_outbound_messenger.py b/src/praisonai-bot/praisonai_bot/bots/_outbound_messenger.py index f1cb9df9c..9bbb3ca52 100644 --- a/src/praisonai-bot/praisonai_bot/bots/_outbound_messenger.py +++ b/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 ) diff --git a/src/praisonai-bot/praisonai_bot/bots/_session.py b/src/praisonai-bot/praisonai_bot/bots/_session.py index 6ef3060b9..ad6f12741 100644 --- a/src/praisonai-bot/praisonai_bot/bots/_session.py +++ b/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 diff --git a/src/praisonai-bot/tests/unit/bots/test_outbound_messenger.py b/src/praisonai-bot/tests/unit/bots/test_outbound_messenger.py index c0ca7b410..084e58997 100644 --- a/src/praisonai-bot/tests/unit/bots/test_outbound_messenger.py +++ b/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")]