Skip to content
academia.sh

Lesson 10 / 22

Code Execution Tools

Measuring, through its resource limits, a tool that runs the program an agent wrote inside a sandbox: three ways of writing a program that computes the same amount spend 15, 172, and 19 steps, hold 2 to 5 cells, and produce 1 to 5 outputs. At the loose limit no call exceeds the declared limit, but three tasks pass the 96-step task budget while the per-call limit stops nothing. The medium limit stops 24 of 28 code calls, and the classes it stops are 20 for steps and 4 for cells; at the tight limit the same 24 calls stop this time on 20 step and 4 output classes. Stopping does not drop a task directly; it drops it by consuming the step budget: on code tasks task success falls from 0.7778 to 0.3333, and in the second run from 1.0000 to 0.5556. The tight limit gives the same task success as the medium limit while cutting wasted steps from 1356 to 552. Running code is entirely on the defensive side.

Published Updated

In this lesson

  • Writes a small execution tool that enforces declared step, cell, and output limits.
  • Counts separately the calls that exceed the declared limit and the calls the limit stopped.
  • Measures through which mechanism stopping lowers task success.
  • Shows with a separate count the task total that a per-call limit cannot see.
Contents

In the previous two lessons every tool did a fixed job and returned, so what a call would cost was never asked. In a tool that runs a program the agent wrote itself, this assumption lifts: two programs that give the same result can spend a hundred times different resources, and that difference is not written in the definition.

The lesson builds such a tool inside a sandbox and is not an attack lesson: no escape, bypass, or privilege escalation routine is written. Running code is entirely on the defensive side — it counts, limits, and stops. Restricting privileges was measured in the Linux and System Administration and Cybersecurity curricula as the principle of least privilege and is not repeated; the only things counted are what the limit stops and what stopping costs.

  • TM17. The environment, the tool set, the task set, the code execution tool, and the agent loop are fictional; no real runtime, execution service, or model endpoint is called. The tariff computation stops being a fixed tool; tariff drops out and run_code takes its place, so the tool set stays at five. Seed 20260218, second run 20260219.
  • TM18. The sandbox is a small register machine and carries three declared resource limits: steps, cells, outputs. When a limit is exceeded the run stops and the exceeded class is reported.
  • TM19. The program the agent writes takes one of three forms, the form is drawn once per task, and all three forms compute the same amount.
  • TM20. The actual consumption of a call is measured with a dry run: the program is run once more with a nearly unlimited setting, and the result enters only the count.
  • TM21. The task budget is 96 execution steps and looks at the total of all the code calls of the task; the sandbox does not see this, because its limit is per call.
  • TM22. The baseline is the loose limit, the step budget is eight, and a code task is one whose sequence contains run_code: nine of the 24 tasks.

The Sandbox and the Three Ways of Writing

The code below builds the shared environment and the task set, then adds the register machine, the declared limits, and the compiler.

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


# ---- FICTIONAL environment: records of the measurement network
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_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},
    "search_docs":   {"function": t_document, "fields": ("query",), "side_effect": False},
    "write_record":  {"function": t_record, "fields": ("subscriber", "column", "value"),
                      "side_effect": True},
}


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)


# ---- FICTIONAL tasks: each one is solved by a known tool sequence
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"])


# ---- FICTIONAL agent: at each step it picks a tool or finishes
def selection(d):
    return max(0.34, 0.80 - 0.09 * (d - 1))       # share of picking the right tool, falls with depth


# ---- FICTIONAL code execution tool: register machine and declared resource limits.
# Running code is on the defensive side: it counts , limits , stops.
LIMITS = {"loose":  {"steps": 512, "cells": 16, "outputs": 16},
          "medium": {"steps": 64,  "cells": 4,  "outputs": 8},
          "tight":  {"steps": 24,  "cells": 4,  "outputs": 2}}
TASK_BUDGET = 96                      # total execution steps per task


def val(x, h):
    return h[x] if isinstance(x, str) else x


def execute(program, limit):
    """When a limit is exceeded the run stops and the exceeded class is reported."""
    cells, outputs, ip, step = {}, [], 0, 0
    while ip < len(program):
        step += 1
        if step > limit["steps"]:
            return {"error": "limit exceeded", "class": "step", "spent": step}
        k = program[ip]
        if k[0] == "set":
            cells[k[1]] = val(k[2], cells)
        elif k[0] == "add":
            cells[k[1]] += val(k[2], cells)
        elif k[0] == "mul":
            cells[k[1]] *= val(k[2], cells)
        elif k[0] == "dec":
            cells[k[1]] -= 1
        elif k[0] == "write":
            outputs.append(cells[k[1]])
        elif k[0] == "branch" and cells[k[1]] > 0:
            ip = k[2]
            continue
        if len(cells) > limit["cells"]:
            return {"error": "limit exceeded", "class": "cell", "spent": step}
        if len(outputs) > limit["outputs"]:
            return {"error": "limit exceeded", "class": "output", "spent": step}
        ip += 1
    return {"amount": outputs[-1] if outputs else None, "spent": step,
            "cells": len(cells), "outputs": len(outputs)}


def compile_program(style, m3, type_):
    """The FICTIONAL program the agent writes; all three compute the same amount."""
    parts, remaining, lower = [], m3, 0
    for upper, unit in TIERS:                    # split into tariff tiers
        share = min(remaining, upper - lower)
        if share <= 0:
            break
        parts.append((share, unit))
        remaining, lower = remaining - share, upper
    if remaining > 0:
        parts.append((remaining, ABOVE))
    p = [("set", "t", 0)]
    for i, (count, unit) in enumerate(parts):
        if style == "short":
            p += [("set", "x", count), ("mul", "x", unit), ("add", "t", "x")]
        elif style == "unit":
            p += [("set", "s", count), ("add", "t", unit), ("dec", "s"),
                  ("branch", "s", len(p) + 1)]
        else:                                    # explicit : a cell and an output per tier
            p += [("set", f"d{i}", count), ("mul", f"d{i}", unit),
                  ("add", "t", f"d{i}"), ("write", f"d{i}")]
    if type_ == "commercial":
        p.append(("mul", "t", COMMERCIAL_FACTOR))
    return p + [("write", "t")]


SETTING = ["loose"]                              # the limit the sandbox declares


def t_code(m3=None, type="residential", program=None, **_):
    if m3 is None or program is None:
        return {"error": "missing field"}
    return execute(program, LIMITS[SETTING[0]])


# The tariff computation is not a fixed tool , it is a program the agent writes; the tool count stays five.
TOOLS["run_code"] = {"function": t_code, "fields": ("m3", "type", "program"), "side_effect": False}
TOOL_NAMES = list(TOOLS)
for g in TASKS:
    g["sequence"] = tuple("run_code" if a == "tariff" else a for a in g["sequence"])
STYLE, STYLE_SHARE = ("short", "unit", "explicit"), [0.40, 0.30, 0.30]

for style in STYLE:
    p = compile_program(style, 55, "commercial")
    s = execute(p, LIMITS["loose"])
    d = "  ".join(f"{a} -> {execute(p, LIMITS[a]).get('class', 'ok')}"
                  for a in ("medium", "tight"))
    print(f"{style:<8} commands {len(p):>2}  steps {s['spent']:>3}  cells {s['cells']}"
          f"  outputs {s['outputs']}  amount {s['amount']}  |  {d}")
short    commands 15  steps  15  cells 2  outputs 1  amount 1100  |  medium -> ok  tight -> ok
unit     commands 19  steps 172  cells 2  outputs 1  amount 1100  |  medium -> step  tight -> step
explicit commands 19  steps  19  cells 5  outputs 5  amount 1100  |  medium -> cell  tight -> output

The three rows compute the same amount, 1100, with three different resource profiles. The command counts are nearly equal — 15, 19, 19 — and no limit can be read from that number. The form that adds unit by unit spends 172 steps, because four of its commands are a backward branch; the explicit form finishes in nineteen steps but opens five cells and produces five outputs. A resource limit is therefore not singular: the step limit stops the second form, and the third is stopped by the cell limit at the medium setting and by the output limit at the tight setting.

What the Limit Stops and What It Cannot

The same 24 tasks run at three limit settings and with two seeds; the calls that exceed the limit are found with a dry run, while the active limit really stops.

def agent_code(g, max_steps=8, seed=SEED):
    """The shared agent loop; the only difference is that the program the agent wrote
    runs in the sandbox instead of the tariff, and every run is recorded."""
    r = rng(seed ^ trace_id(g["name"]))
    style = STYLE[pick(r(), STYLE_SHARE)]          # a single way of writing per task
    state = {"subscriber": g["subscriber"], "query": g["request"], "column": "charge"}
    code_trace, step, seq, k = [], 0, g["sequence"], 0
    while step < max_steps:
        step += 1
        target = seq[k] if k < len(seq) else None
        if target is None:
            if r() < 0.88:
                break
            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 choice == "write_record":
            state["value"] = state.get("outcome")
        prog = None
        if choice == "run_code" and "m3" in state:
            prog = state["program"] = compile_program(style, state["m3"],
                                                      state.get("type", "residential"))
        s = call_tool(choice, state)
        if prog is not None:
            code_trace.append((prog, s))
        if "error" in s:
            continue
        state.update({kk: vv for kk, vv in s.items() if kk != "error"})
        if choice == "consumption":
            state["outcome"] = s["m3"]
        if choice == "run_code":
            state["outcome"] = s["amount"]
        if choice == "search_docs":
            state["outcome"] = s["code"]
        if choice == target:
            k += 1
    return {"steps": step, "code_trace": code_trace, "answer": state.get("outcome")}


def measure(setting, seed=SEED):
    """Limit check: the dry run gives the actual consumption , the active limit stops."""
    SETTING[0], lim, classes = setting, LIMITS[setting], {}
    ok = code = exceeding = stopped = wasted = over_budget = ok_code = n_code = 0
    for g in TASKS:
        s = agent_code(g, seed=seed)
        d = s["answer"] == correct_answer(g)
        ok += d
        if "run_code" in g["sequence"]:
            n_code, ok_code = n_code + 1, ok_code + d
        total = 0
        for prog, outcome in s["code_trace"]:
            dry = execute(prog, LIMITS["loose"])            # for counting only
            code += 1
            exceeding += (dry["spent"] > lim["steps"] or dry["cells"] > lim["cells"]
                          or dry["outputs"] > lim["outputs"])
            total += outcome["spent"]
            if "error" in outcome:
                stopped += 1
                wasted += outcome["spent"]
                classes[outcome["class"]] = classes.get(outcome["class"], 0) + 1
        over_budget += total > TASK_BUDGET
    return code, exceeding, stopped, wasted, over_budget, ok / len(TASKS), ok_code / n_code, classes


print("limit    code calls exceeding stopped wasted steps over budget  success  code tasks")
for setting in LIMITS:
    for seed, e in ((SEED, "A"), (SEED + 1, "B")):
        a = measure(setting, seed)
        print(f"{setting:<7}{e} {a[0]:>10} {a[1]:>9} {a[2]:>7} {a[3]:>12} {a[4]:>11}"
              f"  {a[5]:>7.4f}  {a[6]:>10.4f}")
for setting in ("medium", "tight"):
    print(f"  {setting:<6} stopping classes :", measure(setting)[7])
limit    code calls exceeding stopped wasted steps over budget  success  code tasks
loose  A          9         0       0            0           3   0.8750      0.7778
loose  B         10         0       0            0           1   1.0000      1.0000
medium A         28        24      24         1356           4   0.7083      0.3333
medium B         22        17      17          646           2   0.8333      0.5556
tight  A         28        24      24          552           4   0.7083      0.3333
tight  B         22        17      17          317           1   0.8333      0.5556
  medium stopping classes : {'step': 20, 'cell': 4}
  tight  stopping classes : {'step': 20, 'output': 4}

The second and third columns are equal in every row: the number of calls that exceed the declared limit and the number of calls stopped are the same. A limit placed per call catches every per-call violation; this is not an achievement but the requirement of the definition. What is worth counting is the place the limit does not see.

What it does not see is in the fifth column. At the loose setting no call is stopped, yet three tasks pass the 96-step budget; a limit that allows 512 steps per call does not count how many times a 172-step program was run. At the medium and tight settings the stopped calls are repeated, so the number of tasks that exceed the budget rises to four. Both say the same thing: a per-call limit does not see the task total.

The cost of stopping is in the last column. At the loose setting, task success on code tasks is 0.7778 and 1.0000; at the medium setting it is 0.3333 and 0.5556. The drop is close to 0.4445 in both runs and is twice the run-to-run difference at the loose setting (0.2222), so it is measured. The mechanism is indirect: a stopped call does not drop the task, it consumes the step budget. Because the way of writing is fixed per task, the same program is written again and the budget runs out; this is how code calls rise from 9 to 28.

The only column that separates the medium and tight rows is wasted steps: 1356 against 552. With the stopped calls and the task success the same, the tight limit recovers three fifths of the discarded work; the gain is not in accuracy but in resources.

Summary

  • Three programs that give the same amount spend 15, 172, and 19 steps, hold 2 to 5 cells, and produce 1 to 5 outputs; a resource limit is therefore plural, not singular.
  • In every row the number of exceeding calls equals the number of stopped calls: a per-call limit catches every per-call violation. The place it does not see is the task total: at the loose setting three tasks pass the 96-step budget with no call stopped, and at the limited settings four tasks do.
  • Stopping does not drop a task directly, it drops it by consuming the step budget: on code tasks task success falls from 0.7778 and 1.0000 to 0.3333 and 0.5556, and code calls rise from 9 to 28. The tight limit gives the same task success as the medium limit and cuts wasted steps from 1356 to 552.

Next Step

What was limited here was consumption: how many steps a call would take, how many cells it would hold, how many outputs it would produce. A resource limit does not touch a fourth question at all: what the tool can reach. The program in the sandbox had no outside; the other surfaces the agent touches are not like that. A file tree or a shell can do an irreversible job without lengthening the execution at all. The next lesson counts the permission surface a tool declares.

To keep your progress and take notes, Log in

My notes

Log in to take notes.

Start typing to search.

↑↓ ↵ Esc navigate · open · close