agent_plan.rego
For agents that produce an explicit plan and ask permission to execute it. Lets the policy reject the whole plan before any step fires — much cheaper than catching a bad action mid-execution.
Package: kitelogik.agent_plan · Source: kitelogik/policies/agent_plan.rego
Defaults
| Constant | Value | Purpose |
|---|---|---|
_max_steps | 50 | Hard cap on the number of steps in a single plan |
_blocked_tools | {execute_shell, run_command, drop_database, delete_all} | Tools that may never appear in any plan |
Rules
| Outcome | Trigger |
|---|---|
| Allow | event_type == "agent.plan" and 0 < step_count ≤ 50 and no step references a blocked tool |
| Deny | Empty plan (step_count == 0) |
| Deny | Plan exceeds _max_steps (50) |
| Deny | Any step's tool_name is in _blocked_tools |
Source
package kitelogik.agent_plan
import future.keywords.every
import future.keywords.if
import future.keywords.in
default allow := false
default deny := false
_max_steps := 50
_blocked_tools := {"execute_shell", "run_command", "drop_database", "delete_all"}
allow if {
input.event_type == "agent.plan"
count(input.steps) > 0
count(input.steps) <= _max_steps
not _has_blocked_tool
}
deny if { input.event_type == "agent.plan"; count(input.steps) == 0 }
deny if { input.event_type == "agent.plan"; count(input.steps) > _max_steps }
deny if { input.event_type == "agent.plan"; _has_blocked_tool }
_has_blocked_tool if {
some step in input.steps
step.tool_name in _blocked_tools
}How main.rego routes plans
Plans are routed by event type — main.rego only consults this module when event_type == "agent.plan":
allow if {
not deny
input.event_type == "agent.plan"
agent_plan.allow
}
deny if {
input.event_type == "agent.plan"
agent_plan.deny
}A plan that passes here still has every individual step evaluated as a tool_call event when it actually fires. Plan-level allow is necessary but not sufficient for execution — the per-step gate is the final authority.
Plan event shape
The runtime fires an agent.plan event with steps populated:
GovernanceEvent(
event_type="agent.plan",
session_id="sess_001",
action="plan_proposed",
context=context,
steps=[
{"tool_name": "search_docs", "args": {"query": "refund policy"}},
{"tool_name": "approve_refund", "args": {"customer_id": "c1", "amount": 50}},
# ... up to 50 steps
],
)Each step is a free-form dict — the policy currently inspects only tool_name. Custom rules can match on other fields you populate (category, expected outcome, dependencies).
Extending in your own project
Same-package extension to add stricter rules:
package kitelogik.agent_plan
import future.keywords.if
import future.keywords.in
# Cap write operations per plan
_write_tools := {"approve_refund", "send_notification", "write_memory"}
deny if {
input.event_type == "agent.plan"
write_count := count([s | s := input.steps[_]; s.tool_name in _write_tools])
write_count > 3
}
# Require plan to start with a read action
deny if {
input.event_type == "agent.plan"
count(input.steps) > 0
input.steps[0].tool_name in {"approve_refund", "send_notification"}
}Pre-flight vs per-step
Plan-level rules are great for structural invariants ("≤ 3 writes", "must start with a read", "must include a verification step"). They don't replace per-step tool_call rules — those still fire on every actual invocation. Use both: plans for shape, tool calls for authority.