API reference

Import path: fg_env_kernel. Everything below is re-exported from fg_env_kernel/__init__.py (the import firewall) unless noted otherwise.

Loading worlds (pipeline/loader.py)

load_world(
    template: dict | WorldTemplate,
    *, seed: int = 0, decision_fn=None, on_event=None,
) -> tuple[WorldState, SimulationEngine]
# Build state + engine from a template (raw dict or WorldTemplate).

build_world_state(schema: dict) -> WorldState
# Canonical JSON→state builder — the single source of truth.

load_world_parts(
    schema: dict, rules: dict, viz: dict | None = None,
    *, seed: int = 0, decision_fn=None, on_event=None,
) -> tuple[WorldState, SimulationEngine]
# Three-document form (schema + rules + viz merged).

WorldTemplate

Pydantic BaseModel with extra="allow" — the canonical JSON contract.

Note: a physics block is read by the loader from the schema dict but is not an explicit WorldTemplate field — it flows through via extra="allow".

Supporting Pydantic specs: PropertySpec, EntityTypeSpec, ResourceTypeSpec, PreconditionSpec, EffectConditionSpec, EffectSpec, ActionSpec, EntitySpec, TerminationSpec, DomainModuleSpec.

ActionSpec — the action-authoring contract

name, description, actor_type, target_type,
preconditions,
resolution_archetype="deterministic", resolution_params,
effects_on_success, effects_on_failure, effects_on_partial,
parameters,
requires_action, requires_action_success=True,
cooldown_rounds=0, sequence_rounds=0,
interruptible=True, message_action=False
# `duration` is accepted for continuous mode via extra="allow"

Effect specs name an operation from the built-in EffectOperation verb set (action.py; custom verbs register via @effect): set, add, subtract, multiply, transfer_resource, set_relation, modify_relation, move_to, kill, emit_event, apply_status, remove_status, spawn_entity, despawn_entity, give_item, take_item, drop_item, pickup_item, award_xp, craft_item, social verbs (post_content, share_content, react_content, follow_entity, unfollow_entity), and game-token/board/deck verbs (grant_token, consume_token, advance_track, place_on_board, reset_board, draw_from_deck).

When a resolution returns partial=True, the action's effects_on_partial list is applied, and effects with scale_by_magnitude=True multiply numeric values by success_degree.

SimulationEngine (runtime/engine.py)

SimulationEngine(
    state: WorldState,
    decision_fn=None,        # fn(entity_id, perception, valid_actions) -> ActionInstance | None
    outcome_fn=None,         # fn(entity_id, action_name, success, narrative, details) -> None
    narrative_fn=None,
    max_rounds: int = 100,
    seed: int | None = None, # unseeded → minted from entropy, recorded on engine.seed
    on_round_start=None, on_round_end=None,
    on_event=None,           # fn(event_dict) -> None — real-time streaming
    termination_conditions: list[TerminationCondition] | None = None,
    invariant_checker=None,
    world_event_engine=None,
    continuous_time=None,    # ContinuousTemporalModel → continuous mode
    parallel_decisions: int = 0,  # 0 = sequential; N = max concurrent LLM calls
)

engine.run() -> WorldState   # dispatches _run_discrete() or _run_continuous()
engine.pause()
engine.resume() -> WorldState  # resumes from the in-flight schedule/round
engine.is_paused() -> bool
engine.stop()

Determinism: one seeded random.Random(self.seed) is shared into property dynamics, world events, and domain-module reseed(rng) hooks so all draws stay reproducible.

The decision_fn contract

decision_fn(entity_id: str, perception: dict, valid_actions: list) -> ActionInstance | None

valid_actions is a list of action-name strings (WorldState.get_valid_actions, optionally narrowed by domain modules). Return ActionInstance(action_name=..., actor_id=..., target_id=..., reasoning=...) (from fg_env_kernel.action) or None to pass. Optional outcome_fn and narrative_fn callbacks receive resolution results and narration.

Resolution (resolution.py)

@dataclass
class ResolutionResult:
    success: bool; magnitude: float = 1.0; narrative: str = ""; details: dict = {}
    partial: bool = False          # graduated outcome between success and failure
    success_degree: float = 1.0    # 0.0 (total failure) … 1.0 (total success)

class ResolutionArchetype(ABC):
    def resolve(self, actor_properties, target_properties, params,
                action_params, rng=None) -> ResolutionResult

get_resolution(name: str) -> ResolutionArchetype   # raises ValueError on unknown
RESOLUTION_REGISTRY: dict[str, ResolutionArchetype]

Registry keys (13 built-ins): deterministic, skill_check, contest, deterministic_math, voting, evidence_chain, order_book, cpmm, market_impact, auction_sealed_first, auction_sealed_second, auction_english, auction_dutch.

Physics (physics.py)

class PhysicsModel:
    MAX_SUBSTEPS = 4096
    def __init__(self, variables=None, params=None, substeps=4, time=0.0)
    values: dict[str, float]        # property
    integrated: list[str]           # property — names of rate-driven variables
    def is_empty(self) -> bool
    def integrate(self, dt: float, state=None) -> list[dict]
    def tick(self, state, dt: float) -> list[dict]     # engine entry point
    def to_dict(self) / from_dict(cls, d)
    @classmethod
    def from_schema(cls, spec) -> PhysicsModel | None

@dataclass
class PhysicsVariable:
    name: str; value: float = 0.0; rate: str | None = None
    min: float | None = None; max: float | None = None
    source: EntitySource | None = None
    writeback: EntityWriteback | None = None
    # rate (integrated ODE) XOR source (algebraic read) — never both

@dataclass
class EntitySource:    entity_type: str; property: str; reduce: str = "sum"
@dataclass
class EntityWriteback: entity_type: str; property: str; mode: str = "broadcast"

class PhysicsExprError(ValueError): ...   # raised at build time for invalid rate exprs

Registry & extension decorators (registry.py, effect_context.py)

registry: KernelRegistry   # global singleton
# namespaces: effects, preconditions, resolutions, phases,
#             terminations, modules, target_selectors

@effect(name, *, aliases=None, replace=False)        # handler(ctx: EffectContext, spec: dict) -> dict | None
@precondition(name, *, aliases=None, replace=False)  # fn(...) -> bool
@resolution(name, *, aliases=None, replace=False)    # class or instance
@phase(name, *, aliases=None, replace=False)
@termination(name, ...)      # exported as `termination_decorator`
@module(name, ...)           # factory
@target_selector(name, ...)

KernelRegistry.scoped() gives context-managed temporary registration. Each namespace offers register / get / try_get / has / keys / unregister / __contains__ / __iter__ / __len__.

EffectContext — the argument bundle passed to @effect handlers (as handler(ctx, spec), where spec is the effect-spec dict). Fields: state, actor, target, params, result (the ResolutionResult that produced the effect), rng, emit (event-emit callable), changes. Methods: record(change) appends a state-change record for the round transcript; resolve(value) evaluates $-expressions. There are no property getters/setters — mutate entities directly (e.g. ctx.actor.set(...)) and log via record.

Termination (termination submodule)

termination.evaluate(state, condition, rng=None) -> bool
termination.check_all(state, conditions, rng=None) -> condition | None  # first that fires
termination.resolve_winner(state, condition, rng=None) -> dict  # {"winner_id", "winner_name"}
termination.register_winner_resolver(name, fn) -> None

Built-in check_types: expr, compound_and, compound_or, all_dead, resource_exhausted, round_limit, rounds_idle, property_threshold, all_goals_complete, event_triggered, last_one_standing, first_to_score, score_after_n_rounds, count_property, bankruptcy, faction_win, vote_threshold. Winner resolvers ship for last_one_standing, first_to_score, score_after_n_rounds, bankruptcy, faction_win, vote_threshold.

Expression language (predicates.py, effects.py)

evaluate_predicate(expr, state=..., rng=...) -> bool   # predicates.evaluate
resolve_expression_value(...)                          # predicates.resolve

The safe grammar powering guards, effects, and terminations — $actor.gold >= 100 && $count(player, alive) > 1. effects.py resolves $-expressions: $actor.x, $random(...), $count(...), $lookup against schema tables.

Pipeline & tooling (pipeline/)

compile_template(raw, *, seed=0, decision_fn=None, on_event=None,
                 skip_lint=False, smoke=False, smoke_rounds=20) -> CompileResult
lint_template(template) -> list[CompileIssue]
smoke_test(engine, *, rounds=20, seed=42,
           decisions="random" | "first" | "round_robin" | Callable) -> SmokeReport
replay(...) -> ReplayTrace              # deterministic step trace (list of ReplayStep)
export_kernel_contract() -> dict        # JSON Schema + live capabilities; CONTRACT_VERSION

Env packages (.simworld archives): EnvPackage, PACKAGE_EXTENSION, load_env_package, save_env_package, scaffold_env(directory, name), pack_env, unpack_env, validate_package. Result types: CompileIssue, CompileResult, SmokeReport, ReplayStep, ReplayTrace. pipeline/versioning.py handles contract-version upgrades.

Domain modules (domain/base.py)

class DomainModule(ABC):
    # lifecycle & action hooks
    tick(state, round); validate_action(...); filter_valid_actions(...)
    modify_resolution(...); post_resolution(...)
    get_perception_data(...); get_constraints(...)
    required_properties; custom_actions
    to_dict() / from_dict()

DomainModuleRegistry   # singleton: get_instance, register, create, list_available
DomainModuleManager    # lifecycle management

Full implementations in domain/markets.py: PredictionMarketModule, SecuritiesTradingModule. domain/stubs.py holds Economic/Political/Ecological/Health placeholders.

Module lifecycle & observability

• KernelModule (Protocol) + dispatchers dispatch_spawn, dispatch_despawn, dispatch_round_start, collect_snapshots. Subsystems register via state.register_module(name, instance).
• Metrics: Counter, Gauge, Histogram, MetricsRegistry, metrics, timed, enable_engine_metrics, engine_metrics_enabled, get_logger.
• Temporal: TemporalModel, TimeMode (DISCRETE / CONTINUOUS), Phase.

Continuous time (continuous_time.py)

@dataclass
class ScheduledEvent:
    fire_time; priority; event_type  # "agent_turn" | "environment" | "custom"
    entity_id; data; id; cancelled
    # __lt__: time → priority → entity_id → id (deterministic total ordering)

class EventQueue:  # heap-based priority queue
    schedule; pop; peek; cancel; cancel_for_entity
    get_events_for_entity; is_empty; to_dict / from_dict

class ContinuousTemporalModel:
    # the clock: pops events, advances current_time, reschedules agent turns
    # by action duration, fires recurring environment ticks (environment_interval)
    MIN_ACTION_DURATION = 1e-9     # forces forward progress
    DEFAULT_MAX_EVENTS = 1_000_000 # termination backstop; also max_time

Errors

• PhysicsExprError(ValueError) — invalid physics rate expression, raised at build time.
• get_resolution(name) raises ValueError for an unknown archetype.
• Numeric physics failures at runtime never raise: the step is skipped and a physics_error change is emitted (graceful degradation).