Examples

Runnable recipes. All examples mirror the repo's README and test suite (tests/test_engine.py, tests/test_physics_integration.py, et al.).

Minimal build-and-run

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. Drop such files into kernel_primitives/*.py and they are auto-imported at startup.

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.action 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 and tests/test_termination.py 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.