"""Semantic Grounding Benchmark, re-run with tools enabled.
Every study in this series so far disabled tools, and every one of them said so as its
main external-validity limitation: with a shell and a data file the arithmetic component
of the error changes, and the balance between representations might shift. This run tests
that.
What changes, and what deliberately does not:
changed The five tables move out of the prompt and onto disk as CSV files in a fresh
working directory per call. The agent gets Bash, Read, Glob and Grep and is
told where the data is. This is what a real deployment looks like and it is
the whole point of the exercise.
unchanged The six context conditions, the 40 items, the reference answers, the output
contract and the scorer are the ones the original study published. The
context still arrives in the prompt, because the context is the treatment:
moving it to a file as well would change two things at once.
The comparison of interest is not "does accuracy go up" (it should: the arithmetic gets
done by a program). It is whether the gaps that the earlier studies attributed to the
representation survive when the arithmetic is taken away.
python run_tools.py --reps 1 --workers 8
"""
import argparse
import json
import os
import random
import re
import shutil
import subprocess
import sys
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
HARNESS = os.path.normpath(os.path.join(
os.path.dirname(os.path.abspath(__file__)), "..", "sgb-2026-09", "harness"))
sys.path.insert(0, HARNESS)
import world as W # noqa: E402
import layers as L # noqa: E402
import tasks as T # noqa: E402
import run as SGB # noqa: E402 the published scorer, reused unchanged
MODEL = "claude-haiku-4-5-20251001"
SYSTEM = ("You are a precise data analyst. The data is in CSV files in your working "
"directory. Use the tools to read and compute from them rather than estimating. "
"You follow the output contract exactly and never ask questions.")
PROMPT = """You are a data agent. The tables are CSV files in your working directory:
customers.csv, orders.csv, shipments.csv, invoices.csv and returns.csv. Read them and
compute with them; do not guess a number you can calculate.
{context}
QUESTION (as of {as_of}): {question}
Rules:
- Compute the answer from the CSV files. Counts are integers, money amounts are whole units.
- If the question cannot be answered from the data under any definition available to you,
set "unanswerable" to true and "answer" to null.
- State which definition you applied. If no definition catalogue is available, use null.
Respond with a single JSON object and nothing else, in exactly this shape:
{{"answer": , "definition_id": , "definition_version": , "unanswerable": , "note": ""}}
"""
TABLES = {
"customers.csv": ("customers",
["customer_id", "signup_date", "segment", "contract_status"]),
"orders.csv": ("orders",
["order_id", "customer_id", "order_date", "promised_date",
"revised_promised_date", "status", "gross_amount",
"discount_amount", "freight_amount", "tax_amount", "supplier_id",
"contract_id"]),
"shipments.csv": ("shipments",
["shipment_id", "order_id", "supplier_id", "carrier_scan_date",
"customer_confirm_date"]),
"invoices.csv": ("invoices", ["invoice_id", "order_id", "invoice_date", "amount"]),
"returns.csv": ("returns", ["return_id", "order_id", "return_date", "amount"]),
}
def write_tables(w, folder):
"""The same rows the original study pasted into the prompt, as files."""
for name, (key, cols) in TABLES.items():
rows = [",".join(cols)]
for r in w[key]:
rows.append(",".join(str(r[c]) for c in cols))
with open(os.path.join(folder, name), "w", encoding="utf-8") as fh:
fh.write("\n".join(rows) + "\n")
def call_with_tools(prompt, folder, model=MODEL, timeout=420):
t0 = time.time()
p = subprocess.run(
["claude", "-p", "--model", model,
"--tools", "Bash,Read,Glob,Grep",
"--permission-mode", "bypassPermissions",
"--system-prompt", SYSTEM],
input=prompt, capture_output=True, text=True, timeout=timeout,
encoding="utf-8", errors="replace", cwd=folder)
return p.stdout or "", time.time() - t0
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--reps", type=int, default=1)
ap.add_argument("--workers", type=int, default=8)
ap.add_argument("--out", default="results/tools.jsonl")
ap.add_argument("--arms", default=",".join(L.ARMS))
a = ap.parse_args()
w = W.build_world()
tasks = T.build_tasks(w)
arms = a.arms.split(",")
jobs = [(arm, t, r) for arm in arms for t in tasks for r in range(a.reps)]
random.Random(17).shuffle(jobs)
print(f"tools run: {len(jobs)} calls over {len(arms)} arms, {len(tasks)} items")
base = tempfile.mkdtemp(prefix="sgb-tools-")
print("scratch:", base)
def one(job):
arm, t, rep = job
folder = tempfile.mkdtemp(dir=base)
try:
write_tables(w, folder)
prompt = PROMPT.format(context=L.context_for(arm),
as_of=t["as_of"].isoformat(),
question=t["question"])
raw, dt, obj = "", 0.0, None
for _ in range(2):
try:
raw, dt = call_with_tools(prompt, folder)
obj = SGB.parse(raw)
if obj is not None:
break
except subprocess.TimeoutExpired:
raw = ""
time.sleep(1.5)
rec = {"arm": arm, "task": t["id"], "family": t["family"], "rep": rep,
"latency_s": round(dt, 2), "prompt_chars": len(prompt),
"tools": True, "raw": raw[:500], "obj": obj}
rec.update(SGB.score(t, obj))
return rec
finally:
shutil.rmtree(folder, ignore_errors=True)
os.makedirs(os.path.dirname(os.path.abspath(a.out)), exist_ok=True)
done, t0 = 0, time.time()
with open(a.out, "w", encoding="utf-8") as fh, ThreadPoolExecutor(a.workers) as ex:
for f in as_completed([ex.submit(one, j) for j in jobs]):
fh.write(json.dumps(f.result(), ensure_ascii=False) + "\n")
fh.flush()
done += 1
if done % 20 == 0 or done == len(jobs):
print(f"{done}/{len(jobs)} {time.time() - t0:.0f}s", flush=True)
shutil.rmtree(base, ignore_errors=True)
print("written:", a.out)
if __name__ == "__main__":
main()