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 world model.

`World.phase` is the proof-friendly state machine: 0=boot, 1=instructions,
2=workspace, 3=goal, 4=artifact, 5=done. The policy converges with a bounded
`loop_until`; phase 5 is absorbing and never touches the host filesystem.
"""

from std.agent_loop import loop_until

assure silver

struct World:
    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 Transition:
    world: World
    command: str
    output: str
    valid: bool
    changed: bool
    invariant len(command) > 0
    invariant len(output) > 0

struct Simulation:
    world: World
    trace: list[str]
    invariant len(trace) == world.steps

def initial_world() -> World !{}:
    return World(cwd="/root", phase=0, steps=0)

def advanced(world: World, cwd: str, phase: int) -> World !{}:
    return World(cwd=cwd, phase=phase, steps=world.steps + 1)

def transition(world: World, command: str, output: str, valid: bool, changed: bool) -> Transition !{}:
    return Transition(world=world, command=command, output=output, valid=valid, changed=changed)

def reject(world: World, command: str, reason: str) -> Transition !{}:
    return transition(advanced(world, world.cwd, world.phase), command, reason, false, false)

def apply_action(world: World, command: str) -> Transition !{}:
    require len(command) > 0
    if world.phase == 5:
        if command == "pwd":
            return transition(world, command, "fixed point: goal already satisfied", true, false)
        if command == "rm -rf /":
            return transition(world, command, "blocked: high-risk command", false, false)
        return transition(world, command, "invalid at fixed point: " + command, false, false)
    if command == "pwd":
        return transition(advanced(world, world.cwd, world.phase), command, world.cwd, true, false)
    if command == "ls":
        listing = "README.txt\nworkspace" if world.cwd == "/root" else "goal.txt"
        return transition(advanced(world, world.cwd, world.phase), command, listing, true, false)
    if command == "cat README.txt" and world.phase == 0:
        return transition(advanced(world, "/root", 1), command, "Inspect workspace/goal.txt and create solved.flag.", true, true)
    if command == "cd workspace" and world.phase == 1:
        return transition(advanced(world, "/root/workspace", 2), command, "cwd=/root/workspace", true, true)
    if command == "cat goal.txt" and world.phase == 2:
        return transition(advanced(world, world.cwd, 3), command, "Create solved.flag, then finish.", true, true)
    if command == "touch solved.flag" and world.phase == 3:
        return transition(advanced(world, world.cwd, 4), command, "created solved.flag", true, true)
    if command == "finish" and world.phase == 4:
        return transition(advanced(world, world.cwd, 5), command, "goal satisfied", true, true)
    if command == "rm -rf /":
        return reject(world, command, "blocked: high-risk command")
    if command == "touch solved.flag":
        return reject(world, command, "invalid: inspect goal.txt first")
    if command == "finish":
        return reject(world, command, "invalid: goal artifact missing")
    return reject(world, command, "invalid action: " + command)

def render(item: Transition) -> str !{}:
    validity = "ok" if item.valid else "invalid"
    mutation = "changed" if item.changed else "stable"
    goal = "done" if item.world.phase == 5 else "open"
    return str(item.world.steps) + "|" + item.command + "|" + validity + "|" + mutation + "|" + item.world.cwd + "|" + goal + "|" + item.output

def record(simulation: Simulation, command: str) -> Simulation !{}:
    if simulation.world.phase == 5:
        return simulation
    item = apply_action(simulation.world, command)
    mut trace = simulation.trace
    trace.append(render(item))
    return Simulation(world=item.world, trace=trace)

def choose_action(world: World) -> str !{}:
    if world.phase == 0:
        return "cat README.txt"
    if world.phase == 1:
        return "cd workspace"
    if world.phase == 2:
        return "cat goal.txt"
    if world.phase == 3:
        return "touch solved.flag"
    return "finish"

def advance_policy(simulation: Simulation) -> Simulation !{}:
    if simulation.world.phase == 5:
        return simulation
    return record(simulation, choose_action(simulation.world))

def converge(simulation: Simulation) -> Simulation !{}:
    return loop_until(simulation, 8, advance_policy, lambda state: state.world.phase == 5)

test "invalid actions fail closed":
    unsafe = apply_action(initial_world(), "rm -rf /")
    ensure not unsafe.valid
    ensure not unsafe.changed
    ensure unsafe.world.phase == 0
    unknown = apply_action(initial_world(), "launch rocket")
    ensure not unknown.valid
    ensure unknown.output == "invalid action: launch rocket"
    premature = apply_action(initial_world(), "touch solved.flag")
    ensure not premature.valid
    ensure premature.world.phase == 0

test "bounded policy reaches an absorbing fixed point":
    solved = converge(Simulation(world=initial_world(), trace=[]))
    ensure solved.world.phase == 5
    ensure solved.world.steps == 5
    ensure len(solved.trace) == 5
    fixed = apply_action(solved.world, "pwd")
    ensure fixed.world.steps == solved.world.steps
    ensure fixed.world.phase == solved.world.phase
    ensure not fixed.changed
    unsafe = apply_action(solved.world, "rm -rf /")
    ensure not unsafe.valid
    ensure unsafe.world.steps == solved.world.steps

def main() -> str !{}:
    start = Simulation(world=initial_world(), trace=[])
    solved = converge(record(start, "rm -rf /"))
    fixed = apply_action(solved.world, "pwd")
    ensure solved.world.phase == 5
    ensure fixed.world.steps == solved.world.steps
    mut lines = solved.trace
    lines.append(render(fixed))
    lines.append("goal=solved")
    lines.append("steps=" + str(solved.world.steps))
    return "\n".join(lines)

Reflected API

main

A deterministic, fail-closed shell world model.

World.phase is the proof-friendly state machine: 0=boot, 1=instructions, 2=workspace, 3=goal, 4=artifact, 5=done. The policy converges with a bounded loop_until; phase 5 is absorbing and never touches the host filesystem.

struct World

Fields

field	type	descriptor
cwd	str
phase	int
steps	int

struct Transition

Fields

field	type	descriptor
world	World
command	str
output	str
valid	bool
changed	bool

struct Simulation

Fields

field	type	descriptor
world	World
trace	list[str]

def initial_world

def initial_world() -> World !{}

Returns World

Effects !{}

def advanced

def advanced(world: World, cwd: str, phase: int) -> World !{}

Parameters

name	type
world	World
cwd	str
phase	int

Returns World

Effects !{}

def transition

def transition(world: World, command: str, output: str, valid: bool, changed: bool) -> Transition !{}

Parameters

name	type
world	World
command	str
output	str
valid	bool
changed	bool

Returns Transition

Effects !{}

def reject

def reject(world: World, command: str, reason: str) -> Transition !{}

Parameters

name	type
world	World
command	str
reason	str

Returns Transition

Effects !{}

def apply_action

def apply_action(world: World, command: str) -> Transition !{}

Parameters

name	type
world	World
command	str

Returns Transition

Effects !{}

def render

def render(item: Transition) -> str !{}

Parameters

name	type
item	Transition

Returns str

Effects !{}

def record

def record(simulation: Simulation, command: str) -> Simulation !{}

Parameters

name	type
simulation	Simulation
command	str

Returns Simulation

Effects !{}

def choose_action

def choose_action(world: World) -> str !{}

Parameters

name	type
world	World

Returns str

Effects !{}

def advance_policy

def advance_policy(simulation: Simulation) -> Simulation !{}

Parameters

name	type
simulation	Simulation

Returns Simulation

Effects !{}

def converge

def converge(simulation: Simulation) -> Simulation !{}

Parameters

name	type
simulation	Simulation

Returns Simulation

Effects !{}

def main

def main() -> str !{}

Returns str

Effects !{}