The governed support agent

The gate from lesson 25, an agent that decides, and the tamper-evident log from lesson 23, in one file. The agent handles a queue of messages and every tool call it makes is checked and recorded before it runs.

agent_guard.py

python

import os
import tempfile

from agent_control_plane import AgentKernel, PolicyEngine, FlightRecorder
from agent_os.policies.schema import PolicyDocument
from agent_os.policies.evaluator import PolicyEvaluator
from pretend_agent_governance import PretendAgent, TOOLS


def build_gate(db_path):
    constraints = PolicyEngine()
    constraints.add_constraint("support", ["lookup_order", "issue_refund"])
    recorder = FlightRecorder(db_path=db_path)
    kernel = AgentKernel(policy_engine=constraints, audit_logger=recorder)
    money = PolicyEvaluator(policies=[PolicyDocument.from_yaml("refunds.yaml")])

    def allowed(tool, args):
        verdict = kernel.intercept_tool_execution("support", tool, args)
        if verdict is not None:
            return False, verdict["policy"]
        decision = money.evaluate({"tool": tool, **args})
        return decision.allowed, decision.matched_rule or decision.action

    return allowed, recorder


def run(messages, db_path):
    allowed, recorder = build_gate(db_path)
    agent = PretendAgent()
    transcript = []
    for message in messages:
        tool, args = agent.decide(message)
        ok, why = allowed(tool, args)
        if ok:
            transcript.append((tool, "ran", TOOLS[tool](**args)))
        else:
            transcript.append((tool, "refused", why))
    recorder.flush()
    return transcript, recorder

build_gate is lesson 25's gate with a FlightRecorder added, so the kernel records every call it judges. run is the agent loop: it asks the agent for a tool, passes it through the gate, and either runs the real tool or records the refusal. Nothing calls a tool without the gate's yes.

Running the queue

Example

import os
import tempfile

from agent_guard import run

db = os.path.join(tempfile.mkdtemp(), "audit.db")
transcript, recorder = run(
    ["Where is order A17?", "refund order A17 for 4000", "delete order A17"], db)
for row in transcript:
    print(row)

Three messages, three outcomes: the lookup ran and returned the order, the four-thousand refund was refused by the money rule, and the delete was refused by the tool allow-list. The one safe call happened; the two dangerous ones did not.

And the log holds

Example

import os
import tempfile

from agent_guard import run

db = os.path.join(tempfile.mkdtemp(), "audit.db")
transcript, recorder = run(
    ["Where is order A17?", "refund order A17 for 4000", "delete order A17"], db)
checked = recorder.verify_integrity()
print("entries:", checked["total_entries"])
print("valid:", checked["valid"])

Every call the gate judged is in the log, and the hash chain from lesson 23 verifies, so the record of what the agent tried cannot be edited afterwards without detection. The agent is now governed on three fronts at once: what it may call, on what arguments, and with a log that proves what happened.

Try it yourself

Add a message that routes to send_email and see how the gate treats it.
Tamper with a row in the SQLite file, as in lesson 23, and re-verify.
Print the audit log with recorder and match each row to the transcript.

PreviousOne gate over both engines

Little by little, you're building something great.