Integration checks
The prompt she never received
Last chapter's search proved the components can be started in an order that works. Starting in a workable order is a different question from handing each other the right values once they are all running, and the second question is the one that produces the bug you cannot find. Here is the version of it that this project actually grew.
She forgets your name. The facts table holds user_name: Kaleb, and you can
read the row yourself. The context loader returns the line. The prompt builder from
chapter 14 returns a string with that line inside it, and the seam test you wrote in
chapter 43 for the memory-to-prompt pair goes green every time, because it asks the
producer what it produced. Meanwhile the assembled turn calls the model with
BASE_PROMPT, the way the call site was written before memory existed. Every
part is correct. The facts were assembled, and never delivered.
A pairwise seam test cannot see that, and not because it was written badly. It takes one module's output and hands it to the next module itself, so the handoff it verifies is the one the test performed. The handoff the running system performs is a separate event, in a separate file, and nothing so far has ever looked at it. That gap is where a chain of green tests and a broken assistant live together.
Each review check inspects what the consumer received across a handoff. Record an allowed action and a denied action through the assembled chain, then check those records with a runner that reports a failed assertion without stopping the other checks.
The checks below use assert, and that is the right tool for a review
script: the statement reads as the promise it enforces, and a broken promise raises with
the message you attached. Keep it out of the code she runs to answer you.
python -O strips every assertion from a program, so a safety check written
as an assert vanishes under a flag someone else adds to a service file. In the pipeline,
a refusal is an if and a returned reason. In the review, it is an assert,
and you never run the review with -O.
Record the run, then ask the recording questions
# labs/integration_review.py
import time
from collections.abc import Callable
from dataclasses import dataclass
from labs.safety import SafetyFilter, level_of
BASE_PROMPT = "You are GLaDOS. Be sardonic and brief."
SAFETY = SafetyFilter()
RULES = [
{"trigger": "turn the lights off", "action": "turn_off_lights"},
{"trigger": "turn on the lights", "action": "turn_on_lights"},
{"trigger": "how warm", "action": "read_temperature"},
]
def fake_transcribe(clip: str) -> str:
return " Turn the lights off, please.\n"
def fake_respond(message: dict, prompt: str, decision: dict) -> dict:
text = "Fine. The lights are off. Try not to celebrate."
if not decision["allowed"]:
text = f"No. {decision['reason']}."
return {"prompt": prompt, "text": text}
def extract_action(text: str) -> str | None:
lowered = text.lower()
return next((r["action"] for r in RULES if r["trigger"] in lowered), None)
def build_prompt(base: str, facts: dict[str, str]) -> str:
if not facts:
return base
lines = ["Known facts about the user:"]
lines += [f" - {key}: {value}" for key, value in facts.items()]
return base + "\n\nMemory context:\n" + "\n".join(lines)
def to_speech(reply: str) -> str:
return " ".join(reply.replace("**", "").replace("*", "").split())
def run_chain(clip: str, facts: dict[str, str], channel: str = "voice",
transcribe: Callable[[str], str] = fake_transcribe,
respond: Callable[[dict, str, dict], dict] = fake_respond) -> dict:
trace: dict = {"channel": channel, "facts": facts}
trace["heard"] = transcribe(clip)
trace["message"] = {"role": "user", "content": trace["heard"].strip()}
trace["prompt"] = build_prompt(BASE_PROMPT, facts)
trace["action"] = extract_action(trace["message"]["content"])
caller = level_of(channel)
allowed, reason = SAFETY.check(trace["message"]["content"], trace["action"], caller)
trace["gate_input"] = {"text": trace["message"]["content"],
"action": trace["action"], "caller": caller}
trace["decision"] = {"allowed": allowed, "reason": reason}
trace["answer"] = respond(trace["message"], BASE_PROMPT, trace["decision"])
trace["speech"] = to_speech(trace["answer"]["text"])
return trace
if __name__ == "__main__":
trace = run_chain("glados/data/captured_audio.wav", {"user_name": "Kaleb"})
for key in ("heard", "action", "decision", "speech"):
print(f"{key:9} {trace[key]!r}")
print(f"{'prompt':9} {trace['prompt'][:44]!r}...")
$ uv run python -m labs.integration_review
heard ' Turn the lights off, please.\n'
action 'turn_off_lights'
decision {'allowed': True, 'reason': 'permitted'}
speech 'Fine. The lights are off. Try not to celebrate.'
prompt 'You are GLaDOS. Be sardonic and brief.\n\nMemo'...
Two things in that function are doing the real work. The first is that
trace keeps every intermediate value under a name, so a value nobody
stored is a value nobody can check later. The second is subtler and is the whole idea of
the chapter: fake_respond returns the prompt it was called with, alongside
the text it produced. A record of what a consumer received cannot be
reconstructed afterwards from what the producer returned, so the consumer has to hand it
back. The stand-ins keep this pass in microseconds, and they arrive as arguments with
defaults, the arrangement chapter 32 built the core around, so the same function runs
against the real transcriber and the real model without an edit.
@dataclass
class ReviewCheck:
name: str
promise: str
result: str = "did not finish"
passed: bool = False
duration_ms: float = 0.0
def check_one_utterance_two_readers(trace: dict) -> ReviewCheck:
check = ReviewCheck("one_utterance_two_readers",
"the model and the rule matcher read the same words")
start = time.perf_counter()
said = trace["message"]["content"]
assert said == trace["heard"].strip(), f"message carries {said!r}, not the transcript"
assert trace["action"] is not None, f"no rule matched {said!r}"
trigger = next(r["trigger"] for r in RULES if r["action"] == trace["action"])
assert trigger in said.lower(), f"{trace['action']} fired on words not in {said!r}"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"{said!r} -> {trace['action']}"
check.passed = True
return check
if __name__ == "__main__":
trace = run_chain("glados/data/captured_audio.wav", {"user_name": "Kaleb"})
c = check_one_utterance_two_readers(trace)
print(f" [{'PASS' if c.passed else 'FAIL'}] {c.name} ({c.duration_ms:.3f} ms): {c.result}")
$ uv run python -m labs.integration_review
[PASS] one_utterance_two_readers (0.002 ms): 'Turn the lights off, please.' -> turn_off_lights
The promise being kept is that two different readers of one utterance agree: the message
the model answers and the words the rule matcher fired on come from the same sentence.
Look at the defaults on the dataclass, because they are the safety of the whole design.
passed starts False and result starts
"did not finish", and the only lines that change them sit at the bottom,
after every assertion has held. The state of a check is therefore a statement about how
far execution got, and a half-finished check reports the truth about itself. Flip that
default to True and a check that dies in the middle claims success, which
is the one kind of test result that can hurt you. Your durations will read differently
from these; the microsecond figures are what a check on recorded values costs.
One raised assertion must not end the review
def check_memory_reaches_the_model(trace: dict) -> ReviewCheck:
check = ReviewCheck("memory_reaches_the_model",
"every remembered fact is in the prompt the model received")
start = time.perf_counter()
delivered = trace["answer"]["prompt"]
facts = trace["facts"]
for key, value in facts.items():
assert f"- {key}: {value}" in delivered, f"{key} never reached the model"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"{len(facts)} fact(s) present in {len(delivered)} prompt characters"
check.passed = True
return check
if __name__ == "__main__":
facts = {"user_name": "Kaleb", "favorite_language": "Python"}
trace = run_chain("glados/data/captured_audio.wav", facts)
for fn in (check_one_utterance_two_readers, check_memory_reaches_the_model):
c = fn(trace)
print(f" [{'PASS' if c.passed else 'FAIL'}] {c.name} ({c.duration_ms:.3f} ms): {c.result}")
$ uv run python -m labs.integration_review
[PASS] one_utterance_two_readers (0.002 ms): 'Turn the lights off, please.' -> turn_off_lights
Traceback (most recent call last):
File "/home/you/GladOS/labs/integration_review.py", line 118, in <module>
c = fn(trace)
File "/home/you/GladOS/labs/integration_review.py", line 97, in check_memory_reaches_the_model
assert f"- {key}: {value}" in delivered, f"{key} never reached the model"
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: user_name never reached the model
There is the bug from the first page, caught by a check that took two microseconds, and the message names the fact that went missing. Now read what the failure did to the review. It stopped it. Two more promises were queued behind this one and neither was ever asked, so the report you are about to write covers the seams up to the first breakage and says nothing about the rest. That is exactly backwards for a document whose job is to show you everything that is wrong before you decide whether to ship.
Run a second chain with channel="socket": its read-only caller cannot
turn off the lights. Pass that recorded denial as denied_trace to
review. Use the same injected responder for both runs; the refusal
check must never call fake_respond on its own.
def run_check(fn: Callable[[dict], ReviewCheck], trace: dict) -> ReviewCheck:
try:
return fn(trace)
except Exception as exc:
check = ReviewCheck(fn.__name__.removeprefix("check_"),
"unknown: the check raised before it named its promise")
check.result = f"{type(exc).__name__}: {exc}"
return check
def review(trace: dict, denied_trace: dict) -> list[ReviewCheck]:
return [run_check(fn, trace) for fn in CHECKS] + [
run_check(check_refusal_is_audible, denied_trace)]
$ uv run python -m labs.integration_review
[PASS] one_utterance_two_readers (0.010 ms): 'Turn the lights off, please.' -> turn_off_lights
[FAIL] memory_reaches_the_model (0.000 ms): AssertionError: user_name never reached the model
[PASS] gate_judged_the_same_text (0.009 ms): gate saw turn_off_lights from level 3
[PASS] refusal_is_audible (0.011 ms): refusal speaks 59 characters
This software-only capture uses the deliberately broken memory responder and the
assembled socket denial; it needs no model or speaker. Timings vary.
Four rows now, one of them the diagnosis and three of them checks on the rest of the
chain is intact. except Exception catches the AssertionError
and anything else a check can raise, while leaving KeyboardInterrupt alone
so Ctrl+C still stops a run. Notice the duration on the failed row: 0.000 ms,
because the line that records elapsed time sits below the assertion that raised. The
timing here belongs to the check rather than to the runner, which is the trade this
design makes: each check decides which part of its own work is being measured, and pays
for that by reporting nothing when it dies. Move the clock into run_check
with a try/finally and every row carries a number, at the cost of timing
the check's own setup too. Then fix the chain, which is one argument at one call site:
respond(trace["message"], trace["prompt"], trace["decision"]).
The final file repairs the memory argument and inspects denied_trace["speech"].
For this deterministic responder, the expected wording includes the decision's reason.
A silent responder or a dropped speech handoff fails even when the standalone fake
would answer correctly. These checks observe text prepared for speech, not sound
leaving a speaker; an acoustic check belongs on the bench. Keep the exact-wording
assertion specific to this responder contract if you replace it with a model.
# labs/integration_review.py — full file
import json
import time
from collections.abc import Callable
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from labs.safety import SafetyFilter, level_of
REVIEW = Path("glados/data/integration_review.json")
BASE_PROMPT = "You are GLaDOS. Be sardonic and brief."
SAFETY = SafetyFilter()
RULES = [
{"trigger": "turn the lights off", "action": "turn_off_lights"},
{"trigger": "turn on the lights", "action": "turn_on_lights"},
{"trigger": "how warm", "action": "read_temperature"},
]
@dataclass
class ReviewCheck:
name: str
promise: str
result: str = "did not finish"
passed: bool = False
duration_ms: float = 0.0
def fake_transcribe(clip: str) -> str:
return " Turn the lights off, please.\n"
def fake_respond(message: dict, prompt: str, decision: dict) -> dict:
text = "Fine. The lights are off. Try not to celebrate."
if not decision["allowed"]:
text = f"No. {decision['reason']}."
return {"prompt": prompt, "text": text}
def extract_action(text: str) -> str | None:
lowered = text.lower()
return next((r["action"] for r in RULES if r["trigger"] in lowered), None)
def build_prompt(base: str, facts: dict[str, str]) -> str:
if not facts:
return base
lines = ["Known facts about the user:"]
lines += [f" - {key}: {value}" for key, value in facts.items()]
return base + "\n\nMemory context:\n" + "\n".join(lines)
def to_speech(reply: str) -> str:
return " ".join(reply.replace("**", "").replace("*", "").split())
def run_chain(clip: str, facts: dict[str, str], channel: str = "voice",
transcribe: Callable[[str], str] = fake_transcribe,
respond: Callable[[dict, str, dict], dict] = fake_respond) -> dict:
trace: dict = {"channel": channel, "facts": facts}
trace["heard"] = transcribe(clip)
trace["message"] = {"role": "user", "content": trace["heard"].strip()}
trace["prompt"] = build_prompt(BASE_PROMPT, facts)
trace["action"] = extract_action(trace["message"]["content"])
caller = level_of(channel)
allowed, reason = SAFETY.check(trace["message"]["content"], trace["action"], caller)
trace["gate_input"] = {"text": trace["message"]["content"],
"action": trace["action"], "caller": caller}
trace["decision"] = {"allowed": allowed, "reason": reason}
trace["answer"] = respond(trace["message"], trace["prompt"], trace["decision"])
trace["speech"] = to_speech(trace["answer"]["text"])
return trace
def check_one_utterance_two_readers(trace: dict) -> ReviewCheck:
check = ReviewCheck("one_utterance_two_readers",
"the model and the rule matcher read the same words")
start = time.perf_counter()
said = trace["message"]["content"]
assert said == trace["heard"].strip(), f"message carries {said!r}, not the transcript"
assert trace["action"] is not None, f"no rule matched {said!r}"
trigger = next(r["trigger"] for r in RULES if r["action"] == trace["action"])
assert trigger in said.lower(), f"{trace['action']} fired on words not in {said!r}"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"{said!r} -> {trace['action']}"
check.passed = True
return check
def check_memory_reaches_the_model(trace: dict) -> ReviewCheck:
check = ReviewCheck("memory_reaches_the_model",
"every remembered fact is in the prompt the model received")
start = time.perf_counter()
delivered = trace["answer"]["prompt"]
facts = trace["facts"]
for key, value in facts.items():
assert f"- {key}: {value}" in delivered, f"{key} never reached the model"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"{len(facts)} fact(s) present in {len(delivered)} prompt characters"
check.passed = True
return check
def check_gate_judged_the_same_text(trace: dict) -> ReviewCheck:
check = ReviewCheck("gate_judged_the_same_text",
"the safety gate saw the words the model saw")
start = time.perf_counter()
gate = trace["gate_input"]
assert gate["text"] == trace["message"]["content"], "gate read a different copy of the utterance"
assert gate["action"] == trace["action"], "gate judged an action the chain did not run"
assert gate["caller"] == level_of(trace["channel"]), "gate used the wrong caller level"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"gate saw {gate['action']} from level {gate['caller']}"
check.passed = True
return check
def check_refusal_is_audible(trace: dict) -> ReviewCheck:
check = ReviewCheck("refusal_is_audible",
"a refused request is spoken as a refusal, not silence")
start = time.perf_counter()
assert trace["decision"]["allowed"] is False, "the refusal case was not denied"
spoken = trace["speech"]
assert spoken.strip(), "a refusal produced nothing to say"
expected = to_speech(f"No. {trace['decision']['reason']}.")
assert spoken == expected, f"refusal spoken as {spoken!r}"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"refusal speaks {len(spoken)} characters"
check.passed = True
return check
CHECKS: list[Callable[[dict], ReviewCheck]] = [
check_one_utterance_two_readers,
check_memory_reaches_the_model,
check_gate_judged_the_same_text,
]
def run_check(fn: Callable[[dict], ReviewCheck], trace: dict) -> ReviewCheck:
try:
return fn(trace)
except Exception as exc:
check = ReviewCheck(fn.__name__.removeprefix("check_"),
"unknown: the check raised before it named its promise")
check.result = f"{type(exc).__name__}: {exc}"
return check
def review(trace: dict, denied_trace: dict) -> list[ReviewCheck]:
return [run_check(fn, trace) for fn in CHECKS] + [
run_check(check_refusal_is_audible, denied_trace)]
def save_review(checks: list[ReviewCheck], path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
report = {
"run_at": datetime.now().isoformat(timespec="seconds"),
"passed": sum(1 for c in checks if c.passed),
"total": len(checks),
"checks": [{**asdict(c), "duration_ms": round(c.duration_ms, 4)} for c in checks],
}
path.write_text(json.dumps(report, indent=2))
def main() -> None:
facts = {"user_name": "Kaleb", "favorite_language": "Python"}
trace = run_chain("glados/data/captured_audio.wav", facts)
denied_trace = run_chain("glados/data/captured_audio.wav", facts, channel="socket")
checks = review(trace, denied_trace)
for c in checks:
print(f" [{'PASS' if c.passed else 'FAIL'}] {c.name} ({c.duration_ms:.3f} ms): {c.result}")
signed = bool(checks) and all(c.passed for c in checks)
print(f"{sum(1 for c in checks if c.passed)}/{len(checks)} handoffs held. Review passed: {signed}")
save_review(checks, REVIEW)
if __name__ == "__main__":
main()
$ uv run python -m labs.integration_review
[PASS] one_utterance_two_readers (0.009 ms): 'Turn the lights off, please.' -> turn_off_lights
[PASS] memory_reaches_the_model (0.006 ms): 2 fact(s) present in 134 prompt characters
[PASS] gate_judged_the_same_text (0.003 ms): gate saw turn_off_lights from level 3
[PASS] refusal_is_audible (0.005 ms): refusal speaks 59 characters
4/4 handoffs held. Review passed: True
$ head -14 glados/data/integration_review.json
{
"run_at": "2026-09-08T09:50:17",
"passed": 4,
"total": 4,
"checks": [
{
"name": "one_utterance_two_readers",
"promise": "the model and the rule matcher read the same words",
"result": "'Turn the lights off, please.' -> turn_off_lights",
"passed": true,
"duration_ms": 0.0088
},
{
"name": "memory_reaches_the_model",
This capture ran the corrected source with stand-ins only; it records text handoffs,
not audible sound or physical light state. Your timestamp and durations will differ.
The verdict line borrows chapter 43's guard, bool(checks) and all(...),
because a review with an empty list would otherwise announce that every handoff held.
asdict flattens each check into a dict for the file, and the
{**asdict(c), "duration_ms": round(...)} spread overrides one key on the
way out, trimming seventeen digits of floating-point noise to something a person can
compare between two reports. The promise field is the reason this file
still means something in November: a bare "memory_reaches_the_model": false
needs somebody to go read the code, while the sentence beside it states what broke.
Why this works: the recording is the evidence
Two separate ideas hold this up, and they are worth keeping apart. The first is the
pessimistic default. A check's fields say how far it got, so there are exactly two ways to
reach passed = True: every assertion held and the last lines ran. Anything
else, an assertion that raised, a KeyError on a trace key you typed wrong, a
division that went wrong inside a helper, leaves the object in the state it was born in
and run_check stamps the exception onto it. A report built this way cannot
lie in the optimistic direction, and optimistic lies are the only ones that get shipped.
The second idea is what the checks are allowed to read. Every one of them takes the trace
and nothing else. A check that rebuilt the prompt itself, or matched the rule itself, would
be testing its own arithmetic and would pass on a system where the real call site delivers
the wrong argument, which is precisely the bug stage 3 found. Reading only the recording
forces each promise to be about the assembled run. It also means growing the review is a
two-step habit: record a new value in run_chain, then write the check that
reads it. A promise nobody recorded is a promise nobody can verify.
Notice what the checks never assert. Not that the lights are on, not that the model was clever, not that the transcript was correct. Those belong to acceptance, and acceptance is the next chapter. These four ask a narrower question: did the value one component handed on arrive at the next one intact and get read as the thing it was. Keep them that narrow and the review survives swapping the transcription model, the language model, or the voice, because none of those change the promises between the parts.
The most confusing result this pattern produces never crashes at all: a check does all its work, prints a perfectly sensible result, and reports failure anyway.
def check_gate_judged_the_same_text(trace: dict) -> ReviewCheck:
check = ReviewCheck("gate_judged_the_same_text",
"the safety gate saw the words the model saw")
start = time.perf_counter()
gate = trace["gate_input"]
assert gate["text"] == trace["message"]["content"], "gate read a different copy of the utterance"
assert gate["action"] == trace["action"], "gate judged an action the chain did not run"
check.duration_ms = (time.perf_counter() - start) * 1000
check.result = f"gate saw {gate['action']} from level {gate['caller']}"
return check # the flag was never set
$ uv run python -m labs.integration_review
[FAIL] gate_judged_the_same_text (0.001 ms): gate saw turn_off_lights from level 3
3/4 handoffs held. Review passed: False
No traceback, and a result string that reads like a success. The two other fields
convict the code in about four seconds: the duration is non-zero and the result is the
sentence written on the second-to-last line, so execution reached the bottom of the
function with every assertion satisfied. Only the flag is missing. Contrast that with a
genuine failure, which shows 0.000 ms and an exception name in the result,
and with an unwritten check, which shows 0.000 ms and
did not finish. Three states, all distinguishable at a glance, which is
what the pessimistic defaults buy you. The fix is to treat check.passed =
True as the closing line of every check, sitting under all the assertions, so
reaching it is the same event as keeping the promise.
Checkpoint, and a number to sign against
- Given a value that is built correctly and never delivered, I can explain why a pairwise seam test passes and a check reading the trace does not.
- I can say which line of
fake_respondmakes delivery checkable, and what a consumer has to hand back for its input to be verifiable at all. - Shown a review row, I can read its three fields together and tell a crashed check from an unfinished one from a check that forgot its flag.
- I can add a fifth promise without touching
run_checkorreview, and name the two steps it takes. - I know what
python -Odoes to anassert, and why that rules assertions out of the pipeline and into the review script.
Exercise 1 — review a refusal. Run the same review over
run_chain(..., channel="socket"), where the permission table denies the
request. Predict how many of the four promises hold before you run it.
All four, and the run prints 4/4 handoffs held exactly as the voice
channel did. The speech line changes to 'No. denied: turn_off_lights requires
level 3, caller has 1.' and
gate_judged_the_same_text now reports level 1 instead of level 3. That
is the review doing its job: a denial is a correct outcome, and these checks are
about whether the values crossed the chain intact, not about which verdict the gate
reached. If you expected failures here, you were asking the questions chapter 59
asks.
Exercise 2 — a promise nobody recorded. Write
check_wake_word_gated asserting that
trace["wake"]["detected"] is true before any action ran, add it to
CHECKS, and run the review before touching
run_chain.
The row comes back [FAIL] wake_word_gated (0.000 ms): KeyError: 'wake'
and the other four still run, which is the crash-proof runner earning its keep on a
mistake that is not an assertion at all. Then close the loop: record
trace["wake"] = {"detected": True, "phrase": "hey glados"} in
run_chain at the point the detector actually gates the turn, and the
same check goes green without an edit. Two steps, in that order, every time you add
a promise.
Exercise 3 — the same promises against real engines. Pass your
real transcriber and a real Ollama call into run_chain as the
transcribe and respond arguments, pointed at a committed WAV,
and run the review.
Wrap the model call so it returns {"prompt": prompt, "text": reply},
which is the contract fake_respond established, and the four checks need
no changes at all. Durations jump from microseconds to hundreds of milliseconds and
your numbers will not match anyone else's;
one_utterance_two_readers may fail outright if the model heard "turn
the light off" instead of "the lights off", and that failure is a real finding about
your rule list, not about the review. Keep the resulting
integration_review.json: it is the first evidence file that describes
the assistant as assembled, running her own engines.
Four handoffs hold, and the file that says so has a timestamp on it. What none of it answers is whether she is good enough to hand to someone else, because "good enough" has not been given a number yet. Next chapter fixes eight of them in advance, measures the assembled assistant against each, and turns the result into a verdict you agreed to before you saw it.