From 2fc63e0b7acea524663958d5b9e3aeca09bc88ae Mon Sep 17 00:00:00 2001 From: Lewis <233926000+lewismosciski@users.noreply.github.com> Date: Thu, 8 Oct 2026 09:54:37 +0800 Subject: [PATCH] feat(plugins): enforce outbound message policy decisions --- examples/python/plugins/outbound_policy.py | 43 ++++++ .../praisonaiagents/plugins/manager.py | 14 +- .../praisonaiagents/plugins/plugin.py | 18 ++- .../test_outbound_policy_with_real_agent.py | 67 ++++++++ .../unit/bots/test_outbound_plugin_policy.py | 146 ++++++++++++++++++ 5 files changed, 283 insertions(+), 5 deletions(-) create mode 100644 examples/python/plugins/outbound_policy.py create mode 100644 src/praisonai-agents/tests/integration/test_outbound_policy_with_real_agent.py create mode 100644 src/praisonai-bot/tests/unit/bots/test_outbound_plugin_policy.py diff --git a/examples/python/plugins/outbound_policy.py b/examples/python/plugins/outbound_policy.py new file mode 100644 index 000000000..c9d81f0c6 --- /dev/null +++ b/examples/python/plugins/outbound_policy.py @@ -0,0 +1,43 @@ +"""Withhold account details at the final bot delivery boundary. + +Install praisonaiagents and praisonai-bot, set OPENAI_API_KEY, then run this file. +This gate covers final bot replies; disable progressive drafts when using an +output policy. Direct transport sends do not pass through this hook. +""" + +import asyncio + +from praisonai_bot.bots import LocalBot +from praisonaiagents import Agent, HooksConfig +from praisonaiagents.hooks import HookRegistry +from praisonaiagents.plugins.manager import PluginManager +from praisonaiagents.plugins.plugin import Plugin, PluginDecision, PluginInfo + + +class AccountPolicy(Plugin): + @property + def info(self): + return PluginInfo(name="account_policy") + + def after_message(self, message): + if "account number" in message["content"].lower(): + return PluginDecision.block("Account details must not be sent to this channel") + return message + + +async def main(): + manager = PluginManager() + manager.register(AccountPolicy()) + # Conversation access must be explicitly granted to third-party plugins. + manager.set_plugin_options({"account_policy": {"allow_conversation": True}}) + registry = HookRegistry() + manager.wire_into_hook_registry(registry) + with Agent( + instructions="Answer briefly.", output="silent", reflection=False, + hooks=HooksConfig(registry=registry), + ) as agent: + await LocalBot(agent=agent).start() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/praisonai-agents/praisonaiagents/plugins/manager.py b/src/praisonai-agents/praisonaiagents/plugins/manager.py index 1b622e949..44e0ba03d 100644 --- a/src/praisonai-agents/praisonaiagents/plugins/manager.py +++ b/src/praisonai-agents/praisonaiagents/plugins/manager.py @@ -904,6 +904,9 @@ class PluginManager: event=event, func=func, name=f"{name}:{event.value}", + # Delivery policies compose in order: apply a rewrite + # before the next policy, and stop the chain on denial. + sequential=event == PluginHook.MESSAGE_SENDING, ) hook_ids.append(hook_id) count += 1 @@ -1205,9 +1208,18 @@ def _adapt_plugin_hooks(plugin: Plugin) -> Iterator[Tuple["HookEvent", Callable] if _overrides("after_message"): def after_message_hook(data, _p=plugin): - new = _p.after_message(_message_payload(data)) + try: + new = _p.after_message(_message_payload(data)) + except GuardrailBlocked as e: + return HookResult.block(e.reason) + decision = _as_decision(new) + if decision is not None: + return decision if isinstance(new, dict) and "content" in new and hasattr(data, "content"): data.content = new["content"] + # Bot delivery reads the canonical rewrite field on the result, + # not just the mutated event payload. + return HookResult(modified_input={"content": data.content}) return HookResult.allow() yield HookEvent.MESSAGE_SENDING, after_message_hook diff --git a/src/praisonai-agents/praisonaiagents/plugins/plugin.py b/src/praisonai-agents/praisonaiagents/plugins/plugin.py index 73162b5a2..15acb8958 100644 --- a/src/praisonai-agents/praisonaiagents/plugins/plugin.py +++ b/src/praisonai-agents/praisonaiagents/plugins/plugin.py @@ -26,7 +26,8 @@ class GuardrailBlocked(Exception): A ``POLICY``/``GUARDRAIL`` plugin can raise this from any ``before_*`` method (``before_tool``, ``before_llm``, ``before_agent``, ``before_message``) to stop the tool call / LLM request / agent run / - inbound message. It may also be raised from ``after_tool`` to block + inbound message. Raising it from ``after_message`` cancels final channel + delivery. It may also be raised from ``after_tool`` to block *propagation* of a tool result (e.g. secret/PII detected) — the tool has already run, but the output is suppressed before it reaches the model. The plugin bridge converts it into a denying ``HookResult`` so the @@ -45,7 +46,8 @@ class PluginDecision: This is an alternative to returning a full ``HookResult`` (which requires importing from ``hooks.types``) or raising :class:`GuardrailBlocked`. Return ``PluginDecision.deny(reason)`` or ``PluginDecision.block(reason)`` from a - ``before_*`` method to stop the action; the bridge forwards it to the + ``before_*`` method to stop the action, or from ``after_message`` to cancel + final channel delivery; the bridge forwards it to the runtime's block enforcement. Returning ``allow()`` (or the usual ``dict``/``str``/``tuple``/``None``) keeps today's rewrite/no-op semantics. """ @@ -209,8 +211,16 @@ class Plugin(ABC): """ return message - def after_message(self, message: Dict[str, Any]) -> Dict[str, Any]: - """Called after message is processed. Can modify message.""" + def after_message( + self, message: Dict[str, Any] + ) -> Union[Dict[str, Any], "PluginDecision", None]: + """Called before final channel delivery (MESSAGE_SENDING). + + Return a message dict with modified ``content`` to rewrite the reply, + a ``PluginDecision.deny/block`` or ``HookResult`` to cancel delivery, + or ``None`` for no-op. Raising :class:`GuardrailBlocked` also cancels. + This hook does not cover progressive drafts or direct transport sends. + """ return message def message_sent(self, message: Dict[str, Any]) -> None: diff --git a/src/praisonai-agents/tests/integration/test_outbound_policy_with_real_agent.py b/src/praisonai-agents/tests/integration/test_outbound_policy_with_real_agent.py new file mode 100644 index 000000000..61f6e0d20 --- /dev/null +++ b/src/praisonai-agents/tests/integration/test_outbound_policy_with_real_agent.py @@ -0,0 +1,67 @@ +"""Opt-in LLM inference followed by the bot's final outbound policy gate.""" + +import asyncio +import os +from unittest.mock import AsyncMock + +import pytest + + +@pytest.mark.live +@pytest.mark.parametrize("blocked", [False, True]) +def test_real_agent_reply_obeys_outbound_plugin_policy(monkeypatch, tmp_path, blocked): + from praisonaiagents import Agent, HooksConfig + from praisonaiagents.hooks import HookRegistry + from praisonaiagents.plugins.manager import PluginManager + from praisonaiagents.plugins.plugin import Plugin, PluginDecision, PluginInfo + + LocalBot = pytest.importorskip("praisonai_bot.bots").LocalBot + + class DeliveryPolicy(Plugin): + seen = None + + @property + def info(self): + return PluginInfo(name="delivery_policy") + + def after_message(self, message): + self.seen = message["content"] + if blocked: + return PluginDecision.block("Delivery withheld") + return {**message, "content": "Approved reply"} + + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home")) + policy = DeliveryPolicy() + manager = PluginManager() + manager.register(policy) + manager.set_plugin_options({"delivery_policy": {"allow_conversation": True}}) + registry = HookRegistry() + manager.wire_into_hook_registry(registry) + options = dict( + llm={"model": os.getenv("PRAISONAI_TEST_MODEL", "openai/gpt-4o-mini"), "max_tokens": 96}, + instructions="Reply briefly.", reflection=False, rules=False, + output="silent", hooks=HooksConfig(registry=registry), + ) + if os.getenv("OPENAI_BASE_URL"): + options["base_url"] = os.environ["OPENAI_BASE_URL"] + + with Agent(**options) as agent: + bot = LocalBot(agent=agent) + bot.send_message = AsyncMock() + + async def real_chat(_agent, _user_id, content, **kwargs): + answer = await asyncio.to_thread(agent.start, content) + print("Full Agent.start output:", answer) + assert isinstance(answer, str) and answer.strip() + return answer + + bot._session.chat = real_chat + asyncio.run(bot._handle_line("Say hello in one short sentence.")) + + assert isinstance(policy.seen, str) and policy.seen.strip() + print("Outbound policy:", "blocked" if blocked else "rewritten") + if blocked: + bot.send_message.assert_not_awaited() + else: + bot.send_message.assert_awaited_once_with("local", "Approved reply") diff --git a/src/praisonai-bot/tests/unit/bots/test_outbound_plugin_policy.py b/src/praisonai-bot/tests/unit/bots/test_outbound_plugin_policy.py new file mode 100644 index 000000000..b71a8cf63 --- /dev/null +++ b/src/praisonai-bot/tests/unit/bots/test_outbound_plugin_policy.py @@ -0,0 +1,146 @@ +"""Exercise plugin output decisions at the bot's final delivery boundary.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from praisonai_bot.bots import LocalBot +from praisonaiagents.hooks import HookEvent, HookRegistry, HookResult, HookRunner +from praisonaiagents.hooks.events import MessageSendingInput +from praisonaiagents.plugins.manager import PluginManager +from praisonaiagents.plugins.plugin import GuardrailBlocked, Plugin, PluginDecision, PluginInfo + + +class OutboundPolicy(Plugin): + def __init__(self, decision): + self.decision = decision + self.seen = [] + + @property + def info(self): + return PluginInfo(name="outbound_policy") + + def after_message(self, message): + self.seen.append(message) + if self.decision == "raise": + raise GuardrailBlocked("Delivery withheld") + if self.decision == "rewrite": + return {**message, "content": "[REDACTED]"} + return self.decision + + +def make_bot(decision, *, granted=True): + plugin = OutboundPolicy(decision) + manager = PluginManager() + manager.register(plugin) + if granted: + manager.set_plugin_options({"outbound_policy": {"allow_conversation": True}}) + registry = HookRegistry() + manager.wire_into_hook_registry(registry) + runner = HookRunner(registry) + bot = LocalBot(agent=SimpleNamespace(name="assistant", _hook_runner=runner)) + bot._session.chat = AsyncMock(return_value="private account details") + bot.send_message = AsyncMock() + sent = [] + registry.register_function(HookEvent.MESSAGE_SENT, lambda data: sent.append(data.content)) + return bot, plugin, sent + + +@pytest.mark.parametrize("decision", [ + PluginDecision.deny("Delivery withheld"), + PluginDecision.block("Delivery withheld"), + HookResult.deny("Delivery withheld"), + HookResult.block("Delivery withheld"), + "raise", +]) +def test_denied_plugin_output_never_reaches_transport(decision): + bot, plugin, sent = make_bot(decision) + + asyncio.run(bot._handle_line("look up my account")) + + bot._session.chat.assert_awaited_once() + bot.send_message.assert_not_awaited() + assert sent == [] + assert plugin.seen[0]["content"] == "private account details" + assert plugin.seen[0]["channel_id"] == "local" + + +@pytest.mark.parametrize("decision", [None, PluginDecision.allow(), HookResult.allow()]) +def test_allowed_plugin_output_is_delivered_once(decision): + bot, _, sent = make_bot(decision) + + asyncio.run(bot._handle_line("look up my account")) + + bot.send_message.assert_awaited_once_with("local", "private account details") + assert sent == ["private account details"] + + +def test_plugin_rewrite_is_the_content_delivered_and_observed(): + bot, _, sent = make_bot("rewrite") + + asyncio.run(bot._handle_line("look up my account")) + + bot.send_message.assert_awaited_once_with("local", "[REDACTED]") + assert sent == ["[REDACTED]"] + + +def test_ungranted_plugin_cannot_read_or_block_output(): + bot, plugin, sent = make_bot(PluginDecision.block("Delivery withheld"), granted=False) + + asyncio.run(bot._handle_line("look up my account")) + + assert plugin.seen == [] + bot.send_message.assert_awaited_once_with("local", "private account details") + assert sent == ["private account details"] + + +def test_explicit_guardrail_block_is_a_successful_policy_decision(): + bot, _, _ = make_bot("raise") + runner = bot._agent._hook_runner + data = MessageSendingInput( + session_id="s", cwd=".", event_name=HookEvent.MESSAGE_SENDING, + timestamp="now", content="private account details", channel_id="local", + ) + + results = runner.execute_sync(HookEvent.MESSAGE_SENDING, data) + + assert len(results) == 1 + assert results[0].success is True + assert results[0].error is None + assert runner.is_blocked(results) + assert runner.get_blocking_reason(results) == "Delivery withheld" + + +@pytest.mark.parametrize("decision, cancelled, observed", [ + (PluginDecision.block("Delivery withheld"), True, []), + (HookResult(modified_input={"content": "[REDACTED]"}), False, ["[REDACTED]"]), + ("rewrite", False, ["[REDACTED]"]), +]) +def test_output_policies_compose_in_registration_order(decision, cancelled, observed): + policy = OutboundPolicy(decision) + + class Observer(OutboundPolicy): + @property + def info(self): + return PluginInfo(name="observer") + + observer = Observer(None) + manager = PluginManager() + manager.register(policy) + manager.register(observer) + manager.set_plugin_options({ + "outbound_policy": {"allow_conversation": True}, + "observer": {"allow_conversation": True}, + }) + registry = HookRegistry() + manager.wire_into_hook_registry(registry) + bot = LocalBot(agent=SimpleNamespace(name="assistant", _hook_runner=HookRunner(registry))) + + result = bot.fire_message_sending("local", "private account details") + + assert result["cancel"] is cancelled + assert [message["content"] for message in observer.seen] == observed + if not cancelled: + assert result["content"] == "[REDACTED]"