os-simulator-world-model

The os-simulator-world-model worked example.

Run it from sema/:

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

Source

src/main.sema

"""A deterministic, fail-closed shell model on `std.world_model`.

`ShellState.phase` is the proof-friendly state machine: 0=boot,
1=instructions, 2=workspace, 3=goal, 4=artifact, 5=done. The agent receives
only `Observation.value`; model-private counters and artifact state remain in
`State.hidden`. The policy converges with bounded steps, and phase 5 is an
absorbing terminal that never touches the host filesystem.
"""

from std.agent_loop import loop_until
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

struct ShellState:
    cwd: str
    phase: int
    steps: int
    invariant 0 <= phase <= 5
    invariant steps >= 0
    invariant phase < 2 or cwd == "/root/workspace"
    invariant phase >= 2 or cwd == "/root"

struct ShellHidden:
    artifact_present: bool
    denied_actions: int
    invariant denied_actions >= 0


def model_bounds() !{}:
    return bounds(8, 8, 64, 32)


def initial_state() !{}:
    return state(
        ShellState(cwd="/root", phase=0, steps=0),
        ShellHidden(artifact_present=false, denied_actions=0),
        "shell-hidden-v1",
        "shell-clock-v1",
        "shell-rng-v1",
    )


def shell_invariant(item: State):
    world = item.visible
    private = item.hidden.value
    if world.steps != item.tick or private.denied_actions < 0:
        return false
    if world.phase < 0 or world.phase > 5:
        return false
    if world.phase >= 2 and world.cwd != "/root/workspace":
        return false
    if world.phase < 2 and world.cwd != "/root":
        return false
    if world.phase >= 4 and not private.artifact_present:
        return false
    return true


def shell_action(command: str) !{}:
    require len(command) > 0
    return action("shell.command", command)


def shell_transition(
    before: State,
    selected: Action,
    cwd: str,
    phase: int,
    output: str,
    valid: bool,
    changed: bool,
) !{}:
    require selected.kind == "shell.command"
    private = before.hidden.value
    denied = private.denied_actions + (0 if valid else 1)
    artifact = private.artifact_present or selected.payload == "touch solved.flag" and valid
    visible = ShellState(cwd=cwd, phase=phase, steps=before.tick + 1)
    hidden = ShellHidden(artifact_present=artifact, denied_actions=denied)
    done = phase == 5
    score = 1.0 if done else 0.0
    return transition(
        before,
        selected,
        visible,
        hidden,
        {"output": output, "changed": changed},
        reward(score, "goal" if done else "step"),
        terminal(done, "goal satisfied" if done else ""),
        valid,
    )


def reject(before: State, selected: Action, reason: str) !{}:
    world = before.visible
    return shell_transition(before, selected, world.cwd, world.phase, reason, false, false)


def apply_action(before: State, selected: Action) !{}:
    require selected.kind == "shell.command"
    world = before.visible
    command = selected.payload
    if command == "pwd":
        return shell_transition(before, selected, world.cwd, world.phase, world.cwd, true, false)
    if command == "ls":
        listing = "README.txt\nworkspace" if world.cwd == "/root" else "goal.txt"
        return shell_transition(before, selected, world.cwd, world.phase, listing, true, false)
    if command == "cat README.txt" and world.phase == 0:
        return shell_transition(before, selected, "/root", 1, "Inspect workspace/goal.txt and create solved.flag.", true, true)
    if command == "cd workspace" and world.phase == 1:
        return shell_transition(before, selected, "/root/workspace", 2, "cwd=/root/workspace", true, true)
    if command == "cat goal.txt" and world.phase == 2:
        return shell_transition(before, selected, world.cwd, 3, "Create solved.flag, then finish.", true, true)
    if command == "touch solved.flag" and world.phase == 3:
        return shell_transition(before, selected, world.cwd, 4, "created solved.flag", true, true)
    if command == "finish" and world.phase == 4:
        return shell_transition(before, selected, world.cwd, 5, "goal satisfied", true, true)
    if command == "rm -rf /":
        return reject(before, selected, "blocked: high-risk command")
    if command == "touch solved.flag":
        return reject(before, selected, "invalid: inspect goal.txt first")
    if command == "finish":
        return reject(before, selected, "invalid: goal artifact missing")
    return reject(before, selected, "invalid action: " + command)


def render(item: Transition):
    validity = "ok" if item.accepted else "invalid"
    mutation = "changed" if item.observation.value["changed"] else "stable"
    goal = "done" if item.terminal.done else "open"
    world = item.state.visible
    return str(world.steps) + "|" + item.action.payload + "|" + validity + "|" + mutation + "|" + world.cwd + "|" + goal + "|" + item.observation.value["output"]


def choose_action(item: State) !{}:
    world = item.visible
    if world.phase == 0:
        return shell_action("cat README.txt")
    if world.phase == 1:
        return shell_action("cd workspace")
    if world.phase == 2:
        return shell_action("cat goal.txt")
    if world.phase == 3:
        return shell_action("touch solved.flag")
    return shell_action("finish")


def advance_policy(trace: Trajectory) !{}:
    if trace.state.visible.phase == 5:
        return trace
    return advance(trace, choose_action(trace.state), apply_action, shell_invariant)


def converge(trace: Trajectory) !{}:
    return loop_until(trace, 8, advance_policy, lambda value: value.state.visible.phase == 5)


def fixed_query(trace: Trajectory):
    world = trace.state.visible
    return str(world.steps) + "|pwd|ok|stable|" + world.cwd + "|done|fixed point: goal already satisfied"


test "invalid actions fail closed without leaking private state":
    initial = start(initial_state(), model_bounds(), shell_invariant)
    unsafe = advance(initial, shell_action("rm -rf /"), apply_action, shell_invariant)
    ensure not unsafe.transitions[0].accepted
    ensure not unsafe.transitions[0].observation.value["changed"]
    ensure unsafe.state.visible.phase == 0
    ensure unsafe.state.hidden.value.denied_actions == 1
    ensure "denied_actions" not in json.dumps(unsafe.transitions[0].observation.value)
    unknown = advance(initial, shell_action("launch rocket"), apply_action, shell_invariant)
    ensure unknown.transitions[0].observation.value["output"] == "invalid action: launch rocket"
    premature = advance(initial, shell_action("touch solved.flag"), apply_action, shell_invariant)
    ensure not premature.transitions[0].accepted
    ensure premature.state.visible.phase == 0


test "bounded policy reaches an exact replayable absorbing fixed point":
    initial = start(initial_state(), model_bounds(), shell_invariant)
    solved = converge(initial)
    ensure solved.state.visible.phase == 5
    ensure solved.state.visible.steps == 5
    ensure len(solved.transitions) == 5
    ensure replay(solved, apply_action, shell_invariant).state.visible.phase == 5
    fixed = advance(solved, shell_action("pwd"), apply_action, shell_invariant)
    ensure len(fixed.transitions) == len(solved.transitions)
    ensure fixed.state.visible.steps == solved.state.visible.steps
    unsafe = advance(solved, shell_action("rm -rf /"), apply_action, shell_invariant)
    ensure len(unsafe.transitions) == len(solved.transitions)


def has_invalid_action(trace: Trajectory):
    for item in trace.transitions:
        if not item.accepted:
            return true
    return false


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


def main() !{observe.record}:
    trace = start(initial_state(), model_bounds(), shell_invariant)
    solved = converge(advance(trace, shell_action("rm -rf /"), apply_action, shell_invariant))
    ensure solved.state.visible.phase == 5
    ensure solved.state.visible.steps == 6
    exact = replay(solved, apply_action, shell_invariant)
    ensure exact.state.visible.steps == 6
    branches = fork(
        world_tree(solved),
        "root",
        "unsafe-probe",
        3,
        [shell_action("rm -rf /"), shell_action("cat goal.txt"), shell_action("touch solved.flag"), shell_action("finish")],
        apply_action,
        shell_invariant,
    )
    ensure len(branches.branches) == 1
    counterexample = start(initial_state(), model_bounds(), shell_invariant)
    for selected in [shell_action("pwd"), shell_action("rm -rf /"), shell_action("ls")]:
        counterexample = advance(counterexample, selected, apply_action, shell_invariant)
    reduced = shrink(counterexample, apply_action, shell_invariant, has_invalid_action)
    ensure len(reduced.trajectory.transitions) == 1
    project_shell(solved, exact, branches, reduced)
    mut lines = []
    for item in solved.transitions:
        lines.append(render(item))
    lines.append(fixed_query(solved))
    lines.append("goal=solved")
    lines.append("steps=" + str(solved.state.visible.steps))
    return "\n".join(lines)

Reflected API

main

A deterministic, fail-closed shell model on std.world_model.

ShellState.phase is the proof-friendly state machine: 0=boot, 1=instructions, 2=workspace, 3=goal, 4=artifact, 5=done. The agent receives only Observation.value; model-private counters and artifact state remain in State.hidden. The policy converges with bounded steps, and phase 5 is an absorbing terminal that never touches the host filesystem.

struct ShellState

Fields

field	type	descriptor
cwd	str
phase	int
steps	int

struct ShellHidden

Fields

field	type	descriptor
artifact_present	bool
denied_actions	int

def model_bounds

def model_bounds() !{}

Effects !{}

def initial_state

def initial_state() !{}

Effects !{}

def shell_invariant

def shell_invariant(item: State)

Parameters

name	type
item	State

def shell_action

def shell_action(command: str) !{}

Parameters

name	type
command	str

Effects !{}

def shell_transition

def shell_transition(before: State, selected: Action, cwd: str, phase: int, output: str, valid: bool, changed: bool) !{}

Parameters

name	type
before	State
selected	Action
cwd	str
phase	int
output	str
valid	bool
changed	bool

Effects !{}

def reject

def reject(before: State, selected: Action, reason: str) !{}

Parameters

name	type
before	State
selected	Action
reason	str

Effects !{}

def apply_action

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

Parameters

name	type
before	State
selected	Action

Effects !{}

def render

def render(item: Transition)

Parameters

name	type
item	Transition

def choose_action

def choose_action(item: State) !{}

Parameters

name	type
item	State

Effects !{}

def advance_policy

def advance_policy(trace: Trajectory) !{}

Parameters

name	type
trace	Trajectory

Effects !{}

def converge

def converge(trace: Trajectory) !{}

Parameters

name	type
trace	Trajectory

Effects !{}

def fixed_query

def fixed_query(trace: Trajectory)

Parameters

name	type
trace	Trajectory

def has_invalid_action

def has_invalid_action(trace: Trajectory)

Parameters

name	type
trace	Trajectory

circuit project_shell

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

Parameters

name	type
solved	Trajectory
exact	Trajectory
branches	any
reduced	any

Returns None

Effects !{observe.record}

def main

def main() !{observe.record}

Effects !{observe.record}