Skip to content

Service integration

Integration API. Use the service layer when a host needs explicit authorization, command routing, or custom process wiring. Most scripts should use the runtime instead.

ServiceContainer 

ServiceContainer()

Wires services together with correct dependency ordering.

Each service owns its internal composition. The container only handles service-to-service wiring.

Usage

container = ServiceContainer() world = await container.world_service.create_world(config, storage_config) await container.simulation_service.step(world.world_id, run_config)

shutdown async 

shutdown()

Gracefully shut down all services.

CommandService 

CommandService(
    mutations,
    worlds,
    simulation,
    queries,
    broker,
    audit=None,
    autoresearch=None,
    ingestion=None,
    evals=None,
)

Policy enforcement point.

Every external mutation, lifecycle operation, and read flows through here. The only service that sees ActorCtx.

create_entities async 

create_entities(ctx, world_id, entities)

Batch-spawn entities. Applies one RBAC gate check for the batch.

reserve_entity_ids 

reserve_entity_ids(ctx, world_id, n)

Reserve n entity IDs without spawning. Synchronous — no round-trip.

spawn_with_reserved_id async 

spawn_with_reserved_id(
    ctx, world_id, entity_id, components
)

Materialise a previously reserved entity ID.

update_entity async 

update_entity(ctx, world_id, entity_id, components)

Overlay values on existing components (same archetype).

discover_worlds async 

discover_worlds(ctx, storage_config)

Return worlds recorded in a storage catalog.

Unlike list_worlds(), this works in a fresh process and includes worlds that are not currently live.

open_world_readonly async 

open_world_readonly(ctx, storage_config, world_id)

Return the durable descriptor for a world without activating it.

The returned descriptor can be used with read paths but cannot mutate the world.

ingest_fact async 

ingest_fact(
    ctx,
    world_id,
    components,
    *,
    external_id,
    producer="default",
    storage_config=None,
)

Persist an external fact exactly once per external identity.

This operation completes directly rather than entering the deferred command queue so the caller always receives its durable receipt.

evaluate async 

evaluate(
    ctx,
    world_id,
    components,
    *,
    contract,
    grader,
    evaluation_id,
    producer="evals",
    storage_config=None,
    ticks=None,
    entity_ids=None,
)

Persist one evaluation receipt for an evaluation identity.

The identity is claimed before grading. A completed matching claim returns its receipt without running the grader again. The receipt is pinned to a snapshot and requires a versioned GraderContract.

resume_world async 

resume_world(ctx, storage_config, world_id)

Resume a durable world as the active writer.

Resuming reconstructs live state from durable rows and invalidates the previous writer. The caller receives a descriptor and continues through the normal command interface.

run_episode async 

run_episode(ctx, world_id, config, **input_kwargs)

Gate, then delegate to SimulationService.run_episode.

run_rollout async 

run_rollout(ctx, world_id, config, **input_kwargs)

Gate, then delegate to SimulationService.run_rollout.

Emits ONE rollout-level audit row, not one per fork. Internal forks are SimulationService's mechanics.

autoresearch async 

autoresearch(
    ctx,
    world_id,
    config,
    evaluator,
    *,
    prepare_candidate=None,
    lab_world_id=None,
    on_iteration=None,
)

Gate, then delegate to AutoResearchService.run.

Emits ONE loop-level audit row. Candidate forks, rollouts, and lab-world lifecycle ticks are AutoResearchService's mechanics; the experiment's own ledger is the fine-grained record of every attempt.

query_components async 

query_components(
    ctx,
    components,
    world_id,
    run_id,
    storage_config=None,
    *,
    ticks=None,
    entity_ids=None,
)

Query entities by component subset. Gated read.

submit async 

submit(ctx, world_id, cmd)

Gate, then enqueue for application at cmd.tick.

submit_batch async 

submit_batch(ctx, world_id, cmds)

Gate all-or-nothing, then enqueue atomically.

submit_spawn async 

submit_spawn(
    ctx, world_id, components, *, tick=0, priority=0
)

Reserve entity_id, gate, enqueue spawn, return id immediately.

drain_and_apply async 

drain_and_apply(world_id, tick)

Drain commands due at tick, apply them via MutationService.

No ActorCtx — commands carry their own context validated at submit.

WorldService 

WorldService(storage_service)

Service-layer facade. Bridges StorageService into the WorldOrchestrator.

Only external dependency: StorageService. Internally owns: WorldFactory, WorldRegistry, WorldOrchestrator.

create_world async 

create_world(
    config,
    storage_config=None,
    cache_config=None,
    system=None,
)

Resolve storage, then delegate world creation to the orchestrator.

storage_record 

storage_record(world_id)

The storage/cache config backing a world, or None if unknown.

This is how readers locate a world's rows without being told the storage config out of band. Records outlive destroy_world: storage is append-only and destroyed worlds remain queryable.

fork_world async 

fork_world(
    source_world_id,
    name=None,
    storage_config=None,
    cache_config=None,
)

Fork a world. Inherits source's storage when no override is given.

Per docs/guide/world-lifecycle.md § 4.5, the fork writes to the same physical store as the source by default; an explicit storage_config argument routes the fork to a different store.

destroy_world async 

destroy_world(world_id)

Destroy a world. In-memory cleanup only. Storage is preserved.

The storage record is retained: destroyed worlds' rows are still in the store (append-only), and readers resolve them through storage_record(). The catalog marks status — append-only holds in the control plane too; nothing is deleted.

discover_worlds async 

discover_worlds(storage_config)

Worlds recorded in a store's control catalog — works cold.

Unlike list_worlds (live process registry), this answers from the durable catalog: a fresh process pointed at the same storage identity sees every world ever registered there, including destroyed ones (their rows are still queryable; append-only).

open_world_readonly async 

open_world_readonly(storage_config, world_id)

Cold, read-only open: the world's durable descriptor.

Returns the existing WorldInfo boundary type — identity, name, run, and (advisory until A2 manifests) tick head. Queryability comes through the ordinary gated query path with this info plus the same storage_config. This never constructs a live mutable world: fenced mutable resume is a separate, A2-gated capability.

open_world_mutable async 

open_world_mutable(storage_config, world_id)

Reconstruct a durable world and acquire its writer lease.

The previous writer becomes stale and its next publish fails. Resume fails instead of returning a world whose state cannot be reconstructed faithfully.

Processors, resources, and hooks are code rather than stored state and must be reinstalled by the caller.

record_step async 

record_step(world_id)

Refresh the world's current run in the catalog after a step.

Called by SimulationService post-step. Since #273, the durable tick head is maintained transactionally by manifest publication inside world.step, and signature registration rides publication in the commit coordinator — so this hook only keeps the world's run_id current. Failures log loudly rather than failing a tick whose data-plane writes and manifest already succeeded.

add_resource async 

add_resource(world_id, resource)

Attach a resource to a world's Resources container.

list_processors 

list_processors(world_id)

Return the world's registered processor instances.

list_hooks 

list_hooks(world_id)

Return registered hooks as (event_type, handle, fn, mode) rows.

list_resources 

list_resources(world_id)

Return (type, instance) pairs from the world's Resources container.

add_hook 

add_hook(world_id, event_type, fn, *, mode='blocking')

Add a hook to a world's hook bus. Returns the HookHandle.

remove_hook 

remove_hook(world_id, handle)

Remove a hook by handle.

SimulationService 

SimulationService(world_service)

Execution engine. Takes configs and drives worlds through ticks.

run_episode and run_rollout are owned here. Internal forks inside run_rollout go through WorldService directly — they are not separately gated. The audit unit is the rollout call, not each fork.

set_quota_reset 

set_quota_reset(reset_quota)

Inject the per-tick RBAC quota reset.

Wired by the container to auth.reset_tick_counters so the gate's per-tick command counter starts fresh each tick instead of accumulating process-wide. Left unset in gate-less service tests (no quota is debited without the gate, so there is nothing to reset). Keeps SimulationService free of an auth import — the same decoupling pattern as set_command_drain.

step async 

step(world_id, run_config, **input_kwargs)

Advance a world by one tick.

run async 

run(world_id, run_config, **input_kwargs)

Execute run_config.num_steps ticks and return a RunResult.

Inv O3: the returned run_id is the world's active run_id — i.e. the run_id actually stamped on persisted rows during this run — so callers can round-trip RunResult.run_id back into a query and find the data they just wrote. RunConfig.run_id seeds the world's run_id only when the world has none set yet (per the "first run pins" semantics that keep cross-run state continuity intact).

run_episode async 

run_episode(world_id, config, **input_kwargs)

Run a bounded episode on a world until termination or max_steps.

Does NOT fork — runs on the given world_id directly. Termination strategies and their order are documented on EpisodeConfig: structural / callable are checked before each step; the value-based "all done" contract is checked after each step against persisted rows.

run_rollout async 

run_rollout(world_id, config, **input_kwargs)

Run N episodes, each on a fork of the base world.

Forks go through WorldService directly — not gated. The rollout is the audit unit, not each fork.

QueryService 

QueryService(storage_service, audit=None)

Direct read path to storage.

Depends only on StorageService. Resolves a store per query via get_or_create_store, so any historical storage location is reachable.

query_archetype async 

query_archetype(
    sig,
    world_id,
    run_id,
    storage_config=None,
    *,
    ticks=None,
    entity_ids=None,
    components=None,
    lineage=None,
)

Query an archetype table by signature, world, and run.

Reads directly from the store. Works for any historical world/run, not just worlds that are currently live. The store is resolved via get_or_create_store, so repeated calls with the same config reuse the cached instance.

When lineage is provided (a fork's ancestor segments), pre-fork ticks are read from the owning ancestor's run and unioned in.

query_components async 

query_components(
    components,
    world_id,
    run_id,
    storage_config=None,
    *,
    ticks=None,
    entity_ids=None,
    lineage=None,
)

Query all entities that contain the requested component types.

Subset matching: finds all archetype signatures that contain the requested types, queries each, projects to requested columns, unions.

When lineage is provided (a fork's ancestor segments), pre-fork ticks are read from the owning ancestor's run and unioned in.

get_lineage async 

get_lineage(world_id, run_id, storage_config=None)

Recover a world's persisted fork ancestry from the store.

Lineage rows are append-only, so this works for destroyed worlds. Returns None for root worlds (nothing was recorded at fork time).

list_signatures async 

list_signatures(storage_config=None)

List all archetype signatures in a store.

get_command_history async 

get_command_history(world_id, limit=100)

Compatibility read for pre-gate callers; queued command history only.

StorageService 

StorageService(session=None)

Creates and pools async stores. Manages storage lifecycle.

Multiton semantics: for any (uri, namespace, backend, cache) tuple, only one store instance is created and reused.

get_control_catalog 

get_control_catalog(storage_config)

The durable control catalog for a storage location (pooled).

get_or_create_store async 

get_or_create_store(storage_config, cache_config=None)

Return the pooled async store for a storage/cache configuration.

Creates the store on first call for a given config key. Subsequent calls with the same config return the cached instance.

shutdown async 

shutdown()

Gracefully shuts down all managed storage backends.

CommandBroker 

CommandBroker(max_dequeue=50000, debug=False)

Universal simulation interface for command-based interaction with worlds.

The broker mediates all external and agent-initiated commands with RBAC enforcement.

enqueue async 

enqueue(world_id, cmd, ctx=None)

Enqueue a single command for a specific world. If ctx is provided, validates RBAC permissions and quotas.

enqueue_bulk async 

enqueue_bulk(world_id, cmds, ctx=None)

Enqueue multiple commands for a specific world. All-or-nothing: validates all commands before enqueueing any, and debits quota only once the entire bulk has passed validation. A partial RBAC or quota failure leaves the actor's counters untouched.

dequeue async 

dequeue(world_id, max_items=None)

Dequeue commands for a specific world (all pending, regardless of tick).

dequeue_due async 

dequeue_due(world_id, tick, limit=None)

Pop all commands where cmd.tick <= tick. Ordered by (tick, priority, seq).

ack async 

ack(cmd_ids)

Remove from pending after successful application.

remove async 

remove(world_id, cmd_id)

Remove a command from the queue and pending dict. Preserves history. Idempotent.

peek async 

peek(world_id, max_items=None)

Peek at commands without removing them.

get_pending_count async 

get_pending_count(world_id=None)

Get count of pending commands.

get_history async 

get_history(world_id, limit=100)

Return recent enqueued commands for a world (most recent last).

clear async 

clear(world_id=None)

Clear pending commands.

Command

arrow_schema classmethod 

arrow_schema()

Return a canonical Arrow schema suitable for Parquet.

CommandType

Command types for the command broker.

The broker is the universal simulation interface supporting: - Entity-level mutations (spawn, despawn, components) - Processor mutations (hot-swap behavior) - Simulation-level operations (recursive simulation, rollouts)