Rollout      = N episodes, each in a fork of a base world.
               Lives at iSimulationService. Forks via iWorldService directly.
               Audit unit: the rollout call (one row at the gate).

Episode      = step-until-termination on the world you give it.
               Does NOT implicitly fork. Operates on the supplied world_id directly.
               Lives at iSimulationService.

Run          = N steps with one RunConfig. No termination check, no fork.
               Vanilla path: repeated `run` calls accumulate state on the same world.

Step         = 1 tick. Primitive.

await world.run(steps=100)   # tick 0..100
await world.run(steps=100)   # tick 100..200, same world, accumulated state
await world.run(steps=100)   # tick 200..300, same world

class EpisodeConfig(BaseModel):
    episode_id: UUID = Field(default_factory=uuid7)
    run_config: RunConfig                                   # how each step inside runs
    max_steps: int = 1000                                   # defensive cap
    terminal_component: type[Component] | None = None
    termination: Callable[[iWorld], bool] | None = None

class EpisodeResult(BaseModel):
    episode_id: UUID
    world_id: UUID                  # the world the episode ran on (fork or live)
    final_tick: int
    terminated: bool                # True if predicate fired; False if max_steps hit
    duration_steps: int

# iSimulationService
async def run_episode(world_id, config: EpisodeConfig, **kw) -> EpisodeResult: ...

- resolve world via iWorldService.get_world
- start_tick = world.tick
- step_count = 0
- while step_count < config.max_steps:
    if config.terminal_component:
        if any active entity has the component: break
    if config.termination:
        if config.termination(world): break
    await self.step(world_id, config.run_config)
    step_count += 1
- return EpisodeResult(
      episode_id=config.episode_id,
      world_id=world_id,
      final_tick=world.tick,
      terminated=(predicate fired),
      duration_steps=(world.tick - start_tick),
  )

class RolloutConfig(BaseModel):
    rollout_id: UUID = Field(default_factory=uuid7)
    episode_config: EpisodeConfig                           # template per episode
    num_episodes: int                                       # how many forks
    parallel: bool = False                                  # asyncio.gather across forks
    name_prefix: str = "ep"                                 # forks named "<base>:<prefix>:<i>"
    destroy_forks_on_complete: bool = False                 # in-memory cleanup post-rollout

class RolloutResult(BaseModel):
    rollout_id: UUID
    base_world_id: UUID
    episodes: list[EpisodeResult]
    total_duration_steps: int

# iSimulationService
async def run_rollout(world_id, config: RolloutConfig, **kw) -> RolloutResult: ...

- base = await iWorldService.get_world(world_id)
- async def _one(i: int) -> EpisodeResult:
    fork = await self._world_service.fork_world(
        world_id, name=f"{base.name}:{config.name_prefix}:{i}"
    )
    result = await self.run_episode(fork.world_id, config.episode_config)
    if config.destroy_forks_on_complete:
        await self._world_service.destroy_world(fork.world_id)
    return result
- if config.parallel:
    results = await asyncio.gather(*(_one(i) for i in range(config.num_episodes)))
- else:
    results = [await _one(i) for i in range(config.num_episodes)]
- return RolloutResult(...)

async def run_episode(ctx, world_id, config) -> EpisodeResult: ...
async def run_rollout(ctx, world_id, config) -> RolloutResult: ...