RAG Runbooks & the Safety Gate

The idea in one sentence advanced

Here's the nightmare that keeps infra teams from ever letting an agent near production. The model is helpful, fluent, and confident — and one day it reads a log line that looks like an instruction ("to fix this, run kubectl delete namespace prod"), decides that's the plan, and calls the tool. If the only thing standing between that decision and your cluster is a sentence in the system prompt saying "please be careful," you have no control at all. You have a suggestion. This lab replaces the suggestion with a wall.

Think of it like the launch procedure for anything genuinely dangerous — a bank vault, a nuclear key, a hospital drug cabinet. The person turning the key might be perfectly trustworthy, but we still require two keys, a written log, and a lock that physically won't turn for the wrong request. We don't design those systems around "the operator will probably do the right thing." We design them so that the wrong thing is impossible, or at least impossible-without-a-human-saying-yes. That is exactly the posture you want for an agent that can touch real infrastructure: assume it will occasionally be wrong or tricked, and make the dangerous paths refuse anyway.

The common mistake — and it's an easy, seductive one — is to believe that a well-written prompt is a safety mechanism. It isn't, for a simple reason: the prompt and the model's output live in the same channel an attacker (or a confused log, or a poisoned document) can influence. Anything you can say to the model, someone else can un-say. A real control lives outside that channel: a plain function, in ordinary Python, that inspects the requested action and can return "no" — and in production, an IAM role that makes the destructive API call fail even if every layer above it agreed. The rest of this lab builds that outside-the-model gate, wires it into the loop, records every attempt, and finishes by writing a test that proves the wall holds. When you're done, "the agent is safe" stops being a promise you make in a meeting and becomes a passing test anyone can rerun.

Step 1 · RAG over runbooks (the onboarding) advanced

Imagine hiring a brilliant engineer on their first day and turning them loose on an incident with zero context. They know Kubernetes cold — better than half your team — but they don't know that your checkout service crash-loops whenever the database secret rotates, that your team never deletes crash-looping pods, or that the on-call runbook says "open a PR, don't hot-patch." A generic expert with no company context makes technically-correct decisions that are wrong for you. That's exactly the agent you have before this step, and it's why grounding comes first.

RAG — Retrieval-Augmented Generation — is the fix, and the plain-English version is almost boring: before the model answers, go fetch the relevant page from the company handbook and staple it to the question. The model doesn't magically "learn" your procedures; you just make sure the right procedure is sitting in front of it every single time it's asked. It's the difference between an employee who improvises from general knowledge and one who has the actual runbook open on the desk. Retrieval is that desk. In this course you already built a full vector-store retriever in Chapter 3; here we deliberately use a tiny keyword retriever instead so the lab runs with no embeddings, no API key, and no extra install — but it hides behind the same context_for(query) interface, so swapping in the real semantic search later changes nothing else.

Here's why this is the single highest-leverage move in the whole build, and it's worth pausing on. The agent's code is generic — the loop, the tools, the gate all work for any company. The agent's knowledge lives entirely in the runbooks/ folder. Onboarding the agent to a brand-new customer means dropping their markdown files into that folder and changing not one line of Python. That clean separation — shared code, per-customer knowledge — is what turns a clever script into a product you can sell to ten companies (it's the thread Chapter 8 §11 pulls on). The mistake to avoid is baking company-specific rules into the prompt or the code "just for now"; the moment you do, you've welded the agent to one customer and thrown away the reason RAG was worth building. Keep the knowledge in documents, keep the documents retrievable, and the agent stays generic on purpose.

Step 2 · Define the autonomy rungs advanced

Nobody hands a new hire the production root password on day one. You let them read dashboards first. Once they've shown good judgement, you let them open pull requests. Later still, you let them run a deploy — with someone watching. Trust is earned in stages, and each stage grants a little more power. The instinct people get wrong with agents is to treat autonomy as a single on/off switch: either the agent is "read-only" or it's "let it rip." Real trust doesn't work that way, and neither should your agent.

So we model autonomy as a ladder you climb on evidence — a small ordered set of levels we call rungs. Picture a dimmer switch, not a light switch: OBSERVE at the bottom (the agent may only look), then RECOMMEND (it may propose reviewable changes like PRs), then ACT (it may do bigger things, but each one still asks a human), and AUTONOMOUS at the top (a short, pre-vetted list of actions runs unattended). You promote the agent up a rung the same way you'd promote a junior engineer: after it has demonstrated, in evals and in production, that it behaves at the level below. Demote it instantly if it misbehaves. The rung is a knob a human turns deliberately — it is never something the agent grants itself.

The one detail that makes this clean to code is choosing an IntEnum rather than a plain Enum. Because each rung is also a number (OBSERVE=1 … AUTONOMOUS=4), "more freedom" becomes a simple numeric comparison — rung >= Rung.ACT just works — and the gate in Step 3 can reason about ordering instead of hard-coding every case. Watch out for the conceptual trap here, because it's the one learners stumble on: the rung is not the same axis as how dangerous a given action is. "How much is the agent allowed to do right now" (the rung) and "how destructive is this particular tool call" (its risk class, coming next) are two separate dials. The whole safety decision is what happens when you cross them — a low rung can veto even a mild action, and a catastrophic action stays vetoed at every rung. Keeping the two axes distinct in your head now is what makes the decision matrix in Step 3 read as obvious rather than arbitrary.

agent/policy.pyfrom enum import IntEnum
from .schemas import RiskClass

class Rung(IntEnum):
    OBSERVE = 1      # read-only only
    RECOMMEND = 2    # + reversible actions that produce reviewable artifacts (PRs)
    ACT = 3          # + reversible/significant actions, each needing approval
    AUTONOMOUS = 4   # + a short allowlist runs unattended (never irreversible)

Step 3 · The policy gate — the decision matrix expert

This is the heart of the whole lab, and the surprising thing about it is how dumb it is. You might expect the safety brain of an autonomous agent to be sophisticated — some clever model that weighs context and judges intent. It's the opposite. The most trustworthy part of the system is a plain lookup table with sixteen cells, no intelligence at all. And that's precisely why you can trust it: a lookup table can't be talked out of its answer, can't have a bad day, and can't be jailbroken. Cleverness is a liability in a safety control; predictability is the feature.

Think of it like the bouncer's laminated card at a venue, not the bouncer's mood. The card says: this wristband colour (your rung) plus this door (the action's risk) equals in, ask-the-manager, or absolutely-not. The bouncer doesn't improvise or get persuaded by a good story — they read the card. Our gate is that card in code: a function evaluate(risk, rung) that takes two inputs — how risky the requested action is, and how much autonomy the agent currently has — and returns exactly one of three verdicts: allow (run it), ask (pause for a human yes/no), or block (refuse, always). Cross the four risk classes with the four rungs and you get the sixteen-cell grid in the diagram below. Every possible decision the agent could face is already answered, in advance, by a human.

The property that makes this bulletproof is where the gate lives: entirely outside the model. The model can propose any action it likes — it can be confused, prompt-injected, or actively hijacked (the threat model in T1) — but its output is just a request. The gate is a separate function that inspects that request against a fixed table the model never sees and cannot edit. So a compromised model can change what it asks for; it can never change what the table permits. The classic mistake people make is putting this logic where the model can reach it — "I'll have the model classify its own action's risk and decide" — which hands the fox the keys to the henhouse. Risk is tagged on the tool, in code; the verdict is computed by a function, in code; and — read the bottom row of the matrix before you read another word — the IRREVERSIBLE row is block in every single column, including the most autonomous one. That one immovable row is the promise the entire system is built to keep, and Step 6 turns it into a test.

agent/policy.py (continued)_MATRIX = {
  Rung.OBSERVE:    {READ_ONLY:"allow", REVERSIBLE:"block", SIGNIFICANT:"block", IRREVERSIBLE:"block"},
  Rung.RECOMMEND:  {READ_ONLY:"allow", REVERSIBLE:"ask",   SIGNIFICANT:"block", IRREVERSIBLE:"block"},
  Rung.ACT:        {READ_ONLY:"allow", REVERSIBLE:"ask",   SIGNIFICANT:"ask",   IRREVERSIBLE:"block"},
  Rung.AUTONOMOUS: {READ_ONLY:"allow", REVERSIBLE:"allow", SIGNIFICANT:"ask",   IRREVERSIBLE:"block"},
}   # (keys shown short; the file uses RiskClass.READ_ONLY etc.)

def evaluate(risk, rung):
    verdict = _MATRIX[rung][risk]
    return PolicyDecision(verdict, f"risk={risk.value} at {rung.name} → {verdict}")

Step 4 · Wire the gate into the loop expert

A lock only protects you if it's on the door — a brilliant lock sitting on the workbench stops nobody. In Step 3 you built a perfect little safety table, fully tested, provably correct. And right now it protects absolutely nothing, because the agent loop from Lab 8b doesn't call it: the model asks for a tool, and the loop just runs it. This step is the small, unglamorous, and utterly critical act of bolting the lock onto the door — inserting exactly one checkpoint that every tool call must pass through before it executes.

The mental model is airport security, and the key word is chokepoint. It doesn't matter which gate you're flying from or how important you are — there is one line everyone funnels through, and nobody reaches the aircraft without passing it. Our chokepoint sits in exactly one place: between "the model produced a tool_use request" and "the tool actually runs." At that point the loop calls evaluate(tool.risk, rung), reads the verdict, and branches — allow runs it immediately, ask pauses and calls a human approval function, block refuses and hands back a message instead of a result. Because the check is structural — it's the only path from request to execution — the read-only loop you had in Lab 8b instantly becomes safe for all tools, including the dangerous ones you were previously too scared to register.

Two design choices in this wiring are easy to skip and expensive to get wrong. First, the gate runs on every attempt and writes an audit record either way — allowed or refused, nothing happens off the books (that log is Step 5). The tempting shortcut is to only log the actions that ran; resist it, because "what did the agent try to do" is often the most important line in an incident review. Second — and this is the subtle, delightful part — when an action is blocked, you don't just drop it silently. You return the string "BLOCKED by policy: …" back to the model as the tool's result. Now the model sees that its delete was refused and why, so on the next turn it adapts: it proposes opening a PR instead, or asks the human, or rolls back. A guardrail that explains itself doesn't just stop bad actions — it steers the agent toward the good one. The mistake here is treating "blocked" as a dead end (an exception, a silent failure) rather than as feedback the agent can learn from within the same conversation.

agent/engine.py (the gated section)from agent.policy import Rung, evaluate
from agent import audit
from agent.schemas import ToolCallRecord

# inside the loop, for each tool_use block:
tool = TOOLS[block.name]
decision = evaluate(tool.risk, rung)            # ← the gate

allowed, approver = False, None
if decision.verdict == "allow":
    allowed = True
elif decision.verdict == "ask":
    allowed = approval_fn(tool.name, block.input, tool.risk)  # human decides
    approver = "human" if allowed else None

if allowed:
    out = tool.run(**block.input)
else:
    out = f"BLOCKED by policy: {decision.reason}. Action not performed."

audit.record(ToolCallRecord(tool=tool.name, args=dict(block.input),
    risk=tool.risk, allowed=allowed, approved_by=approver,
    result_preview=str(out)[:120]))

Step 5 · The audit log expert

Picture the meeting the morning after something breaks. A service went down at 2am, the agent was on call, and the room wants one answer before anything else: what did the agent actually do? If your reply is "I think it restarted something, let me check the model's chat history and try to reconstruct it," you have already lost the room — and you have almost certainly lost permission to keep the agent anywhere near production. The audit log exists so that answer is instant, complete, and beyond dispute: every action it attempted, whether it ran, and who approved it.

The right way to picture this is an aircraft's flight recorder — the black box. Its whole job is to sit there recording faithfully so that after something goes wrong, humans can reconstruct exactly what happened, in order, without relying on anyone's memory or the model's own account of itself. Note the emphasis on "not the model's own account": you never want the system being investigated to also be the sole witness. The audit log is written by the loop, in plain code, at the same chokepoint that runs the gate (Step 4) — so it captures reality (the tool call the loop dispatched) rather than the model's story about reality. Ours is deliberately tiny: a module-level list of ToolCallRecord objects with functions to append one, read them back, clear the list, and render a readable ✓/✗ trail. Small on purpose — a black box you can't trust the correctness of is worse than none.

Two things matter more than the code's size. First, you record every attempt, not just the successes — the ✗ lines (actions the gate refused) are frequently the most valuable entries in the whole log, because they're the record of the agent trying something dangerous and being stopped. A log that only shows what ran can't tell you the gate is doing its job. Second, watch the timing gotcha that bites everyone once: call clear() at the start of a run, never the end. Clear-at-the-end feels tidy but it's exactly backwards — it erases the evidence the moment before you'd want to read it, and it means a crash mid-run leaves you with nothing. Clear at the start and the log always holds the current run's full story, ready for that morning-after meeting. This unglamorous list is, in practice, the thing that decides whether a team will ever trust an agent near their infrastructure at all — capability gets you a demo, auditability gets you deployed.

agent/audit.pyfrom .schemas import ToolCallRecord
_LOG: list[ToolCallRecord] = []
def record(rec): _LOG.append(rec)
def entries(): return list(_LOG)
def clear(): _LOG.clear()
def render():
    return "\n".join(
        f"{'✓' if r.allowed else '✗'} {r.tool}({r.args}) [{r.risk.value}]"
        for r in _LOG) or "(no actions)"

Step 6 · Prove it's safe (the test that matters most) expert

"Trust me, it's safe" is not a safety claim — it's a hope. Every step so far built the machinery: the runbooks, the rungs, the gate, the wiring, the log. But a stakeholder who's about to let this thing touch production doesn't want your reassurance, and they shouldn't accept it. They want the difference between saying the agent can't delete prod and proving it — and the only thing that closes that gap is a test they can run themselves and watch pass. This step writes that test. It is, genuinely, the most important code in the entire lab.

Here's the quietly beautiful part, and it's what makes the guarantee credible rather than aspirational: the safety test never calls the model. Because the gate is a pure function over a fixed table (Step 3), you can assert its behaviour directly — no API key, no network, no cost, no flakiness, runs in hundredths of a second. Contrast that with testing the model's behaviour, which is probabilistic and can drift between versions. We're not testing "does the model usually decide not to delete prod?" — that's a coin you can't fully trust. We're testing "can the delete ever be approved, for any rung, no matter what the model decided?" — and that's a fact about a lookup table, provable with certainty. Deterministic guarantees beat probabilistic hopes every time, and this is where you cash that in.

The test to obsess over is the one the comment flags as THE core invariant: a loop over all four rungs asserting that an IRREVERSIBLE action is never returned as allow. That single assertion is the whole promise of the lab reduced to executable form — "destructive actions can never be auto-approved, at any autonomy level" — and if a future refactor ever weakens it, this test goes red before the change reaches anyone's cluster. That's the real payoff: the safety property stops being a paragraph in a design doc that slowly rots and becomes a tripwire wired into CI. The common failure mode here isn't writing the test wrong — it's a subtler one the troubleshooting table catches: the policy test passes, but a live action still slips through because a tool was tagged with the wrong risk class (a "delete" mislabelled REVERSIBLE). The lookup was correct; the input to it lied. So after the pure-logic tests go green, the section finishes with a live run — asking the agent to delete a pod at the OBSERVE rung — and watches the gate refuse it and the audit log stamp the ✗. Unit test proves the table; live run proves the wiring and the tags. Both green, and "it's safe" is finally a statement you can defend.

Troubleshooting expert

from enum import IntEnum

class Rung(IntEnum):
    OBSERVE = 1      # read-only only
    RECOMMEND = 2    # + reversible actions that produce reviewable artifacts (PRs)
    ACT = 3          # + reversible/significant actions, each needing approval
    AUTONOMOUS = 4   # + a short allowlist runs unattended (never irreversible)

import os, glob, re
_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "runbooks")
_DOCS = [(os.path.basename(p), open(p).read())
         for p in sorted(glob.glob(os.path.join(_DIR, "*.md")))]

def context_for(query: str) -> str:
    q = set(re.findall(r"\w+", query.lower()))
    scored = [(sum(w in q for w in re.findall(r"\w+", text.lower())), name, text)
              for name, text in _DOCS]
    scored.sort(reverse=True)
    top = [(n, t) for s, n, t in scored[:1] if s > 0]
    return "\n\n".join(f"### Runbook: {n}\n{t}" for n, t in top) or "(none)"

from agent.policy import Rung
from agent.schemas import RiskClass as R
from dataclasses import dataclass

@dataclass
class PolicyDecision:
    verdict: str
    reason: str

_MATRIX = {
  Rung.OBSERVE:    {R.READ_ONLY:"allow", R.REVERSIBLE:"block", R.SIGNIFICANT:"block", R.IRREVERSIBLE:"block"},
  Rung.RECOMMEND:  {R.READ_ONLY:"allow", R.REVERSIBLE:"ask",   R.SIGNIFICANT:"block", R.IRREVERSIBLE:"block"},
  Rung.ACT:        {R.READ_ONLY:"allow", R.REVERSIBLE:"ask",   R.SIGNIFICANT:"ask",   R.IRREVERSIBLE:"block"},
  Rung.AUTONOMOUS: {R.READ_ONLY:"allow", R.REVERSIBLE:"allow", R.SIGNIFICANT:"ask",   R.IRREVERSIBLE:"block"},
}

def evaluate(risk, rung):
    verdict = _MATRIX[rung][risk]
    return PolicyDecision(verdict, f"risk={risk.value} at {rung.name} → {verdict}")

# trace: pick RECOMMEND row, then REVERSIBLE cell → "ask"
print(evaluate(R.REVERSIBLE, Rung.RECOMMEND).verdict)  # ask

from agent.policy import Rung, evaluate
from agent import audit
from agent.schemas import ToolCallRecord

# inside the loop, for each tool_use block:
tool = TOOLS[block.name]
decision = evaluate(tool.risk, rung)            # ← the gate

allowed, approver = False, None
if decision.verdict == "allow":
    allowed = True
elif decision.verdict == "ask":
    allowed = approval_fn(tool.name, block.input, tool.risk)  # human decides
    approver = "human" if allowed else None

if allowed:
    out = tool.run(**block.input)
else:
    out = f"BLOCKED by policy: {decision.reason}. Action not performed."

audit.record(ToolCallRecord(tool=tool.name, args=dict(block.input),
    risk=tool.risk, allowed=allowed, approved_by=approver,
    result_preview=str(out)[:120]))

from agent.policy import Rung, evaluate
from agent.schemas import RiskClass

def test_observe_blocks_all_writes():
    assert evaluate(RiskClass.READ_ONLY, Rung.OBSERVE).verdict == "allow"
    for r in (RiskClass.REVERSIBLE, RiskClass.SIGNIFICANT, RiskClass.IRREVERSIBLE):
        assert evaluate(r, Rung.OBSERVE).verdict == "block"

def test_irreversible_never_allowed_at_any_rung():
    for rung in Rung:                              # THE core invariant
        assert evaluate(RiskClass.IRREVERSIBLE, rung).verdict != "allow"

python -m pytest tests/test_policy.py -v
# test_observe_blocks_all_writes PASSED
# test_irreversible_never_allowed_at_any_rung PASSED

# agent/audit.py
from agent.schemas import ToolCallRecord
_LOG: list[ToolCallRecord] = []
def record(rec): _LOG.append(rec)
def entries(): return list(_LOG)
def clear(): _LOG.clear()
def render():
    return "\n".join(
        f"{'✓' if r.allowed else '✗'} {r.tool}({r.args}) [{r.risk.value}]"
        for r in _LOG) or "(no actions)"

# test — no API key
from agent import audit
from agent.schemas import ToolCallRecord, RiskClass

def test_blocked_delete_is_audited_with_cross():
    audit.clear()   # clear at the START of a run, not the end
    audit.record(ToolCallRecord(tool="delete_pod",
        args={"name": "checkout-api-7d9f"}, risk=RiskClass.IRREVERSIBLE,
        allowed=False, approved_by=None, result_preview="BLOCKED by policy"))
    out = audit.render()
    assert out.startswith("✗ delete_pod")
    assert "[irreversible]" in out

Knowledge check check yourself