feat(plugins): enforce outbound message policy decisions

This commit is contained in:
Lewis committed 2026-10-08 09:54:37 +08:00
1 parent c4bdb3ff91
commit 2fc63e0b7a
5 files changed
+283 -5

No files matched your search

@@ -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())
@@ -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
@@ -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:
@@ -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")
@@ -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]"