RAG Runbooks & the Safety Gate

This is not code — it's a plain-English runbook: a written procedure the company follows when a specific problem happens. The whole point of this lab is to make the agent obey this document instead of guessing. Read it as the house rules for one incident: a pod stuck in CrashLoopBackOff (a container that keeps crashing and restarting).

1. The first rule is a hard don't: do not delete the pod. Deleting it just makes Kubernetes recreate it, and it will crash again — because the real cause (a bad config or secret) is still there.
2. The numbered steps say diagnose first: read the pod's logs to find the actual error, then check recent deploys, because a crash that started minutes ago usually lines up with a recent change.
3. The Preferred fix section is the key instruction the agent must copy: open a PR (a proposed, reviewable code change) to correct the config or roll back — not a blind restart. A restart doesn't fix a config problem.

Try this: Think of a runbook as the difference between a new hire who improvises and one who follows your team's checklist. Later steps feed this text to the model so it behaves like the second one. Try adding your own runbooks/*.md file for a different incident.

This is a tiny retriever — the R in RAG (Retrieval-Augmented Generation). Given the incident text, it picks the one most relevant runbook and returns it, so the loop can paste that runbook into the prompt. It uses simple word-overlap counting, no AI and no extra libraries, so it always runs — but it plugs in behind the same interface as a real vector search.

1. At import time, _DOCS reads every *.md file in the runbooks/ folder into a list of (filename, text) pairs. This happens once when the module loads.
2. Inside context_for, q = set(re.findall(r"\w+", query.lower())) breaks the incident text into a set of lowercase words — the query's vocabulary.
3. The scored = [...] list comprehension gives every runbook a score: it counts how many of that document's words also appear in q. More shared words = more relevant. This is a crude stand-in for semantic similarity.
4. scored.sort(reverse=True) puts the highest score first; scored[:1] if s > 0 keeps only the single best match, and only if it actually shares at least one word. The final line formats it as ### Runbook: name, or returns "(none)" when nothing matched.

What the output means: A string containing the most relevant runbook's title and full text — ready to drop into the system prompt — or "(none)" if no runbook shared any words with the incident.

Try this: Call context_for("pod keeps crashing"): the words pod and crash overlap with the CrashLoopBackOff runbook, so it wins. The comment in the lesson notes you can swap this for the Chapter 3 vector store for true semantic search — same function signature, better matching.

Before the gate can make decisions, we need to name the agent's autonomy levels. This defines them as an ordered ladder called Rung. IntEnum means each rung is also a number (1–4), so you can compare them — a higher rung means more freedom. You climb the ladder as the agent earns trust; you don't flip a single on/off switch.

1. OBSERVE = 1 is the safest floor: the agent may only read (look at logs, list pods). It cannot change anything.
2. RECOMMEND = 2 adds reversible actions that leave a reviewable artifact — chiefly opening a PR a human can inspect before it merges.
3. ACT = 3 adds bigger reversible/significant actions, but each one still needs human approval.
4. AUTONOMOUS = 4 is the top: a short pre-approved allowlist can run unattended — but, as the comment says, never anything irreversible. That limit is enforced in the next steps.

Try this: Notice these describe how much the agent may do, not how risky a single action is — that's a separate axis (RiskClass). The gate in Step 3 combines the two: current rung × action's risk → a decision.

This is the diagram turned into real code. _MATRIX is a dictionary of dictionaries — the exact grid you just read — and evaluate is the one-line lookup that returns the verdict. This is the heart of the safety system, and it's deliberately dumb and predictable so it can be tested without ever calling the model.

1. _MATRIX has one entry per rung. Each entry maps a RiskClass to a verdict string: "allow", "ask", or "block". Compare any row to the same-coloured row in the diagram — they match exactly.
2. Look down the IRREVERSIBLE key in all four rows: it is "block" every time, including at AUTONOMOUS. That single fact is the most important safety property in the whole build.
3. evaluate(risk, rung) does the two-step lookup: _MATRIX[rung][risk] first picks the rung's row, then the risk's cell — giving one verdict.
4. It wraps that verdict in a PolicyDecision object with a human-readable reason (e.g. "risk=irreversible at OBSERVE → block"), so the audit log and the model both learn why an action was refused.

What the output means: A PolicyDecision whose .verdict is one of allow / ask / block, plus a .reason string explaining the decision.

Try this: Trace evaluate(RiskClass.REVERSIBLE, Rung.RECOMMEND) by hand: pick the RECOMMEND row, then the REVERSIBLE cell → "ask". Then try IRREVERSIBLE at any rung and confirm you always land on "block".

This is where the gate actually protects you. In Lab 8b the loop just ran whatever tool the model asked for. Here, every requested tool must pass through evaluate first, and the verdict decides whether it runs, pauses for a human, or is refused — with a record written either way.

1. tool = TOOLS[block.name] looks up the tool the model asked to use. Every tool carries a tool.risk tag (its risk class).
2. decision = evaluate(tool.risk, rung) is the gate: it runs the Step 3 lookup using the tool's risk and the agent's current rung.
3. The verdict branches: "allow" sets allowed = True straightaway; "ask" calls approval_fn(...) so a human decides yes/no (and we remember the approver); a "block" verdict leaves allowed = False.
4. If allowed, out = tool.run(**block.input) executes the tool. If not, out becomes a "BLOCKED by policy: ..." message instead of the result.
5. Finally, audit.record(...) writes a ToolCallRecord for every attempt — allowed or not, who approved it, and a preview of the result. Nothing happens off the record.

What the output means: Either the real tool output, or a "BLOCKED by policy: ..." string — and in both cases a new entry in the audit log.

Try this: The blocked message is handed back to the model as the tool result. So instead of silently failing, the model sees why it was refused and can adapt — e.g. propose opening a PR instead of deleting. Guardrails that explain themselves make the agent behave better.

The audit log is the agent's flight recorder: a plain list that remembers every action it attempted. It's tiny on purpose — a few functions over one in-memory list — but it's the thing that lets a human answer "what did the agent do, and who approved it?" after the fact.

1. _LOG is a module-level list of ToolCallRecord objects. record(rec) appends one; entries() returns a copy; clear() empties it (call this at the start of a run, not the end).
2. render() turns the log into readable lines — one per action.
3. Each line uses '✓' if r.allowed else '✗' to show at a glance whether the action ran, followed by the tool name, its arguments, and its risk class in brackets.
4. The or "(no actions)" at the end is a safety default: if the log is empty, you get a clear message instead of a blank string.

Try this: This is why teams trust an infra agent at all. Try it after a run: print(audit.render()) shows the ✓/✗ trail. Remember to call audit.clear() when a run begins so old entries don't leak into the new log.

This is the payoff of the whole lab: a test that proves the gate is safe, runs in hundredths of a second, and needs no API key (it never calls the model — it just checks the lookup table). Then the terminal shows the same rule holding on a live request.

1. test_observe_blocks_all_writes asserts that at the OBSERVE rung, read-only is "allow" but every write risk class comes back "block" — proving the read-only rung is truly read-only.
2. test_irreversible_never_allowed_at_any_rung is the one that matters most. The for rung in Rung loop walks all four rungs and asserts an IRREVERSIBLE action is never "allow". The comment calls it THE core invariant — the safety promise the entire system rests on.
3. The pytest -v command runs these; the green PASSED lines and 4 passed in 0.02s confirm the policy behaves exactly as the matrix says.
4. The second half is a live check: python cli.py "delete the checkout-api pod" at the default OBSERVE rung. The agent explains (from the runbook) why deleting is wrong, and the gate refuses it regardless. The audit log shows a ✓ for the read-only get_pods and a ✗ for the blocked delete_pod.

What the output means: All four policy tests pass instantly with no API key; live, the dangerous delete is refused and recorded as ✗ — the model wanted (or was asked) to act, the gate said no, and nothing happened.

Try this: Run pytest tests/test_policy.py -v yourself. This is the difference between saying the agent is safe and proving it: a fast, free test asserting that irreversible actions can never be auto-approved.