use std::collections::{HashMap, HashSet, VecDeque};
use std::time::Duration;
use bevy_ecs::entity::Entity;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::sync::{broadcast, oneshot};
use crate::components::{
AgentMessage, AgentState, AgentStatus, AwaitingInteraction, ContextWindow, ParentRef,
SubAgentChildren, WaitReason,
};
use crate::interaction_hub::InteractionHub;
use crate::persistence::{RunMetadata, TokenTotals};
use crate::world::{LaneSnapshot, PipelineWorld};
use leviath_core::interaction::{InteractionRequest, InteractionResponse};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct SpawnArgs {
pub run_id: String,
pub blueprint_path: String,
pub task: String,
#[serde(default)]
pub regions: HashMap<String, String>,
#[serde(default)]
pub model: Option<String>,
pub workdir: String,
#[serde(default)]
pub metadata: HashMap<String, String>,
#[serde(default)]
pub callback_url: Option<String>,
#[serde(default)]
pub callback_secret: Option<String>,
#[serde(default)]
pub yolo: bool,
#[serde(default)]
pub no_seed_commands: bool,
#[serde(default)]
pub allow: Vec<String>,
#[serde(default)]
pub max_depth: Option<usize>,
#[serde(default)]
pub parent_run_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct RunListEntry {
pub run_id: String,
pub status: AgentStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub wait_reason: Option<WaitReason>,
pub stage: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stage_index: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub num_stages: Option<usize>,
pub iteration: usize,
pub tool_calls: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_progress_at: Option<i64>,
#[serde(default)]
pub unattended: bool,
#[serde(default)]
pub empty_output: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub read_paths: Option<leviath_core::run_meta::ReadPathGrantCounts>,
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct RunListing {
pub runs: Vec<RunListEntry>,
pub finished: Vec<RunListEntry>,
pub health: DaemonHealth,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct DaemonHealth {
pub agents: crate::world::AgentCounts,
pub inference: Vec<crate::inference_pool::PoolOccupancy>,
pub tools_busy: usize,
pub tools_queued: usize,
pub tools_parked: usize,
pub tools_workers: usize,
pub dead_cycles: u32,
pub relief_granted: usize,
pub redrive_secs: u64,
#[serde(default)]
pub providers_down: Vec<crate::pipeline::ProviderCircuitState>,
}
pub type Spawner = Box<dyn FnMut(&mut PipelineWorld, &SpawnArgs) -> Result<Entity, String> + Send>;
pub type Reloader = Box<dyn FnMut(&mut PipelineWorld, &str) -> Option<Entity> + Send>;
pub type ForceTerminator = Box<dyn FnMut(&str) -> bool + Send>;
pub type Reaper = Box<dyn FnMut(&mut PipelineWorld, Entity) + Send>;
pub type SpawnPreprocessor = Box<
dyn Fn(&SpawnArgs) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send,
>;
pub enum SubAgentOp {
Spawn {
args: Box<SpawnArgs>,
parent_run_id: String,
max_depth: usize,
reply: oneshot::Sender<Result<String, String>>,
},
Check {
run_id: String,
reply: oneshot::Sender<Option<AgentStatus>>,
},
Send {
run_id: String,
caller_run_id: String,
content: String,
target_region: Option<String>,
reply: oneshot::Sender<bool>,
},
Kill {
run_id: String,
caller_run_id: String,
reply: oneshot::Sender<bool>,
},
}
pub enum ControlOp {
Spawn {
args: Box<SpawnArgs>,
reply: oneshot::Sender<Result<String, String>>,
},
Status {
run_id: String,
reply: oneshot::Sender<Option<AgentStatus>>,
},
Pause {
run_id: String,
reply: oneshot::Sender<bool>,
},
Resume {
run_id: String,
reply: oneshot::Sender<bool>,
},
Cancel {
run_id: String,
reply: oneshot::Sender<bool>,
},
List {
reply: oneshot::Sender<RunListing>,
},
Message {
agent_id: String,
content: String,
target_region: Option<String>,
reply: oneshot::Sender<bool>,
},
ListInteractions {
reply: oneshot::Sender<Vec<(String, InteractionRequest)>>,
},
AnswerInteraction {
response: InteractionResponse,
reply: oneshot::Sender<bool>,
},
CancelInteraction {
request_id: String,
reply: oneshot::Sender<bool>,
},
Shutdown {
reply: oneshot::Sender<bool>,
},
}
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum WorldEvent {
Spawned {
run_id: String,
agent_id: String,
blueprint: String,
},
Status {
run_id: String,
agent_id: String,
status: String,
stage: String,
iteration: usize,
tool_calls: usize,
accepts_messages: bool,
},
Tokens {
run_id: String,
agent_id: String,
prompt_tokens: usize,
completion_tokens: usize,
cached_tokens: usize,
cache_write_tokens: usize,
},
Context {
run_id: String,
agent_id: String,
total_tokens: usize,
max_tokens: usize,
},
Interaction {
run_id: String,
agent_id: String,
request: InteractionRequest,
},
Completed {
run_id: String,
agent_id: String,
status: String,
},
StageTransition {
run_id: String,
agent_id: String,
from: String,
to: String,
iteration: usize,
},
ToolCallStarted {
run_id: String,
agent_id: String,
call_id: String,
tool: String,
},
ToolCallFinished {
run_id: String,
agent_id: String,
call_id: String,
tool: String,
ok: bool,
summary: String,
},
Log {
run_id: String,
agent_id: String,
line: String,
},
}
impl WorldEvent {
pub fn run_id(&self) -> &str {
match self {
WorldEvent::Spawned { run_id, .. }
| WorldEvent::Status { run_id, .. }
| WorldEvent::Tokens { run_id, .. }
| WorldEvent::Context { run_id, .. }
| WorldEvent::Interaction { run_id, .. }
| WorldEvent::Completed { run_id, .. }
| WorldEvent::StageTransition { run_id, .. }
| WorldEvent::ToolCallStarted { run_id, .. }
| WorldEvent::ToolCallFinished { run_id, .. }
| WorldEvent::Log { run_id, .. } => run_id,
}
}
}
#[derive(bevy_ecs::resource::Resource, Clone)]
pub struct WorldEventSink(pub broadcast::Sender<WorldEvent>);
fn status_str(status: &AgentStatus) -> &'static str {
status.label()
}
#[derive(Clone, Hash)]
struct Emitted {
status: &'static str,
stage: String,
iteration: usize,
tool_calls: usize,
accepts_messages: bool,
prompt_tokens: usize,
completion_tokens: usize,
cached_tokens: usize,
cache_write_tokens: usize,
context_tokens: usize,
terminal: bool,
}
pub struct WorldHost {
world: PipelineWorld,
by_run_id: HashMap<String, Entity>,
interactions: InteractionHub,
spawner: Option<Spawner>,
spawn_preprocessor: Option<SpawnPreprocessor>,
reloader: Option<Reloader>,
force_terminator: Option<ForceTerminator>,
reaper: Option<Reaper>,
events: broadcast::Sender<WorldEvent>,
emitted: HashMap<String, Emitted>,
emitted_interactions: HashSet<String>,
subagent_tx: UnboundedSender<SubAgentOp>,
subagent_rx: UnboundedReceiver<SubAgentOp>,
redrive: Duration,
dead_cycles: u32,
last_progress: Option<u64>,
relief_granted: usize,
dead_cycles_before_relief: u32,
finished: VecDeque<(i64, RunListEntry)>,
finished_retention_secs: u64,
}
const DEFAULT_REDRIVE_INTERVAL: Duration = Duration::from_secs(30);
pub const DEFAULT_DEAD_CYCLES_BEFORE_RELIEF: u32 = 10;
pub const DEFAULT_FINISHED_RETENTION_SECS: u64 = 300;
const MAX_RETAINED_FINISHED: usize = 256;
impl WorldHost {
pub fn new(world: PipelineWorld) -> Self {
Self::with_interactions(world, InteractionHub::new())
}
pub fn with_interactions(mut world: PipelineWorld, interactions: InteractionHub) -> Self {
let (events, _) = broadcast::channel(1024);
world
.world_mut()
.insert_resource(WorldEventSink(events.clone()));
let (subagent_tx, subagent_rx) = tokio::sync::mpsc::unbounded_channel();
Self {
world,
by_run_id: HashMap::new(),
interactions,
spawner: None,
spawn_preprocessor: None,
reloader: None,
force_terminator: None,
reaper: None,
events,
emitted: HashMap::new(),
emitted_interactions: HashSet::new(),
subagent_tx,
subagent_rx,
redrive: DEFAULT_REDRIVE_INTERVAL,
dead_cycles: 0,
last_progress: None,
relief_granted: 0,
dead_cycles_before_relief: DEFAULT_DEAD_CYCLES_BEFORE_RELIEF,
finished: VecDeque::new(),
finished_retention_secs: DEFAULT_FINISHED_RETENTION_SECS,
}
}
fn observe_redrive(&mut self) {
let snapshot = self.world.lane_snapshot();
let progress = self.progress_fingerprint();
let went_nowhere = snapshot.is_under_pressure() && self.last_progress == Some(progress);
self.last_progress = Some(progress);
self.dead_cycles = match went_nowhere {
true => self.dead_cycles.saturating_add(1),
false => 0,
};
self.log_lane_pressure(&snapshot);
let relief = self.relieve_if_wedged(&snapshot);
self.observe_lanes(&snapshot, relief);
}
fn relieve_if_wedged(&mut self, snapshot: &LaneSnapshot) -> usize {
let threshold = self.dead_cycles_before_relief;
if threshold == 0 || self.dead_cycles < threshold || !snapshot.tools_saturated {
return 0;
}
let configured = snapshot.tools_workers.saturating_sub(self.relief_granted);
let remaining = configured.saturating_sub(self.relief_granted);
let granted = self
.world
.relieve_tool_lane(remaining.min(snapshot.tools_queued));
self.relief_granted += granted;
tracing::error!(
dead_cycles = self.dead_cycles,
granted,
relief_granted = self.relief_granted,
tools_queued = snapshot.tools_queued,
tools_parked = snapshot.tools_parked,
"the tool lane has not drained in {} cycles; widening it by {granted}",
self.dead_cycles
);
self.dead_cycles = 0;
granted
}
pub fn set_finished_retention_secs(&mut self, secs: u64) {
self.finished_retention_secs = secs;
}
fn record_finished(&mut self, mut entry: RunListEntry, at: i64) {
if self.finished_retention_secs == 0 {
return;
}
entry.last_progress_at.get_or_insert(at);
self.finished
.retain(|(_, held)| held.run_id != entry.run_id);
self.finished.push_back((at, entry));
while self.finished.len() > MAX_RETAINED_FINISHED {
self.finished.pop_front();
}
}
fn prune_finished(&mut self, now: i64) {
let window = self.finished_retention_secs as i64;
while let Some(&(at, _)) = self.finished.front() {
if now.saturating_sub(at) <= window {
break;
}
self.finished.pop_front();
}
}
pub fn set_dead_cycles_before_relief(&mut self, cycles: u32) {
self.dead_cycles_before_relief = cycles;
}
fn observe_lanes(&self, snapshot: &LaneSnapshot, relief: usize) {
self.world
.world()
.resource::<crate::telemetry::Telemetry>()
.0
.observe_lanes(leviath_core::telemetry::LaneHealth {
agents_active: snapshot.agents.active,
agents_waiting: snapshot.agents.waiting,
tools_busy: snapshot.tools_busy,
tools_queued: snapshot.tools_queued,
tools_parked: snapshot.tools_parked,
tools_workers: snapshot.tools_workers,
dead_cycles: self.dead_cycles,
relief_granted: relief,
});
let down: Vec<leviath_core::telemetry::ProviderHealth> = self
.world
.open_circuits()
.into_iter()
.map(|c| leviath_core::telemetry::ProviderHealth {
provider: c.provider,
reason: c.reason.label().to_string(),
consecutive_failures: c.consecutive_failures,
retry_in_secs: c.retry_in_secs,
})
.collect();
self.world
.world()
.resource::<crate::telemetry::Telemetry>()
.0
.observe_providers(&down);
}
pub fn health(&self) -> DaemonHealth {
let snapshot = self.world.lane_snapshot();
DaemonHealth {
agents: snapshot.agents,
inference: snapshot.inference,
tools_busy: snapshot.tools_busy,
tools_queued: snapshot.tools_queued,
tools_parked: snapshot.tools_parked,
tools_workers: snapshot.tools_workers,
dead_cycles: self.dead_cycles,
relief_granted: self.relief_granted,
redrive_secs: self.redrive.as_secs(),
providers_down: self.world.open_circuits(),
}
}
fn progress_fingerprint(&self) -> u64 {
use std::hash::{Hash, Hasher};
let mut total = self.emitted.len() as u64;
for entry in &self.emitted {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
entry.hash(&mut hasher);
total = total.wrapping_add(hasher.finish());
}
total
}
fn log_lane_pressure(&self, snapshot: &LaneSnapshot) {
let agents = snapshot.agents.to_string();
let inference = snapshot.inference_summary();
if self.dead_cycles > 0 {
tracing::warn!(
dead_cycles = self.dead_cycles,
agents = %agents,
inference = %inference,
tools_busy = snapshot.tools_busy,
tools_workers = snapshot.tools_workers,
tools_queued = snapshot.tools_queued,
tools_parked = snapshot.tools_parked,
"no progress while the lanes are full"
);
} else if snapshot.is_under_pressure() {
tracing::info!(
agents = %agents,
inference = %inference,
tools_busy = snapshot.tools_busy,
tools_workers = snapshot.tools_workers,
tools_queued = snapshot.tools_queued,
tools_parked = snapshot.tools_parked,
"lane heartbeat: at capacity with work queued"
);
} else {
tracing::debug!(
agents = %agents,
inference = %inference,
tools_busy = snapshot.tools_busy,
tools_workers = snapshot.tools_workers,
tools_queued = snapshot.tools_queued,
tools_parked = snapshot.tools_parked,
"lane heartbeat"
);
}
}
pub fn set_redrive_interval(&mut self, every: Duration) {
self.redrive = every;
}
pub fn subagent_sender(&self) -> UnboundedSender<SubAgentOp> {
self.subagent_tx.clone()
}
pub fn subscribe(&self) -> broadcast::Receiver<WorldEvent> {
self.events.subscribe()
}
pub fn event_sender(&self) -> broadcast::Sender<WorldEvent> {
self.events.clone()
}
fn emit_events(&mut self) {
self.adopt_unregistered_runs();
let pairs: Vec<(String, Entity)> = self
.by_run_id
.iter()
.map(|(k, &v)| (k.clone(), v))
.collect();
let mut to_reap: Vec<(String, Entity, RunListEntry)> = Vec::new();
let now = chrono::Utc::now().timestamp();
for (run_id, entity) in pairs {
let Some(state) = self.world.world().get::<AgentState>(entity) else {
continue; };
let agent_id = state.agent_id.clone();
let status = status_str(&state.status);
let terminal = matches!(
state.status,
AgentStatus::Complete | AgentStatus::Error { .. } | AgentStatus::Cancelled
);
let cur = {
let totals = self
.world
.world()
.get::<TokenTotals>(entity)
.copied()
.unwrap_or_default();
let (context_tokens, _) = self
.world
.world()
.get::<ContextWindow>(entity)
.map(|w| (w.current_tokens, w.max_tokens))
.unwrap_or((0, 0));
Emitted {
status,
stage: state.current_stage.clone(),
iteration: state.iteration,
tool_calls: totals.tool_calls,
accepts_messages: state.accepts_messages,
prompt_tokens: totals.prompt_tokens,
completion_tokens: totals.completion_tokens,
cached_tokens: totals.cached_tokens,
cache_write_tokens: totals.cache_write_tokens,
context_tokens,
terminal,
}
};
let max_tokens = self
.world
.world()
.get::<ContextWindow>(entity)
.map(|w| w.max_tokens)
.unwrap_or(0);
let prev = self.emitted.get(&run_id).cloned();
if prev.is_none() {
let blueprint = self
.world
.world()
.get::<RunMetadata>(entity)
.map(|m| m.agent_name.clone())
.unwrap_or_default();
let _ = self.events.send(WorldEvent::Spawned {
run_id: run_id.clone(),
agent_id: agent_id.clone(),
blueprint,
});
}
let status_key = |e: &Emitted| {
(
e.status,
e.stage.clone(),
e.iteration,
e.tool_calls,
e.accepts_messages,
)
};
if prev.as_ref().map(status_key) != Some(status_key(&cur)) {
let _ = self.events.send(WorldEvent::Status {
run_id: run_id.clone(),
agent_id: agent_id.clone(),
status: status.to_string(),
stage: cur.stage.clone(),
iteration: cur.iteration,
tool_calls: cur.tool_calls,
accepts_messages: cur.accepts_messages,
});
}
let token_key = |e: &Emitted| {
(
e.prompt_tokens,
e.completion_tokens,
e.cached_tokens,
e.cache_write_tokens,
)
};
if prev.as_ref().map(token_key) != Some(token_key(&cur)) {
let _ = self.events.send(WorldEvent::Tokens {
run_id: run_id.clone(),
agent_id: agent_id.clone(),
prompt_tokens: cur.prompt_tokens,
completion_tokens: cur.completion_tokens,
cached_tokens: cur.cached_tokens,
cache_write_tokens: cur.cache_write_tokens,
});
}
if prev.as_ref().map(|e| e.context_tokens) != Some(cur.context_tokens) {
let _ = self.events.send(WorldEvent::Context {
run_id: run_id.clone(),
agent_id: agent_id.clone(),
total_tokens: cur.context_tokens,
max_tokens,
});
}
let was_terminal = prev.as_ref().map(|e| e.terminal) == Some(true);
if cur.terminal && !was_terminal {
let _ = self.events.send(WorldEvent::Completed {
run_id: run_id.clone(),
agent_id: agent_id.clone(),
status: status.to_string(),
});
}
if cur.terminal && was_terminal && self.no_live_parent(entity) {
let entry = self.entry_for(&run_id, entity, state);
to_reap.push((run_id.clone(), entity, entry));
}
self.emitted.insert(run_id, cur);
}
let mut reaper = self.reaper.take();
for (run_id, entity, entry) in to_reap {
if let Some(reaper) = reaper.as_mut() {
reaper(&mut self.world, entity);
}
self.world.world_mut().despawn(entity);
self.by_run_id.remove(&run_id);
self.emitted.remove(&run_id);
self.record_finished(entry, now);
}
self.reaper = reaper;
self.prune_finished(now);
for (agent_id, request) in self.interactions.pending() {
if self.emitted_interactions.insert(request.id.clone()) {
let _ = self.events.send(WorldEvent::Interaction {
run_id: agent_id.clone(),
agent_id,
request,
});
}
}
}
fn adopt_unregistered_runs(&mut self) {
let live: Vec<(String, Entity)> = self
.world
.world_mut()
.query::<(Entity, &RunMetadata)>()
.iter(self.world.world())
.map(|(entity, md)| (md.run_id.clone(), entity))
.collect();
for (run_id, entity) in live {
if self.live_entity(&run_id) != Some(entity) {
self.by_run_id.insert(run_id, entity);
}
}
}
fn no_live_parent(&self, entity: Entity) -> bool {
let world = self.world.world();
match world.get::<crate::components::ParentRef>(entity) {
None => true,
Some(parent_ref) => match world.get::<AgentState>(parent_ref.parent_entity) {
None => true,
Some(state) => matches!(
state.status,
AgentStatus::Complete | AgentStatus::Error { .. } | AgentStatus::Cancelled
),
},
}
}
pub fn set_spawner(&mut self, spawner: Spawner) {
self.spawner = Some(spawner);
}
pub fn set_spawn_preprocessor(&mut self, pp: SpawnPreprocessor) {
self.spawn_preprocessor = Some(pp);
}
pub fn set_reloader(&mut self, reloader: Reloader) {
self.reloader = Some(reloader);
}
pub fn set_force_terminator(&mut self, force_terminator: ForceTerminator) {
self.force_terminator = Some(force_terminator);
}
pub fn set_reaper(&mut self, reaper: Reaper) {
self.reaper = Some(reaper);
}
fn resolve_or_reload(&mut self, run_id: &str) -> Option<Entity> {
if let Some(entity) = self.live_entity(run_id) {
return Some(entity);
}
let entity = (self.reloader.as_mut()?)(&mut self.world, run_id)?;
self.by_run_id.insert(run_id.to_string(), entity);
Some(entity)
}
pub fn interactions(&self) -> InteractionHub {
self.interactions.clone()
}
pub fn world_mut(&mut self) -> &mut PipelineWorld {
&mut self.world
}
pub fn register(&mut self, run_id: impl Into<String>, entity: Entity) {
self.by_run_id.insert(run_id.into(), entity);
}
fn live_entity(&self, run_id: &str) -> Option<Entity> {
let entity = *self.by_run_id.get(run_id)?;
self.world.world().get::<AgentState>(entity).map(|_| entity)
}
fn handle_subagent(&mut self, op: SubAgentOp) {
match op {
SubAgentOp::Spawn {
args,
parent_run_id,
max_depth,
reply,
} => {
let _ = reply.send(self.spawn_child(*args, &parent_run_id, max_depth));
}
SubAgentOp::Check { run_id, reply } => {
let status = self
.live_entity(&run_id)
.and_then(|e| self.world.agent_status(e));
let _ = reply.send(status);
}
SubAgentOp::Send {
run_id,
caller_run_id,
content,
target_region,
reply,
} => {
if !self.is_within_tree(&run_id, &caller_run_id) {
let _ = reply.send(false);
return;
}
self.resolve_or_reload(&run_id);
let ok = self
.world
.send_message(AgentMessage {
agent_id: run_id,
content,
target_region,
})
.is_ok();
let _ = reply.send(ok);
}
SubAgentOp::Kill {
run_id,
caller_run_id,
reply,
} => {
let within = self.is_within_tree(&run_id, &caller_run_id);
let _ = reply.send(within && self.cancel_tree(&run_id));
}
}
}
fn spawn_child(
&mut self,
mut args: SpawnArgs,
parent_run_id: &str,
max_depth: usize,
) -> Result<String, String> {
args.parent_run_id = Some(parent_run_id.to_string());
let parent = self
.live_entity(parent_run_id)
.ok_or_else(|| format!("parent run '{parent_run_id}' is not live"))?;
let parent_depth = self
.world
.world()
.get::<ParentRef>(parent)
.map_or(0, |p| p.depth);
let child_depth = parent_depth + 1;
if child_depth > max_depth {
return Err(format!(
"sub-agent depth limit ({max_depth}) reached; not spawning deeper"
));
}
let run_id = args.run_id.clone();
let child = match self.spawner.as_mut() {
Some(spawner) => spawner(&mut self.world, &args)?,
None => return Err("this daemon cannot spawn agents".to_string()),
};
let world = self.world.world_mut();
world.entity_mut(child).insert(ParentRef {
parent_entity: parent,
parent_agent_id: parent_run_id.to_string(),
depth: child_depth,
});
match world.get_mut::<SubAgentChildren>(parent) {
Some(mut kids) => kids.children.push(child),
None => {
world.entity_mut(parent).insert(SubAgentChildren {
children: vec![child],
max_child_depth: max_depth,
});
}
}
world
.get_mut::<crate::components::AgentState>(parent)
.expect("a spawning parent always has AgentState")
.spawned_children_ids
.push(run_id.clone());
crate::context_transform::apply_context_transforms(world, parent, child);
self.by_run_id.insert(run_id.clone(), child);
Ok(run_id)
}
fn is_within_tree(&mut self, run_id: &str, ancestor: &str) -> bool {
if run_id == ancestor {
return true;
}
let (Some(target), Some(root)) = (
self.resolve_or_reload(run_id),
self.resolve_or_reload(ancestor),
) else {
return false;
};
let mut stack = vec![root];
while let Some(e) = stack.pop() {
if e == target {
return true;
}
if let Some(kids) = self.world.world().get::<SubAgentChildren>(e) {
stack.extend(kids.children.iter().copied());
}
}
false
}
fn cancel_tree(&mut self, run_id: &str) -> bool {
let Some(root) = self.resolve_or_reload(run_id) else {
return false;
};
let mut subtree = Vec::new();
let mut stack = vec![root];
while let Some(e) = stack.pop() {
subtree.push(e);
if let Some(kids) = self.world.world().get::<SubAgentChildren>(e) {
stack.extend(kids.children.iter().copied());
}
}
let mut cancelled = false;
for e in subtree {
let agent_id = self
.world
.world()
.get::<AgentState>(e)
.map(|s| s.agent_id.clone());
cancelled |= self.world.cancel(e);
if let Some(agent_id) = agent_id {
self.interactions.cancel_for_agent(&agent_id);
let still_open: HashSet<String> = self
.interactions
.pending()
.into_iter()
.map(|(_, req)| req.id)
.collect();
self.emitted_interactions
.retain(|id| still_open.contains(id));
}
}
cancelled
}
pub fn wait_reason(&self, entity: Entity) -> Option<WaitReason> {
let world = self.world.world();
let state = world.get::<AgentState>(entity)?;
if state.status != AgentStatus::Waiting {
return None;
}
if world
.get::<crate::gate_prompt::AwaitingGatePrompt>(entity)
.is_some()
{
return Some(WaitReason::TaintGate);
}
if world
.get::<crate::interaction_points::AwaitingInteractionPoint>(entity)
.is_some()
{
return Some(WaitReason::InteractionPoint);
}
if let Some(fanout) = world.get::<crate::fanout::FanOutWaiting>(entity) {
return Some(WaitReason::FanOutWorkers {
outstanding: fanout.outstanding(),
});
}
if world
.get::<crate::pipeline::WaitingForChildren>(entity)
.is_some()
{
let outstanding = world
.get::<SubAgentChildren>(entity)
.map(|c| {
c.children
.iter()
.filter(|&&child| {
world
.get::<AgentState>(child)
.is_some_and(|s| !crate::pipeline::is_terminal_status(&s.status))
})
.count()
})
.unwrap_or(0);
return Some(WaitReason::Children { outstanding });
}
if world.get::<AwaitingInteraction>(entity).is_some() {
let kind = self
.interactions
.pending()
.into_iter()
.find(|(agent_id, _)| *agent_id == state.agent_id)
.map(|(_, req)| req.kind);
return Some(match kind {
Some(leviath_core::interaction::InteractionKind::ToolApproval) => {
WaitReason::ToolApproval
}
_ => WaitReason::UserPrompt,
});
}
None
}
fn entry_for(&self, run_id: &str, entity: Entity, state: &AgentState) -> RunListEntry {
let world = self.world.world();
let metadata = world.get::<RunMetadata>(entity);
RunListEntry {
run_id: run_id.to_string(),
status: state.status.clone(),
wait_reason: self.wait_reason(entity),
stage: state.current_stage.clone(),
stage_index: world
.get::<crate::pipeline::StageCursor>(entity)
.map(|c| c.index),
num_stages: metadata.map(|m| m.num_stages),
iteration: state.iteration,
tool_calls: world.get::<TokenTotals>(entity).map_or(0, |t| t.tool_calls),
last_progress_at: world
.get::<crate::pipeline::PersistWatermark>(entity)
.and_then(|w| w.last_progress_at()),
unattended: metadata.is_some_and(|m| m.unattended),
empty_output: world
.get::<crate::persistence::RunOutcomeFlags>(entity)
.is_some_and(|f| crate::persistence::is_empty_output(&state.status, &f.0)),
read_paths: metadata.and_then(|m| m.read_paths),
}
}
fn list(&self) -> Vec<RunListEntry> {
let world = self.world.world();
self.by_run_id
.iter()
.filter_map(|(run_id, &entity)| {
let state = world.get::<AgentState>(entity)?;
Some(self.entry_for(run_id, entity, state))
})
.collect()
}
fn finished(&self) -> Vec<RunListEntry> {
self.finished
.iter()
.map(|(_, entry)| entry.clone())
.collect()
}
pub fn handle(&mut self, op: ControlOp) {
match op {
ControlOp::Spawn { args, reply } => {
let result = match self.spawner.as_mut() {
Some(spawner) => {
match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
spawner(&mut self.world, &args)
})) {
Ok(Ok(entity)) => {
self.by_run_id.insert(args.run_id.clone(), entity);
Ok(args.run_id.clone())
}
Ok(Err(e)) => Err(e),
Err(_) => Err("agent spawn panicked".to_string()),
}
}
None => Err("this daemon cannot spawn agents".to_string()),
};
if let Err(error) = &result {
tracing::error!(
run_id = %args.run_id,
blueprint = %args.blueprint_path,
workdir = %args.workdir,
error = %error,
"agent spawn failed"
);
}
let _ = reply.send(result);
}
ControlOp::Status { run_id, reply } => {
let status = self
.live_entity(&run_id)
.and_then(|e| self.world.agent_status(e))
.or_else(|| {
self.finished
.iter()
.find(|(_, e)| e.run_id == run_id)
.map(|(_, e)| e.status.clone())
});
let _ = reply.send(status);
}
ControlOp::Pause { run_id, reply } => {
let ok = self
.resolve_or_reload(&run_id)
.is_some_and(|e| self.world.pause(e));
let _ = reply.send(ok);
}
ControlOp::Resume { run_id, reply } => {
let ok = self
.resolve_or_reload(&run_id)
.is_some_and(|e| self.world.resume(e));
let _ = reply.send(ok);
}
ControlOp::Cancel { run_id, reply } => {
let ok = self.cancel_tree(&run_id)
|| self
.force_terminator
.as_mut()
.is_some_and(|terminate| terminate(&run_id));
let _ = reply.send(ok);
}
ControlOp::List { reply } => {
let _ = reply.send(RunListing {
runs: self.list(),
finished: self.finished(),
health: self.health(),
});
}
ControlOp::Message {
agent_id,
content,
target_region,
reply,
} => {
self.resolve_or_reload(&agent_id);
let ok = self
.world
.send_message(AgentMessage {
agent_id,
content,
target_region,
})
.is_ok();
let _ = reply.send(ok);
}
ControlOp::ListInteractions { reply } => {
let _ = reply.send(self.interactions.pending());
}
ControlOp::AnswerInteraction { response, reply } => {
let _ = reply.send(self.interactions.answer(response));
}
ControlOp::CancelInteraction { request_id, reply } => {
let _ = reply.send(self.interactions.cancel(&request_id));
}
ControlOp::Shutdown { reply } => {
let _ = reply.send(true);
self.world.shutdown();
}
}
}
pub async fn flush_and_stop(&mut self) {
self.world.flush_and_stop().await;
}
pub async fn serve(&mut self, mut control_rx: UnboundedReceiver<ControlOp>) {
let wake = self.world.wake_handle();
let shutdown = self.world.shutdown_handle();
let mut redrive =
tokio::time::interval_at(tokio::time::Instant::now() + self.redrive, self.redrive);
redrive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
'serve: loop {
self.world.run_to_fixed_point();
self.emit_events();
tokio::select! {
_ = wake.notified() => {}
_ = shutdown.notified() => break 'serve,
_ = redrive.tick() => self.observe_redrive(),
op = control_rx.recv() => {
match op {
Some(op) => {
let pre = match &op {
ControlOp::Spawn { args, .. } => {
self.spawn_preprocessor.as_ref().map(|pp| pp(args))
}
_ => None,
};
if let Some(fut) = pre {
fut.await;
}
self.handle(op);
}
None => break 'serve, }
}
Some(sub) = self.subagent_rx.recv() => {
let pre = match &sub {
SubAgentOp::Spawn { args, .. } => {
self.spawn_preprocessor.as_ref().map(|pp| pp(args))
}
_ => None,
};
if let Some(fut) = pre {
fut.await;
}
self.handle_subagent(sub);
}
}
}
self.flush_and_stop().await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dynamic_interaction::InteractionBackend;
use crate::inference_pool::InferencePoolConfig;
use crate::pipeline::{
AgentBlueprint, ReadyToInfer, StageCursor, StageInference, StageInferences, StageProgress,
StageSetup, StageSetups, ToolService, VisitCounts, WaitingForChildren,
};
use crate::tool_bridge::BoxedToolExec;
use leviath_core::{Region, RegionKind};
use leviath_providers::{
FinishReason, InferenceRequest, InferenceResponse, ModelCapabilities, Provider,
ProviderError, TokenUsage,
};
use std::sync::Arc;
use std::sync::Mutex;
use tokio::runtime::Handle;
use tokio::sync::mpsc;
struct Script {
responses: Mutex<std::collections::VecDeque<InferenceResponse>>,
}
#[async_trait::async_trait]
impl Provider for Script {
async fn infer(
&self,
_req: InferenceRequest,
) -> leviath_providers::Result<InferenceResponse> {
self.responses
.lock()
.unwrap()
.pop_front()
.ok_or_else(|| ProviderError::Other("exhausted".to_string()))
}
async fn count_tokens(&self, _t: &str, _m: &str) -> usize {
1
}
fn max_context_tokens(&self, _m: &str) -> usize {
100_000
}
fn name(&self) -> &str {
"script"
}
fn capabilities(&self, _m: &str) -> ModelCapabilities {
ModelCapabilities::default()
}
}
struct NoTools;
impl ToolService for NoTools {
fn exec_for(
&self,
_e: Entity,
calls: Vec<leviath_providers::ToolCall>,
_progress: crate::pipeline::ToolProgress,
) -> BoxedToolExec {
Box::new(move || {
Box::pin(async move { calls.into_iter().map(|c| (c.id, String::new())).collect() })
})
}
}
fn text(content: &str) -> InferenceResponse {
InferenceResponse {
content: content.to_string(),
tool_calls: vec![],
tokens_used: TokenUsage {
prompt_tokens: 1,
completion_tokens: 1,
total_tokens: 2,
cached_tokens: 0,
cache_write_tokens: 0,
},
finish_reason: FinishReason::Complete,
}
}
fn host_with(responses: Vec<InferenceResponse>) -> WorldHost {
let mut registry = crate::providers::ProviderRegistry::new();
registry.register(
"script".to_string(),
Arc::new(Script {
responses: Mutex::new(responses.into_iter().collect()),
}),
);
let world = PipelineWorld::new(
registry,
Arc::new(NoTools),
InferencePoolConfig::new(),
1,
None,
Handle::current(),
);
WorldHost::new(world)
}
fn blueprint() -> leviath_core::Blueprint {
let layout = leviath_core::layout::ContextLayout::new(
vec![leviath_core::layout::RegionDefinition::new(
"conversation".to_string(),
RegionKind::Clearable,
10_000,
)],
12_000,
);
let s = leviath_core::Stage::new(
"s".to_string(),
leviath_core::blueprint::ModelConfig::new("script".to_string(), "m".to_string()),
);
leviath_core::Blueprint::new("t".to_string(), "d".to_string(), vec![s], layout)
}
fn window() -> crate::components::ContextWindow {
let mut w = crate::components::ContextWindow::new(10_000);
w.add_region(Region::new(
"conversation".to_string(),
RegionKind::Clearable,
10_000,
));
w
}
fn agent_state(agent_id: &str) -> AgentState {
AgentState {
agent_id: agent_id.to_string(),
current_stage: "s".to_string(),
iteration: 0,
status: AgentStatus::Active,
spawned_children_ids: vec![],
pending_wait: None,
accepts_messages: true,
}
}
fn si() -> StageInference {
StageInference {
provider_name: "script".to_string(),
model: "m".to_string(),
tools: vec![],
tool_filter: None,
fallbacks: Vec::new(),
}
}
fn setup() -> StageSetup {
StageSetup {
inference_config: crate::components::InferenceConfig {
temperature: None,
max_output_tokens: None,
extra_params: Default::default(),
batch_tool_hint: false,
shell_hint: false,
request_timeout_secs: None,
},
routing: None,
accepts_messages: true,
context_layout: None,
system_prompt: None,
}
}
fn spawn(host: &mut WorldHost, run_id: &str, agent_id: &str) -> Entity {
let e = host.world_mut().spawn_agent((
AgentBlueprint(blueprint()),
StageCursor { index: 0 },
agent_state(agent_id),
crate::components::MessageInbox::default(),
StageProgress::default(),
StageInferences(vec![si()]),
StageSetups(vec![setup()]),
VisitCounts::default(),
window(),
si(),
setup().inference_config,
ReadyToInfer,
));
host.register(run_id, e);
e
}
fn recording_terminator(seen: Arc<Mutex<Vec<String>>>) -> ForceTerminator {
Box::new(move |run_id| {
seen.lock().unwrap().push(run_id.to_string());
run_id != "never-existed"
})
}
fn paging_reloader() -> Reloader {
Box::new(|world, run_id| Some(world.spawn_agent((agent_state(run_id),))))
}
async fn ask<T>(host: &mut WorldHost, make: impl FnOnce(oneshot::Sender<T>) -> ControlOp) -> T {
let (tx, rx) = oneshot::channel();
host.handle(make(tx));
rx.await.unwrap()
}
struct Hangs {
hang: bool,
}
#[async_trait::async_trait]
impl Provider for Hangs {
async fn infer(
&self,
_req: InferenceRequest,
) -> leviath_providers::Result<InferenceResponse> {
if self.hang {
std::future::pending().await
} else {
Err(ProviderError::Other("not hanging".to_string()))
}
}
async fn count_tokens(&self, _t: &str, _m: &str) -> usize {
1
}
fn max_context_tokens(&self, _m: &str) -> usize {
100_000
}
fn name(&self) -> &str {
"hangs"
}
fn capabilities(&self, _m: &str) -> ModelCapabilities {
ModelCapabilities::default()
}
}
#[tokio::test]
async fn the_hanging_provider_answers_everything_except_a_hanging_infer() {
fn request() -> InferenceRequest {
InferenceRequest {
system: vec![],
messages: vec![],
model: "m".to_string(),
max_tokens: 1,
temperature: 0.0,
tools: vec![],
extra: serde_json::Value::Null,
request_timeout_secs: None,
}
}
let p = Hangs { hang: true };
assert_eq!(p.name(), "hangs");
assert_eq!(p.count_tokens("t", "m").await, 1);
assert_eq!(p.max_context_tokens("m"), 100_000);
let _ = p.capabilities("m");
assert!(
tokio::time::timeout(std::time::Duration::from_millis(20), p.infer(request()))
.await
.is_err(),
"hanging: the whole point is that the call never lands"
);
assert!(Hangs { hang: false }.infer(request()).await.is_err());
}
fn host_with_full_pool(limit: usize) -> WorldHost {
let mut registry = crate::providers::ProviderRegistry::new();
registry.register("script".to_string(), Arc::new(Hangs { hang: true }));
let mut pools = InferencePoolConfig::new();
pools.set_limit("m", limit);
WorldHost::new(PipelineWorld::new(
registry,
Arc::new(NoTools),
pools,
1,
None,
Handle::current(),
))
}
const PARK: std::time::Duration = std::time::Duration::from_millis(250);
async fn serve_until_inferring(
host: &mut WorldHost,
rounds: usize,
park: std::time::Duration,
entity: Entity,
) -> bool {
let wake = host.world_mut().wake_handle();
for _ in 0..rounds {
host.world_mut().run_to_fixed_point();
if is_inferring(host, entity) {
return true;
}
if tokio::time::timeout(park, wake.notified()).await.is_err() {
break; }
}
false
}
fn is_inferring(host: &mut WorldHost, entity: Entity) -> bool {
host.world_mut()
.world()
.get::<crate::pipeline::AwaitingInference>(entity)
.is_some()
}
#[tokio::test]
async fn releasing_a_cancelled_runs_permit_wakes_the_starved_agent_behind_it() {
let mut host = host_with_full_pool(1);
let holder = spawn(&mut host, "run-a", "agent-a");
host.world_mut().run_to_fixed_point();
assert!(is_inferring(&mut host, holder), "the holder takes the slot");
let starved = spawn(&mut host, "run-b", "agent-b");
host.world_mut().run_to_fixed_point();
assert!(
!is_inferring(&mut host, starved),
"the second agent is starved on the full pool"
);
assert!(
!serve_until_inferring(&mut host, 3, PARK, starved).await,
"no slot, no dispatch"
);
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "run-a".to_string(),
reply,
})
.await
);
assert!(
serve_until_inferring(&mut host, 8, PARK, starved).await,
"the freed slot must wake the loop so the starved agent can take it; \
without that wake the daemon parks with capacity it cannot see"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_redrives_the_world_on_its_own_timer_with_no_wake() {
use std::sync::atomic::{AtomicUsize, Ordering};
static TICKS: AtomicUsize = AtomicUsize::new(0);
TICKS.store(0, Ordering::SeqCst);
fn count_ticks() {
TICKS.fetch_add(1, Ordering::SeqCst);
}
let mut host = host_with(vec![]);
host.world_mut().add_test_system(count_ticks);
host.set_redrive_interval(std::time::Duration::from_millis(20));
let shutdown = host.world_mut().shutdown_handle();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move {
host.serve(op_rx).await;
});
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
let ticks = TICKS.load(Ordering::SeqCst);
shutdown.notify_one();
drop(op_tx);
handle.await.unwrap();
assert!(
ticks > 3,
"the timer must keep driving the world with nothing waking it; saw {ticks} ticks"
);
}
fn two_stage_blueprint() -> leviath_core::Blueprint {
let layout = leviath_core::layout::ContextLayout::new(
vec![leviath_core::layout::RegionDefinition::new(
"conversation".to_string(),
RegionKind::Clearable,
10_000,
)],
12_000,
);
let model =
leviath_core::blueprint::ModelConfig::new("script".to_string(), "m".to_string());
let mut one = leviath_core::Stage::new("one".to_string(), model.clone());
one.max_iterations = Some(1);
let mut two = leviath_core::Stage::new("two".to_string(), model);
two.max_iterations = Some(1);
let stages = vec![one, two];
leviath_core::Blueprint::new("t".to_string(), "d".to_string(), stages, layout)
}
fn spawn_two_stage(host: &mut WorldHost, run_id: &str, agent_id: &str) -> Entity {
let mut state = agent_state(agent_id);
state.current_stage = "one".to_string();
let e = host.world_mut().spawn_agent((
AgentBlueprint(two_stage_blueprint()),
StageCursor { index: 0 },
state,
crate::components::MessageInbox::default(),
StageProgress::default(),
StageInferences(vec![si(), si()]),
StageSetups(vec![setup(), setup()]),
VisitCounts::default(),
window(),
si(),
setup().inference_config,
ReadyToInfer,
));
host.register(run_id, e);
e
}
fn tool_call(id: &str) -> InferenceResponse {
InferenceResponse {
tool_calls: vec![leviath_providers::ToolCall {
id: id.to_string(),
name: "noop".to_string(),
arguments: serde_json::Value::Null,
thought_signature: None,
}],
..text("working")
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stage_boundary_is_crossed_without_waiting_for_the_redrive() {
let mut host = host_with(vec![tool_call("c1"), tool_call("c2")]);
host.set_redrive_interval(std::time::Duration::from_secs(3600));
spawn_two_stage(&mut host, "run-a", "agent-a");
let mut events = host.subscribe();
let shutdown = host.world_mut().shutdown_handle();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move { host.serve(op_rx).await });
let completed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
let event = events
.recv()
.await
.expect("the event stream must outlive the run");
if let WorldEvent::Completed { status, .. } = event {
break status;
}
}
})
.await;
shutdown.notify_one();
drop(op_tx);
handle.await.unwrap();
assert_eq!(
completed.expect("the run must reach stage two and finish on wakes alone"),
"complete"
);
}
#[tokio::test]
async fn the_lane_heartbeat_distinguishes_pressure_from_idle() {
leviath_testkit::with_tracing(|| async {
let mut host = host_with_full_pool(1);
let idle = host.world_mut().lane_snapshot();
assert!(!idle.is_under_pressure(), "an empty world is not pressured");
assert_eq!(idle.inference_summary(), "none");
host.log_lane_pressure(&idle);
spawn(&mut host, "run-a", "agent-a");
spawn(&mut host, "run-b", "agent-b");
host.world_mut().run_to_fixed_point();
let busy = host.world_mut().lane_snapshot();
assert_eq!(busy.agents.active, 2);
assert_eq!(busy.inference_summary(), "m=1/1");
assert!(
busy.is_under_pressure(),
"a full pool with active agents is exactly the state worth reporting"
);
host.log_lane_pressure(&busy); })
.await;
}
#[tokio::test]
async fn re_drives_that_go_nowhere_under_pressure_count_as_dead_cycles() {
leviath_testkit::with_tracing(|| async {
let mut host = host_with_full_pool(1);
spawn(&mut host, "run-a", "agent-a");
spawn(&mut host, "run-b", "agent-b");
host.world_mut().run_to_fixed_point();
host.emit_events();
host.observe_redrive();
assert_eq!(host.dead_cycles, 0, "the first cycle sets the baseline");
host.observe_redrive();
assert_eq!(host.dead_cycles, 1, "a whole interval, nothing moved");
host.observe_redrive();
assert_eq!(host.dead_cycles, 2, "and another - this is the `warn` arm");
})
.await;
}
#[tokio::test]
async fn a_run_that_moves_clears_the_dead_cycle_count() {
let mut host = host_with_full_pool(1);
let entity = spawn(&mut host, "run-a", "agent-a");
spawn(&mut host, "run-b", "agent-b");
host.world_mut().run_to_fixed_point();
host.emit_events();
host.observe_redrive();
host.observe_redrive();
assert_eq!(host.dead_cycles, 1, "wedged to begin with");
host.world_mut()
.world_mut()
.get_mut::<AgentState>(entity)
.expect("the agent is loaded")
.iteration += 1;
host.emit_events();
host.observe_redrive();
assert_eq!(host.dead_cycles, 0, "something moved");
}
async fn wedge_the_tool_lane(host: &mut WorldHost) -> crate::cancel::CancelToken {
let snapshot = host.world_mut().lane_snapshot();
let stage = host
.world_mut()
.world()
.resource::<crate::pipeline::ToolStage>()
.clone();
let release = crate::cancel::CancelToken::new();
let submit = |exec: crate::tool_bridge::BoxedToolExec| {
stage.stats.enqueued();
stage
.jobs
.send(crate::tool_bridge::ToolJob {
entity: Entity::from_raw_u32(9_001).expect("a small index is a valid id"),
exec,
cancel: crate::cancel::CancelToken::new(),
})
.expect("the lane is serving");
};
let blocker = || {
let held = release.clone();
submit(Box::new(move || {
Box::pin(async move {
held.cancelled().await;
Vec::new()
})
}));
};
for _ in 0..snapshot.tools_workers.saturating_sub(snapshot.tools_busy) {
blocker();
}
await_full_lane(host).await;
blocker(); await_saturation(host).await;
release
}
async fn await_full_lane(host: &mut WorldHost) {
await_lane(host, "the lane filled up", |snapshot| {
snapshot.tools_busy >= snapshot.tools_workers
})
.await;
}
async fn await_saturation(host: &mut WorldHost) {
await_lane(host, "the lane saturated", |snapshot| {
snapshot.tools_saturated
})
.await;
}
async fn await_drained_queue(host: &mut WorldHost) {
await_lane(host, "the queued batch got in", |snapshot| {
snapshot.tools_queued == 0
})
.await;
}
async fn await_lane(
host: &mut WorldHost,
context: &str,
done: fn(&crate::world::LaneSnapshot) -> bool,
) {
tokio::time::timeout(std::time::Duration::from_secs(30), async {
while !done(&host.world_mut().lane_snapshot()) {
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect(context);
}
async fn release_the_lane(host: &mut WorldHost, releases: &[crate::cancel::CancelToken]) {
for release in releases {
release.cancel();
}
await_lane(host, "the lane emptied", |snapshot| {
snapshot.tools_busy == 0 && snapshot.tools_queued == 0
})
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_lane_that_never_drains_is_widened_rather_than_emptied() {
leviath_testkit::with_tracing(|| async {
let mut host = host_with_full_pool(1);
host.set_dead_cycles_before_relief(2);
let release = wedge_the_tool_lane(&mut host).await;
host.observe_redrive(); host.observe_redrive(); assert_eq!(host.relief_granted, 0, "still inside the grace period");
host.observe_redrive(); assert_eq!(host.relief_granted, 1, "the lane got wider");
assert_eq!(
host.dead_cycles, 0,
"the streak restarts so relief is not granted again immediately"
);
assert_eq!(host.health().tools_workers, 2);
await_drained_queue(&mut host).await;
assert_eq!(host.world_mut().lane_snapshot().tools_busy, 2);
release_the_lane(&mut host, &[release]).await;
})
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn relief_stops_after_one_extra_lane_s_worth() {
leviath_testkit::with_tracing(|| async {
let mut host = host_with_full_pool(1);
host.set_dead_cycles_before_relief(1);
let release = wedge_the_tool_lane(&mut host).await;
host.observe_redrive();
host.observe_redrive();
assert_eq!(host.relief_granted, 1);
let release_two = wedge_the_tool_lane(&mut host).await;
for _ in 0..4 {
host.observe_redrive();
}
assert_eq!(host.relief_granted, 1, "the budget was already spent");
release_the_lane(&mut host, &[release, release_two]).await;
})
.await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn relief_can_be_turned_off_without_turning_off_detection() {
leviath_testkit::with_tracing(|| async {
let mut host = host_with_full_pool(1);
host.set_dead_cycles_before_relief(0);
let release = wedge_the_tool_lane(&mut host).await;
for _ in 0..4 {
host.observe_redrive();
}
assert_eq!(host.relief_granted, 0, "relief is disabled");
assert_eq!(host.dead_cycles, 3, "but the streak is still counted");
release_the_lane(&mut host, &[release]).await;
})
.await;
}
#[tokio::test]
async fn each_re_drive_reports_lane_health_to_the_telemetry_sink() {
let sink = Arc::new(leviath_core::telemetry::MemorySink::default());
let mut host = host_with_full_pool(1);
host.world_mut()
.world_mut()
.insert_resource(crate::telemetry::Telemetry(sink.clone()));
spawn(&mut host, "run-a", "agent-a");
spawn(&mut host, "run-b", "agent-b");
host.world_mut().run_to_fixed_point();
host.emit_events();
host.observe_redrive();
host.observe_redrive();
let samples = sink.lane_samples();
assert_eq!(samples.len(), 2, "one per re-drive");
assert_eq!(samples[0].dead_cycles, 0);
assert_eq!(samples[1].dead_cycles, 1, "the streak is carried through");
assert_eq!(samples[1].agents_active, 2);
}
#[tokio::test]
async fn each_re_drive_reports_providers_out_of_service() {
let sink = Arc::new(leviath_core::telemetry::MemorySink::default());
let mut host = host_with(vec![]);
host.world_mut()
.world_mut()
.insert_resource(crate::telemetry::Telemetry(sink.clone()));
let policy = crate::pipeline::CircuitPolicy {
failures_before_open: 1,
cooldown_secs: 300,
};
let mut circuits = crate::pipeline::ProviderCircuits::default();
circuits.record_failure(
"openrouter",
leviath_providers::UnavailableReason::CreditsExhausted,
chrono::Utc::now().timestamp(),
&policy,
);
host.world_mut().world_mut().insert_resource(circuits);
host.world_mut().world_mut().insert_resource(policy);
host.observe_redrive();
let samples = sink.provider_samples();
assert_eq!(samples.len(), 1);
assert_eq!(samples[0].len(), 1);
assert_eq!(samples[0][0].provider, "openrouter");
assert_eq!(samples[0][0].reason, "credits-exhausted");
assert_eq!(samples[0][0].consecutive_failures, 1);
assert!(samples[0][0].retry_in_secs > 0);
assert_eq!(host.health().providers_down.len(), 1);
host.world_mut()
.world_mut()
.resource_mut::<crate::pipeline::ProviderCircuits>()
.record_success("openrouter");
host.observe_redrive();
assert!(sink.provider_samples()[1].is_empty());
assert!(host.health().providers_down.is_empty());
}
#[tokio::test]
async fn an_idle_daemon_never_counts_a_dead_cycle() {
let mut host = host_with_full_pool(1);
host.emit_events();
for _ in 0..3 {
host.observe_redrive();
}
assert_eq!(host.dead_cycles, 0, "no pressure, no dead cycles");
}
#[tokio::test]
async fn the_lane_snapshot_counts_agents_by_status() {
let mut host = host_with(vec![]);
let active = spawn(&mut host, "run-active", "a");
let paused = spawn(&mut host, "run-paused", "b");
let waiting = spawn(&mut host, "run-waiting", "c");
let done = spawn(&mut host, "run-done", "d");
let idle = spawn(&mut host, "run-idle", "e");
host.world_mut().set_status(paused, AgentStatus::Paused);
host.world_mut().set_status(waiting, AgentStatus::Waiting);
host.world_mut().set_status(done, AgentStatus::Complete);
host.world_mut().set_status(idle, AgentStatus::Idle);
let counts = host.world_mut().lane_snapshot().agents;
assert_eq!(counts.active, 1);
assert_eq!(counts.paused, 1);
assert_eq!(counts.waiting, 1);
assert_eq!(counts.terminal, 1);
assert_eq!(counts.idle, 1);
assert_eq!(
counts.to_string(),
"active=1 waiting=1 paused=1 idle=1 terminal=1"
);
host.world_mut().set_status(active, AgentStatus::Cancelled);
host.world_mut().set_status(
paused,
AgentStatus::Error {
message: "boom".to_string(),
},
);
assert_eq!(host.world_mut().lane_snapshot().agents.terminal, 3);
}
#[tokio::test]
async fn status_and_list_reflect_registered_runs() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "agent-a");
let status = ask(&mut host, |reply| ControlOp::Status {
run_id: "run-a".to_string(),
reply,
})
.await;
assert_eq!(status, Some(AgentStatus::Active));
let list = ask(&mut host, |reply| ControlOp::List { reply }).await.runs;
assert_eq!(list.len(), 1);
assert_eq!(list[0].run_id, "run-a");
assert_eq!(list[0].status, AgentStatus::Active);
assert_eq!(list[0].wait_reason, None);
let none = ask(&mut host, |reply| ControlOp::Status {
run_id: "ghost".to_string(),
reply,
})
.await;
assert_eq!(none, None);
}
#[tokio::test]
async fn pause_resume_cancel_by_run_id() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "agent-a");
assert!(
ask(&mut host, |reply| ControlOp::Pause {
run_id: "run-a".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(host.by_run_id["run-a"]),
Some(AgentStatus::Paused)
);
assert!(
!ask(&mut host, |reply| ControlOp::Pause {
run_id: "run-a".to_string(),
reply
})
.await
);
assert!(
ask(&mut host, |reply| ControlOp::Resume {
run_id: "run-a".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(host.by_run_id["run-a"]),
Some(AgentStatus::Active)
);
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "run-a".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(host.by_run_id["run-a"]),
Some(AgentStatus::Cancelled)
);
assert!(
!ask(&mut host, |reply| ControlOp::Pause {
run_id: "ghost".to_string(),
reply
})
.await
);
assert!(
!ask(&mut host, |reply| ControlOp::Resume {
run_id: "ghost".to_string(),
reply
})
.await
);
assert!(
!ask(&mut host, |reply| ControlOp::Cancel {
run_id: "ghost".to_string(),
reply
})
.await
);
}
#[tokio::test]
async fn spawn_op_uses_installed_spawner_and_registers() {
let mut host = host_with(vec![]);
host.set_spawner(Box::new(|world, args| {
Ok(world.spawn_agent((agent_state(&args.run_id),)))
}));
let result = ask(&mut host, |reply| ControlOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "r1".to_string(),
..Default::default()
}),
reply,
})
.await;
assert_eq!(result, Ok("r1".to_string()));
let status = ask(&mut host, |reply| ControlOp::Status {
run_id: "r1".to_string(),
reply,
})
.await;
assert_eq!(status, Some(AgentStatus::Active));
}
#[tokio::test]
async fn spawn_op_propagates_spawner_error() {
let mut host = host_with(vec![]);
host.set_spawner(Box::new(|_world, _args| Err("bad blueprint".to_string())));
let result = ask(&mut host, |reply| ControlOp::Spawn {
args: Box::new(SpawnArgs::default()),
reply,
})
.await;
assert_eq!(result, Err("bad blueprint".to_string()));
}
#[tokio::test]
async fn spawn_op_contains_a_panicking_spawner() {
let mut host = host_with(vec![]);
host.set_spawner(Box::new(|_world, _args| panic!("simulated spawn panic")));
let (tx, rx) = oneshot::channel();
crate::test_support::with_silenced_panics(|| {
host.handle(ControlOp::Spawn {
args: Box::new(SpawnArgs::default()),
reply: tx,
});
});
assert_eq!(rx.await.unwrap(), Err("agent spawn panicked".to_string()));
let status = ask(&mut host, |reply| ControlOp::Status {
run_id: SpawnArgs::default().run_id,
reply,
})
.await;
assert!(status.is_none());
}
#[tokio::test]
async fn spawn_op_errors_without_a_spawner() {
let mut host = host_with(vec![]);
let result = ask(&mut host, |reply| ControlOp::Spawn {
args: Box::new(SpawnArgs::default()),
reply,
})
.await;
assert!(result.unwrap_err().contains("cannot spawn"));
}
async fn ask_sub<T>(
host: &mut WorldHost,
make: impl FnOnce(oneshot::Sender<T>) -> SubAgentOp,
) -> T {
let (tx, rx) = oneshot::channel();
host.handle_subagent(make(tx));
rx.await.unwrap()
}
fn child_spawner() -> Spawner {
Box::new(|world, args| Ok(world.spawn_agent((agent_state(&args.run_id),))))
}
#[tokio::test]
async fn subagent_spawn_links_child_and_registers() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
let parent = spawn(&mut host, "parent", "parent");
let result = ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "child".to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await;
assert_eq!(result, Ok("child".to_string()));
let child = host.by_run_id["child"];
let pref = host.world.world().get::<ParentRef>(child).unwrap();
assert_eq!(pref.parent_entity, parent);
assert_eq!(pref.depth, 1);
let kids = host.world.world().get::<SubAgentChildren>(parent).unwrap();
assert_eq!(kids.children, vec![child]);
}
#[tokio::test]
async fn subagent_spawn_appends_to_existing_children() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
spawn(&mut host, "parent", "parent");
for id in ["c1", "c2"] {
let r = ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: id.to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await;
assert!(r.is_ok());
}
let parent = host.by_run_id["parent"];
let kids = host.world.world().get::<SubAgentChildren>(parent).unwrap();
assert_eq!(kids.children.len(), 2);
}
#[tokio::test]
async fn subagent_spawn_rejects_beyond_max_depth() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
spawn(&mut host, "parent", "parent");
let result = ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "child".to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 0, reply,
})
.await;
assert!(result.unwrap_err().contains("depth limit"));
assert!(!host.by_run_id.contains_key("child"));
}
#[tokio::test]
async fn subagent_spawn_unknown_parent_and_no_spawner_and_spawner_error() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
let r = ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs::default()),
parent_run_id: "ghost".to_string(),
max_depth: 3,
reply,
})
.await;
assert!(r.unwrap_err().contains("not live"));
let mut host2 = host_with(vec![]);
spawn(&mut host2, "parent", "parent");
let r = ask_sub(&mut host2, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs::default()),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await;
assert!(r.unwrap_err().contains("cannot spawn"));
let mut host3 = host_with(vec![]);
host3.set_spawner(Box::new(|_w, _a| Err("bad blueprint".to_string())));
spawn(&mut host3, "parent", "parent");
let r = ask_sub(&mut host3, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs::default()),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await;
assert_eq!(r, Err("bad blueprint".to_string()));
}
#[tokio::test]
async fn subagent_check_reports_status_or_none() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "run-a");
let status = ask_sub(&mut host, |reply| SubAgentOp::Check {
run_id: "run-a".to_string(),
reply,
})
.await;
assert_eq!(status, Some(AgentStatus::Active));
let none = ask_sub(&mut host, |reply| SubAgentOp::Check {
run_id: "ghost".to_string(),
reply,
})
.await;
assert_eq!(none, None);
}
#[tokio::test]
async fn subagent_ops_reach_a_run_the_caller_spawned() {
let mut host = host_with(vec![]);
let parent = spawn(&mut host, "parent", "parent");
let child = spawn(&mut host, "child", "child");
host.world_mut()
.world_mut()
.entity_mut(parent)
.insert(SubAgentChildren {
children: vec![child],
max_child_depth: 3,
});
let delivered = ask_sub(&mut host, |reply| SubAgentOp::Send {
run_id: "child".to_string(),
caller_run_id: "parent".to_string(),
content: "carry on".to_string(),
target_region: None,
reply,
})
.await;
assert!(delivered, "a run we spawned is ours to message");
}
#[tokio::test]
async fn subagent_ops_refuse_a_run_outside_the_callers_tree() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "run-a");
spawn(&mut host, "outsider", "outsider");
let delivered = ask_sub(&mut host, |reply| SubAgentOp::Send {
run_id: "outsider".to_string(),
caller_run_id: "run-a".to_string(),
content: "take this".to_string(),
target_region: None,
reply,
})
.await;
assert!(!delivered, "a run we did not spawn is not ours to message");
let killed = ask_sub(&mut host, |reply| SubAgentOp::Kill {
run_id: "outsider".to_string(),
caller_run_id: "run-a".to_string(),
reply,
})
.await;
assert!(!killed, "nor ours to cancel");
let phantom = ask_sub(&mut host, |reply| SubAgentOp::Send {
run_id: "no-such-run".to_string(),
caller_run_id: "run-a".to_string(),
content: "hello?".to_string(),
target_region: None,
reply,
})
.await;
assert!(!phantom, "an unknown run id is in nobody's tree");
}
#[tokio::test]
async fn subagent_send_delivers_to_inbox() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "run-a");
let ok = ask_sub(&mut host, |reply| SubAgentOp::Send {
run_id: "run-a".to_string(),
caller_run_id: "run-a".to_string(),
content: "hello child".to_string(),
target_region: None,
reply,
})
.await;
assert!(ok);
}
#[tokio::test]
async fn subagent_send_delivers_into_the_target_region() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
host.world
.world_mut()
.get_mut::<crate::components::ContextWindow>(e)
.unwrap()
.add_region(Region::new(
"notes".to_string(),
RegionKind::Clearable,
5000,
));
let ok = ask_sub(&mut host, |reply| SubAgentOp::Send {
run_id: "run-a".to_string(),
caller_run_id: "run-a".to_string(),
content: "filed under notes".to_string(),
target_region: Some("notes".to_string()),
reply,
})
.await;
assert!(ok);
host.world.tick(); let window = host
.world
.world()
.get::<crate::components::ContextWindow>(e)
.unwrap();
assert!(window.get_region("notes").unwrap().current_tokens > 0);
assert_eq!(window.get_region("conversation").unwrap().current_tokens, 0);
}
#[tokio::test]
async fn subagent_kill_cancels_the_whole_tree() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
spawn(&mut host, "parent", "parent");
ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "child".to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await
.unwrap();
let ok = ask_sub(&mut host, |reply| SubAgentOp::Kill {
run_id: "parent".to_string(),
caller_run_id: "parent".to_string(),
reply,
})
.await;
assert!(ok);
assert_eq!(
host.world.agent_status(host.by_run_id["parent"]),
Some(AgentStatus::Cancelled)
);
assert_eq!(
host.world.agent_status(host.by_run_id["child"]),
Some(AgentStatus::Cancelled)
);
let miss = ask_sub(&mut host, |reply| SubAgentOp::Kill {
run_id: "ghost".to_string(),
caller_run_id: "ghost".to_string(),
reply,
})
.await;
assert!(!miss);
}
#[tokio::test]
async fn cancel_cascades_to_the_whole_tree() {
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
spawn(&mut host, "parent", "parent");
ask_sub(&mut host, |reply| SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "child".to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply,
})
.await
.unwrap();
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "parent".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(host.by_run_id["child"]),
Some(AgentStatus::Cancelled),
"cancelling the parent cancels its children"
);
}
#[tokio::test]
async fn cancel_tolerates_a_child_that_has_already_been_reaped() {
let mut host = host_with(vec![]);
let parent = spawn(&mut host, "parent", "parent");
let ghost = host.world_mut().spawn_agent((agent_state("ghost"),));
host.world_mut()
.world_mut()
.entity_mut(parent)
.insert(SubAgentChildren {
children: vec![ghost],
max_child_depth: 3,
});
host.world_mut().world_mut().despawn(ghost);
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "parent".to_string(),
reply
})
.await,
"the parent is still cancelled"
);
assert_eq!(
host.world.agent_status(parent),
Some(AgentStatus::Cancelled)
);
}
#[tokio::test]
async fn cancel_closes_the_runs_open_interactions() {
let mut host = host_with(vec![]);
let hub = host.interactions();
spawn(&mut host, "run-a", "agent-a");
let backend = hub.backend_for("agent-a");
let asking = tokio::spawn(async move {
backend
.ask(InteractionRequest::free_text("q", "ask", "stage", true))
.await
});
while hub.pending().is_empty() {
tokio::task::yield_now().await;
}
host.emit_events();
assert!(
!host.emitted_interactions.is_empty(),
"the open request was emitted"
);
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "run-a".to_string(),
reply,
})
.await;
tokio::time::timeout(std::time::Duration::from_secs(5), asking)
.await
.expect("cancelling the run releases its blocked ask")
.expect("the ask task did not panic");
assert!(hub.pending().is_empty(), "no orphaned prompt is left open");
assert!(
host.emitted_interactions.is_empty(),
"and it is pruned from the emitted set, not re-announced forever"
);
}
#[tokio::test]
async fn cancel_falls_back_to_the_force_terminator_when_the_world_cannot_hold_the_run() {
let mut host = host_with(vec![]);
host.set_reloader(Box::new(|_world, _run_id| None));
let terminated = Arc::new(Mutex::new(Vec::new()));
host.set_force_terminator(recording_terminator(terminated.clone()));
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "unreloadable".to_string(),
reply
})
.await,
"a run that can't be reloaded is still terminated"
);
assert!(
!ask(&mut host, |reply| ControlOp::Cancel {
run_id: "never-existed".to_string(),
reply
})
.await,
"`false` is reserved for a run that exists nowhere"
);
assert_eq!(
*terminated.lock().unwrap(),
vec!["unreloadable".to_string(), "never-existed".to_string()]
);
}
#[tokio::test]
async fn cancel_does_not_force_terminate_a_run_it_could_cancel() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "agent-a");
let terminated = Arc::new(Mutex::new(Vec::new()));
host.set_force_terminator(recording_terminator(terminated.clone()));
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "run-a".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(host.by_run_id["run-a"]),
Some(AgentStatus::Cancelled)
);
assert!(
terminated.lock().unwrap().is_empty(),
"the disk fallback stayed unused"
);
}
#[tokio::test]
async fn unregistered_world_agents_are_adopted_and_become_cancellable() {
let mut host = host_with(vec![]);
let entity = host.world_mut().spawn_agent((
agent_state("worker"),
RunMetadata {
run_id: "worker-run".to_string(),
agent_name: "w".to_string(),
agent_path: String::new(),
task: String::new(),
model: None,
workdir: String::new(),
num_stages: 1,
started_at: 0,
parent_run_id: None,
metadata: Default::default(),
callback_url: None,
callback_secret: None,
title: None,
unattended: false,
read_paths: None,
},
));
assert!(
!host.by_run_id.contains_key("worker-run"),
"not registered by the spawn itself"
);
host.emit_events();
assert_eq!(host.live_entity("worker-run"), Some(entity), "adopted");
host.set_reloader(paging_reloader());
assert!(
ask(&mut host, |reply| ControlOp::Cancel {
run_id: "worker-run".to_string(),
reply
})
.await
);
assert_eq!(
host.world.agent_status(entity),
Some(AgentStatus::Cancelled),
"the original entity is cancelled, not a reloaded copy"
);
}
#[tokio::test]
async fn interaction_ops_list_answer_and_cancel() {
let mut host = host_with(vec![]);
let hub = host.interactions();
let backend = hub.backend_for("agent-a");
let asking = tokio::spawn(async move {
backend
.ask(leviath_core::interaction::InteractionRequest::free_text(
"q1", "prompt?", "stage", true,
))
.await
});
for _ in 0..8 {
tokio::task::yield_now().await;
}
let list = ask(&mut host, |reply| ControlOp::ListInteractions { reply }).await;
assert_eq!(list.len(), 1);
assert_eq!(list[0].0, "agent-a");
let ok = ask(&mut host, |reply| ControlOp::AnswerInteraction {
response: leviath_core::interaction::InteractionResponse::text("q1", "hi"),
reply,
})
.await;
assert!(ok);
assert_eq!(asking.await.unwrap().value.as_deref(), Some("hi"));
let cancelled = ask(&mut host, |reply| ControlOp::CancelInteraction {
request_id: "gone".to_string(),
reply,
})
.await;
assert!(!cancelled);
}
#[tokio::test]
async fn cancel_interaction_op_wakes_asker() {
let mut host = host_with(vec![]);
let backend = host.interactions().backend_for("agent-a");
let asking = tokio::spawn(async move {
backend
.ask(leviath_core::interaction::InteractionRequest::free_text(
"q2", "p", "s", true,
))
.await
});
for _ in 0..8 {
tokio::task::yield_now().await;
}
let ok = ask(&mut host, |reply| ControlOp::CancelInteraction {
request_id: "q2".to_string(),
reply,
})
.await;
assert!(ok);
assert_eq!(asking.await.unwrap().request_id, "q2");
}
#[tokio::test]
async fn message_op_is_delivered() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "agent-a");
let ok = ask(&mut host, |reply| ControlOp::Message {
agent_id: "agent-a".to_string(),
content: "hi".to_string(),
target_region: Some("conversation".to_string()),
reply,
})
.await;
assert!(ok);
host.world_mut().tick();
assert!(
host.world
.world()
.get::<crate::components::ContextWindow>(e)
.unwrap()
.get_region("conversation")
.unwrap()
.current_tokens
> 0
);
}
#[tokio::test]
async fn serve_drives_agents_and_handles_ops_until_shutdown() {
let mut host = host_with(vec![text("t1"), text("t2"), text("t3"), text("t4")]);
spawn(&mut host, "run-a", "agent-a");
let shutdown = host.world_mut().shutdown_handle();
let mut events = host.subscribe();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move {
host.serve(op_rx).await;
});
let (tx, rx) = oneshot::channel();
op_tx
.send(ControlOp::Status {
run_id: "run-a".to_string(),
reply: tx,
})
.unwrap();
let _ = rx.await.unwrap();
let completed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
if let Ok(WorldEvent::Completed { run_id, status, .. }) = events.recv().await {
return (run_id, status);
}
}
})
.await
.expect("the serve loop must drive the agent to a terminal status");
assert_eq!(completed, ("run-a".to_string(), "complete".to_string()));
shutdown.notify_one();
handle.await.unwrap();
}
#[tokio::test]
async fn serve_awaits_spawn_preprocessor_before_spawning() {
use std::sync::atomic::{AtomicBool, Ordering};
let mut host = host_with(vec![]);
let ran = Arc::new(AtomicBool::new(false));
let ran_pp = ran.clone();
host.set_spawn_preprocessor(Box::new(move |_args| {
let ran = ran_pp.clone();
Box::pin(async move {
ran.store(true, Ordering::SeqCst);
})
}));
let ran_spawn = ran.clone();
host.set_spawner(Box::new(move |world, args| {
assert!(ran_spawn.load(Ordering::SeqCst));
Ok(world.spawn_agent((agent_state(&args.run_id),)))
}));
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move {
host.serve(op_rx).await;
});
let (tx, rx) = oneshot::channel();
op_tx
.send(ControlOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "rp".to_string(),
..Default::default()
}),
reply: tx,
})
.unwrap();
let result = rx.await.unwrap();
drop(op_tx); handle.await.unwrap();
assert_eq!(result, Ok("rp".to_string()));
assert!(ran.load(Ordering::SeqCst), "preprocessor ran");
}
#[tokio::test]
async fn serve_awaits_preprocessor_for_subagent_spawn() {
use std::sync::atomic::{AtomicUsize, Ordering};
let mut host = host_with(vec![]);
host.set_spawner(child_spawner());
let parent = host.world_mut().spawn_agent((agent_state("parent"),));
host.register("parent", parent);
let calls = Arc::new(AtomicUsize::new(0));
let calls_pp = calls.clone();
host.set_spawn_preprocessor(Box::new(move |_args| {
let calls = calls_pp.clone();
Box::pin(async move {
calls.fetch_add(1, Ordering::SeqCst);
})
}));
let sub_tx = host.subagent_sender();
let shutdown = host.world_mut().shutdown_handle();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move {
host.serve(op_rx).await;
});
let (ctx, crx) = oneshot::channel();
sub_tx
.send(SubAgentOp::Check {
run_id: "parent".to_string(),
reply: ctx,
})
.unwrap();
let _ = crx.await.unwrap();
let (stx, srx) = oneshot::channel();
sub_tx
.send(SubAgentOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "child".to_string(),
..Default::default()
}),
parent_run_id: "parent".to_string(),
max_depth: 3,
reply: stx,
})
.unwrap();
assert_eq!(srx.await.unwrap(), Ok("child".to_string()));
shutdown.notify_one();
drop(op_tx);
handle.await.unwrap();
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"only the Spawn preprocessed"
);
}
#[tokio::test]
async fn serve_spawns_without_a_preprocessor() {
let mut host = host_with(vec![]);
host.set_spawner(Box::new(|world, args| {
Ok(world.spawn_agent((agent_state(&args.run_id),)))
}));
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move {
host.serve(op_rx).await;
});
let (tx, rx) = oneshot::channel();
op_tx
.send(ControlOp::Spawn {
args: Box::new(SpawnArgs {
run_id: "np".to_string(),
..Default::default()
}),
reply: tx,
})
.unwrap();
let result = rx.await.unwrap();
drop(op_tx);
handle.await.unwrap();
assert_eq!(result, Ok("np".to_string()));
}
#[tokio::test]
async fn shutdown_op_stops_the_serve_loop() {
let mut host = host_with(vec![]);
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move { host.serve(op_rx).await });
let (tx, rx) = oneshot::channel();
op_tx.send(ControlOp::Shutdown { reply: tx }).unwrap();
assert!(rx.await.unwrap());
handle.await.unwrap();
}
#[tokio::test]
async fn flush_and_stop_delegates_to_the_world() {
let mut host = host_with(vec![]);
host.flush_and_stop().await;
host.flush_and_stop().await; }
#[tokio::test]
async fn serve_loop_services_subagent_ops_via_the_sender() {
let mut host = host_with(vec![]);
spawn(&mut host, "run-a", "run-a");
let sub_tx = host.subagent_sender();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let handle = tokio::spawn(async move { host.serve(op_rx).await });
let (tx, rx) = oneshot::channel();
sub_tx
.send(SubAgentOp::Check {
run_id: "run-a".to_string(),
reply: tx,
})
.unwrap();
assert!(rx.await.unwrap().is_some());
let (stx, srx) = oneshot::channel();
op_tx.send(ControlOp::Shutdown { reply: stx }).unwrap();
assert!(srx.await.unwrap());
handle.await.unwrap();
}
#[test]
fn status_str_covers_all_variants() {
assert_eq!(status_str(&AgentStatus::Idle), "idle");
assert_eq!(status_str(&AgentStatus::Active), "active");
assert_eq!(status_str(&AgentStatus::Paused), "paused");
assert_eq!(status_str(&AgentStatus::Waiting), "waiting");
assert_eq!(status_str(&AgentStatus::Complete), "complete");
assert_eq!(
status_str(&AgentStatus::Error {
message: "x".to_string()
}),
"error"
);
assert_eq!(status_str(&AgentStatus::Cancelled), "cancelled");
}
#[tokio::test]
async fn emit_events_broadcasts_agent_changes() {
let mut host = host_with(vec![text("done")]);
let mut rx = host.subscribe();
let entity = spawn(&mut host, "run-a", "agent-a");
host.world_mut()
.world_mut()
.entity_mut(entity)
.insert(RunMetadata {
run_id: "run-a".to_string(),
agent_name: "coder".to_string(),
agent_path: "/a".to_string(),
task: "t".to_string(),
model: None,
workdir: "/w".to_string(),
num_stages: 1,
started_at: 0,
parent_run_id: None,
metadata: std::collections::HashMap::new(),
callback_url: None,
callback_secret: None,
title: None,
unattended: false,
read_paths: None,
});
host.emit_events();
let first: Vec<WorldEvent> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
assert!(
first
.iter()
.any(|e| matches!(e, WorldEvent::Spawned { .. }))
);
assert!(first.iter().any(|e| matches!(e, WorldEvent::Status { .. })));
assert!(first.iter().any(|e| matches!(e, WorldEvent::Tokens { .. })));
assert!(
first
.iter()
.any(|e| matches!(e, WorldEvent::Context { .. }))
);
host.emit_events();
assert!(rx.try_recv().is_err());
host.world_mut().run_until_idle(20).await;
host.emit_events();
let done: Vec<WorldEvent> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
assert!(
done.iter()
.any(|e| matches!(e, WorldEvent::Completed { .. }))
);
host.emit_events();
assert!(
std::iter::from_fn(|| rx.try_recv().ok())
.collect::<Vec<_>>()
.is_empty()
);
}
#[tokio::test]
async fn emit_events_unloads_terminal_agents_when_safe() {
let mut host = host_with(vec![]);
let root = {
let mut s = agent_state("root");
s.status = AgentStatus::Complete;
host.world.world_mut().spawn(s).id()
};
host.register("root", root);
host.emit_events();
assert!(
host.live_entity("root").is_some(),
"not reaped on the first terminal pass (event must go out first)"
);
host.emit_events();
assert!(host.live_entity("root").is_none(), "reaped after emit");
assert!(
host.world.world().get::<AgentState>(root).is_none(),
"entity despawned"
);
let parent = host.world.world_mut().spawn(agent_state("parent")).id();
host.register("parent", parent);
let child = {
let mut s = agent_state("child");
s.status = AgentStatus::Complete;
host.world
.world_mut()
.spawn((
s,
ParentRef {
parent_entity: parent,
parent_agent_id: "parent".to_string(),
depth: 1,
},
))
.id()
};
host.register("child", child);
host.emit_events();
host.emit_events();
assert!(
host.live_entity("child").is_some(),
"not reaped while its parent is live"
);
host.world
.world_mut()
.get_mut::<AgentState>(parent)
.unwrap()
.status = AgentStatus::Complete;
host.emit_events();
host.emit_events();
assert!(
host.live_entity("child").is_none(),
"reaped once its parent is terminal"
);
let ghost = host.world.world_mut().spawn_empty().id();
host.world.world_mut().despawn(ghost);
let orphan = {
let mut s = agent_state("orphan");
s.status = AgentStatus::Complete;
host.world
.world_mut()
.spawn((
s,
ParentRef {
parent_entity: ghost,
parent_agent_id: "gone".to_string(),
depth: 1,
},
))
.id()
};
host.register("orphan", orphan);
host.emit_events();
host.emit_events();
assert!(
host.live_entity("orphan").is_none(),
"reaped: parent entity despawned"
);
}
#[tokio::test]
async fn emit_events_does_not_reap_non_terminal_agents() {
let mut host = host_with(vec![]);
let active = host.world.world_mut().spawn(agent_state("active")).id();
host.register("active", active);
host.emit_events();
host.emit_events();
assert!(host.live_entity("active").is_some());
}
#[tokio::test]
async fn reaper_runs_once_per_agent_before_despawn() {
use std::sync::atomic::{AtomicUsize, Ordering};
let mut host = host_with(vec![]);
static SEEN_LIVE: AtomicUsize = AtomicUsize::new(0);
SEEN_LIVE.store(0, Ordering::SeqCst);
host.set_reaper(Box::new(|world, entity| {
let live = world.world().get::<AgentState>(entity).is_some();
SEEN_LIVE.fetch_add(live as usize, Ordering::SeqCst);
}));
let root = {
let mut s = agent_state("root");
s.status = AgentStatus::Complete;
host.world.world_mut().spawn(s).id()
};
host.register("root", root);
host.emit_events(); assert_eq!(SEEN_LIVE.load(Ordering::SeqCst), 0);
host.emit_events(); assert!(host.live_entity("root").is_none(), "reaped after emit");
assert_eq!(
SEEN_LIVE.load(Ordering::SeqCst),
1,
"reaper ran exactly once, while the entity was still live"
);
}
fn unload_with(host: &mut WorldHost, run_id: &str, status: AgentStatus) {
let mut s = agent_state(run_id);
s.status = status;
let e = host.world.world_mut().spawn(s).id();
host.register(run_id, e);
host.emit_events();
host.emit_events();
}
#[tokio::test]
async fn an_unloaded_run_stays_in_the_listing_with_the_reason_it_ended() {
let mut host = host_with(vec![]);
let died = AgentStatus::Error {
message: "HTTP 402 Payment Required".to_string(),
};
unload_with(&mut host, "worker-1", died.clone());
assert!(host.live_entity("worker-1").is_none(), "unloaded");
let listing = ask(&mut host, |reply| ControlOp::List { reply }).await;
assert!(listing.runs.is_empty(), "nothing is running");
assert_eq!(listing.finished.len(), 1);
assert_eq!(listing.finished[0].run_id, "worker-1");
assert_eq!(listing.finished[0].status, died);
assert!(listing.finished[0].last_progress_at.is_some());
}
#[tokio::test]
async fn an_unloaded_run_leaves_the_listing_once_it_is_stale() {
let mut host = host_with(vec![]);
unload_with(&mut host, "worker-1", AgentStatus::Complete);
let window = DEFAULT_FINISHED_RETENTION_SECS as i64;
let at = host.finished.front().expect("just unloaded").0;
host.prune_finished(at + window);
assert_eq!(host.finished().len(), 1);
host.prune_finished(at + window + 1);
assert!(host.finished().is_empty());
}
#[tokio::test]
async fn a_zero_window_keeps_nothing() {
let mut host = host_with(vec![]);
host.set_finished_retention_secs(0);
unload_with(&mut host, "worker-1", AgentStatus::Complete);
assert!(host.live_entity("worker-1").is_none(), "still unloaded");
assert!(host.finished().is_empty());
}
#[tokio::test]
async fn a_run_is_listed_once_however_often_it_is_recorded() {
let mut host = host_with(vec![]);
let entry = |status| RunListEntry {
run_id: "worker-1".to_string(),
status,
wait_reason: None,
stage: "work".to_string(),
stage_index: None,
num_stages: None,
iteration: 0,
tool_calls: 0,
last_progress_at: None,
unattended: false,
empty_output: false,
read_paths: None,
};
host.record_finished(entry(AgentStatus::Cancelled), 100);
host.record_finished(entry(AgentStatus::Complete), 200);
let finished = host.finished();
assert_eq!(finished.len(), 1);
assert_eq!(finished[0].status, AgentStatus::Complete);
}
#[tokio::test]
async fn the_listing_of_finished_runs_is_capped() {
let mut host = host_with(vec![]);
for i in 0..=MAX_RETAINED_FINISHED {
host.record_finished(
RunListEntry {
run_id: format!("worker-{i}"),
status: AgentStatus::Complete,
wait_reason: None,
stage: "work".to_string(),
stage_index: None,
num_stages: None,
iteration: 0,
tool_calls: 0,
last_progress_at: None,
unattended: false,
empty_output: false,
read_paths: None,
},
100,
);
}
let finished = host.finished();
assert_eq!(finished.len(), MAX_RETAINED_FINISHED);
assert_eq!(
finished[0].run_id, "worker-1",
"the oldest is the one dropped"
);
}
#[tokio::test]
async fn the_status_of_an_unloaded_run_is_still_answerable() {
let mut host = host_with(vec![]);
unload_with(&mut host, "worker-1", AgentStatus::Complete);
let status = ask(&mut host, |reply| ControlOp::Status {
run_id: "worker-1".to_string(),
reply,
})
.await;
assert_eq!(status, Some(AgentStatus::Complete));
}
fn register_waiting(host: &mut WorldHost, run_id: &str) -> Entity {
let mut s = agent_state(run_id);
s.status = AgentStatus::Waiting;
let e = host.world.world_mut().spawn(s).id();
host.register(run_id, e);
e
}
#[tokio::test]
async fn emit_events_never_unloads_waiting_agents() {
use crate::components::AwaitingInteraction;
let mut host = host_with(vec![]);
let asking = register_waiting(&mut host, "asking");
host.world
.world_mut()
.entity_mut(asking)
.insert(AwaitingInteraction);
let gated = register_waiting(&mut host, "gated");
host.world
.world_mut()
.entity_mut(gated)
.insert(WaitingForChildren);
register_waiting(&mut host, "parked");
for _ in 0..5 {
host.emit_events();
}
for run_id in ["asking", "gated", "parked"] {
assert!(
host.live_entity(run_id).is_some(),
"a Waiting agent was unloaded and can no longer be resumed"
);
}
}
#[tokio::test]
async fn resolve_or_reload_pages_in_and_registers() {
let mut host = host_with(vec![]);
assert!(host.resolve_or_reload("ghost").is_none());
host.set_reloader(Box::new(|_world, _run_id| None));
assert!(host.resolve_or_reload("gone").is_none());
assert!(
host.live_entity("gone").is_none(),
"a declined reload registers nothing"
);
host.set_reloader(Box::new(|world, run_id| {
Some(world.spawn_agent((agent_state(run_id),)))
}));
let paged = host.resolve_or_reload("paged").expect("reloaded");
assert_eq!(
host.live_entity("paged"),
Some(paged),
"registered after reload"
);
assert_eq!(host.resolve_or_reload("paged"), Some(paged));
}
#[tokio::test]
async fn cancel_pages_in_an_unloaded_run() {
let mut host = host_with(vec![]);
host.set_reloader(paging_reloader());
let cancelled = ask(&mut host, |reply| ControlOp::Cancel {
run_id: "unloaded".to_string(),
reply,
})
.await;
assert!(cancelled, "reloaded then cancelled");
assert_eq!(
host.world
.agent_status(host.live_entity("unloaded").unwrap()),
Some(AgentStatus::Cancelled)
);
}
#[tokio::test]
async fn emit_events_broadcasts_new_interactions_once() {
let mut host = host_with(vec![]);
let mut rx = host.subscribe();
let backend = host.interactions().backend_for("agent-a");
let asking = tokio::spawn(async move {
backend
.ask(leviath_core::interaction::InteractionRequest::free_text(
"q1", "p", "s", true,
))
.await
});
for _ in 0..8 {
tokio::task::yield_now().await;
}
host.emit_events();
let evs: Vec<WorldEvent> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
assert!(
evs.iter()
.any(|e| matches!(e, WorldEvent::Interaction { .. }))
);
host.emit_events();
assert!(rx.try_recv().is_err());
assert!(
host.interactions()
.answer(leviath_core::interaction::InteractionResponse::text(
"q1", "ok"
))
);
let _ = asking.await;
}
#[tokio::test]
async fn event_sender_feeds_subscribers() {
let host = host_with(vec![]);
let mut rx = host.subscribe();
let event = WorldEvent::Completed {
run_id: "r".to_string(),
agent_id: "a".to_string(),
status: "complete".to_string(),
};
host.event_sender().send(event.clone()).unwrap();
assert_eq!(rx.try_recv().unwrap(), event);
}
#[tokio::test]
async fn emit_events_skips_despawned_agents() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "agent-a");
host.world_mut().world_mut().despawn(e);
host.emit_events();
}
#[tokio::test]
async fn serve_returns_when_control_channel_closes() {
let mut host = host_with(vec![text("done")]);
let (op_tx, op_rx) = mpsc::unbounded_channel();
drop(op_tx); host.serve(op_rx).await; }
#[tokio::test]
async fn mock_helpers_are_exercised() {
let p = Script {
responses: Mutex::new(std::collections::VecDeque::new()),
};
assert_eq!(p.name(), "script");
assert_eq!(p.count_tokens("t", "m").await, 1);
assert_eq!(p.max_context_tokens("m"), 100_000);
let _ = p.capabilities("m");
let req = InferenceRequest {
system: vec![],
messages: vec![],
model: "m".to_string(),
max_tokens: 1,
temperature: 0.0,
tools: vec![],
extra: serde_json::Value::Null,
request_timeout_secs: None,
};
assert!(p.infer(req).await.is_err());
let exec = NoTools.exec_for(
Entity::from_raw_u32(1).expect("a small literal index is always a valid entity id"),
vec![leviath_providers::ToolCall {
id: "c".to_string(),
name: "n".to_string(),
arguments: serde_json::Value::Null,
thought_signature: None,
}],
crate::pipeline::noop_progress(),
);
assert_eq!(exec().await, vec![("c".to_string(), String::new())]);
}
#[tokio::test]
async fn list_skips_despawned_entity() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "agent-a");
host.world_mut().world_mut().despawn(e);
let list = ask(&mut host, |reply| ControlOp::List { reply }).await.runs;
assert!(list.is_empty()); let status = ask(&mut host, |reply| ControlOp::Status {
run_id: "run-a".to_string(),
reply,
})
.await;
assert_eq!(status, None);
}
fn waiting_because(
host: &mut WorldHost,
entity: Entity,
attach: impl FnOnce(&mut bevy_ecs::world::EntityWorldMut),
) -> Option<WaitReason> {
{
let world = host.world_mut().world_mut();
world
.get_mut::<AgentState>(entity)
.expect("spawned agent has state")
.status = AgentStatus::Waiting;
let mut e = world.entity_mut(entity);
attach(&mut e);
}
host.wait_reason(entity)
}
#[tokio::test]
async fn wait_reason_is_none_unless_the_agent_is_waiting() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
host.world_mut()
.world_mut()
.entity_mut(e)
.insert(crate::pipeline::WaitingForChildren);
assert_eq!(host.wait_reason(e), None);
}
#[tokio::test]
async fn wait_reason_is_none_for_an_unknown_entity() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
host.world_mut().world_mut().despawn(e);
assert_eq!(host.wait_reason(e), None);
}
#[tokio::test]
async fn wait_reason_is_none_when_nothing_claims_the_wait() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
assert_eq!(waiting_because(&mut host, e, |_| {}), None);
}
#[tokio::test]
async fn wait_reason_reports_a_taint_gate() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let reason = waiting_because(&mut host, e, |entity| {
entity.insert(crate::gate_prompt::AwaitingGatePrompt(1));
});
assert_eq!(reason, Some(WaitReason::TaintGate));
}
#[tokio::test]
async fn wait_reason_reports_an_interaction_point() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let reason = waiting_because(&mut host, e, |entity| {
entity.insert(crate::interaction_points::AwaitingInteractionPoint);
});
assert_eq!(reason, Some(WaitReason::InteractionPoint));
}
#[tokio::test]
async fn wait_reason_counts_unfinished_children() {
let mut host = host_with(vec![]);
let parent = spawn(&mut host, "run-a", "run-a");
let running = spawn(&mut host, "run-b", "run-b");
let done = spawn(&mut host, "run-c", "run-c");
{
let world = host.world_mut().world_mut();
world
.get_mut::<AgentState>(done)
.expect("child has state")
.status = AgentStatus::Complete;
}
let reason = waiting_because(&mut host, parent, |entity| {
entity.insert((
crate::pipeline::WaitingForChildren,
SubAgentChildren {
children: vec![running, done],
max_child_depth: 3,
},
));
});
assert_eq!(reason, Some(WaitReason::Children { outstanding: 1 }));
}
#[tokio::test]
async fn wait_reason_reports_children_with_none_recorded() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let reason = waiting_because(&mut host, e, |entity| {
entity.insert(crate::pipeline::WaitingForChildren);
});
assert_eq!(reason, Some(WaitReason::Children { outstanding: 0 }));
}
fn open_prompt(
host: &WorldHost,
agent_id: &str,
request: InteractionRequest,
) -> tokio::task::JoinHandle<InteractionResponse> {
let backend = host.interactions().backend_for(agent_id.to_string());
tokio::spawn(async move {
use crate::dynamic_interaction::InteractionBackend;
backend.ask(request).await
})
}
async fn await_pending(host: &WorldHost, agent_id: &str) {
for _ in 0..8 {
tokio::task::yield_now().await;
}
assert!(
host.interactions()
.pending()
.iter()
.any(|(id, _)| id == agent_id),
"the hub registered a request for {agent_id}"
);
}
#[tokio::test]
async fn wait_reason_distinguishes_a_tool_approval_from_a_question() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let approval = open_prompt(
&host,
"run-a",
InteractionRequest::tool_approval("req-1", "shell", serde_json::json!({}), "implement"),
);
await_pending(&host, "run-a").await;
let reason = waiting_because(&mut host, e, |entity| {
entity.insert(AwaitingInteraction);
});
assert_eq!(reason, Some(WaitReason::ToolApproval));
assert_eq!(host.interactions().cancel_for_agent("run-a"), 1);
approval.await.expect("the asking task finishes");
let question = open_prompt(
&host,
"run-a",
InteractionRequest::free_text("req-2", "which one?", "implement", true),
);
await_pending(&host, "run-a").await;
assert_eq!(host.wait_reason(e), Some(WaitReason::UserPrompt));
assert_eq!(host.interactions().cancel_for_agent("run-a"), 1);
question.await.expect("the asking task finishes");
}
#[tokio::test]
async fn wait_reason_falls_back_to_user_prompt_without_a_hub_entry() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let reason = waiting_because(&mut host, e, |entity| {
entity.insert(AwaitingInteraction);
});
assert_eq!(reason, Some(WaitReason::UserPrompt));
}
#[tokio::test]
async fn a_gate_outranks_the_generic_interaction_marker() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
let reason = waiting_because(&mut host, e, |entity| {
entity.insert((
AwaitingInteraction,
crate::gate_prompt::AwaitingGatePrompt(1),
));
});
assert_eq!(reason, Some(WaitReason::TaintGate));
}
#[tokio::test]
async fn wait_reason_counts_outstanding_fan_out_workers() {
let mut host = host_with(vec![]);
let parent = spawn(&mut host, "run-a", "run-a");
let worker = spawn(&mut host, "run-b", "run-b");
{
let world = host.world_mut().world_mut();
world
.get_mut::<AgentState>(parent)
.expect("parent has state")
.status = AgentStatus::Waiting;
crate::fanout::restore_fan_out_waiting(
world,
parent,
crate::fanout::FanOutState {
config: leviath_core::blueprint::FanOutConfig {
worker_agent: None,
worker_stage: Some("work".to_string()),
worker_query: None,
merge_stage: None,
max_workers: 2,
on_worker_failure: Default::default(),
split_prompt: String::new(),
},
max_workers: 2,
pending: vec![
crate::fanout::WorkItem::default(),
crate::fanout::WorkItem::default(),
],
active: vec![("item-1".to_string(), "run-b".to_string())],
summaries: Vec::new(),
failures: Vec::new(),
},
&|run_id| (run_id == "run-b").then_some(worker),
);
}
assert_eq!(
host.wait_reason(parent),
Some(WaitReason::FanOutWorkers { outstanding: 3 })
);
}
#[tokio::test]
async fn list_reports_blueprint_shape_and_unattended() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
host.world_mut().world_mut().entity_mut(e).insert((
RunMetadata {
run_id: "run-a".to_string(),
agent_name: "coder".to_string(),
agent_path: "/tmp/agent".to_string(),
task: "t".to_string(),
model: None,
workdir: "/tmp".to_string(),
num_stages: 3,
started_at: 0,
parent_run_id: None,
metadata: HashMap::new(),
callback_url: None,
callback_secret: None,
title: None,
unattended: true,
read_paths: None,
},
TokenTotals {
tool_calls: 9,
..Default::default()
},
{
let mut watermark = crate::pipeline::PersistWatermark::default();
watermark.backdate(1_700);
watermark
},
));
let list = ask(&mut host, |reply| ControlOp::List { reply }).await.runs;
assert_eq!(list[0].num_stages, Some(3));
assert_eq!(list[0].tool_calls, 9);
assert!(list[0].unattended);
assert_eq!(list[0].last_progress_at, Some(1_700));
assert!(!list[0].empty_output);
}
#[tokio::test]
async fn list_reports_a_finished_run_that_produced_nothing() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
host.world_mut()
.world_mut()
.entity_mut(e)
.insert(crate::persistence::RunOutcomeFlags::default());
assert!(!ask(&mut host, |reply| ControlOp::List { reply }).await.runs[0].empty_output);
host.world_mut()
.world_mut()
.get_mut::<AgentState>(e)
.expect("spawned agent has state")
.status = AgentStatus::Complete;
assert!(ask(&mut host, |reply| ControlOp::List { reply }).await.runs[0].empty_output);
host.world_mut()
.world_mut()
.get_mut::<crate::persistence::RunOutcomeFlags>(e)
.expect("just inserted")
.0
.no_output_tools = true;
assert!(!ask(&mut host, |reply| ControlOp::List { reply }).await.runs[0].empty_output);
}
#[tokio::test]
async fn list_explains_a_waiting_run() {
let mut host = host_with(vec![]);
let e = spawn(&mut host, "run-a", "run-a");
waiting_because(&mut host, e, |entity| {
entity.insert(crate::pipeline::WaitingForChildren);
});
let list = ask(&mut host, |reply| ControlOp::List { reply }).await.runs;
assert_eq!(list.len(), 1);
assert_eq!(
list[0].wait_reason,
Some(WaitReason::Children { outstanding: 0 })
);
assert_eq!(list[0].stage_index, Some(0));
assert_eq!(list[0].num_stages, None);
assert!(!list[0].unattended);
}
#[test]
fn every_world_event_variant_carries_its_run_id() {
let rid = "run-x".to_string();
let aid = "agent-x".to_string();
let events = vec![
WorldEvent::Spawned {
run_id: rid.clone(),
agent_id: aid.clone(),
blueprint: "b".to_string(),
},
WorldEvent::Status {
run_id: rid.clone(),
agent_id: aid.clone(),
status: "active".to_string(),
stage: "s".to_string(),
iteration: 1,
tool_calls: 0,
accepts_messages: false,
},
WorldEvent::Tokens {
run_id: rid.clone(),
agent_id: aid.clone(),
prompt_tokens: 1,
completion_tokens: 2,
cached_tokens: 0,
cache_write_tokens: 0,
},
WorldEvent::Context {
run_id: rid.clone(),
agent_id: aid.clone(),
total_tokens: 3,
max_tokens: 4,
},
WorldEvent::Interaction {
run_id: rid.clone(),
agent_id: aid.clone(),
request: InteractionRequest::free_text("i", "p", "s", true),
},
WorldEvent::Completed {
run_id: rid.clone(),
agent_id: aid.clone(),
status: "complete".to_string(),
},
WorldEvent::StageTransition {
run_id: rid.clone(),
agent_id: aid.clone(),
from: "a".to_string(),
to: "b".to_string(),
iteration: 1,
},
WorldEvent::ToolCallStarted {
run_id: rid.clone(),
agent_id: aid.clone(),
call_id: "c".to_string(),
tool: "t".to_string(),
},
WorldEvent::ToolCallFinished {
run_id: rid.clone(),
agent_id: aid.clone(),
call_id: "c".to_string(),
tool: "t".to_string(),
ok: true,
summary: "s".to_string(),
},
WorldEvent::Log {
run_id: rid.clone(),
agent_id: aid.clone(),
line: "l".to_string(),
},
];
for ev in events {
assert_eq!(ev.run_id(), "run-x");
}
}
}