Struct PipelineWorld 

pub struct PipelineWorld { /* private fields */ }

The shared ECS world that hosts and drives every agent.

Implementations

impl PipelineWorld

pub fn new( providers: ProviderRegistry, tool_service: Arc<dyn ToolService>, pool_config: InferencePoolConfig, tool_concurrency: usize, runs_dir: PathBuf, runtime: Handle, ) -> Self

Build a world: wire the pool/bridge resources, register the providers and tool service, spawn the tool worker onto runtime, and assemble the tick schedule. Agents are added later via Self::spawn_agent.

pub fn world_mut(&mut self) -> &mut World

Mutable access to the underlying ECS world, for spawning agents (the CLI / daemon builds each agent’s component bundle) and inspection.

pub fn world(&self) -> &World

Read-only access to the underlying ECS world.

pub fn set_exact_token_counting(&mut self, enabled: bool)

Enable (or disable) the opt-in exact pre-inference budget guard for this world - see inference_bridge::InferenceJob::exact_token_counting. Call once at startup when the run config requests it, before serving.

pub fn insert_interaction_hub(&mut self, hub: InteractionHub)

Install the shared interaction hub as a world resource and attach this world’s wake handle to it, so opening/answering a prompt wakes the driver and reflect_interaction_status mirrors the change into agent status. Call once at startup, before serving. Without this, that system is a no-op (test worlds).

pub fn spawn_agent(&mut self, bundle: impl Bundle) -> Entity

Spawn an agent from its pre-built component bundle and wake the driver so the next fixed-point picks it up. Returns the new entity.

pub fn spawn_from_blueprint( &mut self, agent_id: String, blueprint: Blueprint, task: &str, stages: Vec<ResolvedStage>, global_batch_tool_hint: bool, ) -> Result<Entity, String>

Spawn an agent from a blueprint + task + per-stage resolution (see crate::pipeline::spawn_agent) and wake the driver. Returns the new entity, or an error if the first stage’s system prompt doesn’t fit.

pub fn send_message(&self, msg: AgentMessage) -> Result<(), ProviderError>

Deliver a message to a running agent (routed to its inbox on the next tick) and wake the driver.

pub fn wake_handle(&self) -> Arc<Notify>

A clone of the wake handle, so external producers (e.g. a control socket) can nudge the driver after mutating the world directly.

pub fn shutdown(&self)

Request the Self::run loop to stop after its current fixed point.

pub fn shutdown_handle(&self) -> Arc<Notify>

A clone of the shutdown handle, so a supervisor can stop a Self::run loop that has taken ownership of the world on another task.

pub async fn flush_and_stop(&mut self)

Cleanly stop the world, guaranteeing every queued snapshot reaches disk.

The persistence lane is async and fire-and-forget, so a plain shutdown (the Self::run/serve loop returning, then the world dropping) can lose snapshots still queued in the channel. This method closes that gap: it signals shutdown, drives one last fixed point so any state that settled after the loop parked is dispatched to the lane, then closes the lane and awaits the worker so all queued writes (meta.json / context.json / run.lvr) land before it returns.

Call it after the serve loop has returned (the tokio runtime must still be alive for the worker to be scheduled). Idempotent: a second call is a no-op because the persistence resource is already removed and the task taken.

pub fn agent_status(&self, entity: Entity) -> Option<AgentStatus>

The status of an agent, if it still exists.

pub fn set_status(&mut self, entity: Entity, status: AgentStatus) -> bool

Set an agent’s status and wake the driver. Returns false if the agent no longer exists. The async-starting dispatchers only act on Active agents, so this is how the world pauses/resumes/cancels an agent - a non-Active agent is simply data the systems skip until it is Active again.

pub fn pause(&mut self, entity: Entity) -> bool

Pause an agent (it finishes any in-flight step, then stops before starting new work). Returns false if the agent no longer exists.

pub fn resume(&mut self, entity: Entity) -> bool

Resume a paused agent.

pub fn cancel(&mut self, entity: Entity) -> bool

Cancel an agent (it stops starting new work; in-flight results still land).

pub fn tick(&mut self) -> TickOutcome

Run one schedule tick over every agent, catching a panic from any system so one bad agent can’t crash the daemon and take every other hosted agent with it.

When the panic can be traced to a specific agent (the usual case - see tick_scope), that agent is failed with the panic message so it stops being driven, its run is persisted as errored, and the host reaps it. Without that, the world would re-tick the same unchanged state on every wake and panic again indefinitely.

pub fn run_to_fixed_point(&mut self)

Drive the schedule until a tick changes nothing (quiescence). Public so a host loop can interleave control operations between quiescent points.

pub async fn run_until_idle(&mut self, max_waits: usize)

Drive every agent as far as it can go right now, then, while async work is in flight, wait for each completion and drive again - returning once the world is fully quiescent with nothing in flight. Bounded by max_waits wake-waits as a safety valve so a lost/never-arriving wake can’t hang a caller (e.g. a test) forever.

pub async fn run(&mut self)

Run forever: drive to quiescence, then park until an async completion or an external send_message/spawn_agent wakes the driver. Returns when Self::shutdown is signalled.