Service Layer

Services

The service layer (src/service/) is the transport-agnostic boundary between the core engine and any external client. Transport adapters (WebSocket, gRPC, Unity bridges) depend on this layer exclusively — they never import from src/engine/ directly.

The service layer provides three things:

SimulationSession — wraps a Simulation with lifecycle management, tick modes, and event subscriptions
DTOs — data transfer objects for the tick request/response boundary
EventStreamAdapter — bridges the engine's synchronous EventBus to service-level handlers

Tick mode

TickMode is declared at session creation and cannot change at runtime. It controls who drives simulation timing. Source: src/service/interfaces.py.

class TickMode(str, Enum): HOST_DRIVEN = "host_driven" AUTONOMOUS = "autonomous"

Mode	Who ticks	Valid calls	Invalid calls
`HOST_DRIVEN`	Client calls `step()`	`step()`, `snapshot()`, `restore()`, events, agent management	`pause()`, `resume()`, `run()` → `SessionError`
`AUTONOMOUS`	Backend `session.run()`	`pause()`, `resume()`, `snapshot()`, `restore()`, events, agent management	`step()` → `SessionError`

Session state

class SessionState(Enum): CREATED = "created" # initialised, not yet ticked RUNNING = "running" # autonomous loop active, or processing a host tick PAUSED = "paused" # autonomous loop suspended (AUTONOMOUS mode only) STOPPED = "stopped" # shutdown() called; no further ticks ERROR = "error" # unrecoverable error

SimulationSession

The primary object transport adapters interact with. Source: src/service/session.py.

Factory

python

from src.service import create_session, TickMode
from src.engine.simulation import Simulation
from src.plugins.external.world import HostedWorld

# Host-driven (default): client calls step() on demand
world   = HostedWorld()
sim     = Simulation(agents=[...], world=world, registry=registry)
session = create_session(sim, tick_mode=TickMode.HOST_DRIVEN)

# Autonomous: backend self-ticks
sim     = Simulation.from_config("examples/engine_owned/sim.yaml")
session = create_session(sim, tick_mode=TickMode.AUTONOMOUS)

Host-driven pattern

python

from src.service import StepRequest, AgentObservationDTO

# Subscribe to events before first tick
session.subscribe("tick_end", lambda ev: print(f"tick {ev.tick}"))
session.subscribe("action_completed", lambda ev: print(f"  {ev.agent_id}: {ev.data['action']}"))

# Each game tick:
req = StepRequest(
    agent_observations=[
        AgentObservationDTO("agent_001", {"location": "market", "health": 95}),
        AgentObservationDTO("agent_002", {"location": "castle", "health": 100}),
    ],
    world_metadata={"weather": "clear", "time_of_day": "noon"},
)
response = await session.step(req)

print(f"Tick {response.tick}: {len(response.decisions)} decisions")
for d in response.decisions:
    print(f"  {d.agent_id}: {d.action} — {d.outcome_text}")

# Relay engine_commands back to host
commands = response.engine_commands()
for cmd in commands:
    relay_to_host(cmd)

Autonomous pattern

python

session = create_session(sim, tick_mode=TickMode.AUTONOMOUS)
session.subscribe("tick_end", lambda ev: print(f"tick {ev.tick}"))

# Start the autonomous loop (non-blocking from caller's perspective)
import asyncio
loop_task = asyncio.create_task(session.run())

# Control the loop from another coroutine
await asyncio.sleep(5)
session.pause()    # suspends after current tick
await asyncio.sleep(2)
session.resume()   # resumes

session.shutdown() # stops permanently
await loop_task    # wait for cleanup

Session API reference

Method/Property	Mode	Description
`session_id`	both	UUID string identifying this session
`state`	both	`SessionState` enum value
`tick`	both	Current world tick number
`tick_mode`	both	`TickMode` enum value
`await step(request?)`	HOST_DRIVEN only	Execute one tick, return `StepResponse`
`await run()`	AUTONOMOUS only	Run all ticks to completion
`pause()`	AUTONOMOUS only	Suspend loop after current tick
`resume()`	AUTONOMOUS only	Resume a paused loop
`shutdown()`	both	Stop permanently, detach event adapter
`snapshot()`	both	Return `SimulationSnapshot`
`restore(snapshot)`	both	Restore from snapshot
`subscribe(event_type, handler)`	both	Returns subscription_id
`unsubscribe(subscription_id)`	both	Remove subscription
`status()`	both	Return `SimulationStatus`

Data transfer objects

DTOs are plain dataclasses in src/service/dto.py. The core engine never imports them — they exist only at the service boundary.

StepRequest

@dataclass class AgentObservationDTO: agent_id: str observation: dict[str, Any] = @dataclass class StepRequest: agent_observations: list[AgentObservationDTO] = [] world_metadata: dict[str, Any] =

StepResponse

@dataclass class AgentDecisionDTO: agent_id: str agent_name: str action: str parameters: dict[str, Any] = outcome_text: str = "" engine_commands: list[dict[str, Any]] = [] error: str | None = None @dataclass class StepResponse: tick: int decisions: list[AgentDecisionDTO] = [] errors: list[tuple[str, str]] = [] def engine_commands(self) -> list[dict[str, Any]]: ...

ServiceEvent

@dataclass class ServiceEvent: session_id: str event_type: str tick: int agent_id: str data: dict[str, Any] =

SimulationStatus

@dataclass class SimulationStatus: session_id: str state: str # SessionState.value tick: int config_ticks: int # total ticks configured agent_count: int has_world_snap: bool # True if world implements Snapshotable tick_mode: str = "host_driven"

EventStreamAdapter

EventStreamAdapter bridges the engine's synchronous EventBus to service-layer ServiceEvent handlers. One adapter is created per session at construction time. Source: src/service/events.py.

How it works

The adapter subscribes to the EventBus as a single wildcard listener ("*"). When an engine event arrives, it translates it to a ServiceEvent and fans it out to all registered handlers by event type. Handlers are bucketed by event type at subscribe time, so delivery is O(matching handlers), not O(all handlers).

python

from src.service.events import EventStreamAdapter
from src.engine.event_bus import EventBus

bus     = EventBus()
adapter = EventStreamAdapter(bus, session_id="sess-1")

# Subscribe to a specific event type
sub_id = adapter.subscribe("tick_end", lambda ev: print(f"tick {ev.tick}"))

# Subscribe to all events
all_id = adapter.subscribe(None, lambda ev: print(ev.event_type))

# Unsubscribe
adapter.unsubscribe(sub_id)

# Detach from the bus (call on session shutdown)
adapter.close()

Data isolation

EventStreamAdapter makes a defensive copy of event.data before delivering it to handlers. Mutations by a handler cannot leak back into the engine's event data.

SimulationController protocol

Transport adapters should depend on the SimulationController protocol rather than SimulationSession directly. This allows alternative implementations (mock sessions for testing, multi-session routers). Source: src/service/interfaces.py.

python

from src.service.interfaces import SimulationController
from src.service.session import SimulationSession

sim     = Simulation.from_config("sim.yaml")
session = SimulationSession(sim)

# Structural check — SimulationSession satisfies SimulationController
assert isinstance(session, SimulationController)   # True