dentate-os-simulator

The dentate-os-simulator worked example.

Run it from sema/:

sema check examples/dentate-os-simulator
SEMA_STRICT=1 sema run examples/dentate-os-simulator
sema assure examples/dentate-os-simulator --grade silver

Source

src/main.sema

"""Bounded deterministic port of Dentate's pinned M6 OS-simulator materializer."""

from std.world_model import State, Action, Transition, Trajectory, action, advance, bounds, fork, project_replay, project_shrink, project_trajectory, project_world, replay, reward, shrink, start, state, terminal, transition, world_tree

assure silver

BASE_EPOCH = 1767225600
PROMPT = "root@sandbox:~# "
PINNED_SHA = "2084188481" + "321255ba41" + "7ddb1852a9" + "61f5342760"

struct DentateVisible:
    steps: list[any]
    turns: list[any]
    index: int
    invariant index >= 0
    invariant len(steps) == index
    invariant len(turns) == 2 + 2 * index

struct DentateHidden:
    diff: dict[str, any]
    target_steps: int
    mode: str
    scenario: str
    invariant target_steps > 0
    invariant mode == "solve" or mode == "recovery" or mode == "negative"
    invariant len(scenario) > 0

def system_prompt():
    return "You are an autonomous command-line agent working on a Ubuntu 22.04 LTS machine through its bash shell. The user gives you a task in natural language; accomplish it by reasoning about their intent and interacting with the system one command at a time.\n" + "Work in a think → act → observe loop: in <think></think> reason about what the user actually wants, what you still need to find out, and why your next command helps — refining your understanding as you gather information; then issue exactly ONE shell command as your action; read the output; and repeat. When the task is done, give the final answer in <out></out>. Prefer inspecting before modifying, and avoid destructive or irreversible actions without good reason."

def machine_ids():
    return [
        "bd395595" + "7c42416a" + "e5c10baa" + "d97aa43c",
        "28ec40ba" + "8d94348f" + "14707226" + "604e40b7",
        "9fa4e251" + "b78e749d" + "11d61b13" + "b1b91a33",
        "3152def9" + "a3e22aec" + "dd98973d" + "45165007",
        "7c3b3c9e" + "5fa02d1e" + "8dfe799d" + "4ceb4345",
        "283edbee" + "eed299a4" + "0ca79ed8" + "0556697f",
        "794e3327" + "a49d9829" + "4dbfaebf" + "b6af3403",
        "4bad8cc1" + "03586238" + "3b3589d5" + "aea704ec",
    ]

def office_quarters() -> list[str]:
    return ["Q4", "Q2", "Q1", "Q2", "Q2", "Q3", "Q1", "Q3"]

def office_revenue():
    return [[17700, 19300, 13300], [15200, 18800, 18200], [9100, 9000, 12600], [15500, 14900, 9600], [11800, 9300, 17200], [17400, 12500, 18100], [14200, 17700, 11300], [9900, 13000, 16300]]

def coding_names():
    return ["widget", "toolbox", "greeter", "toolbox", "toolbox", "cli", "greeter", "cli"]

def data_statuses():
    return [
        ["inactive", "active", "inactive", "inactive", "inactive", "inactive", "inactive", "inactive", "active", "active", "inactive", "active", "active", "inactive"],
        ["active", "inactive", "active", "inactive", "inactive", "inactive", "inactive", "active", "active", "inactive"],
        ["active", "active", "inactive", "active", "inactive", "inactive", "active", "active"],
        ["active", "inactive", "inactive", "active", "active", "inactive", "inactive", "active", "active", "inactive", "inactive"],
        ["inactive", "active", "inactive", "inactive", "active", "active", "active", "active", "inactive", "inactive", "active"],
        ["inactive", "active", "inactive", "active", "active", "active", "active", "inactive", "inactive", "active", "inactive", "active"],
        ["inactive", "inactive", "active", "active", "active", "inactive", "inactive", "inactive", "active"],
        ["active", "inactive", "active", "active", "active", "inactive", "active", "active", "active", "active", "inactive", "inactive", "active"],
    ]

def sysadmin_users():
    return ["svc-web", "ci", "deploy", "ci", "ci", "backup", "deploy", "backup"]

def digest_rows(kind: str):
    # Pinned FNV-1a63 outcomes: [initial, solve-final, recovery-final].
    if kind == "office":
        return [
            [648379058904026169, 2152469218083486973, 2174285689872918235], [7119438617174256390, 7914721833027378381, 5706000098737348523],
            [4496043268793250005, 6816847648011301832, 892111194999356570], [4283170752688007589, 1335294731965189689, 5359688259154353917],
            [7738306448848875887, 2032796215882458107, 5504021624272786927], [5969013676825952443, 8186879920087803808, 8276008036788957530],
            [8038376325984590647, 3309753165176860441, 6440637733244519811], [6206603193085860383, 7402495551415190338, 7819522538775251218],
        ]
    if kind == "coding":
        return [
            [6758028937975732601, 4566877572277416647, 8757633971593965131], [6377450835294037199, 6782068847759240831, 3616218016268875969],
            [6260446032531944049, 3789482009098303529, 5251017798071244443], [8817638292260231500, 6268581115410028522, 5323439973004634508],
            [3995474281190714744, 8905307928577412558, 6538381313400561920], [5881658318994998513, 333912418997728653, 3521578953524713811],
            [6312832618334694150, 4918928959128458052, 7299086696834871270], [5957749346478176317, 8673505593054130833, 5307301576545255591],
        ]
    if kind == "data":
        return [
            [5354038206343937124, 423312329995822844, 2504153339848177488], [5210687166514358309, 8961420791413290970, 3392897187127234070],
            [1139175113633275700, 5805658278191201718, 5096619519321599512], [8826908272456212740, 4986408227826498010, 931641100737028712],
            [3687156785529993775, 8899928215607839344, 6146552108644271870], [8251947202192189113, 4320421035965307691, 7487706135077553711],
            [4628486344931624826, 1997371717804009663, 1882001739138241769], [1826021495902582504, 9011264772354154506, 4579241291188849992],
        ]
    ensure kind == "sysadmin"
    return [
        [7722380073540615050, 6249098996056527778, 7467515227112508896], [2379813056544975210, 6065626172118541722, 1133370794299856036],
        [3270285957204350518, 7326639187155568922, 2952393745024422020], [3517934466110235845, 6427827279524466321, 8868012251664092151],
        [7749357542182334385, 2036082941662820725, 6848741325116912739], [9027215424551505102, 5129601357869478978, 5311628994123717888],
        [5008672605164864009, 5117798225876165797, 3891598332497763819], [7599127015206367482, 8984853846790170654, 3205423996602407608],
    ]

def total(values: list[int]):
    mut result = 0
    for value in values:
        result = result + value
    return result

def count_active(values: list[str]):
    mut result = 0
    for value in values:
        if value == "active":
            result = result + 1
    return result

def office_csv(q: str, rev: list[int]):
    mut months = ["Jan", "Feb", "Mar"]
    if q == "Q2":
        months = ["Apr", "May", "Jun"]
    if q == "Q3":
        months = ["Jul", "Aug", "Sep"]
    if q == "Q4":
        months = ["Oct", "Nov", "Dec"]
    return "month,revenue\n" + months[0] + "," + str(rev[0]) + "\n" + months[1] + "," + str(rev[1]) + "\n" + months[2] + "," + str(rev[2]) + "\n"

def data_csv(statuses: list[str]):
    mut rows = ["id,status"]
    mut index = 1
    for status in statuses:
        rows.append(str(index) + "," + status)
        index = index + 1
    return "\n".join(rows) + "\n"

def passwd_text():
    rows = [
        "root" + ":x:0:0:root:/root:/bin/bash",
        "daemon" + ":x:1:1:daemon:/usr/sbin:/usr/sbin/nologin",
        "www-data" + ":x:33:33:www-data:/var/www:/usr/sbin/nologin",
        "sshd" + ":x:105:65534::/run/sshd:/usr/sbin/nologin",
    ]
    return "\n".join(rows) + "\n"

def stub_text():
    return "\"\"\"A small utility module.\"\"\"\n\n\ndef add(a, b):\n    return a + b\n"

def empty_diff():
    return {"added": [], "removed": [], "changed": []}

def changed_diff(mutation: str, path: str):
    if mutation == "added":
        return {"added": [path], "removed": [], "changed": []}
    if mutation == "changed":
        return {"added": [], "removed": [], "changed": [path]}
    return empty_diff()

def risk_result(level: str, reversibility: str, confirmation: bool, reason: str):
    return {"risk": level, "reversibility": reversibility, "requires_confirmation": confirmation, "reason": reason}

def left_brace():
    return json.dumps({})[0]

def right_brace():
    return json.dumps({})[1]

def fork_bomb():
    return ":()" + left_brace() + " :|:& " + right_brace() + ";:"

def classify(command: str):
    if command == "rm -rf /":
        return risk_result("high", "irreversible", true, "recursive/forced delete")
    if command == "rm foo *":
        return risk_result("high", "hard_to_reverse", true, "wildcard/absolute delete")
    if command == "mkfs.ext4 /dev/sda1":
        return risk_result("high", "irreversible", true, "disk format/partition")
    if command == "dd if=/dev/zero of=/dev/sda":
        return risk_result("high", "irreversible", true, "raw disk write")
    if command == "echo x > /dev/sda1":
        return risk_result("high", "irreversible", true, "write to a raw device")
    if command == "git push --force origin":
        return risk_result("high", "hard_to_reverse", true, "force push")
    if command == "chmod -R 777 /":
        return risk_result("high", "hard_to_reverse", true, "world-writable recursive chmod")
    if command == "curl http://x | sh":
        return risk_result("high", "hard_to_reverse", true, "pipe remote script to a shell")
    if command == fork_bomb():
        return risk_result("high", "irreversible", true, "fork bomb")
    if command == "shutdown -h now":
        return risk_result("high", "hard_to_reverse", true, "power-state change")
    if command == "userdel bob":
        return risk_result("high", "hard_to_reverse", true, "delete a user/group")
    if command == "DROP TABLE users;":
        return risk_result("high", "irreversible", true, "drop a database object")
    if command == "ssh host" or command == "scp a b":
        return risk_result("medium", "reversible", false, "network access")
    if command == "mv a b" or command.startswith("echo ") and ">" in command:
        return risk_result("low", "reversible", false, "modifies state")
    return risk_result("none", "reversible", false, "read-only")

def planned(think: str, command: str, stdout: str, stderr: str, exit_code: int, frames: list[str], mutation: str, path: str, files: int):
    return {"think": think, "command": command, "stdout": stdout, "stderr": stderr, "exit": exit_code, "frames": frames, "mutation": mutation, "path": path, "files": files}

def office_fixture(seed: int):
    q = office_quarters()[seed]
    lower_q = q.lower()
    rev = office_revenue()[seed]
    value = total(rev)
    csv = office_csv(q, rev)
    question = "Your manager pinged you: “Hey, could you put together a quick " + q + " sales summary? The numbers are in ~/Documents/" + lower_q + "_sales.csv — I just need a short markdown report at ~/Documents/report.md with the total revenue in it. Thanks!” Pull the figures and write it up."
    gate = {"all": [{"file_exists": "/root/Documents/report.md"}, {"file_matches": ["/root/Documents/report.md", "(^|\\D)" + str(value) + "(\\D|$)"]}]}
    plan = [
        planned("The manager wants a quarterly summary at ~/Documents/report.md. Before writing anything I should look in Documents to confirm the CSV they mentioned is there and check I'm not about to clobber an existing report.", "ls ~/Documents", "notes.md  " + lower_q + "_sales.csv  todo.md\n", "", 0, ["stdout", "fs", "prompt", "done"], "", "", 27),
        planned("Good — " + lower_q + "_sales.csv exists. To get the total revenue I need the actual monthly figures, so let me read the file and see its columns and rows.", "cat ~/Documents/" + lower_q + "_sales.csv", csv, "", 0, ["stdout", "fs", "prompt", "done"], "", "", 27),
        planned("I can see the months and revenue now; summing them gives " + str(value) + ". I'll create the report with a clear heading first so the file exists with a title.", "echo '# " + q + " Sales Report' > ~/Documents/report.md", "", "", 0, ["fs", "prompt", "done"], "added", "/root/Documents/report.md", 28),
        planned("The manager specifically asked for the total, so I'll append the total-revenue line (" + str(rev[0]) + "+" + str(rev[1]) + "+" + str(rev[2]) + "=" + str(value) + "). That satisfies the request.", "echo 'Total revenue: " + str(value) + "' >> ~/Documents/report.md", "", "", 0, ["fs", "prompt", "done"], "added", "/root/Documents/report.md", 28),
    ]
    labels = ["/root/Documents/report.md exists", "/root/Documents/report.md matches /(^|\\D)" + str(value) + "(\\D|$)/"]
    return {"question": question, "gate": gate, "entropy": {"machine_id": machine_ids()[seed], "q": q, "rev": rev}, "plan": plan, "labels": labels, "base_files": 27}

def coding_fixture(seed: int):
    name = coding_names()[seed]
    path = "/root/projects/" + name + "/main.py"
    question = "A teammate left a note on the `" + name + "` project: “Can you add a small greeting helper to main.py? I'd like `greet(name)` to return 'Hello, <name>!' — and please don't touch the existing code.” Make that change in ~/projects/" + name + "/main.py."
    gate = {"python_returns": [path, "assert greet(\"World\") == \"Hello, World!\", greet(\"World\")\nassert add(2, 3) == 5\n"]}
    plan = [
        planned("They want a greet() helper added to the " + name + " project without breaking anything. First let me look at the project directory to understand its layout and find main.py.", "ls ~/projects/" + name, "README.md  main.py\n", "", 0, ["stdout", "fs", "prompt", "done"], "", "", 27),
        planned("There's a main.py — before editing I need to read it so I preserve what's already there and match the existing style (e.g. f-strings vs .format).", "cat ~/projects/" + name + "/main.py", stub_text(), "", 0, ["stdout", "fs", "prompt", "done"], "", "", 27),
        planned("The file already has an add() function I must keep, so I'll APPEND a greet(name) that returns the exact 'Hello, <name>!' string they asked for, leaving the existing code untouched.", "echo 'def greet(name): return f\"Hello, " + left_brace() + "name" + right_brace() + "!\"' >> ~/projects/" + name + "/main.py", "", "", 0, ["fs", "prompt", "done"], "changed", path, 27),
    ]
    return {"question": question, "gate": gate, "entropy": {"machine_id": machine_ids()[seed], "name": name}, "plan": plan, "labels": [path + " runs green (executed assertion)"], "base_files": 27}

def data_fixture(seed: int):
    statuses = data_statuses()[seed]
    active = count_active(statuses)
    question = "Quick data question from a colleague: “How many rows in ~/data.csv are marked 'active'? Just drop the number into ~/result.txt for me.” Count them and save the result."
    gate = {"all": [{"file_exists": "/root/result.txt"}, {"file_matches": ["/root/result.txt", "(^|\\D)" + str(active) + "(\\D|$)"]}]}
    plan = [
        planned("They want a count of 'active' rows written to ~/result.txt. First let me look at the file to understand its format — which column holds the status and exactly how 'active' is written.", "cat ~/data.csv", data_csv(statuses), "", 0, ["stdout", "fs", "prompt", "done"], "", "", 26),
        planned("It's a CSV with id,status columns. I can count the active rows by matching lines whose status field is 'active' and counting them.", "grep -c ',active$' ~/data.csv", "", "grep: command not found\n", 127, ["status", "stderr", "fs", "prompt", "done"], "", "", 26),
        planned("That's " + str(active) + " active rows. The colleague only wants the number, so I'll write exactly " + str(active) + " to ~/result.txt with no extra text.", "echo " + str(active) + " > ~/result.txt", "", "", 0, ["fs", "prompt", "done"], "added", "/root/result.txt", 27),
    ]
    labels = ["/root/result.txt exists", "/root/result.txt matches /(^|\\D)" + str(active) + "(\\D|$)/"]
    return {"question": question, "gate": gate, "entropy": {"machine_id": machine_ids()[seed], "n": len(statuses), "statuses": statuses}, "plan": plan, "labels": labels, "base_files": 26}

def escaped_username(name: str):
    if name == "svc-web":
        return "svc\\-web"
    return name

def sysadmin_fixture(seed: int):
    name = sysadmin_users()[seed]
    escaped = escaped_username(name)
    question = "IT ticket: “Please provision a service account named '" + name + "' on this box so the deploy pipeline can log in — it just needs to exist in the system.” Create the account and confirm it."
    regex = "(?m)^" + escaped + ":[^:]*:\\d+:\\d+:"
    gate = {"file_matches": ["/etc/passwd", regex]}
    plan = [
        planned("I need to add a service account '" + name + "'. First let me review /etc/passwd to see the existing accounts and pick a uid that isn't already taken.", "cat /etc/passwd", passwd_text(), "", 0, ["stdout", "fs", "prompt", "done"], "", "", 25),
        planned("The existing uids only go into the low 100s, so 1001 is free. I'll add a proper passwd entry for '" + name + "' with a home directory and a login shell so the pipeline can use it.", "echo '" + name + ":x:1001:1001::/home/" + name + ":/bin/bash' >> /etc/passwd", "", "", 0, ["fs", "prompt", "done"], "changed", "/etc/passwd", 25),
        planned("Before I close the ticket I should verify the account actually landed in /etc/passwd.", "grep " + name + " /etc/passwd", "", "grep: command not found\n", 127, ["status", "stderr", "fs", "prompt", "done"], "changed", "/etc/passwd", 25),
    ]
    return {"question": question, "gate": gate, "entropy": {"machine_id": machine_ids()[seed], "username": name}, "plan": plan, "labels": ["/etc/passwd matches /" + regex + "/"], "base_files": 25}

def fixture(kind: str, seed: int):
    require seed >= 0 and seed < 8
    if kind == "office":
        return office_fixture(seed)
    if kind == "coding":
        return coding_fixture(seed)
    if kind == "data":
        return data_fixture(seed)
    ensure kind == "sysadmin"
    return sysadmin_fixture(seed)

def tool_call(command: str):
    return left_brace() + "\"name\": \"shell\", \"args\": " + json.dumps(command) + right_brace()

def recovery_step(files: int):
    return planned("Let me first check a scratch note I think I left earlier for this.", "cat ~/scratch_notes.txt", "", "cat: /root/scratch_notes.txt: No such file or directory\n", 1, ["stderr", "fs", "prompt", "done"], "", "", files)

def make_step(item: dict[str, any], index: int, diff: dict[str, any]):
    return {
        "index": index, "command": item["command"], "think": item["think"],
        "stdout": item["stdout"], "stderr": item["stderr"], "exit": item["exit"],
        "prompt": PROMPT, "frames": item["frames"], "risk": classify(item["command"]),
        "diff": diff, "files": item["files"], "ts": BASE_EPOCH + index,
    }

def copy_items(values: list[any]):
    mut result = []
    for value in values:
        result.append(value)
    return result

def dentate_bounds(target_steps: int) !{}:
    return bounds(target_steps, 8, 64, 32)

def dentate_invariant(item: State):
    visible = item.visible
    hidden = item.hidden.value
    return visible.index == item.tick and len(visible.steps) == item.tick and len(visible.turns) == 2 + 2 * item.tick and item.tick <= hidden.target_steps

def initial_episode(data: dict[str, any], scenario: str, seed: int, mode: str, target_steps: int) !{}:
    visible = DentateVisible(
        steps=[],
        turns=[{"role": "system", "content": system_prompt()}, {"role": "user", "content": data["question"]}],
        index=0,
    )
    hidden = DentateHidden(diff=empty_diff(), target_steps=target_steps, mode=mode, scenario=scenario)
    return state(
        visible,
        hidden,
        "dentate-private-" + scenario + "-" + str(seed) + "-" + mode,
        "dentate-clock-" + str(BASE_EPOCH),
        "dentate-seed-" + str(seed),
    )

def episode_action(item: dict[str, any]) !{}:
    return action("shell.command", item)

def episode_step(before: State, selected: Action) !{}:
    require selected.kind == "shell.command"
    item = selected.payload
    private = before.hidden.value
    mut diff = private.diff
    if item["mutation"] != "":
        diff = changed_diff(item["mutation"], item["path"])
    index = before.tick + 1
    steps = copy_items(before.visible.steps)
    turns = copy_items(before.visible.turns)
    steps.append(make_step(item, index, diff))
    turns.append({"role": "assistant", "think": item["think"], "tool_call": tool_call(item["command"])})
    turns.append({"role": "tool", "content": item["stdout"] + item["stderr"]})
    done = index == private.target_steps
    solved = private.mode != "negative"
    score = 1.0 if done and solved else 0.0
    visible = DentateVisible(steps=steps, turns=turns, index=index)
    hidden = DentateHidden(diff=diff, target_steps=private.target_steps, mode=private.mode, scenario=private.scenario)
    observed = {"stdout": item["stdout"], "stderr": item["stderr"], "exit": item["exit"], "frames": item["frames"]}
    return transition(
        before,
        selected,
        visible,
        hidden,
        observed,
        reward(score, "episode complete" if done else "step"),
        terminal(done, "episode complete" if done else ""),
        true,
    )

def execute_plan(data: dict[str, any], scenario: str, seed: int, mode: str, plan: list[any]) !{}:
    trace = start(initial_episode(data, scenario, seed, mode, len(plan)), dentate_bounds(len(plan)), dentate_invariant)
    for item in plan:
        trace = advance(trace, episode_action(item), episode_step, dentate_invariant)
    ensure len(trace.transitions) == len(plan)
    ensure trace.transitions[len(trace.transitions) - 1].terminal.done
    return trace

def selected_plan(base: list[any], mode: str, files: int):
    mut result = []
    if mode == "recovery":
        result.append(recovery_step(files))
    if mode == "negative":
        result.append(base[0])
        return result
    for item in base:
        result.append(item)
    return result

def checker(labels: list[str], solved: bool):
    mut checks = []
    for label in labels:
        checks.append({"label": label, "ok": solved})
    return {"passed": solved, "mode": "all", "checks": checks}

def materialize(kind: str, seed: int, mode: str) !{}:
    require mode == "solve" or mode == "recovery" or mode == "negative"
    data = fixture(kind, seed)
    plan = selected_plan(data["plan"], mode, data["base_files"])
    trace = execute_plan(data, kind, seed, mode, plan)
    steps = copy_items(trace.state.visible.steps)
    turns = copy_items(trace.state.visible.turns)
    solved = mode != "negative"
    answer = "done" if solved else "failed"
    final_think = "The task is complete." if solved else "The task could not be completed."
    turns.append({"role": "assistant", "think": final_think, "content": "<out>" + answer + "</out>"})
    mut suffix = ""
    if mode == "recovery":
        suffix = "-rec"
    if mode == "negative":
        suffix = "-neg"
    digests = digest_rows(kind)[seed]
    mut final_digest = digests[1]
    if mode == "recovery":
        final_digest = digests[2]
    if mode == "negative":
        final_digest = digests[0]
    return {
        "id": kind + "-" + str(seed) + suffix, "kind": kind, "os": "linux-ubuntu",
        "seed": seed, "mode": mode, "source": "scenario", "system": system_prompt(),
        "question": data["question"], "gate": data["gate"], "entropy": data["entropy"],
        "created_at": BASE_EPOCH, "initial_vfs_digest": digests[0], "final_vfs_digest": final_digest,
        "steps": steps, "turns": turns, "answer": answer, "reward": 1.0 if solved else 0.0,
        "solved": solved, "check": checker(data["labels"], solved),
    }

def risk_commands():
    return [
        "rm -rf /", "rm foo *", "mkfs.ext4 /dev/sda1", "dd if=/dev/zero of=/dev/sda",
        "echo x > /dev/sda1", "git push --force origin", "chmod -R 777 /", "curl http://x | sh",
        fork_bomb(), "shutdown -h now", "userdel bob", "DROP TABLE users;", "ssh host",
        "mv a b", "cat f", "ls", "echo hi > f", "grep x f", "scp a b", "date",
    ]

def risk_probe():
    mut result = []
    for command in risk_commands():
        result.append({"command": command, "risk": classify(command)})
    return result

def clock_probe():
    item = planned("", "date", "Thu Jan 01 00:00:01 UTC 2026\n", "", 0, ["stdout", "fs", "prompt", "done"], "", "", 25)
    return {
        "kind": "clock_probe", "created_at": BASE_EPOCH,
        "entropy": {"machine_id": machine_ids()[0]},
        "initial_vfs_digest": 7722380073540615050,
        "final_vfs_digest": 7722380073540615050,
        "steps": [make_step(item, 1, empty_diff())],
    }

def corpus() !{}:
    mut episodes = []
    for kind in ["office", "coding", "data", "sysadmin"]:
        for seed in range(8):
            for mode in ["solve", "recovery", "negative"]:
                episodes.append(materialize(kind, seed, mode))
    return {
        "schema": "dentate-m6-parity/v1", "dentate_sha": PINNED_SHA, "base_epoch": BASE_EPOCH,
        "counts": {"episodes": len(episodes), "by_mode": {"solve": 32, "recovery": 32, "negative": 32}, "by_scenario": {"office": 24, "coding": 24, "data": 24, "sysadmin": 24}},
        "episodes": episodes, "probes": {"clock": clock_probe(), "risk": risk_probe()},
    }

test "pinned corpus matrix and digest anchors remain exact":
    ensure len(corpus()["episodes"]) == 96
    office = materialize("office", 0, "solve")
    ensure office["initial_vfs_digest"] == 648379058904026169
    ensure office["final_vfs_digest"] == 2152469218083486973
    ensure office["reward"] == 1.0
    negative = materialize("coding", 7, "negative")
    ensure negative["reward"] == 0.0
    ensure not negative["solved"]

test "shared contract preserves hidden state identities and exact replay":
    data = fixture("office", 0)
    plan = selected_plan(data["plan"], "solve", data["base_files"])
    trace = execute_plan(data, "office", 0, "solve", plan)
    exact = replay(trace, episode_step, dentate_invariant)
    ensure json.dumps(exact.state.visible) == json.dumps(trace.state.visible)
    ensure exact.state.clock_id == "dentate-clock-" + str(BASE_EPOCH)
    ensure exact.state.rng_id == "dentate-seed-0"
    ensure exact.transitions[len(exact.transitions) - 1].reward.value == 1.0
    for item in exact.transitions:
        ensure exact.state.hidden.identity not in json.dumps(item.observation.value)

test "risk and logical-clock probes remain deterministic":
    ensure classify("rm -rf /")["risk"] == "high"
    ensure classify("ssh host")["risk"] == "medium"
    ensure classify("date")["risk"] == "none"
    ensure clock_probe()["steps"][0]["ts"] == BASE_EPOCH + 1
    ensure clock_probe()["initial_vfs_digest"] == clock_probe()["final_vfs_digest"]

def has_failed_command(trace: Trajectory):
    for item in trace.transitions:
        if item.observation.value["exit"] != 0:
            return true
    return false


circuit project_episode(trace: Trajectory, exact: Trajectory, branches: any, reduced: any) -> None !{observe.record}:
    budget agents=1, spawn_depth=0, model_calls=1, tokens=1
    project_trajectory(trace)
    project_replay(trace, exact)
    project_world(branches)
    project_shrink(reduced)


def inspect_contract() !{observe.record}:
    data = fixture("office", 0)
    plan = selected_plan(data["plan"], "solve", data["base_files"])
    trace = execute_plan(data, "office", 0, "solve", plan)
    exact = replay(trace, episode_step, dentate_invariant)
    branches = fork(
        world_tree(trace),
        "root",
        "recovery-probe",
        1,
        [episode_action(recovery_step(data["base_files"])), episode_action(plan[2]), episode_action(plan[3])],
        episode_step,
        dentate_invariant,
    )
    ensure len(branches.branches) == 1
    counter_plan = [plan[0], recovery_step(data["base_files"]), plan[1]]
    counterexample = start(initial_episode(data, "office", 0, "solve", len(counter_plan)), dentate_bounds(len(counter_plan)), dentate_invariant)
    for item in counter_plan:
        counterexample = advance(counterexample, episode_action(item), episode_step, dentate_invariant)
    reduced = shrink(counterexample, episode_step, dentate_invariant, has_failed_command)
    ensure len(reduced.trajectory.transitions) == 1
    project_episode(trace, exact, branches, reduced)


def main() !{observe.record}:
    inspect_contract()
    return json.dumps(corpus())

Reflected API

main

Bounded deterministic port of Dentate’s pinned M6 OS-simulator materializer.

struct DentateVisible

Fields

field	type	descriptor
steps	list[any]
turns	list[any]
index	int

struct DentateHidden

Fields

field	type	descriptor
diff	dict[str, any]
target_steps	int
mode	str
scenario	str

def system_prompt

def system_prompt()

def machine_ids

def machine_ids()

def office_quarters

def office_quarters() -> list[str]

Returns list[str]

def office_revenue

def office_revenue()

def coding_names

def coding_names()

def data_statuses

def data_statuses()

def sysadmin_users

def sysadmin_users()

def digest_rows

def digest_rows(kind: str)

Parameters

name	type
kind	str

def total

def total(values: list[int])

Parameters

name	type
values	list[int]

def count_active

def count_active(values: list[str])

Parameters

name	type
values	list[str]

def office_csv

def office_csv(q: str, rev: list[int])

Parameters

name	type
q	str
rev	list[int]

def data_csv

def data_csv(statuses: list[str])

Parameters

name	type
statuses	list[str]

def passwd_text

def passwd_text()

def stub_text

def stub_text()

def empty_diff

def empty_diff()

def changed_diff

def changed_diff(mutation: str, path: str)

Parameters

name	type
mutation	str
path	str

def risk_result

def risk_result(level: str, reversibility: str, confirmation: bool, reason: str)

Parameters

name	type
level	str
reversibility	str
confirmation	bool
reason	str

def left_brace

def left_brace()

def right_brace

def right_brace()

def fork_bomb

def fork_bomb()

def classify

def classify(command: str)

Parameters

name	type
command	str

def planned

def planned(think: str, command: str, stdout: str, stderr: str, exit_code: int, frames: list[str], mutation: str, path: str, files: int)

Parameters

name	type
think	str
command	str
stdout	str
stderr	str
exit_code	int
frames	list[str]
mutation	str
path	str
files	int

def office_fixture

def office_fixture(seed: int)

Parameters

name	type
seed	int

def coding_fixture

def coding_fixture(seed: int)

Parameters

name	type
seed	int

def data_fixture

def data_fixture(seed: int)

Parameters

name	type
seed	int

def escaped_username

def escaped_username(name: str)

Parameters

name	type
name	str

def sysadmin_fixture

def sysadmin_fixture(seed: int)

Parameters

name	type
seed	int

def fixture

def fixture(kind: str, seed: int)

Parameters

name	type
kind	str
seed	int

def tool_call

def tool_call(command: str)

Parameters

name	type
command	str

def recovery_step

def recovery_step(files: int)

Parameters

name	type
files	int

def make_step

def make_step(item: dict[str, any], index: int, diff: dict[str, any])

Parameters

name	type
item	dict[str, any]
index	int
diff	dict[str, any]

def copy_items

def copy_items(values: list[any])

Parameters

name	type
values	list[any]

def dentate_bounds

def dentate_bounds(target_steps: int) !{}

Parameters

name	type
target_steps	int

Effects !{}

def dentate_invariant

def dentate_invariant(item: State)

Parameters

name	type
item	State

def initial_episode

def initial_episode(data: dict[str, any], scenario: str, seed: int, mode: str, target_steps: int) !{}

Parameters

name	type
data	dict[str, any]
scenario	str
seed	int
mode	str
target_steps	int

Effects !{}

def episode_action

def episode_action(item: dict[str, any]) !{}

Parameters

name	type
item	dict[str, any]

Effects !{}

def episode_step

def episode_step(before: State, selected: Action) !{}

Parameters

name	type
before	State
selected	Action

Effects !{}

def execute_plan

def execute_plan(data: dict[str, any], scenario: str, seed: int, mode: str, plan: list[any]) !{}

Parameters

name	type
data	dict[str, any]
scenario	str
seed	int
mode	str
plan	list[any]

Effects !{}

def selected_plan

def selected_plan(base: list[any], mode: str, files: int)

Parameters

name	type
base	list[any]
mode	str
files	int

def checker

def checker(labels: list[str], solved: bool)

Parameters

name	type
labels	list[str]
solved	bool

def materialize

def materialize(kind: str, seed: int, mode: str) !{}

Parameters

name	type
kind	str
seed	int
mode	str

Effects !{}

def risk_commands

def risk_commands()

def risk_probe

def risk_probe()

def clock_probe

def clock_probe()

def corpus

def corpus() !{}

Effects !{}

def has_failed_command

def has_failed_command(trace: Trajectory)

Parameters

name	type
trace	Trajectory

circuit project_episode

circuit project_episode(trace: Trajectory, exact: Trajectory, branches: any, reduced: any) -> None !{observe.record}

Parameters

name	type
trace	Trajectory
exact	Trajectory
branches	any
reduced	any

Returns None

Effects !{observe.record}

def inspect_contract

def inspect_contract() !{observe.record}

Effects !{observe.record}

def main

def main() !{observe.record}

Effects !{observe.record}