--- title: 'Scenario-Based End-to-End Validation' source: 'https://academia.sh/en/courses/ai-agents/scenario-based-end-to-end-validation' course: 'AI Agents' language: en updated: '2026-10-21T05:02:26+00:00' license: 'CC BY-SA 4.0' objectives: - 'Builds the end-to-end flow test over the scenario set and determines the defects caught relative to its defect-free baseline.' - 'Compares the defect sets caught by the two testing layers: counts their intersection, their differences, and their union.' - 'Uses the defect activation count to separate a defect that never runs from a defect that runs but does not reach the result.' - 'Names the residual defect that no run can see and explains why it is invisible.' --- # Scenario-Based End-to-End Validation Counting side by side the defect sets caught by two testing layers: the twenty-four-scenario end-to-end flow test catches three of the four definition defects that the tool-level test cannot see at all, and fails on five of the eleven defects in total. The tool-level layer catches seven, the intersection of the two layers is two, and the union is ten of eleven defects, that is 0.9091. The two separate causes of a miss are told apart by the defect activation count: D2, D4, and D11 never activate once during the loop; D5, D6, and D7 activate but the scenario result does not change. The residual defect D11 is a defect whose function is correct but whose definition declares a field it never uses, and no run can see it. On the defect-free baseline the failing scenario in the two runs is G23 and G15. The previous lesson drew the limit of tool-level testing: seven of eleven defects were caught, and none of the four definition defects showed. Not being seen does not make them harmless; the charge of commercial subscribers is computed wrongly. These defects come out only when the tools are **connected to each other**. This lesson builds the second layer. The end-to-end flow test runs not the tool but the **scenario**, and at the end checks both the sequence and the answer. What is measured is not the layer but **the relation between the defect sets of the two layers**. - **AQ12.** The defect set is **the same** as in the previous lesson; the defects are placed in a copy of the table. **The tool-level column is the measurement of that lesson**; the test suite is not repeated here. - **AQ13.** **End-to-end flow test**: twenty-four scenarios are run with an eight-step budget. A scenario passes only if **both the sequence and the answer are correct**. - **AQ14.** On the defect-free table its baseline is **23 of 24**. A defect counts as caught by the flow layer if its failing scenario set differs from this baseline **in both runs**. - **AQ15.** **Defect activation count**: at every call the same input is also given to the defect-free table, and the counter goes up when the two results differ. **Zero activations** means the defect did not show up in the loop even once. - **AQ16.** The seeds are 20260218 and 20260219; a difference below the run-to-run difference **counts as unmeasured**. ## Fictional Environment and Agent Loop Every scenario has a known tool sequence and a known correct answer; the oracle of the test is this pair. ```python # 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(f"fictional environment {len(SUBSCRIBERS)} subscribers , {len(TOOLS)} tools , " f"{len(TASKS)} scenarios") ``` ``` fictional environment 60 subscribers , 5 tools , 24 scenarios ``` ## Two Layers Side by Side The only thing the flow test sees is the sequence the loop produced and the answer. The defect activation counter puts one more column next to it: in how many calls the defect departed from the defect-free table in that run. The tool-level column comes from the previous lesson as input. ```python DEFECTS = { "D1": ("read_meter", "value", lambda subscriber=None, **_: ({"error": "no such subscriber"} if subscriber not in SUBSCRIBERS else dict(SUBSCRIBERS[subscriber], current=SUBSCRIBERS[subscriber]["previous"]))), "D2": ("consumption", "value", 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", 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", 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", lambda previous=None, current=None, **_: t_consumption(previous or 0, current or 0)), "D6": ("tariff", "contract", lambda m3=None, type="residential", **_: t_tariff(m3 or 0, type)), "D7": ("write_record", "contract", lambda subscriber=None, column=None, value=None, **_: { "written": True, "subscriber": subscriber, "column": column, "value": value}), "D8": ("tariff", "definition", ("m3",)), # does not declare the type field "D9": ("write_record", "definition", ("subscriber", "column")), # does not declare the value field "D10": ("search_docs", "definition", ("subscriber",)), # subscriber instead of query "D11": ("read_meter", "definition", ("subscriber", "column")), # field the function does not use } CLEAN = {a: dict(v) for a, v in TOOLS.items()} 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, p = DEFECTS[code][0], DEFECTS[code][2] t[name]["fields" if isinstance(p, tuple) else "function"] = p return t # The layer of the previous lesson is input: the eleven-case test suite was built there # and these are the defects it failed. The computation is not repeated here. TOOL_LEVEL = {"D1", "D2", "D3", "D4", "D5", "D6", "D7"} ACTIVATIONS = [0] _call_tool = call_tool def call_tool(name, state): """Builds the call with the table in force, also gives the same input to the defect-free table, and counts an activation when the two results differ.""" b = _call_tool(name, state) kw = {a: state[a] for a in CLEAN[name]["fields"] if a in state} ACTIVATIONS[0] += b != CLEAN[name]["function"](**kw) return b def sequence_correct(g, trace): k = 0 for name, failed in trace: if k < len(g["sequence"]) and name == g["sequence"][k] and not failed: k += 1 return k == len(g["sequence"]) def flow_test(table, seed=SEED): """End-to-end scenario: it passes if both the sequence and the answer are correct.""" global TOOLS original, TOOLS, ACTIVATIONS[0] = TOOLS, table, 0 d = [] for g in TASKS: s = agent(g, max_steps=8, seed=seed) if not (sequence_correct(g, s["trace"]) and s["answer"] == correct_answer(g)): d.append(g["name"]) TOOLS = original return d, ACTIVATIONS[0] baseline = [flow_test(CLEAN, t) for t in (SEED, 20260219)] print(f"flow test {len(TASKS)} scenarios , failing scenario on the defect-free table " f"{baseline[0][0]} and {baseline[1][0]}") print("\ncode tool class tool-level flow-1 flow-2 act-1 act-2") by_tool, by_flow = set(), set() for code, (name, cls, p) in DEFECTS.items(): t, at_tool = defective(code), code in TOOL_LEVEL (d1, a1), (d2, a2) = flow_test(t, SEED), flow_test(t, 20260219) if at_tool: by_tool.add(code) if d1 != baseline[0][0] and d2 != baseline[1][0]: by_flow.add(code) print(f"{code:<5} {name:<12} {cls:<10} {'+' if at_tool else '-':>10} {len(d1):>7} " f"{len(d2):>7} {a1:>7} {a2:>7}") n = len(DEFECTS) print() for label, s in (("both", by_tool & by_flow), ("flow only", by_flow - by_tool), ("tool level only", by_tool - by_flow), ("residual", set(DEFECTS) - by_tool - by_flow)): print(f"{label:<16} {' '.join(k for k in DEFECTS if k in s)}") print(f"tool level {len(by_tool)}/{n} = {len(by_tool) / n:.4f} , flow {len(by_flow)}/{n} = " f"{len(by_flow) / n:.4f} , union {len(by_tool | by_flow)}/{n} = " f"{len(by_tool | by_flow) / n:.4f}") ``` ``` flow test 24 scenarios , failing scenario on the defect-free table ['G23'] and ['G15'] code tool class tool-level flow-1 flow-2 act-1 act-2 D1 read_meter value + 17 17 18 20 D2 consumption value + 1 1 0 0 D3 tariff value + 7 7 9 8 D4 search_docs value + 1 1 0 0 D5 consumption contract + 1 1 2 1 D6 tariff contract + 1 1 4 5 D7 write_record contract + 1 1 2 1 D8 tariff definition - 5 6 6 5 D9 write_record definition - 3 3 7 14 D10 search_docs definition - 7 8 44 49 D11 read_meter definition - 1 1 0 0 both D1 D3 flow only D8 D9 D10 tool level only D2 D4 D5 D6 D7 residual D11 tool level 7/11 = 0.6364 , flow 5/11 = 0.4545 , union 10/11 = 0.9091 ``` ## Two Sets That Do Not Intersect The tool-level layer catches **7/11**, the flow layer **5/11**. The intersection is only **two defects**: `D1` and `D3`. `D1` activates on every meter reading and fails **seventeen** of the twenty-four scenarios; `D3` activates only on consumptions that exceed the upper tier, and fails **seven** of them. The real information is outside the intersection. The flow layer catches **`D8`, `D9`, `D10`**, that is, three of the four definition defects the tool level could not see at all. The reason is in the mechanism: the flow test sees the tool not through its function but through **the call the loop builds**, and that call passes through the declared field list. `D10` gives the highest activation count: **44** and **49**. In the opposite direction **`D2`, `D4`, `D5`, `D6`, `D7`** show up only at the tool level. The union is **10/11 = 0.9091**; neither layer alone comes close to it. ## Defects That Do Not Activate, and the Residual The defects the flow misses have **two separate causes**, and the activation column tells them apart. `D2` and `D4` activate **zero** times in both runs: the meter rollover branch is never entered in this task set, and in the document query the first overlapping document is already the one with the highest overlap. The defect did not show up even once; the flow had no chance of finding it. `D5`, `D6`, and `D7`, by contrast, do activate, in 2, 4, and 2 calls respectively, but the failing scenario set does not change: the defect was run, it spoiled the result, and **the scenario passed anyway**. The first is a gap in coverage, the second a gap in observation. The activation count is not read as a magnitude. `D9` activates 7 times in the first run and 14 in the second; `D1` 18 and 20. What is read is **whether it is zero**. The residual defect is a single defect and its name is `D11`. The `read_meter` function is correct, but its definition declares a field the function never uses. The tool-level test cannot see it, because the function works correctly. The flow test cannot see it, because the extra declared field enters the call and is swallowed silently: the activation count is **zero** in both runs. The defect is not in the behavior but **in the contract itself**, and the one who will read it is someone else, who will call the tool. ## Summary - The flow test runs the scenario and sees only the sequence and the answer; the failing scenario on the defect-free baseline is G23 and G15 in the two runs. - The flow layer catches 5 of the 11 defects, the tool-level layer 7; the intersection is only `D1` and `D3`, and the union is 10/11 = **0.9091**. - The flow layer catches three of the four definition defects the tool level could not see, because it passes the call through the declared field list. - The cause of a miss splits in two: `D2`, `D4`, and `D11` never activate; `D5`, `D6`, and `D7` activate but do not change the scenario result. - The residual defect is `D11`: the function is correct, the definition declares an extra field; because the activation count is zero, no run can show this defect. ## Next Step Together the two layers found ten of the eleven defects, but they have one thing in common: we **put the defect in** and knew from the start where it was. Testing sees the runs in which the agent is tested; it says nothing about a loop running in production. There is no failing scenario there, only a user and a wrong answer. The next lesson measures what a run leaves behind: at three trace levels, the share of failed tasks whose cause can be found, and the stored token.