use std::collections::HashMap;
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::Arc;
#[cfg(any(feature = "native", feature = "os-patterns"))]
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use tracing::{debug, info, warn};
use crate::process::CancellationToken;
use clawft_platform::Platform;
use crate::capability::AgentCapabilities;
use crate::error::{KernelError, KernelResult};
use crate::ipc::KernelIpc;
use crate::process::{Pid, ProcessEntry, ProcessState, ProcessTable, ResourceUsage};
#[non_exhaustive]
#[cfg(feature = "os-patterns")]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum RestartStrategy {
#[default]
OneForOne,
OneForAll,
RestForOne,
Permanent,
Transient,
}
#[cfg(feature = "os-patterns")]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RestartBudget {
pub max_restarts: u32,
pub within_secs: u64,
}
#[cfg(feature = "os-patterns")]
impl Default for RestartBudget {
fn default() -> Self {
Self {
max_restarts: 5,
within_secs: 60,
}
}
}
#[cfg(feature = "os-patterns")]
pub struct RestartTracker {
pub restart_count: u32,
pub window_start: std::time::Instant,
pub last_restart: Option<std::time::Instant>,
pub backoff_ms: u64,
}
#[cfg(feature = "os-patterns")]
impl RestartTracker {
pub fn new() -> Self {
Self {
restart_count: 0,
window_start: std::time::Instant::now(),
last_restart: None,
backoff_ms: 0,
}
}
pub fn next_backoff_ms(&self) -> u64 {
let base: u64 = 100;
let exponent = self.restart_count.saturating_sub(1);
let delay = base.saturating_mul(1u64 << exponent.min(20));
delay.min(30_000)
}
pub fn next_backoff_ms_with_model(
&self,
model: Option<&crate::eml_kernel::RestartStrategyModel>,
failure_type: u32,
uptime_secs: f64,
system_load: f64,
) -> u64 {
match model {
Some(m) => {
let (delay, _should_retry) =
m.predict(self.restart_count, failure_type, uptime_secs, system_load);
delay
}
None => self.next_backoff_ms(),
}
}
pub fn check_window(&mut self, budget: &RestartBudget) {
let now = std::time::Instant::now();
if now.duration_since(self.window_start).as_secs() > budget.within_secs {
self.restart_count = 0;
self.window_start = now;
}
}
pub fn record_restart(&mut self, budget: &RestartBudget) -> bool {
self.check_window(budget);
self.restart_count += 1;
self.backoff_ms = self.next_backoff_ms();
self.last_restart = Some(std::time::Instant::now());
self.restart_count <= budget.max_restarts
}
pub fn record_restart_with_model(
&mut self,
budget: &RestartBudget,
model: Option<&crate::eml_kernel::RestartStrategyModel>,
failure_type: u32,
uptime_secs: f64,
system_load: f64,
) -> bool {
self.check_window(budget);
self.restart_count += 1;
self.backoff_ms =
self.next_backoff_ms_with_model(model, failure_type, uptime_secs, system_load);
self.last_restart = Some(std::time::Instant::now());
self.restart_count <= budget.max_restarts
}
pub fn is_exhausted(&self, budget: &RestartBudget) -> bool {
let now = std::time::Instant::now();
if now.duration_since(self.window_start).as_secs() > budget.within_secs {
return false;
}
self.restart_count >= budget.max_restarts
}
pub fn remaining(&self, budget: &RestartBudget) -> u32 {
let now = std::time::Instant::now();
if now.duration_since(self.window_start).as_secs() > budget.within_secs {
return budget.max_restarts;
}
budget.max_restarts.saturating_sub(self.restart_count)
}
pub fn should_restart(strategy: &RestartStrategy, exit_code: i32) -> bool {
match strategy {
RestartStrategy::Permanent => false,
RestartStrategy::Transient => exit_code != 0,
_ => true,
}
}
}
#[cfg(feature = "os-patterns")]
impl Default for RestartTracker {
fn default() -> Self {
Self::new()
}
}
#[non_exhaustive]
#[cfg(feature = "os-patterns")]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ResourceCheckResult {
Ok,
Warning {
resource: String,
current: u64,
limit: u64,
},
Exceeded {
resource: String,
current: u64,
limit: u64,
},
}
#[cfg(feature = "os-patterns")]
pub fn check_resource_usage(
usage: &ResourceUsage,
limits: &crate::capability::ResourceLimits,
) -> Vec<ResourceCheckResult> {
let mut results = Vec::new();
let checks: &[(&str, u64, u64)] = &[
("memory", usage.memory_bytes, limits.max_memory_bytes),
("cpu_time", usage.cpu_time_ms, limits.max_cpu_time_ms),
("messages", usage.messages_sent, limits.max_messages),
("tool_calls", usage.tool_calls, limits.max_tool_calls),
];
for &(name, current, limit) in checks {
if limit == 0 {
continue; }
let ratio = current as f64 / limit as f64;
if ratio >= 1.0 {
results.push(ResourceCheckResult::Exceeded {
resource: name.to_owned(),
current,
limit,
});
} else if ratio >= 0.8 {
results.push(ResourceCheckResult::Warning {
resource: name.to_owned(),
current,
limit,
});
}
}
results
}
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SpawnBackend {
Native,
Wasm {
module: PathBuf,
},
Container {
image: String,
},
Tee {
enclave: EnclaveConfig,
},
Remote {
node_id: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EnclaveConfig {
pub enclave_type: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SpawnRequest {
pub agent_id: String,
#[serde(default)]
pub capabilities: Option<AgentCapabilities>,
#[serde(default)]
pub parent_pid: Option<Pid>,
#[serde(default)]
pub env: HashMap<String, String>,
#[serde(default)]
pub backend: Option<SpawnBackend>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SpawnResult {
pub pid: Pid,
pub agent_id: String,
}
pub struct AgentSupervisor<P: Platform> {
process_table: Arc<ProcessTable>,
kernel_ipc: Arc<KernelIpc>,
default_capabilities: AgentCapabilities,
#[cfg(feature = "native")]
running_agents: Arc<DashMap<Pid, tokio::task::JoinHandle<()>>>,
#[cfg(feature = "native")]
a2a_router: Option<Arc<crate::a2a::A2ARouter>>,
#[cfg(feature = "native")]
cron_service: Option<Arc<crate::cron::CronService>>,
#[cfg(feature = "exochain")]
tree_manager: Option<Arc<crate::tree_manager::TreeManager>>,
#[cfg(feature = "exochain")]
chain_manager: Option<Arc<crate::chain::ChainManager>>,
#[cfg(feature = "os-patterns")]
monitor_registry: Arc<crate::monitor::MonitorRegistry>,
#[cfg(feature = "os-patterns")]
restart_strategy: RestartStrategy,
#[cfg(feature = "os-patterns")]
restart_budget: RestartBudget,
#[cfg(feature = "os-patterns")]
restart_trackers: Arc<DashMap<Pid, RestartTracker>>,
_platform: PhantomData<P>,
}
impl<P: Platform> AgentSupervisor<P> {
pub fn new(
process_table: Arc<ProcessTable>,
kernel_ipc: Arc<KernelIpc>,
default_capabilities: AgentCapabilities,
) -> Self {
Self {
process_table,
kernel_ipc,
default_capabilities,
#[cfg(feature = "native")]
running_agents: Arc::new(DashMap::new()),
#[cfg(feature = "native")]
a2a_router: None,
#[cfg(feature = "native")]
cron_service: None,
#[cfg(feature = "exochain")]
tree_manager: None,
#[cfg(feature = "exochain")]
chain_manager: None,
#[cfg(feature = "os-patterns")]
monitor_registry: Arc::new(crate::monitor::MonitorRegistry::new()),
#[cfg(feature = "os-patterns")]
restart_strategy: RestartStrategy::default(),
#[cfg(feature = "os-patterns")]
restart_budget: RestartBudget::default(),
#[cfg(feature = "os-patterns")]
restart_trackers: Arc::new(DashMap::new()),
_platform: PhantomData,
}
}
#[cfg(feature = "native")]
pub fn with_a2a_router(
mut self,
a2a_router: Arc<crate::a2a::A2ARouter>,
cron_service: Arc<crate::cron::CronService>,
) -> Self {
self.a2a_router = Some(a2a_router);
self.cron_service = Some(cron_service);
self
}
#[cfg(feature = "native")]
pub fn a2a_router(&self) -> Option<&Arc<crate::a2a::A2ARouter>> {
self.a2a_router.as_ref()
}
#[cfg(feature = "native")]
pub fn cron_service(&self) -> Option<&Arc<crate::cron::CronService>> {
self.cron_service.as_ref()
}
#[cfg(feature = "exochain")]
pub fn with_exochain(
mut self,
tree_manager: Option<Arc<crate::tree_manager::TreeManager>>,
chain_manager: Option<Arc<crate::chain::ChainManager>>,
) -> Self {
self.tree_manager = tree_manager;
self.chain_manager = chain_manager;
self
}
#[cfg(feature = "os-patterns")]
pub fn with_restart_config(mut self, strategy: RestartStrategy, budget: RestartBudget) -> Self {
self.restart_strategy = strategy;
self.restart_budget = budget;
self
}
#[cfg(feature = "os-patterns")]
pub fn monitor_registry(&self) -> &Arc<crate::monitor::MonitorRegistry> {
&self.monitor_registry
}
#[cfg(feature = "os-patterns")]
pub fn restart_strategy(&self) -> &RestartStrategy {
&self.restart_strategy
}
#[cfg(feature = "os-patterns")]
pub fn restart_budget(&self) -> &RestartBudget {
&self.restart_budget
}
#[cfg(feature = "os-patterns")]
pub fn handle_exit(&self, pid: Pid, exit_code: i32) -> Vec<(Pid, SpawnResult)> {
use crate::monitor::ExitReason;
let reason = if exit_code == 0 {
ExitReason::Normal
} else {
ExitReason::Crash(format!("exit code {exit_code}"))
};
let (_link_signals, _down_signals) = self.monitor_registry.process_exited(pid, &reason);
let mut restarts = Vec::new();
if !RestartTracker::should_restart(&self.restart_strategy, exit_code) {
debug!(pid, exit_code, strategy = ?self.restart_strategy, "not restarting per strategy");
return restarts;
}
let mut tracker = self.restart_trackers.entry(pid).or_default();
let within_budget = tracker.record_restart(&self.restart_budget);
let backoff_ms = tracker.backoff_ms;
drop(tracker);
if !within_budget {
warn!(pid, "restart budget exhausted for pid, escalating");
return restarts;
}
info!(
pid,
backoff_ms,
strategy = ?self.restart_strategy,
"scheduling restart after backoff"
);
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"supervisor",
"agent.self_heal_restart",
Some(serde_json::json!({
"pid": pid,
"exit_code": exit_code,
"strategy": format!("{:?}", self.restart_strategy),
"backoff_ms": backoff_ms,
})),
);
}
match self.restart_strategy {
RestartStrategy::OneForOne | RestartStrategy::Transient => {
if let Ok(result) = self.restart(pid) {
restarts.push((pid, result));
}
}
RestartStrategy::OneForAll => {
let siblings: Vec<ProcessEntry> = self
.process_table
.list()
.into_iter()
.filter(|e| e.pid != 0 && e.pid != pid && e.state == ProcessState::Running)
.collect();
for sibling in &siblings {
if let Ok(result) = self.restart(sibling.pid) {
restarts.push((sibling.pid, result));
}
}
if let Ok(result) = self.restart(pid) {
restarts.push((pid, result));
}
}
RestartStrategy::RestForOne => {
let mut all: Vec<ProcessEntry> = self
.process_table
.list()
.into_iter()
.filter(|e| e.pid != 0 && e.state == ProcessState::Running)
.collect();
all.sort_by_key(|e| e.pid);
let after: Vec<Pid> = all.iter().filter(|e| e.pid > pid).map(|e| e.pid).collect();
for sibling_pid in &after {
if let Ok(result) = self.restart(*sibling_pid) {
restarts.push((*sibling_pid, result));
}
}
if let Ok(result) = self.restart(pid) {
restarts.push((pid, result));
}
}
RestartStrategy::Permanent => {
}
}
restarts
}
pub fn spawn(&self, request: SpawnRequest) -> KernelResult<SpawnResult> {
match &request.backend {
None | Some(SpawnBackend::Native) => { }
Some(SpawnBackend::Wasm { .. }) => {
return Err(KernelError::BackendNotAvailable {
backend: "wasm".into(),
reason: "WASM sandbox requires K3 (Wasmtime integration)".into(),
});
}
Some(SpawnBackend::Container { .. }) => {
return Err(KernelError::BackendNotAvailable {
backend: "container".into(),
reason: "agents do not run in containers yet; cog workloads run in \
containers through workload_runtime::WorkloadHost (ADR-099 s5)"
.into(),
});
}
Some(SpawnBackend::Tee { .. }) => {
return Err(KernelError::BackendNotAvailable {
backend: "tee".into(),
reason: "TEE runtime requires K6+ and hardware support".into(),
});
}
Some(SpawnBackend::Remote { .. }) => {
return Err(KernelError::BackendNotAvailable {
backend: "remote".into(),
reason: "remote agent delegation requires K6; remote workloads run \
through workload_runtime adapters (remote.api, ADR-099 s5)"
.into(),
});
}
}
let caps = request
.capabilities
.unwrap_or_else(|| self.default_capabilities.clone());
info!(
agent_id = %request.agent_id,
parent_pid = ?request.parent_pid,
"spawning supervised agent"
);
let entry = ProcessEntry {
pid: 0, agent_id: request.agent_id.clone(),
state: ProcessState::Starting,
capabilities: caps,
resource_usage: ResourceUsage::default(),
cancel_token: CancellationToken::new(),
parent_pid: request.parent_pid,
};
let pid = self.process_table.insert(entry)?;
debug!(pid, agent_id = %request.agent_id, "agent spawned");
Ok(SpawnResult {
pid,
agent_id: request.agent_id,
})
}
#[cfg(feature = "native")]
pub fn spawn_and_run<F, Fut>(&self, request: SpawnRequest, work: F) -> KernelResult<SpawnResult>
where
F: FnOnce(Pid, CancellationToken) -> Fut,
Fut: std::future::Future<Output = i32> + Send + 'static,
{
#[cfg(feature = "exochain")]
let parent_pid = request.parent_pid;
let result = self.spawn(request)?;
let pid = result.pid;
let entry = self
.process_table
.get(pid)
.ok_or(KernelError::ProcessNotFound { pid })?;
let cancel_token = entry.cancel_token.clone();
#[cfg(feature = "exochain")]
if let Some(ref tm) = self.tree_manager
&& let Err(e) = tm.register_agent(&result.agent_id, pid, &entry.capabilities)
{
warn!(error = %e, pid, "failed to register agent in resource tree");
}
let _ = self.process_table.update_state(pid, ProcessState::Running);
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"supervisor",
"agent.spawn",
Some(serde_json::json!({
"agent_id": result.agent_id,
"pid": pid,
"parent_pid": parent_pid,
})),
);
}
let process_table = Arc::clone(&self.process_table);
let running_agents = Arc::clone(&self.running_agents);
let agent_id = result.agent_id.clone();
#[cfg(feature = "exochain")]
let tree_manager = self.tree_manager.clone();
#[cfg(feature = "exochain")]
let chain_manager = self.chain_manager.clone();
let future = work(pid, cancel_token);
let handle = tokio::spawn(async move {
let exit_code = future.await;
let _ = process_table.update_state(pid, ProcessState::Exited(exit_code));
#[cfg(feature = "exochain")]
if let Some(ref tm) = tree_manager {
let agent_path = format!("/kernel/agents/{agent_id}");
let rid = exo_resource_tree::ResourceId::new(&agent_path);
let success = exit_code == 0;
let observation = exo_resource_tree::NodeScoring {
trust: if success { 0.8 } else { 0.2 },
performance: if success { 0.7 } else { 0.3 },
difficulty: 0.5,
reward: if success { 0.6 } else { 0.1 },
reliability: if success { 0.9 } else { 0.1 },
velocity: 0.5,
};
if let Err(e) = tm.blend_scoring(&rid, &observation, 0.3) {
debug!(error = %e, pid, "scoring blend skipped (node may be unregistered)");
}
}
#[cfg(feature = "exochain")]
if let Some(ref tm) = tree_manager
&& let Err(e) = tm.unregister_agent(&agent_id, pid, exit_code)
{
tracing::warn!(error = %e, pid, "failed to unregister agent from tree");
}
#[cfg(feature = "exochain")]
if let Some(ref cm) = chain_manager {
cm.append(
"supervisor",
"agent.exit",
Some(serde_json::json!({
"agent_id": agent_id,
"pid": pid,
"exit_code": exit_code,
})),
);
}
running_agents.remove(&pid);
info!(pid, exit_code, agent_id = %agent_id, "agent task completed");
});
self.running_agents.insert(pid, handle);
info!(pid, agent_id = %result.agent_id, "agent spawned and running");
Ok(result)
}
pub fn stop(&self, pid: Pid, graceful: bool) -> KernelResult<()> {
let entry = self
.process_table
.get(pid)
.ok_or(KernelError::ProcessNotFound { pid })?;
if matches!(entry.state, ProcessState::Exited(_)) {
warn!(pid, "stop called on already-exited process");
return Ok(());
}
if graceful {
info!(pid, "gracefully stopping agent");
let _ = self.process_table.update_state(pid, ProcessState::Stopping);
entry.cancel_token.cancel();
} else {
info!(pid, "force stopping agent");
entry.cancel_token.cancel();
let _ = self
.process_table
.update_state(pid, ProcessState::Exited(-1));
#[cfg(feature = "native")]
if let Some((_, handle)) = self.running_agents.remove(&pid) {
handle.abort();
}
#[cfg(feature = "exochain")]
{
if let Some(ref tm) = self.tree_manager {
let _ = tm.unregister_agent(&entry.agent_id, pid, -1);
}
if let Some(ref cm) = self.chain_manager {
cm.append(
"supervisor",
"agent.force_stop",
Some(serde_json::json!({
"agent_id": entry.agent_id,
"pid": pid,
})),
);
}
}
}
Ok(())
}
pub fn restart(&self, pid: Pid) -> KernelResult<SpawnResult> {
let old_entry = self
.process_table
.get(pid)
.ok_or(KernelError::ProcessNotFound { pid })?;
info!(pid, agent_id = %old_entry.agent_id, "restarting agent");
self.stop(pid, true)?;
if !matches!(old_entry.state, ProcessState::Exited(_)) {
let _ = self
.process_table
.update_state(pid, ProcessState::Exited(0));
}
let request = SpawnRequest {
agent_id: old_entry.agent_id.clone(),
capabilities: Some(old_entry.capabilities.clone()),
parent_pid: Some(pid),
env: HashMap::new(),
backend: None, };
let result = self.spawn(request)?;
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"supervisor",
"agent.restart",
Some(serde_json::json!({
"agent_id": result.agent_id,
"old_pid": pid,
"new_pid": result.pid,
})),
);
}
Ok(result)
}
pub fn inspect(&self, pid: Pid) -> KernelResult<ProcessEntry> {
self.process_table
.get(pid)
.ok_or(KernelError::ProcessNotFound { pid })
}
pub fn list_by_state(&self, state: ProcessState) -> Vec<ProcessEntry> {
self.process_table
.list()
.into_iter()
.filter(|e| e.state == state)
.collect()
}
pub fn list_agents(&self) -> Vec<ProcessEntry> {
self.process_table
.list()
.into_iter()
.filter(|e| e.pid != 0)
.collect()
}
pub fn process_table(&self) -> &Arc<ProcessTable> {
&self.process_table
}
pub fn ipc(&self) -> &Arc<KernelIpc> {
&self.kernel_ipc
}
pub fn default_capabilities(&self) -> &AgentCapabilities {
&self.default_capabilities
}
pub fn running_count(&self) -> usize {
self.process_table
.list()
.iter()
.filter(|e| e.pid != 0 && e.state == ProcessState::Running)
.count()
}
#[cfg(feature = "native")]
pub fn running_task_count(&self) -> usize {
self.running_agents.len()
}
#[cfg(feature = "native")]
pub fn abort_all(&self) {
for entry in self.running_agents.iter() {
entry.value().abort();
}
self.running_agents.clear();
}
#[cfg(feature = "native")]
pub async fn watchdog_sweep(&self) -> Vec<(Pid, i32)> {
let mut reaped = Vec::new();
let finished_pids: Vec<Pid> = self
.running_agents
.iter()
.filter(|entry| entry.value().is_finished())
.map(|entry| *entry.key())
.collect();
for pid in finished_pids {
if let Some((_, handle)) = self.running_agents.remove(&pid) {
let exit_code = match handle.await {
Ok(()) => -2, Err(e) if e.is_panic() => -3, Err(_) => -2, };
if let Some(entry) = self.process_table.get(pid)
&& entry.state == ProcessState::Running
{
let _ = self
.process_table
.update_state(pid, ProcessState::Exited(exit_code));
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain_manager {
cm.append(
"watchdog",
"agent.watchdog_reap",
Some(serde_json::json!({
"pid": pid,
"exit_code": exit_code,
"agent_id": entry.agent_id,
})),
);
}
reaped.push((pid, exit_code));
info!(pid, exit_code, agent_id = %entry.agent_id, "watchdog reaped stale agent");
}
}
}
reaped
}
#[cfg(feature = "native")]
pub async fn shutdown_all(&self, timeout: std::time::Duration) -> Vec<(Pid, i32)> {
for entry in self.process_table.list() {
if entry.pid == 0 {
continue; }
entry.cancel_token.cancel();
}
let handles: Vec<(Pid, tokio::task::JoinHandle<()>)> = {
let pids: Vec<Pid> = self.running_agents.iter().map(|e| *e.key()).collect();
let mut collected = Vec::with_capacity(pids.len());
for pid in pids {
if let Some((pid, handle)) = self.running_agents.remove(&pid) {
collected.push((pid, handle));
}
}
collected
};
if handles.is_empty() {
return Vec::new();
}
let process_table = &self.process_table;
let mut results = Vec::with_capacity(handles.len());
match tokio::time::timeout(
timeout,
futures::future::join_all(
handles
.into_iter()
.map(|(pid, handle)| async move { (pid, handle.await) }),
),
)
.await
{
Ok(join_results) => {
for (pid, join_result) in join_results {
let exit_code = match join_result {
Ok(()) => process_table
.get(pid)
.and_then(|e| match e.state {
ProcessState::Exited(code) => Some(code),
_ => None,
})
.unwrap_or(0),
Err(e) if e.is_panic() => -3,
Err(_) => -1,
};
results.push((pid, exit_code));
}
}
Err(_elapsed) => {
info!("shutdown timeout reached, aborting remaining agents");
let remaining: Vec<Pid> = self.running_agents.iter().map(|e| *e.key()).collect();
for pid in remaining {
if let Some((pid, handle)) = self.running_agents.remove(&pid) {
handle.abort();
let _ = process_table.update_state(pid, ProcessState::Exited(-1));
results.push((pid, -1));
}
}
if results.is_empty() {
for entry in process_table.list() {
if entry.pid != 0 && !matches!(entry.state, ProcessState::Exited(_)) {
let _ = process_table.update_state(entry.pid, ProcessState::Exited(-1));
results.push((entry.pid, -1));
}
}
}
}
}
results
}
#[cfg(feature = "exochain")]
pub fn tree_manager(&self) -> Option<&Arc<crate::tree_manager::TreeManager>> {
self.tree_manager.as_ref()
}
#[cfg(feature = "exochain")]
pub fn chain_manager(&self) -> Option<&Arc<crate::chain::ChainManager>> {
self.chain_manager.as_ref()
}
}
#[cfg(test)]
mod tests {
use super::*;
use clawft_core::bus::MessageBus;
fn make_supervisor() -> AgentSupervisor<clawft_platform::NativePlatform> {
let process_table = Arc::new(ProcessTable::new(16));
let bus = Arc::new(MessageBus::new());
let ipc = Arc::new(KernelIpc::new(bus));
AgentSupervisor::new(process_table, ipc, AgentCapabilities::default())
}
fn simple_request(agent_id: &str) -> SpawnRequest {
SpawnRequest {
agent_id: agent_id.to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: None,
}
}
#[test]
fn spawn_creates_process_entry() {
let sup = make_supervisor();
let result = sup.spawn(simple_request("agent-1")).unwrap();
assert!(result.pid > 0);
assert_eq!(result.agent_id, "agent-1");
let entry = sup.inspect(result.pid).unwrap();
assert_eq!(entry.agent_id, "agent-1");
assert_eq!(entry.state, ProcessState::Starting);
}
#[test]
fn spawn_uses_default_capabilities() {
let sup = make_supervisor();
let result = sup.spawn(simple_request("agent-1")).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert!(entry.capabilities.can_spawn);
assert!(entry.capabilities.can_ipc);
assert!(entry.capabilities.can_exec_tools);
}
#[test]
fn spawn_uses_custom_capabilities() {
let sup = make_supervisor();
let caps = AgentCapabilities {
can_spawn: false,
can_ipc: false,
can_exec_tools: true,
can_network: true,
..Default::default()
};
let request = SpawnRequest {
agent_id: "restricted".to_owned(),
capabilities: Some(caps.clone()),
parent_pid: None,
env: HashMap::new(),
backend: None,
};
let result = sup.spawn(request).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert!(!entry.capabilities.can_spawn);
assert!(!entry.capabilities.can_ipc);
assert!(entry.capabilities.can_network);
}
#[test]
fn spawn_with_parent_pid() {
let sup = make_supervisor();
let parent = sup.spawn(simple_request("parent")).unwrap();
let request = SpawnRequest {
agent_id: "child".to_owned(),
capabilities: None,
parent_pid: Some(parent.pid),
env: HashMap::new(),
backend: None,
};
let result = sup.spawn(request).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert_eq!(entry.parent_pid, Some(parent.pid));
}
#[test]
fn spawn_fails_when_table_full() {
let process_table = Arc::new(ProcessTable::new(2));
let bus = Arc::new(MessageBus::new());
let ipc = Arc::new(KernelIpc::new(bus));
let sup: AgentSupervisor<clawft_platform::NativePlatform> =
AgentSupervisor::new(process_table, ipc, AgentCapabilities::default());
sup.spawn(simple_request("a1")).unwrap();
sup.spawn(simple_request("a2")).unwrap();
let result = sup.spawn(simple_request("a3"));
assert!(result.is_err());
}
#[test]
fn stop_graceful() {
let sup = make_supervisor();
let result = sup.spawn(simple_request("agent-1")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Running)
.unwrap();
sup.stop(result.pid, true).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert_eq!(entry.state, ProcessState::Stopping);
assert!(entry.cancel_token.is_cancelled());
}
#[test]
fn stop_force() {
let sup = make_supervisor();
let result = sup.spawn(simple_request("agent-1")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Running)
.unwrap();
sup.stop(result.pid, false).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert!(entry.cancel_token.is_cancelled());
}
#[test]
fn stop_already_exited_is_idempotent() {
let sup = make_supervisor();
let result = sup.spawn(simple_request("agent-1")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Exited(0))
.unwrap();
sup.stop(result.pid, true).unwrap();
}
#[test]
fn stop_nonexistent_pid_fails() {
let sup = make_supervisor();
let result = sup.stop(999, true);
assert!(result.is_err());
}
#[test]
fn restart_creates_new_process() {
let sup = make_supervisor();
let original = sup.spawn(simple_request("agent-1")).unwrap();
sup.process_table()
.update_state(original.pid, ProcessState::Running)
.unwrap();
let restarted = sup.restart(original.pid).unwrap();
assert_ne!(restarted.pid, original.pid);
assert_eq!(restarted.agent_id, "agent-1");
let new_entry = sup.inspect(restarted.pid).unwrap();
assert_eq!(new_entry.parent_pid, Some(original.pid));
}
#[test]
fn restart_preserves_capabilities() {
let sup = make_supervisor();
let caps = AgentCapabilities {
can_spawn: false,
can_network: true,
..Default::default()
};
let request = SpawnRequest {
agent_id: "restricted".to_owned(),
capabilities: Some(caps),
parent_pid: None,
env: HashMap::new(),
backend: None,
};
let original = sup.spawn(request).unwrap();
sup.process_table()
.update_state(original.pid, ProcessState::Running)
.unwrap();
let restarted = sup.restart(original.pid).unwrap();
let entry = sup.inspect(restarted.pid).unwrap();
assert!(!entry.capabilities.can_spawn);
assert!(entry.capabilities.can_network);
}
#[test]
fn list_by_state() {
let sup = make_supervisor();
let r1 = sup.spawn(simple_request("a1")).unwrap();
let r2 = sup.spawn(simple_request("a2")).unwrap();
sup.spawn(simple_request("a3")).unwrap();
sup.process_table()
.update_state(r1.pid, ProcessState::Running)
.unwrap();
sup.process_table()
.update_state(r2.pid, ProcessState::Running)
.unwrap();
let running = sup.list_by_state(ProcessState::Running);
assert_eq!(running.len(), 2);
let starting = sup.list_by_state(ProcessState::Starting);
assert_eq!(starting.len(), 1);
}
#[test]
fn list_agents_excludes_kernel() {
let sup = make_supervisor();
let kernel_entry = ProcessEntry {
pid: 0,
agent_id: "kernel".to_owned(),
state: ProcessState::Running,
capabilities: AgentCapabilities::default(),
resource_usage: ResourceUsage::default(),
cancel_token: CancellationToken::new(),
parent_pid: None,
};
sup.process_table().insert_with_pid(kernel_entry).unwrap();
sup.spawn(simple_request("agent-1")).unwrap();
let agents = sup.list_agents();
assert_eq!(agents.len(), 1);
assert_eq!(agents[0].agent_id, "agent-1");
}
#[test]
fn running_count() {
let sup = make_supervisor();
let r1 = sup.spawn(simple_request("a1")).unwrap();
let r2 = sup.spawn(simple_request("a2")).unwrap();
sup.spawn(simple_request("a3")).unwrap();
assert_eq!(sup.running_count(), 0);
sup.process_table()
.update_state(r1.pid, ProcessState::Running)
.unwrap();
assert_eq!(sup.running_count(), 1);
sup.process_table()
.update_state(r2.pid, ProcessState::Running)
.unwrap();
assert_eq!(sup.running_count(), 2);
}
#[test]
fn default_capabilities_accessor() {
let sup = make_supervisor();
let caps = sup.default_capabilities();
assert!(caps.can_spawn);
assert!(caps.can_ipc);
assert!(caps.can_exec_tools);
}
#[test]
fn spawn_request_serde_roundtrip() {
let request = SpawnRequest {
agent_id: "test".to_owned(),
capabilities: Some(AgentCapabilities {
can_spawn: false,
..Default::default()
}),
parent_pid: Some(5),
env: HashMap::from([("KEY".into(), "VALUE".into())]),
backend: Some(SpawnBackend::Native),
};
let json = serde_json::to_string(&request).unwrap();
let restored: SpawnRequest = serde_json::from_str(&json).unwrap();
assert_eq!(restored.agent_id, "test");
assert_eq!(restored.parent_pid, Some(5));
assert!(!restored.capabilities.unwrap().can_spawn);
}
#[test]
fn spawn_result_serde_roundtrip() {
let result = SpawnResult {
pid: 42,
agent_id: "agent-42".to_owned(),
};
let json = serde_json::to_string(&result).unwrap();
let restored: SpawnResult = serde_json::from_str(&json).unwrap();
assert_eq!(restored.pid, 42);
assert_eq!(restored.agent_id, "agent-42");
}
#[tokio::test]
async fn spawn_and_run_executes_work() {
let sup = make_supervisor();
let result = sup
.spawn_and_run(simple_request("runner-1"), |_pid, _cancel| async { 0 })
.unwrap();
assert!(result.pid > 0);
assert_eq!(result.agent_id, "runner-1");
let entry = sup.inspect(result.pid).unwrap();
assert_eq!(entry.state, ProcessState::Running);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let entry = sup.inspect(result.pid).unwrap();
assert!(matches!(entry.state, ProcessState::Exited(0)));
assert_eq!(sup.running_task_count(), 0);
}
#[tokio::test]
async fn spawn_and_run_respects_cancellation() {
let sup = make_supervisor();
let result = sup
.spawn_and_run(simple_request("cancellable"), |_pid, cancel| async move {
cancel.cancelled().await;
42
})
.unwrap();
assert_eq!(sup.running_task_count(), 1);
sup.stop(result.pid, true).unwrap();
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let entry = sup.inspect(result.pid).unwrap();
assert!(matches!(entry.state, ProcessState::Exited(42)));
assert_eq!(sup.running_task_count(), 0);
}
#[tokio::test]
async fn spawn_and_run_force_stop_aborts() {
let sup = make_supervisor();
let result = sup
.spawn_and_run(simple_request("force-me"), |_pid, cancel| async move {
cancel.cancelled().await;
0
})
.unwrap();
sup.stop(result.pid, false).unwrap();
let entry = sup.inspect(result.pid).unwrap();
assert!(matches!(entry.state, ProcessState::Exited(-1)));
assert_eq!(sup.running_task_count(), 0);
}
#[tokio::test]
async fn abort_all_clears_running_agents() {
let sup = make_supervisor();
sup.spawn_and_run(simple_request("a1"), |_pid, cancel| async move {
cancel.cancelled().await;
0
})
.unwrap();
sup.spawn_and_run(simple_request("a2"), |_pid, cancel| async move {
cancel.cancelled().await;
0
})
.unwrap();
assert_eq!(sup.running_task_count(), 2);
sup.abort_all();
assert_eq!(sup.running_task_count(), 0);
}
#[tokio::test]
async fn watchdog_sweep_reaps_finished_task() {
let sup = make_supervisor();
let result = sup
.spawn_and_run(simple_request("instant"), |_pid, _cancel| async { 0 })
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let reaped = sup.watchdog_sweep().await;
let entry = sup.inspect(result.pid).unwrap();
assert!(
matches!(entry.state, ProcessState::Exited(_)),
"process should be Exited after sweep, got {:?}",
entry.state
);
assert_eq!(sup.running_task_count(), 0);
for (pid, code) in &reaped {
assert_eq!(*pid, result.pid);
assert!(*code == -2 || *code == -3);
}
}
#[tokio::test]
async fn shutdown_all_graceful() {
let sup = make_supervisor();
sup.spawn_and_run(simple_request("g1"), |_pid, cancel| async move {
cancel.cancelled().await;
0
})
.unwrap();
sup.spawn_and_run(simple_request("g2"), |_pid, cancel| async move {
cancel.cancelled().await;
42
})
.unwrap();
assert_eq!(sup.running_task_count(), 2);
let results = sup.shutdown_all(std::time::Duration::from_secs(5)).await;
assert_eq!(results.len(), 2);
assert_eq!(sup.running_task_count(), 0);
}
#[tokio::test]
async fn shutdown_all_timeout_aborts() {
let sup = make_supervisor();
sup.spawn_and_run(simple_request("stubborn"), |_pid, _cancel| async move {
tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
0
})
.unwrap();
assert_eq!(sup.running_task_count(), 1);
let results = sup
.shutdown_all(std::time::Duration::from_millis(100))
.await;
assert!(!results.is_empty());
assert_eq!(sup.running_task_count(), 0);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn chain_logs_agent_spawn() {
let process_table = Arc::new(ProcessTable::new(16));
let bus = Arc::new(MessageBus::new());
let ipc = Arc::new(KernelIpc::new(bus));
let cm = Arc::new(crate::chain::ChainManager::new(0, 1000));
let sup: AgentSupervisor<clawft_platform::NativePlatform> =
AgentSupervisor::new(process_table, ipc, AgentCapabilities::default())
.with_exochain(None, Some(cm.clone()));
let request = SpawnRequest {
agent_id: "chain-agent".to_owned(),
capabilities: None,
parent_pid: Some(99),
env: HashMap::new(),
backend: None,
};
let result = sup
.spawn_and_run(request, |_pid, _cancel| async { 0 })
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let events = cm.tail(10);
let spawn_evt = events.iter().find(|e| e.kind == "agent.spawn");
assert!(spawn_evt.is_some(), "expected agent.spawn event on chain");
let payload = spawn_evt.unwrap().payload.as_ref().unwrap();
assert_eq!(payload["agent_id"], "chain-agent");
assert_eq!(payload["pid"], result.pid);
assert_eq!(payload["parent_pid"], 99);
let exit_evt = events.iter().find(|e| e.kind == "agent.exit");
assert!(exit_evt.is_some(), "expected agent.exit event on chain");
}
#[test]
fn spawn_native_explicit() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "native-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: Some(SpawnBackend::Native),
};
let result = sup.spawn(request).unwrap();
assert!(result.pid > 0);
assert_eq!(result.agent_id, "native-agent");
}
#[test]
fn spawn_backend_none_defaults_to_native() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "default-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: None,
};
let result = sup.spawn(request).unwrap();
assert!(result.pid > 0);
assert_eq!(result.agent_id, "default-agent");
}
#[test]
fn spawn_wasm_returns_not_available() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "wasm-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: Some(SpawnBackend::Wasm {
module: PathBuf::from("/tmp/agent.wasm"),
}),
};
let result = sup.spawn(request);
assert!(result.is_err());
let err = result.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("wasm"), "error should mention wasm: {msg}");
assert!(
msg.contains("not available"),
"error should say not available: {msg}"
);
}
#[test]
fn spawn_container_returns_not_available() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "container-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: Some(SpawnBackend::Container {
image: "ghcr.io/test/agent:latest".into(),
}),
};
let result = sup.spawn(request);
assert!(result.is_err());
let msg = result.unwrap_err().to_string();
assert!(
msg.contains("container"),
"error should mention container: {msg}"
);
}
#[test]
fn spawn_tee_returns_not_available() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "tee-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: Some(SpawnBackend::Tee {
enclave: EnclaveConfig {
enclave_type: "sgx".into(),
},
}),
};
let result = sup.spawn(request);
assert!(result.is_err());
let msg = result.unwrap_err().to_string();
assert!(msg.contains("tee"), "error should mention tee: {msg}");
}
#[test]
fn spawn_remote_returns_not_available() {
let sup = make_supervisor();
let request = SpawnRequest {
agent_id: "remote-agent".to_owned(),
capabilities: None,
parent_pid: None,
env: HashMap::new(),
backend: Some(SpawnBackend::Remote {
node_id: "node-42".into(),
}),
};
let result = sup.spawn(request);
assert!(result.is_err());
let msg = result.unwrap_err().to_string();
assert!(msg.contains("remote"), "error should mention remote: {msg}");
}
#[cfg(feature = "os-patterns")]
mod restart_tests {
use super::*;
use crate::capability::ResourceLimits;
use crate::supervisor::{
ResourceCheckResult, RestartBudget, RestartStrategy, RestartTracker,
check_resource_usage,
};
#[test]
fn restart_strategy_default_is_one_for_one() {
assert_eq!(RestartStrategy::default(), RestartStrategy::OneForOne);
}
#[test]
fn restart_strategy_serde_roundtrip() {
let strategies = vec![
RestartStrategy::OneForOne,
RestartStrategy::OneForAll,
RestartStrategy::RestForOne,
];
for strategy in strategies {
let json = serde_json::to_string(&strategy).unwrap();
let restored: RestartStrategy = serde_json::from_str(&json).unwrap();
assert_eq!(restored, strategy);
}
}
#[test]
fn restart_budget_default() {
let budget = RestartBudget::default();
assert_eq!(budget.max_restarts, 5);
assert_eq!(budget.within_secs, 60);
}
#[test]
fn restart_budget_serde_roundtrip() {
let budget = RestartBudget {
max_restarts: 3,
within_secs: 30,
};
let json = serde_json::to_string(&budget).unwrap();
let restored: RestartBudget = serde_json::from_str(&json).unwrap();
assert_eq!(restored.max_restarts, 3);
assert_eq!(restored.within_secs, 30);
}
#[test]
fn tracker_new_starts_at_zero() {
let tracker = RestartTracker::new();
assert_eq!(tracker.restart_count, 0);
assert_eq!(tracker.backoff_ms, 0);
assert!(tracker.last_restart.is_none());
}
#[test]
fn tracker_backoff_exponential() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 10,
within_secs: 60,
};
tracker.record_restart(&budget);
assert_eq!(tracker.backoff_ms, 100);
tracker.record_restart(&budget);
assert_eq!(tracker.backoff_ms, 200);
tracker.record_restart(&budget);
assert_eq!(tracker.backoff_ms, 400);
tracker.record_restart(&budget);
assert_eq!(tracker.backoff_ms, 800);
}
#[test]
fn tracker_backoff_caps_at_30s() {
let mut tracker = RestartTracker::new();
tracker.restart_count = 20;
let delay = tracker.next_backoff_ms();
assert!(delay <= 30_000, "backoff should cap at 30s, got {delay}");
}
#[test]
fn tracker_budget_exceeded_returns_false() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 2,
within_secs: 60,
};
assert!(tracker.record_restart(&budget)); assert!(tracker.record_restart(&budget)); assert!(!tracker.record_restart(&budget)); }
#[test]
fn tracker_budget_within_returns_true() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 5,
within_secs: 60,
};
for _ in 0..5 {
assert!(tracker.record_restart(&budget));
}
}
#[test]
fn tracker_records_last_restart() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget::default();
assert!(tracker.last_restart.is_none());
tracker.record_restart(&budget);
assert!(tracker.last_restart.is_some());
}
#[test]
fn tracker_with_untrained_model_matches_hardcoded() {
let model = crate::eml_kernel::RestartStrategyModel::new();
assert!(!model.is_trained());
let mut tracker = RestartTracker::new();
let budget = RestartBudget::default();
for _ in 0..5 {
tracker.check_window(&budget);
tracker.restart_count += 1;
let hardcoded = tracker.next_backoff_ms();
let via_model = tracker.next_backoff_ms_with_model(Some(&model), 0, 60.0, 0.5);
assert_eq!(
hardcoded, via_model,
"untrained model must reproduce hardcoded backoff at count {}",
tracker.restart_count
);
}
}
#[test]
fn tracker_with_trained_model_uses_prediction() {
let model = crate::eml_kernel::RestartStrategyModel::new();
let mut json = serde_json::to_value(&model).unwrap();
if let Some(inner) = json.get_mut("inner").and_then(|v| v.as_object_mut()) {
inner.insert("trained".into(), serde_json::Value::Bool(true));
}
let forced: crate::eml_kernel::RestartStrategyModel =
serde_json::from_value(json).unwrap();
assert!(forced.is_trained());
let mut tracker = RestartTracker::new();
tracker.restart_count = 3;
let hardcoded = tracker.next_backoff_ms();
let via_model = tracker.next_backoff_ms_with_model(Some(&forced), 0, 60.0, 0.5);
assert_eq!(hardcoded, 400);
assert_ne!(hardcoded, via_model);
}
#[test]
fn tracker_record_with_model_matches_plain_when_untrained() {
let model = crate::eml_kernel::RestartStrategyModel::new();
let budget = RestartBudget::default();
let mut baseline = RestartTracker::new();
let mut wired = RestartTracker::new();
for _ in 0..4 {
baseline.record_restart(&budget);
wired.record_restart_with_model(&budget, Some(&model), 0, 60.0, 0.5);
assert_eq!(baseline.backoff_ms, wired.backoff_ms);
assert_eq!(baseline.restart_count, wired.restart_count);
}
}
#[test]
fn resource_check_within_limits() {
let usage = ResourceUsage {
memory_bytes: 100,
cpu_time_ms: 100,
tool_calls: 10,
messages_sent: 10,
};
let limits = ResourceLimits::default(); let results = check_resource_usage(&usage, &limits);
assert!(results.is_empty());
}
#[test]
fn resource_check_warning_at_80_percent() {
let usage = ResourceUsage {
memory_bytes: 220 * 1024 * 1024, cpu_time_ms: 100,
tool_calls: 10,
messages_sent: 10,
};
let limits = ResourceLimits::default();
let results = check_resource_usage(&usage, &limits);
assert_eq!(results.len(), 1);
assert!(
matches!(&results[0], ResourceCheckResult::Warning { resource, .. } if resource == "memory")
);
}
#[test]
fn resource_check_exceeded_at_100_percent() {
let usage = ResourceUsage {
memory_bytes: 300 * 1024 * 1024, cpu_time_ms: 100,
tool_calls: 10,
messages_sent: 10,
};
let limits = ResourceLimits::default();
let results = check_resource_usage(&usage, &limits);
assert_eq!(results.len(), 1);
assert!(
matches!(&results[0], ResourceCheckResult::Exceeded { resource, .. } if resource == "memory")
);
}
#[test]
fn resource_check_unlimited_skipped() {
let usage = ResourceUsage {
memory_bytes: 999_999_999,
cpu_time_ms: 999_999_999,
tool_calls: 999_999_999,
messages_sent: 999_999_999,
};
let limits = ResourceLimits {
max_memory_bytes: 0,
max_cpu_time_ms: 0,
max_tool_calls: 0,
max_messages: 0,
..Default::default()
};
let results = check_resource_usage(&usage, &limits);
assert!(results.is_empty(), "0 = unlimited should skip enforcement");
}
#[test]
fn resource_check_multiple_exceeded() {
let limits = ResourceLimits {
max_memory_bytes: 100,
max_cpu_time_ms: 100,
max_tool_calls: 10,
max_messages: 10,
..Default::default()
};
let usage = ResourceUsage {
memory_bytes: 200,
cpu_time_ms: 200,
tool_calls: 20,
messages_sent: 20,
};
let results = check_resource_usage(&usage, &limits);
assert_eq!(results.len(), 4);
for r in &results {
assert!(matches!(r, ResourceCheckResult::Exceeded { .. }));
}
}
#[test]
fn resource_check_message_limit() {
let limits = ResourceLimits {
max_messages: 100,
..Default::default()
};
let usage = ResourceUsage {
messages_sent: 100,
..Default::default()
};
let results = check_resource_usage(&usage, &limits);
assert_eq!(results.len(), 1);
assert!(
matches!(&results[0], ResourceCheckResult::Exceeded { resource, .. } if resource == "messages")
);
}
#[test]
fn resource_check_cpu_time_limit() {
let limits = ResourceLimits {
max_cpu_time_ms: 1000,
..Default::default()
};
let usage = ResourceUsage {
cpu_time_ms: 1000,
..Default::default()
};
let results = check_resource_usage(&usage, &limits);
assert_eq!(results.len(), 1);
assert!(
matches!(&results[0], ResourceCheckResult::Exceeded { resource, .. } if resource == "cpu_time")
);
}
#[test]
fn restart_strategy_permanent_serde_roundtrip() {
let strategy = RestartStrategy::Permanent;
let json = serde_json::to_string(&strategy).unwrap();
let restored: RestartStrategy = serde_json::from_str(&json).unwrap();
assert_eq!(restored, RestartStrategy::Permanent);
}
#[test]
fn restart_strategy_transient_serde_roundtrip() {
let strategy = RestartStrategy::Transient;
let json = serde_json::to_string(&strategy).unwrap();
let restored: RestartStrategy = serde_json::from_str(&json).unwrap();
assert_eq!(restored, RestartStrategy::Transient);
}
#[test]
fn should_restart_permanent_never_restarts() {
assert!(!RestartTracker::should_restart(
&RestartStrategy::Permanent,
0
));
assert!(!RestartTracker::should_restart(
&RestartStrategy::Permanent,
1
));
assert!(!RestartTracker::should_restart(
&RestartStrategy::Permanent,
-1
));
}
#[test]
fn should_restart_transient_only_on_abnormal() {
assert!(!RestartTracker::should_restart(
&RestartStrategy::Transient,
0
));
assert!(RestartTracker::should_restart(
&RestartStrategy::Transient,
1
));
assert!(RestartTracker::should_restart(
&RestartStrategy::Transient,
-1
));
assert!(RestartTracker::should_restart(
&RestartStrategy::Transient,
42
));
}
#[test]
fn should_restart_one_for_one_always() {
assert!(RestartTracker::should_restart(
&RestartStrategy::OneForOne,
0
));
assert!(RestartTracker::should_restart(
&RestartStrategy::OneForOne,
1
));
}
#[test]
fn should_restart_one_for_all_always() {
assert!(RestartTracker::should_restart(
&RestartStrategy::OneForAll,
0
));
assert!(RestartTracker::should_restart(
&RestartStrategy::OneForAll,
-1
));
}
#[test]
fn should_restart_rest_for_one_always() {
assert!(RestartTracker::should_restart(
&RestartStrategy::RestForOne,
0
));
assert!(RestartTracker::should_restart(
&RestartStrategy::RestForOne,
1
));
}
#[test]
fn tracker_is_exhausted_when_at_max() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 3,
within_secs: 60,
};
assert!(!tracker.is_exhausted(&budget));
tracker.record_restart(&budget); tracker.record_restart(&budget); assert!(!tracker.is_exhausted(&budget));
tracker.record_restart(&budget); assert!(tracker.is_exhausted(&budget));
}
#[test]
fn tracker_remaining_decreases() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 5,
within_secs: 60,
};
assert_eq!(tracker.remaining(&budget), 5);
tracker.record_restart(&budget);
assert_eq!(tracker.remaining(&budget), 4);
tracker.record_restart(&budget);
assert_eq!(tracker.remaining(&budget), 3);
}
#[test]
fn tracker_remaining_saturates_at_zero() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 2,
within_secs: 60,
};
tracker.record_restart(&budget); tracker.record_restart(&budget); assert_eq!(tracker.remaining(&budget), 0);
tracker.record_restart(&budget); assert_eq!(tracker.remaining(&budget), 0);
}
#[test]
fn tracker_backoff_sequence_matches_spec() {
let mut tracker = RestartTracker::new();
let budget = RestartBudget {
max_restarts: 20,
within_secs: 600,
};
let expected = [100, 200, 400, 800, 1600, 3200, 6400, 12800, 25600, 30000];
for &exp in &expected {
tracker.record_restart(&budget);
assert_eq!(
tracker.backoff_ms, exp,
"restart #{}: expected {}ms, got {}ms",
tracker.restart_count, exp, tracker.backoff_ms
);
}
}
#[test]
fn handle_exit_permanent_no_restart() {
let sup =
make_supervisor_with_strategy(RestartStrategy::Permanent, RestartBudget::default());
let result = sup.spawn(simple_request("perm-agent")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(result.pid, ProcessState::Exited(1));
let restarts = sup.handle_exit(result.pid, 1);
assert!(
restarts.is_empty(),
"Permanent strategy should never restart"
);
}
#[test]
fn handle_exit_transient_normal_no_restart() {
let sup =
make_supervisor_with_strategy(RestartStrategy::Transient, RestartBudget::default());
let result = sup.spawn(simple_request("trans-agent")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(result.pid, ProcessState::Exited(0));
let restarts = sup.handle_exit(result.pid, 0);
assert!(
restarts.is_empty(),
"Transient should not restart on normal exit (code 0)"
);
}
#[test]
fn handle_exit_transient_abnormal_restarts() {
let sup = make_supervisor_with_strategy(
RestartStrategy::Transient,
RestartBudget {
max_restarts: 5,
within_secs: 60,
},
);
let result = sup.spawn(simple_request("trans-crash")).unwrap();
sup.process_table()
.update_state(result.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(result.pid, ProcessState::Exited(1));
let restarts = sup.handle_exit(result.pid, 1);
assert_eq!(
restarts.len(),
1,
"Transient should restart on abnormal exit"
);
assert_eq!(restarts[0].0, result.pid);
}
#[test]
fn handle_exit_one_for_one_restarts_only_failed() {
let sup = make_supervisor_with_strategy(
RestartStrategy::OneForOne,
RestartBudget {
max_restarts: 10,
within_secs: 60,
},
);
let r1 = sup.spawn(simple_request("ofo-a")).unwrap();
let r2 = sup.spawn(simple_request("ofo-b")).unwrap();
sup.process_table()
.update_state(r1.pid, ProcessState::Running)
.unwrap();
sup.process_table()
.update_state(r2.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(r1.pid, ProcessState::Exited(1));
let restarts = sup.handle_exit(r1.pid, 1);
assert_eq!(restarts.len(), 1);
assert_eq!(restarts[0].0, r1.pid);
let r2_entry = sup.inspect(r2.pid).unwrap();
assert_eq!(r2_entry.state, ProcessState::Running);
}
#[test]
fn handle_exit_budget_exhausted_no_restart() {
let sup = make_supervisor_with_strategy(
RestartStrategy::OneForOne,
RestartBudget {
max_restarts: 1,
within_secs: 60,
},
);
let r1 = sup.spawn(simple_request("budget-a")).unwrap();
sup.process_table()
.update_state(r1.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(r1.pid, ProcessState::Exited(1));
let restarts1 = sup.handle_exit(r1.pid, 1);
assert_eq!(restarts1.len(), 1, "first restart should succeed");
let new_pid = restarts1[0].1.pid;
sup.process_table()
.update_state(new_pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(r1.pid, ProcessState::Exited(1));
let restarts2 = sup.handle_exit(r1.pid, 1);
assert!(
restarts2.is_empty(),
"budget should be exhausted after 1 restart"
);
}
#[test]
fn handle_exit_links_notify_monitor_registry() {
let sup = make_supervisor_with_strategy(
RestartStrategy::Permanent, RestartBudget::default(),
);
let r1 = sup.spawn(simple_request("linked-a")).unwrap();
let r2 = sup.spawn(simple_request("linked-b")).unwrap();
sup.monitor_registry().link(r1.pid, r2.pid);
let _ref_id = sup.monitor_registry().monitor(r2.pid, r1.pid);
sup.process_table()
.update_state(r1.pid, ProcessState::Running)
.unwrap();
let _ = sup
.process_table()
.update_state(r1.pid, ProcessState::Exited(1));
let _restarts = sup.handle_exit(r1.pid, 1);
assert!(!sup.monitor_registry().is_linked(r1.pid, r2.pid));
assert!(sup.monitor_registry().get_monitors(r1.pid).is_empty());
}
#[test]
fn supervisor_with_restart_config_builder() {
let process_table = Arc::new(ProcessTable::new(16));
let bus = Arc::new(clawft_core::bus::MessageBus::new());
let ipc = Arc::new(KernelIpc::new(bus));
let sup: AgentSupervisor<clawft_platform::NativePlatform> =
AgentSupervisor::new(process_table, ipc, AgentCapabilities::default())
.with_restart_config(
RestartStrategy::RestForOne,
RestartBudget {
max_restarts: 3,
within_secs: 30,
},
);
assert_eq!(*sup.restart_strategy(), RestartStrategy::RestForOne);
assert_eq!(sup.restart_budget().max_restarts, 3);
assert_eq!(sup.restart_budget().within_secs, 30);
}
fn make_supervisor_with_strategy(
strategy: RestartStrategy,
budget: RestartBudget,
) -> AgentSupervisor<clawft_platform::NativePlatform> {
let process_table = Arc::new(ProcessTable::new(32));
let bus = Arc::new(clawft_core::bus::MessageBus::new());
let ipc = Arc::new(KernelIpc::new(bus));
AgentSupervisor::new(process_table, ipc, AgentCapabilities::default())
.with_restart_config(strategy, budget)
}
}
#[test]
fn spawn_backend_native_serde_roundtrip() {
let backend = SpawnBackend::Native;
let json = serde_json::to_string(&backend).unwrap();
let _: SpawnBackend = serde_json::from_str(&json).unwrap();
}
#[test]
fn spawn_backend_wasm_serde_roundtrip() {
let backend = SpawnBackend::Wasm {
module: PathBuf::from("/opt/modules/agent.wasm"),
};
let json = serde_json::to_string(&backend).unwrap();
let restored: SpawnBackend = serde_json::from_str(&json).unwrap();
assert!(matches!(restored, SpawnBackend::Wasm { .. }));
}
#[test]
fn spawn_backend_container_serde_roundtrip() {
let backend = SpawnBackend::Container {
image: "ghcr.io/org/agent:v1".into(),
};
let json = serde_json::to_string(&backend).unwrap();
let restored: SpawnBackend = serde_json::from_str(&json).unwrap();
assert!(matches!(restored, SpawnBackend::Container { .. }));
}
#[test]
fn spawn_backend_tee_serde_roundtrip() {
let backend = SpawnBackend::Tee {
enclave: EnclaveConfig {
enclave_type: "sgx".into(),
},
};
let json = serde_json::to_string(&backend).unwrap();
let restored: SpawnBackend = serde_json::from_str(&json).unwrap();
assert!(matches!(restored, SpawnBackend::Tee { .. }));
}
#[test]
fn spawn_backend_remote_serde_roundtrip() {
let backend = SpawnBackend::Remote {
node_id: "node-42".into(),
};
let json = serde_json::to_string(&backend).unwrap();
let restored: SpawnBackend = serde_json::from_str(&json).unwrap();
assert!(matches!(restored, SpawnBackend::Remote { .. }));
}
#[test]
fn enclave_config_serde_roundtrip() {
let cfg = EnclaveConfig {
enclave_type: "trustzone".into(),
};
let json = serde_json::to_string(&cfg).unwrap();
let restored: EnclaveConfig = serde_json::from_str(&json).unwrap();
assert_eq!(restored.enclave_type, "trustzone");
}
#[test]
fn spawn_request_with_backend_serde_roundtrip() {
let req = SpawnRequest {
agent_id: "test-agent".into(),
capabilities: None,
parent_pid: Some(42),
env: HashMap::from([("LOG_LEVEL".into(), "debug".into())]),
backend: Some(SpawnBackend::Native),
};
let json = serde_json::to_string(&req).unwrap();
let restored: SpawnRequest = serde_json::from_str(&json).unwrap();
assert_eq!(restored.agent_id, "test-agent");
assert_eq!(restored.parent_pid, Some(42));
assert!(restored.backend.is_some());
}
#[test]
fn spawn_request_minimal_json_deserializes() {
let json = r#"{"agent_id":"minimal"}"#;
let req: SpawnRequest = serde_json::from_str(json).unwrap();
assert_eq!(req.agent_id, "minimal");
assert!(req.capabilities.is_none());
assert!(req.parent_pid.is_none());
assert!(req.env.is_empty());
assert!(req.backend.is_none());
}
}