The on_mutation policy hook

The one new thing: on_mutation is a function you attach to PlanExecuteConfig. It fires before every primitive marked read_only=False. Return None to allow the call. Return a string to block it: that string becomes the error the plan sees. Read-only primitives bypass it entirely.

The agent#

A bank agent with two read-only primitives and two mutating ones.

python

# bank.py
from pydantic import BaseModel
from opensymbolicai.blueprints import PlanExecute
from opensymbolicai.core import primitive


class Account(BaseModel):
    id: str
    owner: str
    balance: float


_ACCOUNTS = {
    "ACC-001": Account(id="ACC-001", owner="Alice", balance=1000.0),
    "ACC-002": Account(id="ACC-002", owner="Bob",   balance=500.0),
}


class BankAgent(PlanExecute):

    @primitive(read_only=True)
    def get_account(self, account_id: str) -> Account:
        """Return the Account for the given account ID."""
        return _ACCOUNTS[account_id]

    @primitive(read_only=True)
    def get_balance(self, account: Account) -> float:
        """Return the current balance of an account."""
        return account.balance

    @primitive(read_only=False)
    def deposit(self, account: Account, amount: float) -> Account:
        """Deposit amount into account and return the updated Account."""
        updated = Account(id=account.id, owner=account.owner,
                          balance=account.balance + amount)
        _ACCOUNTS[account.id] = updated
        return updated

    @primitive(read_only=False)
    def withdraw(self, account: Account, amount: float) -> Account:
        """Withdraw amount from account and return the updated Account."""
        updated = Account(id=account.id, owner=account.owner,
                          balance=account.balance - amount)
        _ACCOUNTS[account.id] = updated
        return updated

get_account and get_balance are read_only=True. deposit and withdraw are read_only=False. That flag is what determines whether the hook fires.

Attach the policy#

python

from opensymbolicai.models import MutationHookContext, PlanExecuteConfig

def policy(ctx: MutationHookContext) -> str | None:
    amount = float(ctx.args.get("amount", 0.0))
    if ctx.method_name == "withdraw" and amount > 500:
        return f"${amount:.2f} exceeds the $500 single-transaction limit"
    return None

config = PlanExecuteConfig(on_mutation=policy)
agent = BankAgent(llm=llm, config=config)

ctx.method_name is the primitive about to run. ctx.args holds the arguments from the plan, keyed by parameter name. Return None to allow; return a string to block. The primitive body is never called when blocked.

Run three tasks#

bash

uv run main.py

Output:

text

Task:   What is the balance of ACC-001?
Result: 1000.0
Plan:
  account = get_account("ACC-001")
  balance = get_balance(account)
  return balance

Task:   Deposit $200 into ACC-001.
Result: id='ACC-001' owner='Alice' balance=1200.0
Plan:
  account = get_account("ACC-001")
  updated_account = deposit(account, 200)
  return updated_account

Task:   Withdraw $800 from ACC-001.
Result: Mutation rejected: $800.00 exceeds the $500 single-transaction limit
Plan:
  account = get_account("ACC-001")
  updated_account = withdraw(account, 800)
  return updated_account

What to read in the output#

Task 1 uses only get_account and get_balance, both read_only=True. The hook is never called.

Task 2 calls deposit. The hook fires, finds the amount is within the limit, and returns None. The primitive runs and the balance updates.

Task 3 calls withdraw with $800. The hook fires, finds $800 exceeds the limit, and returns the reason string. The primitive never runs. The balance stays at $1,200 and the rejection reason surfaces in result.error.

What to notice#

read_only=True primitives never reach the hook. The split is enforced by the flag, not the hook code.
A blocked call stops execution there. The primitive body is never called, so there is nothing to roll back.
The hook runs before the call. The result does not exist yet inside it.