Every example on this page runs end-to-end with a single command. The recommended pattern is ArchetypeRuntime for scripts. A small number of examples intentionally drop lower when the read path is still below the runtime API, most notably the time-travel example.

1. World Mutations¶

Demonstrates every mutation type: spawn entities with components, inject processors at runtime, RBAC permission checks, fork a world, and query the full command audit trail.

uv run python examples/01_world_mutations.py

Source: examples/01_world_mutations.py

This example uses ArchetypeRuntime plus world.as_actor(...) to show multiple ActorCtx roles on one logical world without dropping to the service layer.

What it demonstrates:

• SPAWN / DESPAWN / UPDATE through the brokered runtime surface
• ADD_COMPONENT / REMOVE_COMPONENT with archetype migration at tick boundaries
• ADD_PROCESSOR to inject a MovementProcessor at runtime
• RBAC checks through actor-bound handles: viewer denied spawn, player denied add_processor
• FORK from an actor-bound handle while keeping the same actor binding on the branch
• Command history through world.command_history()

Output:

1. SPAWN + RBAC
   viewer: SPAWN denied (correct)
   player: spawned scout=1, dummy=2

2. UPDATE + COMPONENT MUTATIONS
   scout after update/add_components: pos=(2.0, 1.0), vel=(1.5, 0.5), hp=80

3. PROCESSOR MUTATIONS
   player: ADD_PROCESSOR denied (correct)

Command types (15 total, including run_rollout and run_episode for MCTS):

2. Fork for Counterfactuals¶

Fork a world three times with different parameters, run each branch, compare results.

uv run python examples/02_fork_counterfactual.py

Source: examples/02_fork_counterfactual.py

import asyncio
from dataclasses import dataclass

from archetype import ArchetypeRuntime, Component, StorageConfig


@dataclass
class PhysicsConfig:
    gravity: float = 9.8
    drag: float = 0.1


class Probe(Component):
    label: str = ""


async def main():
    storage = StorageConfig(uri="./archetype_data", namespace="counterfactuals")

    async with ArchetypeRuntime() as runtime:
        base = runtime.world("base", storage=storage)
        await base.spawn(Probe(label="seed"))
        await base.run(steps=1)

        for gravity in [1.0, 9.8, 25.0]:
            fork = await base.fork(f"gravity-{gravity}", storage=storage)
            fork.resources.insert(PhysicsConfig(gravity=gravity))

            result = await fork.run(steps=10)
            rows = (await fork.query(Probe)).collect().to_pylist()
            print(f"gravity={gravity:>5.1f}: tick={result.final_tick}, entities={len(rows)}")

asyncio.run(main())

Output:

gravity=  1.0: tick=11
gravity=  9.8: tick=11
gravity= 25.0: tick=11

All three branches start from the same base state and diverge independently.

3. Time-Travel Queries¶

Run 10 ticks, then query any point in history. Every tick is preserved.

uv run python examples/03_time_travel.py

Source: examples/03_time_travel.py

This example intentionally uses the lower-level QueryService because historical snapshot reads are part of the read path rather than the current top-level runtime sugar. Current-state runtime mutation helpers exist; history snapshot helpers do not yet.

import asyncio

from uuid_utils import uuid7

from archetype.app.auth.models import ActorCtx
from archetype.app.container import ServiceContainer
from archetype.app.models import Command, CommandType
from archetype.core.config import RunConfig, StorageConfig, WorldConfig


async def main():
    container = ServiceContainer()
    ctx = ActorCtx(id=uuid7(), roles={"admin"})

    world = await container.world_service.create_world(
        WorldConfig(name="time-travel-demo"), StorageConfig(),
    )

    # Spawn 3 entities and run 5 ticks
    for _ in range(3):
        cmd = Command(type=CommandType.SPAWN, payload={"components": []})
        await container.command_service.submit(world.world_id, cmd, ctx)
    await container.simulation_service.run(world.world_id, RunConfig(num_steps=5))

    # Spawn 2 more and run 5 more ticks
    for _ in range(2):
        cmd = Command(type=CommandType.SPAWN, payload={"components": []})
        await container.command_service.submit(world.world_id, cmd, ctx)
    await container.simulation_service.run(world.world_id, RunConfig(num_steps=5))

    # Query state at different ticks
    for tick in [1, 5, 10]:
        state = await container.query_service.get_world_state(world.world_id, tick=tick)
        print(f"tick {tick:>2}: {len(state.entities)} entities")

    # Full command audit trail
    history = await container.query_service.get_command_history(world.world_id)
    for cmd in history:
        print(f"  tick={cmd.tick}: {cmd.type.value}")

    await container.shutdown()

asyncio.run(main())

4. Agent Messaging¶

Three agents send greetings to each other via the CommandBroker. Messages are enqueued as MESSAGE commands and delivered at tick boundaries. Mood and energy update based on messages received.

uv run python examples/04_messaging.py

Source: examples/04_messaging.py

What it demonstrates:

• Components: AgentState (name, mood, energy), Inbox, Outbox
• Resources: SimConfig for shared parameters, CommandBroker for message routing
• Processors: GreetingProcessor (sends messages), MessageRealizationProcessor (drains broker into inboxes), MoodProcessor (updates mood based on inbox)
• Hooks: PreTick and PostTick lifecycle callbacks

Output:

Archetype Messaging Demo: Resources + MESSAGE + Hooks

-> Pre-tick 0: Starting processing...
<- Post-tick 1: Completed!
   Messages pending in broker: 6

Final State:
  Alice:   mood=happy, energy=130.0, 2 messages received
  Bob:     mood=happy, energy=130.0, 2 messages received
  Charlie: mood=happy, energy=130.0, 2 messages received

5. LLM-Powered Agents¶

Three agents with different personalities, each calling an LLM every tick via daft.functions.prompt. The ECS handles batching automatically — all entities get LLM calls in parallel because world state is a DataFrame.

export OPENAI_API_KEY=sk-...
uv run python examples/05_llm_agents.py

Source: examples/05_llm_agents.py

What it demonstrates:

• Component: Agent with name, role, and a JSON journal of thoughts
• Processor: ThinkProcessor uses daft.functions.prompt to call an LLM for every agent entity in a single DataFrame operation
• Pattern: ArchetypeRuntime keeps the script surface to world.spawn(...), world.run(...), and world.query(...)

Requires an OpenAI API key (or any provider via daft.set_provider()).

6. Trajectory Analysis¶

Ingest agent trajectories, label them with LLM-based evaluation, and compare techniques via world forking. Demonstrates the full ECS pattern: components, processors, resources, and forking in a single script.

uv run python examples/06_trajectory_analysis.py

Source: examples/06_trajectory_analysis.py

What it demonstrates:

• Components: Trajectory (JSON-encoded turns), Label (evaluation result)
• Processors: SamplingProcessor (filter), LabelingProcessor (LLM eval), ScoringProcessor (normalize)
• Resources: SamplingConfig, LabelingConfig staged on runtime.world(..., resources=[...])
• Fork-based comparison: Clone a world with world.fork(...), swap config, run independently

7. Lifecycle Hooks¶

Record lifecycle audit events, measure tick duration, and publish per-tick metrics without putting side effects inside processors.

uv run python examples/07_hooks.py

Source: examples/07_hooks.py

What it demonstrates:

• Mutation audit: OnSpawn, OnDespawn, OnComponentAdded, and OnComponentRemoved
• Tick telemetry: PreTick starts a timer and PostTick computes metrics from event.results
• Hook handles: unregister a temporary debug hook with world.remove_hook(handle)
• Boundary discipline: hooks emit side effects; processors keep the simulation state deterministic