building a harness — scaffold, tools, loop, gate (step 8/9) · agent harnesses

write build_harness_run(spec, inputs, tools, fake_model, max_iters) end-to-end with the fake_model:

prep: build messages from spec + inputs — include name, trigger, the inputs dict, decisions, and writeback in the user turn.
loop up to max_iters:
- call fake_model(messages)
- parse text vs tool_use blocks
- on end_turn: return {text, iters, tool_calls, gate}
- dispatch only names on spec["allowlist"]. skip the rest. track tool_calls for names you actually ran.
on cap: text is "capped", iters is max_iters.
gate: "fail" if the allowlist was violated (the model named a tool that is not on spec["allowlist"]) or the run hit max_iters (text == "capped" or iters > spec["eval_gate"]["max_iters"]). otherwise "pass" iff spec["eval_gate"]["must_call"] was dispatched and spec["eval_gate"]["must_include"] is in text.

two cases run. the spec is a inbox / queue triage. expected output:

text='ROUTE night.' iters=2 tool_calls=1 gate=pass
text='done.' iters=2 tool_calls=0 gate=fail

TOOLS = {
    "lookup": lambda q: f"hits for {q}",
    "writeback": lambda path, body: f"wrote {path}",
}

SPEC = {
    "name": "inbox triage",
    "trigger": "item landed in the queue",
    "inputs": ["item.body", "queue.history"],
    "decisions": {"route": {"reversible": True}},
    "writeback": "queue/log.jsonl",
    "eval_gate": {"must_call": "lookup", "must_include": "ROUTE", "max_iters": 4},
    "allowlist": ["lookup", "writeback"],
}

INPUTS = {"item.body": "urgent in the queue"}

good_responses = iter([
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "t1", "name": "lookup", "input": {"q": "overtime"}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": "ROUTE night."}]},
])

bad_responses = iter([
    {"stop_reason": "tool_use", "content": [{"type": "tool_use", "id": "b1", "name": "wipe", "input": {}}]},
    {"stop_reason": "end_turn", "content": [{"type": "text", "text": "done."}]},
])

current = {"iter": good_responses}
def fake_model(messages):
    return next(current["iter"])

# define build_harness_run(spec, inputs, tools, fake_model, max_iters) below

current["iter"] = good_responses
r = build_harness_run(SPEC, INPUTS, TOOLS, fake_model, 5)
print(f"text={r['text']!r} iters={r['iters']} tool_calls={r['tool_calls']} gate={r['gate']}")

current["iter"] = bad_responses
r = build_harness_run(SPEC, INPUTS, TOOLS, fake_model, 5)
print(f"text={r['text']!r} iters={r['iters']} tool_calls={r['tool_calls']} gate={r['gate']}")

write build_harness_run(spec, inputs, tools, fake_model, max_iters) end-to-end with the fake_model:

prep: build messages from spec + inputs — include name, trigger, the inputs dict, decisions, and writeback in the user turn.
loop up to max_iters:
- call fake_model(messages)
- parse text vs tool_use blocks
- on end_turn: return {text, iters, tool_calls, gate}
- dispatch only names on spec["allowlist"]. skip the rest. track tool_calls for names you actually ran.
on cap: text is "capped", iters is max_iters.
gate: "fail" if the allowlist was violated (the model named a tool that is not on spec["allowlist"]) or the run hit max_iters (text == "capped" or iters > spec["eval_gate"]["max_iters"]). otherwise "pass" iff spec["eval_gate"]["must_call"] was dispatched and spec["eval_gate"]["must_include"] is in text.

two cases run. the spec is a inbox / queue triage. expected output:

text='ROUTE night.' iters=2 tool_calls=1 gate=pass
text='done.' iters=2 tool_calls=0 gate=fail

full-screen editor opens — close anytime to keep reading.

building a harness — scaffold, tools, loop, gate — step 8 of 9