Examples

Runnable recipes. These mirror the repo's examples/ directory (00_simulate.py, quickstart.py, tic_tac_toe/) and README.

Zero-config — simulate()

The one-liner (examples/00_simulate.py): load a template — dict, WorldTemplate, or a path to a JSON file — run it to completion with the built-in seeded random_policy, and read the wrap-up:

from fg_env_kernel import simulate

world = simulate("examples/tic_tac_toe/template.json", seed=7)

print(world)              # <World 'Tic-Tac-Toe': round 4/9, finished, terminated_by='three_in_a_row'>
print(world.summary())

Same template + same seed = the same random game, every time — the default agent derives a per-entity RNG stream from (seed, entity_id), so replays are exact. Pass agent= to swap in your own decision_fn and keep the one-line ergonomics.

Quickstart — the Kernel facade

Load a template, plug in an agent, step until done, read the outcome — this is the repo's examples/quickstart.py driving the examples/tic_tac_toe/ template (run it from a clone of the repo; the template JSON ships there, not in the pip install):

import json, random
from pathlib import Path

from fg_env_kernel import ActionInstance, Kernel

template = json.loads(Path("examples/tic_tac_toe/template.json").read_text())
rng = random.Random(7)

def decision_fn(entity_id, perception, valid_actions):
    """Toy agent: pick a random cell. Swap in an LLM call here."""
    if "place_mark" not in valid_actions:
        return None
    return ActionInstance(action_name="place_mark", actor_id=entity_id,
                          parameters={"cell": rng.randint(1, 9)})

kernel = Kernel(seed=42)
world = kernel.load(template, decision_fn=decision_fn)

while not world.finished:
    world.step()

print(f"Rounds played: {world.current_round}")
print(f"Terminated by: {world.terminated_by or 'round budget'}")
print(f"Final event:   {world.events[-1].narrative}")

Minimal build-and-run (low-level)

The facade delegates to load_world; call it directly when you want the raw pair:

from fg_env_kernel import load_world

state, engine = load_world(my_world_definition, seed=42, decision_fn=my_agent_brain)
engine.run()

Same template + same seed + same decision_fn = byte-identical run, every time.

Registering a custom effect verb

New mechanics never require editing the engine — register a handler and reference it by name from any template's effect specs:

Handlers receive (ctx, spec) — an EffectContext plus the effect-spec dict (target / field / value / resource …), mirroring the kernel's own handlers in composition.py:

from fg_env_kernel import effect, EffectContext

@effect("grant_gold")
def grant_gold(ctx: EffectContext, spec: dict):
    amount = ctx.resolve(spec.get("value", 0)) or 0
    ctx.actor.set("gold", (ctx.actor.get("gold") or 0) + amount)
    ctx.record({"type": "grant_gold", "entity": ctx.actor.id, "amount": amount})

EffectContext carries state, actor, target, params, result (the ResolutionResult), rng, emit, and changes; use ctx.resolve(value) to evaluate $-expressions and ctx.record(change) to log changes for the round transcript.

Module-level @effect registers into the process-global registry. To keep a custom verb private to one kernel, register it on a fork instead — the fork sees all built-ins, but its registrations are invisible to the parent and to sibling forks:

from fg_env_kernel import Kernel, registry

mine = registry.fork()

@mine.effect("grant_gold")
def grant_gold(ctx, spec): ...

kernel = Kernel(seed=42, registry=mine)   # only this kernel's worlds resolve grant_gold

Drop-in kernel_primitives/*.py files are no longer auto-imported at startup: call discover() explicitly, or set the KERNEL_PRIMITIVES_DIR env var (an explicitly configured directory is honored at import time).

Programmatic world with contest resolution

Two warriors, an opposed-strength attack, and a deterministic rest — built directly against the state API (full walkthrough in Getting started):

state.register_action(ActionDefinition(
    name="attack", description="Strike an opponent (opposed strength check)",
    actor_type="warrior", target_type="warrior",
    preconditions=[Precondition(subject="actor", operator=Operator.GTE,
                                field="stamina", value=10)],
    resolution_archetype="contest",
    resolution_params={"attacker_property": "strength", "defender_property": "strength"},
    effects_on_success=[
        Effect(target="target", operation=EffectOperation.SUBTRACT, field="health", value=20),
        Effect(target="actor",  operation=EffectOperation.SUBTRACT, field="stamina", value=10),
    ],
    effects_on_failure=[
        Effect(target="actor", operation=EffectOperation.SUBTRACT, field="stamina", value=5),
    ],
))

engine = SimulationEngine(state=state, decision_fn=always_rest, max_rounds=3, seed=42)
result = engine.run()
assert result.get_entity("w1").get("stamina") == 95.0   # 50 + 15 * 3

The contest archetype rolls the opposed check with the engine's seeded RNG; the declared effect lists apply based on its ResolutionResult.

Declarative physics: logistic growth with writeback

A herd whose size follows the logistic ODE, integrated by RK4 each round and written back onto the entity so agents perceive it:

from fg_env_kernel import load_world
from fg_env_kernel import ActionInstance

schema = {
    "name": "ecosystem",
    "entity_types": [
        {"name": "ranger", "role": "agent", "properties": []},
        {"name": "herd", "role": "object",
         "properties": [{"name": "size", "type": "float", "default": 10}]},
    ],
    "entities": [
        {"id": "r1", "name": "Ranger", "entity_type": "ranger", "properties": {}},
        {"id": "h1", "name": "Herd",   "entity_type": "herd",   "properties": {"size": 10}},
    ],
    "actions": [{"name": "observe", "actor_type": "ranger",
                 "resolution_archetype": "deterministic"}],
    "temporal": {"phases": [{"name": "action"}], "mode": "discrete"},
    "physics": {
        "params": {"r": 0.8, "K": 1000.0},
        "substeps": 8,
        "variables": [
            {"name": "population", "value": 10.0,
             "rate": "r * population * (1 - population / K)", "min": 0.0,
             "writeback": {"entity_type": "herd", "property": "size",
                           "mode": "broadcast"}},
        ],
    },
}

def idle(entity_id, perception, valid_actions):
    # valid_actions is a list of action-name strings
    if valid_actions:
        return ActionInstance(action_name=valid_actions[0], actor_id=entity_id)
    return None

state, engine = load_world(schema, seed=1, decision_fn=idle)
engine.max_rounds = 30
engine.run()

pop = state.physics.values["population"]           # > 100 — logistic growth occurred
assert state.get_entity("h1").get("size") == pop   # writeback grounded onto the entity

Coupled ODEs: predator/prey

Lotka–Volterra in four lines of template — each rate references the other variable, so the system is genuinely coupled:

"physics": {
    "params": {"alpha": 1.1, "beta": 0.4, "delta": 0.1, "gamma": 0.4},
    "variables": [
        {"name": "prey", "value": 10, "rate": "alpha*prey - beta*prey*pred", "min": 0},
        {"name": "pred", "value": 5,  "rate": "delta*prey*pred - gamma*pred", "min": 0}
    ]
}

Rate expressions are compiled by the safe whitelisted-AST evaluator — attribute access, lambdas, and unknown symbols are rejected at build time with PhysicsExprError, and runtime numeric blow-ups skip the step and emit a physics_error change instead of crashing.

Continuous-time mode

Switch the temporal mode and give actions durations; the event-driven clock does the rest:

"temporal": {"mode": "continuous", "phases": [{"name": "action"}]},
"actions": [
    {"name": "quick_jab",  "actor_type": "fighter", "duration": 0.5, ...},
    {"name": "haymaker",   "actor_type": "fighter", "duration": 3.0, ...},
]

The loader builds a ContinuousTemporalModel: agent turns reschedule after each action's duration, and recurring environment ticks advance physics and property dynamics by the real elapsed dt — a fast fighter simply acts more often. Event ordering is deterministically total (time → priority → entity_id → id), so replays stay exact even under float-time collisions.

Custom termination condition

Built-in check_types cover most games; when they don't, register your own — and a winner resolver to go with it:

Both are called as fn(state, params, rng), where params is the condition's params dict (not the condition object) — the same convention the built-ins use:

from fg_env_kernel import termination_decorator, termination

@termination_decorator("gold_hoard")
def gold_hoard(state, params, rng) -> bool:
    threshold = params.get("threshold", 1000)
    return any(state.resources["gold"].holdings.get(e.id, 0) >= threshold
               for e in state.entities.values())

def richest_wins(state, params, rng):
    holdings = state.resources["gold"].holdings
    winner_id = max(holdings, key=holdings.get)
    return {"winner_id": winner_id,
            "winner_name": state.get_entity(winner_id).name}

termination.register_winner_resolver("gold_hoard", richest_wins)

Templates then reference it declaratively. Check parameters nest under "params", and the validated path (TerminationSpec) requires a name:

"termination_conditions": [
    {"check_type": "gold_hoard", "name": "gold_hoard",
     "params": {"threshold": 1000}},
]

Expression-based conditions need no code at all:

"termination_conditions": [
    {"check_type": "expr", "name": "last_warrior",
     "params": {"expr": "$count(warrior, alive) <= 1"}},
    {"check_type": "round_limit", "name": "round_cap",
     "params": {"rounds": 50}},
]

Validate before you ship

The authoring loop for a new environment:

from fg_env_kernel import compile_template

result = compile_template(raw_json, smoke=True, smoke_rounds=20)
# result: CompileResult — lint issues + a SmokeReport from a stub-agent dry run

Then playtest with a real LLM decision_fn, and package with pack_env into a .simworld archive (or from the CLI: fg-env-kernel compile, fg-env-kernel pack). Use replay to get a deterministic step-by-step trace of any run.