Testing Each Tool Separately from the Loop

The previous lesson split an agent run into two columns: the correctness of the sequence and the correctness of the answer. Both assume the tools work correctly. The moment that assumption is lifted, the four-state table falls short: it says a task failed, it does not say which tool is broken.

This lesson pulls the tools out of the loop and tests them one by one. The test levels were classified in the Quality and Testing Fundamentals course of the Software Quality and Testing curriculum, and the anatomy of the unit test was built in the Unit Testing and Test-Driven Development course; the procedure is not repeated here. The question belongs to measurement: how many of the defects placed in five tools show up in this layer, and which class this layer cannot see at all.

• AQ7. The defects are fictional and are placed in a copy of the tool table; the original table does not change.
• AQ8. There are three defect classes. Value defect: the function gives a wrong result. Contract defect: the function does not report an error where it should. Definition defect: the function is correct, the declared field list is wrong.
• AQ9. Tool-level test tests a tool outside the loop, by calling its function directly; the expected output is written by hand.
• AQ10. A defect counts as caught if it fails at least one case of the test suite.
• AQ11. The test suite uses no seed; the loop measure is run with an eight-step budget and two seeds.

Environment, Tool Definitions, and the Agent Loop

A tool consists of two parts: the function does the computation, the definition declares which fields that function will take, and the loop builds the call only from the declared fields.

# FICTIONAL. Subscriber records, document set, tool set, task set and agent
# loop are generated within the lesson; no real endpoint or tool server is called.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


def tokenize(m):
    b = []
    for s in str(m).lower().replace(",", " , ").split():
        while len(s) > 5:
            b.append(s[:5])
            s = s[5:]
        b.append(s)
    return b


def trace_id(m):
    h = 2166136261
    for c in str(m):
        h = ((h ^ ord(c)) * 16777619) & M32
    return h


REGION = ["east", "west", "north", "south", "central"]
TYPE = ["residential", "commercial"]
SUBSCRIBERS = {}
for i in range(60):
    r = rng(SEED + 31 * i)
    previous = 10000 + int(r() * 80000)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "previous": previous, "current": previous + int(r() * 70)}
TIERS, ABOVE = ((10, 4), (25, 7), (40, 11)), 16
COMMERCIAL_FACTOR = 2
DOCUMENTS = {"TR01": "tarife basamaklari 10 25 40 metrekup ve ustu",
             "TR02": "isyeri abonesinde birim ucret iki katidir",
             "SY01": "endeks farki tuketimi verir sayac donusu 100000 eklenir",
             "AR01": "ariza kaydi acildiginda saha ekibi iki gun icinde gider",
             "KS01": "planli kesinti duyurusu bir hafta once yayimlanir"}

CALLS = [0]                              # tool call counter


def _charge(m3, type_):
    remaining, lower, total = m3, 0, 0
    for upper, unit in TIERS:
        band = min(remaining, upper - lower)
        if band <= 0:
            break
        total += band * unit
        remaining -= band
        lower = upper
    total += remaining * ABOVE
    return total * (COMMERCIAL_FACTOR if type_ == "commercial" else 1)


def t_read_meter(subscriber=None, **_):
    a = SUBSCRIBERS.get(subscriber)
    if a is None:
        return {"error": "no such subscriber"}
    return {"previous": a["previous"], "current": a["current"], "type": a["type"],
            "region": a["region"]}


def t_consumption(previous=None, current=None, **_):
    if previous is None or current is None:
        return {"error": "missing field"}
    f = current - previous
    return {"m3": f + 100000 if f < 0 else f}


def t_tariff(m3=None, type="residential", **_):
    if m3 is None:
        return {"error": "missing field"}
    return {"amount": _charge(m3, type)}


def t_document(query=None, **_):
    if not query:
        return {"error": "missing field"}
    q = set(tokenize(query))
    best, code = -1, None
    for k, m in DOCUMENTS.items():
        o = len(q & set(tokenize(m)))
        if o > best:
            best, code = o, k
    return {"code": code, "text": DOCUMENTS[code]} if best > 0 else {"error": "not found"}


def t_record(subscriber=None, column=None, value=None, **_):
    if not (subscriber and column and value is not None):
        return {"error": "missing field"}
    return {"written": True, "subscriber": subscriber, "column": column, "value": value}


TOOLS = {
    "read_meter":    {"function": t_read_meter, "fields": ("subscriber",), "side_effect": False},
    "consumption":   {"function": t_consumption, "fields": ("previous", "current"),
                      "side_effect": False},
    "tariff":        {"function": t_tariff, "fields": ("m3", "type"), "side_effect": False},
    "search_docs":   {"function": t_document, "fields": ("query",), "side_effect": False},
    "write_record":  {"function": t_record, "fields": ("subscriber", "column", "value"),
                      "side_effect": True},
}
TOOL_NAMES = list(TOOLS)


def call_tool(name, state):
    """Fields outside the schema are filtered out; a missing field is the tool's own error."""
    CALLS[0] += 1
    if name not in TOOLS:
        return {"error": "no such tool"}
    kw = {a: state[a] for a in TOOLS[name]["fields"] if a in state}
    return TOOLS[name]["function"](**kw)


TASKS = []
for i in range(24):
    r = rng(SEED + 9973 * i)
    sub = f"A{1 + int(r() * 60):03d}"
    t = pick(r(), [0.30, 0.22, 0.20, 0.28])
    if t == 0:
        TASKS.append({"name": f"G{i + 1:02d}", "kind": "charge", "subscriber": sub,
                      "request": f"{sub} abonesinin bu donem bedeli nedir",
                      "sequence": ("read_meter", "consumption", "tariff")})
    elif t == 1:
        TASKS.append({"name": f"G{i + 1:02d}", "kind": "consumption", "subscriber": sub,
                      "request": f"{sub} abonesinin bu donem tuketimi kac metrekup",
                      "sequence": ("read_meter", "consumption")})
    elif t == 2:
        TASKS.append({"name": f"G{i + 1:02d}", "kind": "document", "subscriber": sub,
                      "request": "sayac donusunde endeks farki nasil hesaplanir",
                      "sequence": ("search_docs",)})
    else:
        TASKS.append({"name": f"G{i + 1:02d}", "kind": "record", "subscriber": sub,
                      "request": f"{sub} abonesinin bedelini hesapla ve kayda yaz",
                      "sequence": ("read_meter", "consumption", "tariff", "write_record")})


def correct_answer(g):
    a = SUBSCRIBERS[g["subscriber"]]
    f = a["current"] - a["previous"]
    m3 = f + 100000 if f < 0 else f
    if g["kind"] == "consumption":
        return m3
    if g["kind"] == "document":
        return "SY01"
    return _charge(m3, a["type"])


def selection(d):
    return max(0.34, 0.80 - 0.09 * (d - 1))       # share of picking the right tool, falls with depth


def agent(g, max_steps=8, seed=SEED, self_critique=False, approval=False):
    """FICTIONAL agent loop. No real model is called; tool selection
    comes from a single draw that depends on the task's sequence and the depth."""
    r = rng(seed ^ trace_id(g["name"]))
    state = {"subscriber": g["subscriber"], "query": g["request"], "column": "charge"}
    trace, step, tok, errors = [], 0, 0, 0
    seq, k = g["sequence"], 0
    while step < max_steps:
        step += 1
        tok += 60 + 22 * len(trace)
        target = seq[k] if k < len(seq) else None
        if target is None:                         # the job is done , the agent should stop
            if r() < 0.88:
                return {"done": True, "steps": step, "trace": trace, "tokens": tok,
                        "errors": errors, "answer": state.get("result"),
                        "approvals": state.get("approvals", 0)}
            choice = TOOL_NAMES[int(r() * len(TOOL_NAMES))]
        else:
            choice = target if r() < selection(k + 1) else TOOL_NAMES[int(r() * len(TOOL_NAMES))]
        if TOOLS[choice]["side_effect"] and approval:
            state["approvals"] = state.get("approvals", 0) + 1
        if choice == "write_record":
            state["value"] = state.get("result")
        s = call_tool(choice, state)
        trace.append((choice, "error" in s))
        tok += 30
        if "error" in s:
            errors += 1
            if self_critique:
                tok += 40
            continue
        state.update({kk: vv for kk, vv in s.items() if kk != "error"})
        if choice == "consumption":
            state["result"] = s["m3"]
        if choice == "tariff":
            state["result"] = s["amount"]
        if choice == "search_docs":
            state["result"] = s["code"]
        if choice == target:
            k += 1
    return {"done": False, "steps": step, "trace": trace, "tokens": tok, "errors": errors,
            "answer": state.get("result"), "approvals": state.get("approvals", 0)}


print("tool         declared fields             side effect")
for name, v in TOOLS.items():
    print(f"{name:<12} {', '.join(v['fields']):<26}  {v['side_effect']}")

tool         declared fields             side effect
read_meter   subscriber                  False
consumption  previous, current           False
tariff       m3, type                    False
search_docs  query                       False
write_record subscriber, column, value   True

Defect Set and Test Suite

Eleven defects are spread over five tools. Value and contract defects change the function; definition defects leave the function alone and move only the declared field list. The test suite carries one ordinary call and one edge case for each tool.

DEFECTS = {                     # tool , class , description , defective function or field list
    "D1": ("read_meter", "value", "current index same as previous",
           lambda subscriber=None, **_: ({"error": "no such subscriber"}
                                         if subscriber not in SUBSCRIBERS
                                         else dict(SUBSCRIBERS[subscriber],
                                                   current=SUBSCRIBERS[subscriber]["previous"]))),
    "D2": ("consumption", "value", "adds 10000 on meter rollover",
           lambda previous=None, current=None, **_: (
               {"error": "missing field"} if previous is None or current is None
               else {"m3": current - previous + (10000 if current < previous else 0)})),
    "D3": ("tariff", "value", "upper tier unit 11 instead of 16",
           lambda m3=None, type="residential", **_: (
               {"error": "missing field"} if m3 is None else
               {"amount": _charge(m3, type) - max(m3 - 40, 0) * 5
                * (COMMERCIAL_FACTOR if type == "commercial" else 1)})),
    "D4": ("search_docs", "value", "first match instead of highest overlap",
           lambda query=None, **_: next(
               ({"code": k, "text": m} for k, m in DOCUMENTS.items()
                if set(tokenize(query or "")) & set(tokenize(m))),
               {"error": "missing field" if not query else "not found"})),
    "D5": ("consumption", "contract", "zero instead of error on missing field",
           lambda previous=None, current=None, **_: t_consumption(previous or 0, current or 0)),
    "D6": ("tariff", "contract", "zero instead of error on missing field",
           lambda m3=None, type="residential", **_: t_tariff(m3 or 0, type)),
    "D7": ("write_record", "contract", "says written when value is missing",
           lambda subscriber=None, column=None, value=None, **_: {
               "written": True, "subscriber": subscriber, "column": column, "value": value}),
    "D8": ("tariff", "definition", "definition does not declare the type field", ("m3",)),
    "D9": ("write_record", "definition", "definition does not declare the value field",
           ("subscriber", "column")),
    "D10": ("search_docs", "definition", "definition declares subscriber instead of query",
            ("subscriber",)),
    "D11": ("read_meter", "definition", "definition declares an unused field",
            ("subscriber", "column")),
}
CLASSES = ("value", "contract", "definition")


def defective(code):
    """Copy of the table with a defect: if a tuple is given the declared field list,
    if a function is given the tool's function is replaced."""
    t = {a: dict(v) for a, v in TOOLS.items()}
    name, defect = DEFECTS[code][0], DEFECTS[code][3]
    t[name]["fields" if isinstance(defect, tuple) else "function"] = defect
    return t


A1 = "A001"
TESTS = [
    ("read_meter", {"subscriber": A1}, {"previous": SUBSCRIBERS[A1]["previous"],
                                        "current": SUBSCRIBERS[A1]["current"]}),
    ("read_meter", {"subscriber": "A999"}, {"error": "no such subscriber"}),
    ("consumption", {"previous": 100, "current": 150}, {"m3": 50}),
    ("consumption", {"previous": 99980, "current": 30}, {"m3": 50}),
    ("consumption", {"current": 150}, {"error": "missing field"}),
    ("tariff", {"m3": 50, "type": "commercial"}, {"amount": 940}),
    ("tariff", {}, {"error": "missing field"}),
    ("search_docs", {"query": "isyeri abonesinde birim ucret"}, {"code": "TR02"}),
    ("search_docs", {"query": ""}, {"error": "missing field"}),
    ("write_record", {"subscriber": A1, "column": "charge", "value": 500}, {"written": True}),
    ("write_record", {"subscriber": A1, "column": "charge"}, {"error": "missing field"}),
]


def tool_test(table):
    """Tests each tool outside the loop, by calling its function directly."""
    failed = []
    for name, inp, expected in TESTS:
        c = table[name]["function"](**inp)
        if ("error" in c) != ("error" in expected) or any(
                c.get(k) != v for k, v in expected.items()):
            failed.append(name)
    return failed


def loop_success(table, seed=SEED):
    global TOOLS
    original, TOOLS = TOOLS, table
    d = [g["name"] for g in TASKS
         if agent(g, max_steps=8, seed=seed)["answer"] != correct_answer(g)]
    TOOLS = original
    return d


print(f"test suite {len(TESTS)} cases , failing on the defect-free table "
      f"{len(tool_test(TOOLS))}")
print(f"defect set {len(DEFECTS)} defects , {len(TOOLS)} tools , {len(CLASSES)} classes\n")
print(f"{'code':<5} {'tool':<12} {'class':<10} {'failed':>6}  description")
caught = {s: [0, 0] for s in CLASSES}
for code, (name, cls, desc, defect) in DEFECTS.items():
    d = tool_test(defective(code))
    caught[cls][1] += 1
    caught[cls][0] += 1 if d else 0
    print(f"{code:<5} {name:<12} {cls:<10} {len(d):>6}  {desc}")

print("\nclass      defects     caught   share")
for s in CLASSES:
    a, b = caught[s]
    print(f"{s:<10} {b:>7} {a:>10} {a / b:>7.4f}")
tp = sum(a for a, _ in caught.values())
print(f"{'total':<10} {len(DEFECTS):>7} {tp:>10} {tp / len(DEFECTS):>7.4f}")

print("\ntwo runs , defect-free table")
print(f"tool-level test , failing cases {tool_test(TOOLS)} and {tool_test(TOOLS)}")
for t in (SEED, 20260219):
    d = loop_success(TOOLS, t)
    print(f"loop seed {t} , success {1 - len(d) / 24:.4f} , failing task {d}")

test suite 11 cases , failing on the defect-free table 0
defect set 11 defects , 5 tools , 3 classes

code  tool         class      failed  description
D1    read_meter   value           1  current index same as previous
D2    consumption  value           1  adds 10000 on meter rollover
D3    tariff       value           1  upper tier unit 11 instead of 16
D4    search_docs  value           1  first match instead of highest overlap
D5    consumption  contract        1  zero instead of error on missing field
D6    tariff       contract        1  zero instead of error on missing field
D7    write_record contract        1  says written when value is missing
D8    tariff       definition      0  definition does not declare the type field
D9    write_record definition      0  definition does not declare the value field
D10   search_docs  definition      0  definition declares subscriber instead of query
D11   read_meter   definition      0  definition declares an unused field

class      defects     caught   share
value            4          4  1.0000
contract         3          3  1.0000
definition       4          0  0.0000
total           11          7  0.6364

two runs , defect-free table
tool-level test , failing cases [] and []
loop seed 20260218 , success 0.9583 , failing task ['G23']
loop seed 20260219 , success 0.9583 , failing task ['G15']

Seven Caught, Four Unseen

The strength of the layer is complete in the first two classes: all four value defects and all three contract defects fail a case in the test suite. That the number of failing cases is one each is also information; the defect shows in a single case written for its own tool, and the name of the failing case gives the broken tool directly. What the four-state table cannot say is said here in one line.

The unseen class is not chance; it is the result of the call path. The test suite calls the tool’s function directly and never looks at the declared field list. The loop, on the other hand, builds the call through call_tool and filters from the state only the declared fields. A definition defect sits exactly inside this filter: the tariff function applies the commercial factor correctly, but because its definition does not declare the type field, that field never passes in the loop and every commercial subscriber is computed from the residential tariff. A test that calls the function cannot see this, because the function is correct. For all four definition defects the result is 0.0000.

Total catch is 0.6364. This number is specific to this defect set; what carries over is not the rate but how the rate is divided. A layer that tests the function is blind to the mismatch between function and definition, and this blindness does not close by adding cases.

A Layer Independent of the Run

The payoff of the layer is in the last block. On the defect-free tool table the loop gives 0.9583 at both seeds, but the failing task changes: G23 in the first run, G15 in the second. The tool-level test was run twice on the same table, and the set of failing cases came out empty and identical both times, because this layer uses no seed.

The verdict of the tool-level test is repeatable: a defect either shows in every run or never shows. The verdict of the loop measure is banded: the difference between the two runs is a single task, and no sign below the band counts as a finding.

Summary

• A tool is the combination of a function and a definition; the tool-level test calls the function directly and never uses the definition.
• The eleven-case test suite catches all four value defects and all three contract defects; each defect shows in a single case, and the failing case names the broken tool.
• None of the four definition defects is caught: the share is 0.0000, the total catch is 0.6364.
• The cause of the blindness is the call path: the loop filters only the declared fields, and the test suite calls the function directly and skips that filter.
• The payoff of the layer is independence from the run: while the failing task of the loop shifts from G23 to G15 when the seed changes, the set of failing cases of the test suite is empty in both runs.

Next Step

This lesson drew the limit of a layer with a number: the tool-level test says cheaply and exactly whether a tool keeps its own word, but it never sees how the tool is connected to the loop. The four defects that go unseen have to show up somewhere, because the agent writes a wrong charge to commercial subscribers. The next lesson runs scenarios end to end and asks two questions together: how many of the defects that the tool level missed does the flow test catch, and which defect do both layers fail to see.