File System and Terminal Access

The previous lesson tied the code execution tool to a resource limit and counted how many calls the limit stopped. What stopped there was quantity: memory, time, output length. The surface the agent touches does not end there: the same loop reads files, writes files, and runs operating system commands. In these three actions what stops is not quantity but permission.

The lesson gathers that permission in a single concept: the permission surface, which is the capability a tool declares together with the path or command it touches. The principle of least privilege was established in the Introduction to Cybersecurity course, and file permissions and ownership were measured in the Introduction to Linux course; neither routine is repeated. The only things counted are these: under three permission sets, how many tasks the agent finishes, how many calls the gate stops, and who is billed for stopping.

• TM31. The code is a simulator; no real file system or runtime is called. The environment, the tools, and the tasks are fictional, seed 20260218.
• TM32. The agent loop is the shared definition of the course: the share of picking the right tool falls with depth, and when the job is done the agent stops with a share of 0.88.
• TM33. The permission surface is the field the tool declares: each tool declares a capability (read, write, execute); the path or command it touches is read from inside the call. The fictional tree carries five paths, and the fictional terminal surface carries four commands.
• TM34. The gate stands in front of the executor. A call that does not pass never reaches the tool; it returns to the loop as a tool error and consumes a step. The three measured sets are full, read-only, and allowlisted.
• TM35. Two conditions are required to call a task completed: the answer must be correct and the step with a side effect in the sequence must have run. Looking only at the answer counts a task that did not write as finished.
• TM36. The baseline is the full permission set — the loop in which the gate stops nothing. The cost is stopped calls, steps, and dropped tasks.

The Shared Loop and the Declaration of the Surface

The environment, the five basic tools, and the agent loop are the same throughout the course.

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_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)


# ---- 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


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("shared environment:", len(SUBSCRIBERS), "subscribers ,", len(DOCUMENTS), "documents ,",
      len(TOOLS), "tools ,", len(TASKS), "tasks")

shared environment: 60 subscribers , 5 documents , 5 tools , 24 tasks

Three Permission Sets and the Gate

The surface opens with three tools: read_file, write_file, run_command; write_record in the shared set also declares a write capability. The gate compares this declaration with the permission set and only stops; nowhere in the code is there a path that exceeds the limit, and every line that runs is on the side of control. The twenty-four fictional surface tasks are spread over six kinds, and three carry a side effect.

# ---- FICTIONAL file tree and terminal surface (specific to this lesson)
TREE = {"data/index.csv": 60, "data/result.csv": 0, "docs/tariff.txt": 5,
        "config/threshold.cfg": 2, "logs/requests.log": 41}
COMMANDS = {"index_dump": 60, "log_tail": 12, "archive_cleanup": 7, "meter_reset": 0}
PATH_HINT = {"endek": "data/index.csv", "sonuc": "data/result.csv", "tarif": "docs/tariff.txt",
              "esik": "config/threshold.cfg", "gunlu": "logs/requests.log"}
COMMAND_HINT = {"dokum": "index_dump", "kuyru": "log_tail",
                 "temiz": "archive_cleanup", "sifir": "meter_reset"}


def resolve_path(a, mode):
    """Read takes the first match , write takes the last."""
    e = [PATH_HINT[t] for t in tokenize(a) if t in PATH_HINT]
    return (e[0] if mode == "read" else e[-1]) if e else "logs/requests.log"


def resolve_command(a):
    return next((COMMAND_HINT[t] for t in tokenize(a) if t in COMMAND_HINT), None)


def t_read_file(query="", subscriber=None, **_):
    path = resolve_path(query, "read")
    if path == "data/index.csv":
        a = SUBSCRIBERS.get(subscriber)
        return {"error": "missing field"} if a is None else {
            "previous": a["previous"], "current": a["current"], "type": a["type"]}
    if path == "docs/tariff.txt":
        return {"result": "TR01"}
    return {"result": ABOVE if path == "config/threshold.cfg" else TREE[path]}


def t_write_file(query="", result=None, **_):
    return {"error": "missing field"} if result is None else {"written": resolve_path(query, "write")}


def t_command(query="", **_):
    k = resolve_command(query)
    return {"error": "no such command"} if k is None else {"result": COMMANDS[k]}


TOOLS.update({"read_file": {"function": t_read_file, "fields": ("query", "subscriber"),
                            "side_effect": False},
              "write_file": {"function": t_write_file, "fields": ("query", "result"),
                             "side_effect": True},
              "run_command": {"function": t_command, "fields": ("query",),
                              "side_effect": True}})
TOOL_NAMES = list(TOOLS)
CAPABILITY = {"read_file": "read", "write_file": "write", "write_record": "write",
              "run_command": "execute"}
PERMISSION = {"full": {"read": ("",), "write": ("",), "command": None},
              "read-only": {"read": ("",), "write": (), "command": ()},
              "allowlisted": {"read": ("data/", "docs/", "config/"), "write": ("data/result",),
                              "command": ("index_dump", "log_tail")}}


def allow(name, state, p):
    """Compares the surface the tool declares with the permission set given."""
    g = CAPABILITY.get(name)
    if g is None:
        return True
    if g == "execute":
        return p["command"] is None or resolve_command(state.get("query", "")) in p["command"]
    if name == "write_record":
        return bool(p[g])
    return any(resolve_path(state.get("query", ""), g).startswith(o) for o in p[g])


ACTIVE, STOPPED = [PERMISSION["full"]], {"count": 0}
_execute = call_tool


def call_tool(name, state):
    """The gate stands in front of the executor; a call that does not pass never reaches the tool."""
    if allow(name, state, ACTIVE[0]):
        return _execute(name, state)
    STOPPED["count"] += 1
    return {"error": "no permission"}


# ---- FICTIONAL surface tasks: 24 tasks , six kinds
KINDS = [("index", "{a} abonesinin endeks satirini oku ve tuketimi hesapla",
          ("read_file", "consumption")),
         ("document", "tarife basamaklarini belgeden oku", ("read_file",)),
         ("threshold", "ustu birim ucret esik degerini ayardan oku", ("read_file",)),
         ("write", "{a} abonesinin endeks farkini hesapla ve sonuc dosyasina yaz",
          ("read_file", "consumption", "write_file")),
         ("dump", "endeks dokumunu calistir", ("run_command",)),
         ("cleanup", "arsiv temizligini calistir", ("run_command",))]
SURFACE = []
for i in range(24):
    r = rng(SEED + 6421 * i)
    sub = f"A{1 + int(r() * 60):03d}"
    kind, text, seq = KINDS[pick(r(), [0.20, 0.12, 0.14, 0.18, 0.18, 0.18])]
    SURFACE.append({"name": f"Y{i + 1:02d}", "kind": kind, "subscriber": sub,
                    "request": text.format(a=sub), "sequence": seq})


def surface_correct(g):
    a = SUBSCRIBERS[g["subscriber"]]
    return {"index": a["current"] - a["previous"], "write": a["current"] - a["previous"],
            "document": "TR01", "threshold": ABOVE, "dump": COMMANDS["index_dump"],
            "cleanup": COMMANDS["archive_cleanup"]}[g["kind"]]


def run(permission, seed=SEED):
    """For a task to be completed the answer must be correct and the step with a side effect must have run."""
    ACTIVE[0] = PERMISSION[permission]
    CALLS[0] = STOPPED["count"] = 0
    c = {"correct": [], "completed": [], "steps": 0, "share": {}}
    for g in SURFACE:
        before = STOPPED["count"]
        s = agent(g, seed=seed)
        d = s["answer"] == surface_correct(g)
        c["correct"].append(d)
        c["completed"].append(d and all(any(x == t and not h for x, h in s["trace"])
                                        for t in g["sequence"] if TOOLS[t]["side_effect"]))
        c["steps"] += s["steps"]
        c["share"][g["kind"]] = c["share"].get(g["kind"], 0) + STOPPED["count"] - before
    c.update(calls=CALLS[0], stopped=STOPPED["count"])
    return c


print("surface task kinds:", {t[0]: sum(g["kind"] == t[0] for g in SURFACE) for t in KINDS})
print("permission set correct A  correct B  completed  executed  stopped  steps  swapped")
table = {}
for p in PERMISSION:
    a, b = run(p), run(p, SEED + 1)
    table[p] = a
    print(f"{p:<14} {sum(a['correct']) / 24:>9.4f}  {sum(b['correct']) / 24:>9.4f}"
          f"  {sum(a['completed']) / 24:>9.4f}  {a['calls']:>8}  {a['stopped']:>7}"
          f"  {a['steps']:>5}  {sum(x != z for x, z in zip(a['completed'], b['completed'])):>7}")
print("\nstopped calls by task kind")
for p in ("read-only", "allowlisted"):
    print(f"  {p:<13}", {k: v for k, v in table[p]["share"].items() if v})

surface task kinds: {'index': 6, 'document': 3, 'threshold': 3, 'write': 5, 'dump': 4, 'cleanup': 3}
permission set correct A  correct B  completed  executed  stopped  steps  swapped
full              0.9583     0.9583     0.9583        54        0     78        2
read-only         0.6250     0.6250     0.5000        46       71    129        0
allowlisted       0.8333     0.8333     0.8333        51       20     92        2

stopped calls by task kind
  read-only     {'write': 22, 'cleanup': 19, 'dump': 30}
  allowlisted   {'cleanup': 20}

Who Is Billed for Stopping

The baseline is the first row: the gate stops no call, and 0.9583 of the tasks finish. Lowering the permission to read raises the stopped calls to 71 but drops the completed tasks to 0.5000; the allowlist stops 20 calls and stays at 0.8333.

The real information is in the last two lines. The 71 stops of read-only are spread over three kinds: write 22, dump 30, cleanup 19. All of the 20 stops of the allowlist, by contrast, are in the cleanup kind. This is the measured form of the principle of least privilege: cutting permission by capability drops three task kinds at once, while the allowlist loads the whole cost on the kind it wants to shut down. Its difference from full permission is only 0.1250, that is, three cleanup tasks.

Whether this difference is measured is told by the last column. The totals of the two runs are the same in all three rows, but a constant total does not mean run-to-run variability is zero: the swapped column says that under full permission and under the allowlist two tasks each changed their result, and that the total did not move because they canceled each other. The measurement band is $2/24 = 0.0833$; 0.1250 is above the band and is measured, and the 0.4583 drop of read-only is far above it.

Two readings remain. The correct answer and the completed task separate under read-only: 0.6250 correct, 0.5000 completed. In the three tasks in between, the agent produced the right number but the write call was stopped at the gate; a measure that looks only at the answer would have counted all three as finished. The steps column says stopping is not free: an error returned from the gate looks to the loop like a tool error, and the agent retries the step. Read-only spends 129 steps instead of 78, that is 65% more, and finishes less work.

Summary

• The permission surface is the capability a tool declares together with the path or command it touches; the gate compares this declaration with the permission set and only stops.
• Under full permission 0.9583 of the 24 tasks finish and no call stops; read-only stops 71 calls and lowers completed tasks to 0.5000; the allowlist stops 20 and stays at 0.8333.
• All 20 stops of the allowlist are in a single task kind, while those of read-only are spread over three kinds; this is the measurable difference of the principle of least privilege.
• A correct answer does not mean a completed task: under read-only the answer was correct in three tasks but the write was stopped, so the measure reads 0.5000 instead of 0.6250.
• Stopping is not free: read-only spends 129 steps, 65% more than the 78 steps of full permission; and because two tasks swapped places in the two runs, the measurement band is 0.0833.

Next Step

In this lesson every surface the agent touched was ours: the file tree, the command set, and the gate were all inside the measurement network. Being able to stop a call was possible because we knew who was permitted to do what. When the agent brings in information from outside, this ground shifts: the retrieved text is not ours, it cannot be checked by a permission set, and once it enters the state it is carried to all the steps. The next lesson treats outside information as a source classification and counts two things: how the class mix changes the share of correct answers, and how many steps a wrong source is carried.