use std::sync::Arc;
use std::time::Instant;
use serde::{Deserialize, Serialize};
use tracing::{error, info};
#[cfg(feature = "exochain")]
use tracing::warn;
use clawft_core::bootstrap::AppContext;
use clawft_core::bus::MessageBus;
use clawft_platform::Platform;
use clawft_types::config::Config;
#[cfg(feature = "native")]
use crate::a2a::A2ARouter;
use crate::capability::AgentCapabilities;
#[cfg(feature = "native")]
use crate::capability::CapabilityChecker;
use crate::cluster::{ClusterConfig, ClusterMembership};
use crate::console::{BootEvent, BootLog, BootPhase, KernelEventLog};
use crate::error::{KernelError, KernelResult};
use crate::health::HealthSystem;
use crate::ipc::KernelIpc;
use crate::process::{ProcessEntry, ProcessState, ProcessTable, ResourceUsage};
use crate::service::ServiceRegistry;
use crate::supervisor::AgentSupervisor;
#[cfg(feature = "native")]
use crate::topic::TopicRouter;
use clawft_types::config::KernelConfig;
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum KernelState {
Booting,
Running,
ShuttingDown,
Halted,
}
impl std::fmt::Display for KernelState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
KernelState::Booting => write!(f, "booting"),
KernelState::Running => write!(f, "running"),
KernelState::ShuttingDown => write!(f, "shutting_down"),
KernelState::Halted => write!(f, "halted"),
}
}
}
#[cfg(feature = "exochain")]
pub struct ChainSubsystem {
pub(crate) chain_manager: Option<Arc<crate::chain::ChainManager>>,
pub(crate) tree_manager: Option<Arc<crate::tree_manager::TreeManager>>,
pub(crate) governance_gate: Option<Arc<dyn crate::gate::GateBackend>>,
}
#[cfg(feature = "ecc")]
pub struct EccSubsystem {
pub(crate) hnsw: Option<Arc<crate::hnsw_service::HnswService>>,
pub(crate) causal: Option<Arc<crate::causal::CausalGraph>>,
pub(crate) tick: Option<Arc<crate::cognitive_tick::CognitiveTick>>,
pub(crate) crossrefs: Option<Arc<crate::crossref::CrossRefStore>>,
pub(crate) impulses: Option<Arc<crate::impulse::ImpulseQueue>>,
pub(crate) calibration: Option<crate::calibration::EccCalibration>,
}
#[cfg(feature = "os-patterns")]
pub struct ObservabilitySubsystem {
pub(crate) metrics_registry: Option<Arc<crate::metrics::MetricsRegistry>>,
pub(crate) log_service: Option<Arc<crate::log_service::LogService>>,
pub(crate) timer_service: Option<Arc<crate::timer::TimerService>>,
pub(crate) dead_letter_queue: Option<Arc<crate::dead_letter::DeadLetterQueue>>,
}
pub struct Kernel<P: Platform> {
state: KernelState,
config: KernelConfig,
app_context: Option<AppContext<P>>,
bus: Arc<MessageBus>,
process_table: Arc<ProcessTable>,
service_registry: Arc<ServiceRegistry>,
ipc: Arc<KernelIpc>,
#[cfg(feature = "native")]
a2a_router: Arc<A2ARouter>,
#[cfg(feature = "native")]
cron_service: Arc<crate::cron::CronService>,
#[cfg(feature = "native")]
assessment_service: Arc<crate::assessment::AssessmentService>,
health: HealthSystem,
supervisor: AgentSupervisor<P>,
boot_log: BootLog,
event_log: Arc<KernelEventLog>,
boot_time: Instant,
cluster_membership: Arc<ClusterMembership>,
#[cfg(feature = "exochain")]
chain: ChainSubsystem,
#[cfg(feature = "ecc")]
ecc: EccSubsystem,
#[cfg(feature = "os-patterns")]
observability: ObservabilitySubsystem,
}
impl<P: Platform> Kernel<P> {
pub async fn boot(
config: Config,
kernel_config: KernelConfig,
platform: Arc<P>,
) -> KernelResult<Self> {
let boot_time = Instant::now();
let mut boot_log = BootLog::new();
info!("WeftOS kernel booting");
boot_log.push(BootEvent::info(BootPhase::Init, "WeftOS v0.1.0 booting..."));
boot_log.push(BootEvent::info(BootPhase::Init, "PID 0 (kernel)"));
let process_table = Arc::new(ProcessTable::new(kernel_config.max_processes));
let service_registry = Arc::new(ServiceRegistry::new());
boot_log.push(BootEvent::info(
BootPhase::Config,
format!("Max processes: {}", kernel_config.max_processes),
));
boot_log.push(BootEvent::info(
BootPhase::Config,
format!(
"Health check interval: {}s",
kernel_config.health_check_interval_secs
),
));
let app_context = AppContext::new(config, platform)
.await
.map_err(|e| KernelError::Boot(format!("AppContext init failed: {e}")))?;
let bus = app_context.bus().clone();
let ipc = Arc::new(KernelIpc::new(bus.clone()));
let health = HealthSystem::new(kernel_config.health_check_interval_secs);
let kernel_entry = ProcessEntry {
pid: 0,
agent_id: "kernel".to_owned(),
state: ProcessState::Running,
capabilities: AgentCapabilities::default(),
resource_usage: ResourceUsage::default(),
cancel_token: crate::process::CancellationToken::new(),
parent_pid: None,
};
process_table
.insert_with_pid(kernel_entry)
.map_err(|e| KernelError::Boot(format!("failed to register kernel process: {e}")))?;
boot_log.push(BootEvent::info(
BootPhase::Services,
"Service registry ready",
));
#[cfg(feature = "native")]
let capability_checker = Arc::new(CapabilityChecker::new(process_table.clone()));
#[cfg(feature = "native")]
let topic_router = Arc::new(TopicRouter::new(process_table.clone()));
#[cfg(feature = "native")]
let a2a_router = Arc::new(A2ARouter::new(
process_table.clone(),
capability_checker,
topic_router,
));
#[cfg(feature = "native")]
boot_log.push(BootEvent::info(BootPhase::Services, "A2A router ready"));
#[cfg(feature = "native")]
let cron_svc = Arc::new(crate::cron::CronService::new());
#[cfg(feature = "native")]
if let Err(e) = service_registry.register(cron_svc.clone()) {
error!(error = %e, "failed to register cron service");
} else {
boot_log.push(BootEvent::info(BootPhase::Services, "Cron service registered"));
}
#[cfg(feature = "native")]
let assessment_svc = Arc::new(crate::assessment::AssessmentService::new());
#[cfg(feature = "native")]
if let Err(e) = service_registry.register(assessment_svc.clone()) {
error!(error = %e, "failed to register assessment service");
} else {
boot_log.push(BootEvent::info(BootPhase::Services, "Assessment service registered"));
}
let container_manager = std::sync::Arc::new(
crate::container::ContainerManager::new(crate::container::ContainerConfig::default()),
);
let container_service = std::sync::Arc::new(
crate::container::ContainerService::new(container_manager.clone()),
);
if let Err(e) = service_registry.register(container_service) {
error!(error = %e, "failed to register container service");
} else {
boot_log.push(BootEvent::info(
BootPhase::Services,
"Container service registered",
));
}
let cluster_config = ClusterConfig {
node_id: uuid::Uuid::new_v4().to_string(),
node_name: kernel_config
.cluster
.as_ref()
.and_then(|c| c.node_name.clone())
.unwrap_or_else(|| "local".into()),
heartbeat_interval_secs: kernel_config
.cluster
.as_ref()
.map(|c| c.heartbeat_interval_secs)
.unwrap_or(5),
..ClusterConfig::default()
};
let cluster_membership = Arc::new(ClusterMembership::new(cluster_config));
boot_log.push(BootEvent::info(
BootPhase::Network,
format!(
"Cluster membership ready (node {})",
cluster_membership.local_node_id()
),
));
#[cfg(feature = "cluster")]
{
use crate::cluster::ClusterService;
use ruvector_cluster::StaticDiscovery;
let net_config = kernel_config
.cluster
.clone()
.unwrap_or_default();
let seed_addrs: Vec<std::net::SocketAddr> = net_config
.seed_nodes
.iter()
.filter_map(|s| s.parse().ok())
.collect();
let seed_nodes: Vec<ruvector_cluster::ClusterNode> = seed_addrs
.into_iter()
.map(|addr| ruvector_cluster::ClusterNode::new(addr.to_string(), addr))
.collect();
let discovery = Box::new(StaticDiscovery::new(seed_nodes));
let node_id = cluster_membership.local_node_id().to_owned();
match ClusterService::new(
net_config,
node_id,
discovery,
Arc::clone(&cluster_membership),
) {
Ok(cluster_svc) => {
let svc = Arc::new(cluster_svc);
if let Err(e) = service_registry.register(svc) {
error!(error = %e, "failed to register cluster service");
} else {
boot_log.push(BootEvent::info(
BootPhase::Network,
"Cluster service registered (ruvector)",
));
}
}
Err(e) => {
error!(error = %e, "failed to create cluster service");
boot_log.push(BootEvent::info(
BootPhase::Network,
format!("Cluster service failed: {e}"),
));
}
}
}
service_registry
.start_all()
.await
.map_err(|e| KernelError::Boot(format!("service start failed: {e}")))?;
#[cfg(feature = "exochain")]
let chain_manager = {
let chain_config = kernel_config.chain.clone().unwrap_or_default();
if chain_config.enabled {
let signing_key = if let Some(ref ckpt_path) = chain_config.effective_checkpoint_path() {
let key_path = std::path::PathBuf::from(ckpt_path).with_extension("key");
match crate::chain::ChainManager::load_or_create_key(&key_path) {
Ok(key) => {
boot_log.push(BootEvent::info(
BootPhase::Services,
format!("Ed25519 signing key loaded: {}", key_path.display()),
));
Some(key)
}
Err(e) => {
warn!(error = %e, "failed to load/create signing key");
boot_log.push(BootEvent::info(
BootPhase::Services,
format!("Signing key unavailable: {e} — chain will be unsigned"),
));
None
}
}
} else {
None
};
let cm = if let Some(ref ckpt_path) = chain_config.effective_checkpoint_path() {
let json_path = std::path::PathBuf::from(ckpt_path);
let rvf_path = json_path.with_extension("rvf");
if rvf_path.exists() {
match crate::chain::ChainManager::load_from_rvf(&rvf_path, chain_config.checkpoint_interval) {
Ok(restored) => {
let seq = restored.sequence();
boot_log.push(BootEvent::info(
BootPhase::Services,
format!(
"Chain restored from RVF (seq={}, chain_id={})",
seq, chain_config.chain_id,
),
));
Arc::new(restored)
}
Err(e) => {
error!(error = %e, "failed to restore RVF chain, trying JSON fallback");
if json_path.exists() {
match crate::chain::ChainManager::load_from_file(&json_path, chain_config.checkpoint_interval) {
Ok(restored) => {
let seq = restored.sequence();
boot_log.push(BootEvent::info(
BootPhase::Services,
format!(
"Chain restored from JSON fallback (seq={}, chain_id={})",
seq, chain_config.chain_id,
),
));
Arc::new(restored)
}
Err(e2) => {
error!(error = %e2, "JSON fallback also failed, starting fresh");
boot_log.push(BootEvent::info(
BootPhase::Services,
format!("Chain restore failed (RVF: {e}, JSON: {e2}), starting fresh"),
));
Arc::new(crate::chain::ChainManager::new(
chain_config.chain_id,
chain_config.checkpoint_interval,
))
}
}
} else {
boot_log.push(BootEvent::info(
BootPhase::Services,
format!("RVF restore failed: {e}, starting fresh"),
));
Arc::new(crate::chain::ChainManager::new(
chain_config.chain_id,
chain_config.checkpoint_interval,
))
}
}
}
} else if json_path.exists() {
match crate::chain::ChainManager::load_from_file(&json_path, chain_config.checkpoint_interval) {
Ok(restored) => {
let seq = restored.sequence();
boot_log.push(BootEvent::info(
BootPhase::Services,
format!(
"Chain restored from JSON (seq={}, chain_id={}, will migrate to RVF)",
seq, chain_config.chain_id,
),
));
Arc::new(restored)
}
Err(e) => {
error!(error = %e, "failed to restore chain, starting fresh");
boot_log.push(BootEvent::info(
BootPhase::Services,
format!("Chain restore failed: {e}, starting fresh"),
));
Arc::new(crate::chain::ChainManager::new(
chain_config.chain_id,
chain_config.checkpoint_interval,
))
}
}
} else {
Arc::new(crate::chain::ChainManager::new(
chain_config.chain_id,
chain_config.checkpoint_interval,
))
}
} else {
Arc::new(crate::chain::ChainManager::new(
chain_config.chain_id,
chain_config.checkpoint_interval,
))
};
let mut cm = cm;
if let Some(key) = signing_key
&& let Some(inner) = Arc::get_mut(&mut cm)
{
let ml_dsa_seed = key.to_bytes();
let (ml_key, _ml_vk) = weftos_rvf_crypto::MlDsa65Key::generate(&ml_dsa_seed);
inner.set_signing_key(key);
inner.set_ml_dsa_key(ml_key);
boot_log.push(BootEvent::info(
BootPhase::Services,
"Dual signing enabled (Ed25519 + ML-DSA-65)",
));
}
boot_log.push(BootEvent::info(
BootPhase::Services,
format!(
"Local chain ready (chain_id={}, seq={}, signed={})",
chain_config.chain_id,
cm.sequence(),
cm.has_signing_key(),
),
));
cm.append(
"kernel",
"boot.init",
Some(serde_json::json!({"version": "0.1.0"})),
);
cm.append(
"kernel",
"boot.config",
Some(serde_json::json!({
"max_processes": kernel_config.max_processes,
"health_interval": kernel_config.health_check_interval_secs,
})),
);
cm.append(
"kernel",
"boot.services",
Some(serde_json::json!({
"count": service_registry.len(),
})),
);
Some(cm)
} else {
boot_log.push(BootEvent::info(
BootPhase::Services,
"Local chain disabled",
));
None
}
};
#[cfg(feature = "exochain")]
let tree_manager = {
let rt_config = kernel_config.resource_tree.clone().unwrap_or_default();
if rt_config.enabled {
if let Some(ref cm) = chain_manager {
let tm = Arc::new(crate::tree_manager::TreeManager::new(Arc::clone(cm)));
let chain_cfg = kernel_config.chain.clone().unwrap_or_default();
let tree_ckpt_path = chain_cfg
.effective_checkpoint_path()
.map(|p| std::path::PathBuf::from(p).with_extension("tree.json"));
let mut restored_from_checkpoint = false;
if let Some(ref tree_path) = tree_ckpt_path
&& tree_path.exists()
{
match tm.load_checkpoint(tree_path) {
Ok(()) => {
let stats = tm.stats();
let chain_tree_hash = cm.last_tree_root_hash();
if let Some(ref expected) = chain_tree_hash {
if stats.root_hash == *expected {
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
format!(
"Resource tree restored from checkpoint ({} nodes, root={}..., hash verified)",
stats.node_count,
&stats.root_hash[..12],
),
));
restored_from_checkpoint = true;
} else {
warn!(
expected = %expected,
actual = %stats.root_hash,
"tree checkpoint root hash mismatch — falling back to fresh bootstrap"
);
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
format!(
"Tree checkpoint hash mismatch (expected={}..., got={}...), bootstrapping fresh",
&expected[..std::cmp::min(12, expected.len())],
&stats.root_hash[..12],
),
));
}
} else {
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
format!(
"Resource tree restored from checkpoint ({} nodes, root={}...)",
stats.node_count,
&stats.root_hash[..12],
),
));
restored_from_checkpoint = true;
}
}
Err(e) => {
error!(error = %e, "failed to restore tree checkpoint, bootstrapping fresh");
}
}
}
if !restored_from_checkpoint {
if let Err(e) = tm.bootstrap() {
error!(error = %e, "failed to bootstrap resource tree");
} else {
let stats = tm.stats();
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
format!(
"Resource tree bootstrapped ({} nodes, root={}...)",
stats.node_count,
&stats.root_hash[..12],
),
));
}
}
if let Err(e) = tm.register_service_with_manifest("cron", "scheduler") {
tracing::debug!(error = %e, "failed to register cron in tree (may already exist)");
}
{
use exo_resource_tree::model::{ResourceId, ResourceKind};
if let Err(e) = tm.insert(
ResourceId::new("/kernel/containers"),
ResourceKind::Namespace,
ResourceId::new("/kernel"),
) {
tracing::debug!(
error = %e,
"failed to register containers namespace (may already exist)"
);
}
}
#[cfg(feature = "ecc")]
{
use exo_resource_tree::model::{ResourceId, ResourceKind};
let ecc_namespaces = [
("/kernel/services/ecc", "/kernel/services"),
("/kernel/services/ecc/hnsw", "/kernel/services/ecc"),
("/kernel/services/ecc/causal", "/kernel/services/ecc"),
("/kernel/services/ecc/tick", "/kernel/services/ecc"),
("/kernel/services/ecc/calibration", "/kernel/services/ecc"),
("/kernel/services/ecc/crossrefs", "/kernel/services/ecc"),
];
for (path, parent) in &ecc_namespaces {
if let Err(e) = tm.insert(
ResourceId::new(*path),
ResourceKind::Namespace,
ResourceId::new(*parent),
) {
tracing::debug!(
error = %e, path = *path,
"failed to register ECC namespace (may already exist)"
);
}
}
boot_log.push(BootEvent::info(
BootPhase::Ecc,
"ECC resource tree namespaces registered",
));
}
{
use exo_resource_tree::model::{ResourceId, ResourceKind};
let tool_namespaces = [
("/kernel/tools", "/kernel"),
("/kernel/tools/fs", "/kernel/tools"),
("/kernel/tools/agent", "/kernel/tools"),
("/kernel/tools/sys", "/kernel/tools"),
("/kernel/tools/ipc", "/kernel/tools"),
#[cfg(feature = "ecc")]
("/kernel/tools/ecc", "/kernel/tools"),
];
for (path, parent) in &tool_namespaces {
if let Err(e) = tm.insert(
ResourceId::new(*path),
ResourceKind::Namespace,
ResourceId::new(*parent),
) {
tracing::debug!(
error = %e, path = *path,
"failed to register tool namespace (may already exist)"
);
}
}
let catalog = crate::wasm_runner::builtin_tool_catalog();
for spec in &catalog {
let cat_path = match spec.category {
crate::wasm_runner::ToolCategory::Filesystem => "/kernel/tools/fs",
crate::wasm_runner::ToolCategory::Agent => "/kernel/tools/agent",
crate::wasm_runner::ToolCategory::System => "/kernel/tools/sys",
crate::wasm_runner::ToolCategory::Ecc => "/kernel/tools/ecc",
crate::wasm_runner::ToolCategory::User => "/kernel/tools",
};
let tool_path = format!("{}/{}", cat_path, spec.name.replace('.', "/"));
if let Err(e) = tm.insert(
ResourceId::new(&tool_path),
ResourceKind::Tool,
ResourceId::new(cat_path),
) {
tracing::debug!(
error = %e, tool = %spec.name,
"failed to register tool node (may already exist)"
);
}
}
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
format!("Registered {} built-in tools in resource tree", catalog.len()),
));
}
Some(tm)
} else {
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
"Resource tree requires chain — skipped",
));
None
}
} else {
boot_log.push(BootEvent::info(
BootPhase::ResourceTree,
"Resource tree disabled",
));
None
}
};
#[cfg(feature = "exochain")]
let governance_gate: Option<Arc<dyn crate::gate::GateBackend>> = {
if let Some(ref cm) = chain_manager {
use crate::governance::{GovernanceBranch, GovernanceRule, RuleSeverity};
use crate::gate::GovernanceGate;
let risk_threshold = 0.7;
let human_approval = false;
let mut gate = GovernanceGate::new(risk_threshold, human_approval)
.with_chain(Arc::clone(cm));
let genesis_rules = vec![
GovernanceRule {
id: "GOV-001".into(),
description: "High-risk operations require elevated review".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Blocking,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-002".into(),
description: "Security-sensitive actions must not exceed security threshold".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Blocking,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-003".into(),
description: "Privacy-impacting operations flagged for review".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Warning,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-004".into(),
description: "Novel/unprecedented actions require advisory logging".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Advisory,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-005".into(),
description: "Filesystem write operations scored for risk".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Warning,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-006".into(),
description: "Agent spawn operations require governance clearance".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Blocking,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "GOV-007".into(),
description: "IPC messages between agents logged for audit trail".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Advisory,
active: true,
reference_url: None,
sop_category: None,
},
GovernanceRule {
id: "SOP-L001".into(),
description: "AI-IRB approval required before high-impact deployments".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Blocking,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1300-01-AI_IRB_Approval.md".into()),
sop_category: Some("governance".into()),
},
GovernanceRule {
id: "SOP-L002".into(),
description: "Version control and branching policies must be enforced".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1003-01-AI_Version_Control.md".into()),
sop_category: Some("governance".into()),
},
GovernanceRule {
id: "SOP-L003".into(),
description: "Requirements must include AI-IRB ethical review".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1040-01-AI_Requirements.md".into()),
sop_category: Some("engineering".into()),
},
GovernanceRule {
id: "SOP-L004".into(),
description: "Release planning must follow structured lifecycle gates".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Advisory,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1005-01-AI_Release_Planning.md".into()),
sop_category: Some("lifecycle".into()),
},
GovernanceRule {
id: "SOP-L005".into(),
description: "Data protection and PII handling must comply with policy".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Blocking,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1303-01-AI_Data_Protection.md".into()),
sop_category: Some("ethics".into()),
},
GovernanceRule {
id: "SOP-L006".into(),
description: "Risk register must be maintained and reviewed".into(),
branch: GovernanceBranch::Legislative,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1062-01-AI_Risk_Register.md".into()),
sop_category: Some("governance".into()),
},
GovernanceRule {
id: "SOP-E001".into(),
description: "Secure coding standards must be followed".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1200-01-AI_Secure_Coding.md".into()),
sop_category: Some("engineering".into()),
},
GovernanceRule {
id: "SOP-E002".into(),
description: "Deployment requires governance clearance checkpoint".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Blocking,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1220-01-AI_Deployment_Clearance.md".into()),
sop_category: Some("lifecycle".into()),
},
GovernanceRule {
id: "SOP-E003".into(),
description: "Incident response procedures must be documented and followed".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1008-01-AI_Incident_Response.md".into()),
sop_category: Some("security".into()),
},
GovernanceRule {
id: "SOP-E004".into(),
description: "Decommissioning must follow structured teardown procedure".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Advisory,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1011-01-AI_Decommissioning.md".into()),
sop_category: Some("lifecycle".into()),
},
GovernanceRule {
id: "SOP-E005".into(),
description: "Third-party AI procurement requires screening".into(),
branch: GovernanceBranch::Executive,
severity: RuleSeverity::Advisory,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1004-01-AI_Procurement_Screening.md".into()),
sop_category: Some("governance".into()),
},
GovernanceRule {
id: "SOP-J001".into(),
description: "Bias and fairness assessments required for model outputs".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Blocking,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1301-01-AI_Bias_Fairness.md".into()),
sop_category: Some("ethics".into()),
},
GovernanceRule {
id: "SOP-J002".into(),
description: "Explainability documentation required for decision systems".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1302-01-AI_Explainability.md".into()),
sop_category: Some("ethics".into()),
},
GovernanceRule {
id: "SOP-J003".into(),
description: "Model drift detection and monitoring must be active".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Warning,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-1009-01-AI_Drift_Detection.md".into()),
sop_category: Some("lifecycle".into()),
},
GovernanceRule {
id: "SOP-J004".into(),
description: "Quality records must be maintained for audit compliance".into(),
branch: GovernanceBranch::Judicial,
severity: RuleSeverity::Advisory,
active: true,
reference_url: Some("https://github.com/AISDLC/AI-SDLC-SOPs/blob/main/sops/SOP-2002-01-AI_Quality_Records.md".into()),
sop_category: Some("quality".into()),
},
];
let genesis_seq = cm.sequence();
let rules_json: Vec<serde_json::Value> = genesis_rules.iter().map(|r| {
serde_json::json!({
"id": r.id,
"description": r.description,
"branch": format!("{}", r.branch),
"severity": format!("{}", r.severity),
"reference_url": r.reference_url,
"sop_category": r.sop_category,
})
}).collect();
cm.append(
"governance",
"governance.genesis",
Some(serde_json::json!({
"version": "2.0.0",
"risk_threshold": risk_threshold,
"human_approval_required": human_approval,
"rules": rules_json,
"rule_count": genesis_rules.len(),
"genesis_seq": genesis_seq,
})),
);
for rule in &genesis_rules {
cm.append(
"governance",
"governance.rule",
Some(serde_json::json!({
"rule_id": rule.id,
"branch": format!("{}", rule.branch),
"severity": format!("{}", rule.severity),
"genesis_seq": genesis_seq,
})),
);
}
for rule in genesis_rules {
gate = gate.add_rule(rule);
}
boot_log.push(BootEvent::info(
BootPhase::Services,
format!(
"Governance genesis anchored (seq={}, {} rules, threshold={:.1})",
genesis_seq,
gate.engine().rule_count(),
risk_threshold,
),
));
Some(Arc::new(gate) as Arc<dyn crate::gate::GateBackend>)
} else {
boot_log.push(BootEvent::info(
BootPhase::Services,
"Governance: no chain available, using open governance",
));
None
}
};
#[cfg(feature = "exochain")]
if let Some(ref gate) = governance_gate {
a2a_router.set_gate(Arc::clone(gate));
}
#[cfg(feature = "exochain")]
if let Some(ref cm) = chain_manager {
cm.append(
"kernel",
"boot.cluster",
Some(serde_json::json!({
"node_id": cluster_membership.local_node_id(),
})),
);
let elapsed_ms = boot_time.elapsed().as_millis() as u64;
let mut ready_payload = serde_json::json!({
"elapsed_ms": elapsed_ms,
"processes": process_table.len(),
"services": service_registry.len(),
});
if let Some(ref tm) = tree_manager {
let root_hash = tm.stats().root_hash;
ready_payload
.as_object_mut()
.unwrap()
.insert("tree_root_hash".to_string(), serde_json::json!(root_hash));
}
cm.append("kernel", "boot.ready", Some(ready_payload));
let mut manifest = serde_json::json!({
"version": env!("CARGO_PKG_VERSION"),
"node_id": cluster_membership.local_node_id(),
"process_count": process_table.len(),
"service_count": service_registry.len(),
"chain_sequence": cm.sequence(),
"boot_elapsed_ms": boot_time.elapsed().as_millis() as u64,
});
if let Some(ref tm) = tree_manager {
let stats = tm.stats();
manifest.as_object_mut().unwrap().insert(
"tree_root_hash".to_string(),
serde_json::json!(stats.root_hash),
);
manifest.as_object_mut().unwrap().insert(
"tree_node_count".to_string(),
serde_json::json!(stats.node_count),
);
}
cm.append("kernel", "boot.manifest", Some(manifest));
}
#[cfg(feature = "ecc")]
let (ecc_hnsw, ecc_causal, ecc_tick, ecc_crossrefs, ecc_impulses, ecc_calibration) = {
use crate::calibration::{EccCalibrationConfig, run_calibration};
use crate::causal::CausalGraph;
use crate::cognitive_tick::{CognitiveTick, CognitiveTickConfig};
use crate::crossref::CrossRefStore;
use crate::hnsw_service::{HnswService, HnswServiceConfig};
use crate::impulse::ImpulseQueue;
boot_log.push(BootEvent::info(BootPhase::Ecc, "Initializing ECC cognitive substrate"));
let hnsw = Arc::new(HnswService::new(HnswServiceConfig::default()));
let causal = Arc::new(CausalGraph::new());
let crossrefs = Arc::new(CrossRefStore::new());
let impulses = Arc::new(ImpulseQueue::new());
let cal_config = EccCalibrationConfig::default();
let calibration = run_calibration(&hnsw, &causal, &cal_config);
boot_log.push(BootEvent::info(
BootPhase::Ecc,
format!(
"ECC calibration complete (p50={}us, p95={}us, tick={}ms, spectral={})",
calibration.compute_p50_us,
calibration.compute_p95_us,
calibration.tick_interval_ms,
calibration.spectral_capable,
),
));
let tick_config = CognitiveTickConfig {
tick_interval_ms: calibration.tick_interval_ms,
..CognitiveTickConfig::default()
};
let tick = Arc::new(CognitiveTick::new(tick_config));
if let Err(e) = service_registry.register(hnsw.clone()) {
tracing::debug!(error = %e, "failed to register HNSW service");
}
if let Err(e) = service_registry.register(tick.clone()) {
tracing::debug!(error = %e, "failed to register cognitive tick service");
}
#[cfg(feature = "exochain")]
if let Some(ref cm) = chain_manager {
cm.append(
"ecc",
"ecc.boot.calibration",
Some(serde_json::json!({
"compute_p50_us": calibration.compute_p50_us,
"compute_p95_us": calibration.compute_p95_us,
"tick_interval_ms": calibration.tick_interval_ms,
"spectral_capable": calibration.spectral_capable,
})),
);
}
boot_log.push(BootEvent::info(
BootPhase::Ecc,
format!(
"ECC ready (hnsw={}, causal={} nodes, tick={}ms)",
hnsw.len(),
causal.node_count(),
calibration.tick_interval_ms,
),
));
(
Some(hnsw),
Some(causal),
Some(tick),
Some(crossrefs),
Some(impulses),
Some(calibration),
)
};
#[cfg(feature = "os-patterns")]
let (metrics_registry, log_svc, timer_svc, dead_letter_queue) = {
use crate::dead_letter::DeadLetterQueue;
use crate::log_service::LogService;
use crate::metrics::MetricsRegistry;
use crate::timer::TimerService;
boot_log.push(BootEvent::info(
BootPhase::Services,
"Initializing os-patterns observability modules",
));
let registry = Arc::new(MetricsRegistry::with_builtins());
registry.gauge_set("kernel.process.count", process_table.len() as i64);
registry.gauge_set("kernel.agent.count", 0);
registry.gauge_set("kernel.ipc.messages_sent", 0);
registry.counter_add("kernel.ipc.messages_failed", 0);
registry.gauge_set("kernel.chain.height", 0);
registry.gauge_set("kernel.uptime_secs", 0);
boot_log.push(BootEvent::info(
BootPhase::Services,
"MetricsRegistry ready (built-in + kernel gauges)",
));
let log_svc = Arc::new(LogService::with_default_capacity());
boot_log.push(BootEvent::info(BootPhase::Services, "LogService ready"));
let timer_svc = Arc::new(TimerService::new());
boot_log.push(BootEvent::info(BootPhase::Services, "TimerService ready"));
let dlq = Arc::new(DeadLetterQueue::with_default_capacity());
boot_log.push(BootEvent::info(
BootPhase::Services,
"DeadLetterQueue ready",
));
a2a_router.set_dead_letter_queue(Arc::clone(&dlq));
(Some(registry), Some(log_svc), Some(timer_svc), Some(dlq))
};
let elapsed = boot_time.elapsed();
boot_log.push(BootEvent::info(
BootPhase::Ready,
format!(
"Boot complete in {:.1}s ({} processes, {} services)",
elapsed.as_secs_f64(),
process_table.len(),
service_registry.len(),
),
));
info!(
elapsed_ms = elapsed.as_millis(),
processes = process_table.len(),
services = service_registry.len(),
"kernel boot complete"
);
let supervisor = AgentSupervisor::new(
process_table.clone(),
ipc.clone(),
AgentCapabilities::default(),
);
#[cfg(feature = "native")]
let supervisor = supervisor.with_a2a_router(a2a_router.clone(), cron_svc.clone());
#[cfg(feature = "exochain")]
let supervisor = supervisor.with_exochain(
tree_manager.clone(),
chain_manager.clone(),
);
let event_log = Arc::new(KernelEventLog::new());
event_log.ingest_boot_log(&boot_log);
Ok(Self {
state: KernelState::Running,
config: kernel_config,
app_context: Some(app_context),
bus,
process_table,
service_registry,
ipc,
#[cfg(feature = "native")]
a2a_router,
#[cfg(feature = "native")]
cron_service: cron_svc,
#[cfg(feature = "native")]
assessment_service: assessment_svc,
health,
supervisor,
boot_log,
event_log,
boot_time,
cluster_membership,
#[cfg(feature = "exochain")]
chain: ChainSubsystem {
chain_manager,
tree_manager,
governance_gate,
},
#[cfg(feature = "ecc")]
ecc: EccSubsystem {
hnsw: ecc_hnsw,
causal: ecc_causal,
tick: ecc_tick,
crossrefs: ecc_crossrefs,
impulses: ecc_impulses,
calibration: ecc_calibration,
},
#[cfg(feature = "os-patterns")]
observability: ObservabilitySubsystem {
metrics_registry,
log_service: log_svc,
timer_service: timer_svc,
dead_letter_queue,
},
})
}
pub async fn shutdown(&mut self) -> KernelResult<()> {
if self.state != KernelState::Running {
return Err(KernelError::WrongState {
expected: "Running".into(),
actual: self.state.to_string(),
});
}
info!("kernel shutting down");
self.state = KernelState::ShuttingDown;
self.event_log.info("kernel", "shutdown initiated");
if let Err(e) = self.service_registry.stop_all().await {
error!(error = %e, "error stopping services during shutdown");
}
#[cfg(feature = "exochain")]
if let Some(ref tm) = self.chain.tree_manager
&& let Some(ref cm) = self.chain.chain_manager
{
let stats = tm.stats();
cm.append(
"kernel",
"shutdown",
Some(serde_json::json!({
"tree_root_hash": stats.root_hash,
"chain_seq": cm.sequence(),
"tree_nodes": stats.node_count,
})),
);
}
#[cfg(feature = "native")]
self.supervisor.abort_all();
for entry in self.process_table.list() {
if entry.pid == 0 {
continue; }
entry.cancel_token.cancel();
#[cfg(feature = "exochain")]
if let Some(ref tm) = self.chain.tree_manager {
let _ = tm.unregister_agent(&entry.agent_id, entry.pid, 0);
}
let _ = self
.process_table
.update_state(entry.pid, ProcessState::Exited(0));
}
#[cfg(feature = "exochain")]
if let Some(ref cm) = self.chain.chain_manager {
let chain_config = self.config.chain.clone().unwrap_or_default();
if let Some(ref ckpt_path) = chain_config.effective_checkpoint_path() {
let json_path = std::path::PathBuf::from(ckpt_path);
let rvf_path = json_path.with_extension("rvf");
match cm.save_to_rvf(&rvf_path) {
Ok(()) => info!(path = %rvf_path.display(), "chain saved to RVF checkpoint"),
Err(e) => {
error!(error = %e, "failed to save RVF checkpoint, falling back to JSON");
match cm.save_to_file(&json_path) {
Ok(()) => info!(path = %json_path.display(), "chain saved to JSON checkpoint (fallback)"),
Err(e2) => error!(error = %e2, "failed to save JSON checkpoint fallback"),
}
}
}
if let Some(ref tm) = self.chain.tree_manager {
let tree_path = json_path.with_extension("tree.json");
match tm.save_checkpoint(&tree_path) {
Ok(()) => {
info!(path = %tree_path.display(), "tree checkpoint saved");
cm.append(
"tree",
"tree.checkpoint",
Some(serde_json::json!({
"path": tree_path.display().to_string(),
"root_hash": tm.stats().root_hash,
})),
);
}
Err(e) => error!(error = %e, "failed to save tree checkpoint"),
}
}
}
}
self.state = KernelState::Halted;
self.event_log.info("kernel", "halted");
info!("kernel halted");
Ok(())
}
pub fn state(&self) -> &KernelState {
&self.state
}
pub fn kernel_config(&self) -> &KernelConfig {
&self.config
}
pub fn process_table(&self) -> &Arc<ProcessTable> {
&self.process_table
}
pub fn services(&self) -> &Arc<ServiceRegistry> {
&self.service_registry
}
pub fn ipc(&self) -> &Arc<KernelIpc> {
&self.ipc
}
pub fn bus(&self) -> &Arc<MessageBus> {
&self.bus
}
#[cfg(feature = "native")]
pub fn a2a_router(&self) -> &Arc<A2ARouter> {
&self.a2a_router
}
#[cfg(feature = "native")]
pub fn cron_service(&self) -> &Arc<crate::cron::CronService> {
&self.cron_service
}
#[cfg(feature = "native")]
pub fn assessment_service(&self) -> &Arc<crate::assessment::AssessmentService> {
&self.assessment_service
}
pub fn health(&self) -> &HealthSystem {
&self.health
}
pub fn supervisor(&self) -> &AgentSupervisor<P> {
&self.supervisor
}
pub fn boot_log(&self) -> &BootLog {
&self.boot_log
}
pub fn event_log(&self) -> &Arc<KernelEventLog> {
&self.event_log
}
pub fn uptime(&self) -> std::time::Duration {
self.boot_time.elapsed()
}
pub fn cluster_membership(&self) -> &Arc<ClusterMembership> {
&self.cluster_membership
}
#[cfg(feature = "exochain")]
pub fn chain_manager(&self) -> Option<&Arc<crate::chain::ChainManager>> {
self.chain.chain_manager.as_ref()
}
#[cfg(feature = "exochain")]
pub fn tree_manager(&self) -> Option<&Arc<crate::tree_manager::TreeManager>> {
self.chain.tree_manager.as_ref()
}
#[cfg(feature = "exochain")]
pub fn governance_gate(&self) -> Option<&Arc<dyn crate::gate::GateBackend>> {
self.chain.governance_gate.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_hnsw(&self) -> Option<&Arc<crate::hnsw_service::HnswService>> {
self.ecc.hnsw.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_causal(&self) -> Option<&Arc<crate::causal::CausalGraph>> {
self.ecc.causal.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_tick(&self) -> Option<&Arc<crate::cognitive_tick::CognitiveTick>> {
self.ecc.tick.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_calibration(&self) -> Option<&crate::calibration::EccCalibration> {
self.ecc.calibration.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_crossrefs(&self) -> Option<&Arc<crate::crossref::CrossRefStore>> {
self.ecc.crossrefs.as_ref()
}
#[cfg(feature = "ecc")]
pub fn ecc_impulses(&self) -> Option<&Arc<crate::impulse::ImpulseQueue>> {
self.ecc.impulses.as_ref()
}
#[cfg(feature = "os-patterns")]
pub fn metrics_registry(&self) -> Option<&Arc<crate::metrics::MetricsRegistry>> {
self.observability.metrics_registry.as_ref()
}
#[cfg(feature = "os-patterns")]
pub fn log_service(&self) -> Option<&Arc<crate::log_service::LogService>> {
self.observability.log_service.as_ref()
}
#[cfg(feature = "os-patterns")]
pub fn timer_service(&self) -> Option<&Arc<crate::timer::TimerService>> {
self.observability.timer_service.as_ref()
}
#[cfg(feature = "os-patterns")]
pub fn dead_letter_queue(&self) -> Option<&Arc<crate::dead_letter::DeadLetterQueue>> {
self.observability.dead_letter_queue.as_ref()
}
pub fn take_app_context(&mut self) -> Option<AppContext<P>> {
self.app_context.take()
}
}
#[cfg(test)]
mod tests {
use super::*;
use clawft_platform::NativePlatform;
use clawft_types::config::{AgentDefaults, AgentsConfig};
fn test_config() -> Config {
Config {
agents: AgentsConfig {
defaults: AgentDefaults {
workspace: "~/.clawft/workspace".into(),
model: "test/model".into(),
max_tokens: 1024,
temperature: 0.5,
max_tool_iterations: 5,
memory_window: 10,
},
},
..Config::default()
}
}
fn test_kernel_config() -> KernelConfig {
KernelConfig {
enabled: true,
max_processes: 16,
health_check_interval_secs: 5,
cluster: None,
chain: None,
resource_tree: None,
}
}
#[tokio::test]
async fn boot_and_shutdown() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert_eq!(*kernel.state(), KernelState::Running);
let _uptime = kernel.uptime();
let kernel_proc = kernel.process_table().get(0).unwrap();
assert_eq!(kernel_proc.agent_id, "kernel");
assert_eq!(kernel_proc.state, ProcessState::Running);
kernel.shutdown().await.unwrap();
assert_eq!(*kernel.state(), KernelState::Halted);
}
#[tokio::test]
async fn double_shutdown_fails() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
kernel.shutdown().await.unwrap();
let result = kernel.shutdown().await;
assert!(result.is_err());
}
#[tokio::test]
async fn boot_log_has_events() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let log = kernel.boot_log();
assert!(!log.is_empty());
let formatted = log.format_all();
assert!(formatted.contains("WeftOS v0.1.0"));
assert!(formatted.contains("Boot complete"));
}
#[tokio::test]
async fn process_table_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let pt = kernel.process_table();
assert_eq!(pt.len(), 1); assert_eq!(pt.max_processes(), 16);
}
#[tokio::test]
async fn services_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let count = kernel.services().len();
#[cfg(all(feature = "ecc", feature = "cluster"))]
assert_eq!(count, 6, "expected cron+containers+assess+cluster+hnsw+cognitive_tick");
#[cfg(all(feature = "ecc", not(feature = "cluster")))]
assert_eq!(count, 5, "expected cron+containers+assess+hnsw+cognitive_tick");
#[cfg(all(not(feature = "ecc"), feature = "cluster"))]
assert_eq!(count, 4, "expected cron+containers+assess+cluster");
#[cfg(all(not(feature = "ecc"), not(feature = "cluster")))]
assert_eq!(count, 3, "expected cron+containers+assess");
}
#[tokio::test]
async fn take_app_context() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let ctx = kernel.take_app_context();
assert!(ctx.is_some());
let ctx2 = kernel.take_app_context();
assert!(ctx2.is_none());
}
#[tokio::test]
async fn ipc_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let ipc = kernel.ipc();
assert!(Arc::ptr_eq(ipc.bus(), kernel.bus()));
}
#[test]
fn kernel_state_display() {
assert_eq!(KernelState::Booting.to_string(), "booting");
assert_eq!(KernelState::Running.to_string(), "running");
assert_eq!(KernelState::ShuttingDown.to_string(), "shutting_down");
assert_eq!(KernelState::Halted.to_string(), "halted");
}
#[cfg(all(feature = "exochain", feature = "ecc", feature = "wasm-sandbox"))]
fn test_kernel_config_full_stack() -> KernelConfig {
use clawft_types::config::{ChainConfig, ResourceTreeConfig};
KernelConfig {
enabled: true,
max_processes: 32,
health_check_interval_secs: 5,
cluster: None,
chain: Some(ChainConfig {
enabled: true,
checkpoint_interval: 10_000,
chain_id: 0,
checkpoint_path: None,
}),
resource_tree: Some(ResourceTreeConfig {
enabled: true,
checkpoint_path: None,
}),
}
}
#[tokio::test]
#[cfg(all(feature = "exochain", feature = "ecc", feature = "wasm-sandbox"))]
async fn integration_full_stack_kernel() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_full_stack(), platform)
.await
.unwrap();
assert_eq!(*kernel.state(), KernelState::Running);
let svc_count = kernel.services().len();
assert!(svc_count >= 4, "expected >= 4 services, got {svc_count}");
let spawn_result = kernel.supervisor().spawn(
crate::supervisor::SpawnRequest {
agent_id: "integration-agent-1".into(),
capabilities: None,
parent_pid: None,
env: std::collections::HashMap::new(),
backend: None, },
);
assert!(spawn_result.is_ok(), "native agent spawn failed: {:?}", spawn_result.err());
let agent_pid = spawn_result.unwrap().pid;
assert!(agent_pid > 0, "agent should get PID > 0");
let processes = kernel.process_table().list();
assert!(
processes.iter().any(|p| p.pid == agent_pid),
"agent not in process table"
);
let a2a = kernel.a2a_router();
let _inbox = a2a.create_inbox(agent_pid);
let msg = crate::ipc::KernelMessage::new(
0, crate::ipc::MessageTarget::Process(agent_pid),
crate::ipc::MessagePayload::Json(serde_json::json!({"cmd": "ping"})),
);
let send_result = a2a.send(msg).await;
assert!(send_result.is_ok(), "A2A send failed: {:?}", send_result.err());
let wasm_config = crate::wasm_runner::WasmSandboxConfig::default();
let wasm_runner = crate::wasm_runner::WasmToolRunner::new(wasm_config);
let noop_wat = r#"(module (func (export "_start")))"#;
let result = wasm_runner
.execute_bytes("integration-tool", noop_wat.as_bytes(), serde_json::json!({}))
.await;
assert!(result.is_ok(), "WASM execution failed: {:?}", result.err());
let wasm_result = result.unwrap();
assert_eq!(wasm_result.exit_code, 0, "WASM tool should exit cleanly");
assert!(wasm_result.fuel_consumed > 0, "WASM should consume fuel");
let wasm_runner_arc = Arc::new(
crate::wasm_runner::WasmToolRunner::new(
crate::wasm_runner::WasmSandboxConfig::default(),
),
);
let mut registry = crate::wasm_runner::ToolRegistry::new();
registry
.register_wasm_tool(
"demo-tool",
"A demo WASM tool for integration testing",
noop_wat.as_bytes().to_vec(),
wasm_runner_arc,
)
.unwrap();
assert!(registry.get("demo-tool").is_some(), "WASM tool should be in registry");
let tool_list = registry.list();
assert!(
tool_list.contains(&"demo-tool".to_string()),
"WASM tool should appear in listing"
);
let svc_list = kernel.services().list();
assert!(
svc_list.iter().any(|(name, _)| name == "containers"),
"container service should be registered"
);
let ecc_hnsw = kernel.ecc_hnsw().expect("HNSW service should be present");
assert_eq!(ecc_hnsw.len(), 0, "HNSW should be empty after calibration cleanup");
let ecc_causal = kernel.ecc_causal().expect("causal graph should be present");
assert_eq!(ecc_causal.node_count(), 0, "causal graph should be empty after cleanup");
let _ecc_tick = kernel.ecc_tick().expect("cognitive tick should be present");
let calibration = kernel.ecc_calibration().expect("calibration should exist");
assert!(calibration.compute_p95_us > 0, "calibration should have run");
assert!(calibration.tick_interval_ms > 0, "tick interval should be auto-calibrated");
ecc_hnsw.insert(
"test-doc".into(),
vec![1.0, 0.0, 0.0, 0.0],
serde_json::json!({"text": "hello"}),
);
assert_eq!(ecc_hnsw.len(), 1);
let results = ecc_hnsw.search(&[1.0, 0.0, 0.0, 0.0], 5);
assert_eq!(results.len(), 1);
assert_eq!(results[0].id, "test-doc");
let node1 = ecc_causal.add_node("concept-A".into(), serde_json::json!({}));
let node2 = ecc_causal.add_node("concept-B".into(), serde_json::json!({}));
ecc_causal.link(
node1,
node2,
crate::causal::CausalEdgeType::Causes,
1.0,
0,
0,
);
assert_eq!(ecc_causal.node_count(), 2);
assert_eq!(ecc_causal.edge_count(), 1);
let path = ecc_causal.find_path(node1, node2, 5);
assert!(path.is_some(), "should find path between linked nodes");
let chain = kernel.chain_manager().expect("chain should exist");
assert!(
chain.sequence() > 5,
"chain should have boot events, got seq={}",
chain.sequence()
);
let verify_result = chain.verify_integrity();
assert!(verify_result.valid, "chain should be valid: {:?}", verify_result.errors);
assert!(verify_result.errors.is_empty(), "no integrity errors");
let _governance = kernel.governance_gate().expect("governance gate should exist");
let all_events = chain.tail(0);
let genesis_events: Vec<_> = all_events.iter()
.filter(|e| e.kind == "governance.genesis")
.collect();
assert!(!genesis_events.is_empty(), "governance genesis should be on chain");
let genesis_payload = genesis_events.iter()
.filter_map(|e| e.payload.as_ref())
.find(|p| p.get("version").and_then(|v| v.as_str()) == Some("2.0.0"))
.expect("should find v2.0.0 governance genesis on chain");
assert_eq!(
genesis_payload["rule_count"].as_u64().unwrap(),
22,
"genesis should contain 22 rules"
);
assert_eq!(
genesis_payload["version"].as_str().unwrap(),
"2.0.0",
"genesis version should be 2.0.0"
);
let rule_events: Vec<_> = all_events.iter()
.filter(|e| e.kind == "governance.rule")
.collect();
assert!(
rule_events.len() >= 22,
"at least 22 genesis rules should be individually anchored, got {}",
rule_events.len(),
);
let rule_ids: Vec<&str> = rule_events.iter()
.filter_map(|e| e.payload.as_ref()?.get("rule_id")?.as_str())
.collect();
for expected_id in &[
"GOV-001", "GOV-002", "GOV-003", "GOV-004", "GOV-005", "GOV-006", "GOV-007",
"SOP-L001", "SOP-L002", "SOP-L003", "SOP-L004", "SOP-L005", "SOP-L006",
"SOP-E001", "SOP-E002", "SOP-E003", "SOP-E004", "SOP-E005",
"SOP-J001", "SOP-J002", "SOP-J003", "SOP-J004",
] {
assert!(
rule_ids.contains(expected_id),
"{expected_id} should be anchored on chain"
);
}
let ecc_events: Vec<_> = all_events.iter().filter(|e| e.kind.starts_with("ecc.")).collect();
assert!(
!ecc_events.is_empty(),
"ECC boot calibration should be logged to chain"
);
let tree = kernel.tree_manager().expect("tree should exist");
let stats = tree.stats();
assert!(
stats.node_count > 20,
"tree should have many nodes (got {})",
stats.node_count
);
assert!(kernel.services().get("cron").is_some(), "cron service should be accessible");
assert!(
kernel.services().get("containers").is_some(),
"container service should be accessible"
);
let mut kernel = kernel;
kernel.shutdown().await.unwrap();
assert_eq!(*kernel.state(), KernelState::Halted);
}
#[test]
#[cfg(feature = "wasm-sandbox")]
fn integration_cross_backend_tools() {
use crate::wasm_runner::{
BuiltinTool, BuiltinToolSpec, ToolCategory, ToolError, ToolRegistry,
WasmSandboxConfig, WasmToolRunner,
};
use crate::governance::EffectVector;
struct EchoTool;
impl BuiltinTool for EchoTool {
fn name(&self) -> &str {
"native.echo"
}
fn spec(&self) -> &BuiltinToolSpec {
static SPEC: std::sync::OnceLock<BuiltinToolSpec> = std::sync::OnceLock::new();
SPEC.get_or_init(|| BuiltinToolSpec {
name: "native.echo".into(),
category: ToolCategory::System,
description: "Echoes input back".into(),
parameters: serde_json::json!({}),
gate_action: "tool.native.echo".into(),
effect: EffectVector::default(),
native: true,
})
}
fn execute(&self, args: serde_json::Value) -> Result<serde_json::Value, ToolError> {
Ok(serde_json::json!({"echo": args}))
}
}
let mut registry = ToolRegistry::new();
registry.register(Arc::new(EchoTool));
let runner = Arc::new(WasmToolRunner::new(WasmSandboxConfig::default()));
let noop_wat = r#"(module (func (export "_start")))"#;
registry
.register_wasm_tool(
"wasm.noop",
"A no-op WASM tool",
noop_wat.as_bytes().to_vec(),
runner,
)
.unwrap();
assert_eq!(registry.list().len(), 2);
let native = registry.get("native.echo").expect("native tool should exist");
assert!(native.spec().native, "native tool should be marked native");
let wasm = registry.get("wasm.noop").expect("wasm tool should exist");
assert!(!wasm.spec().native, "wasm tool should NOT be marked native");
let result = native.execute(serde_json::json!({"hello": "world"})).unwrap();
assert_eq!(result["echo"]["hello"], "world");
let parent = Arc::new(registry);
let mut child = ToolRegistry::with_parent(parent);
assert!(child.get("native.echo").is_some(), "child sees parent native tool");
assert!(child.get("wasm.noop").is_some(), "child sees parent wasm tool");
struct OverrideTool;
impl BuiltinTool for OverrideTool {
fn name(&self) -> &str {
"native.echo"
}
fn spec(&self) -> &BuiltinToolSpec {
static SPEC: std::sync::OnceLock<BuiltinToolSpec> = std::sync::OnceLock::new();
SPEC.get_or_init(|| BuiltinToolSpec {
name: "native.echo".into(),
category: ToolCategory::System,
description: "Overridden echo".into(),
parameters: serde_json::json!({}),
gate_action: "tool.native.echo".into(),
effect: EffectVector::default(),
native: true,
})
}
fn execute(&self, _args: serde_json::Value) -> Result<serde_json::Value, ToolError> {
Ok(serde_json::json!({"overridden": true}))
}
}
child.register(Arc::new(OverrideTool));
let result = child.get("native.echo").unwrap().execute(serde_json::json!({})).unwrap();
assert_eq!(result["overridden"], true, "child should shadow parent tool");
assert!(child.get("wasm.noop").is_some(), "WASM tool still reachable via parent");
}
#[tokio::test]
async fn boot_reaches_running_state() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert_eq!(*kernel.state(), KernelState::Running);
}
#[tokio::test]
async fn boot_registers_cron_service() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.services().get("cron").is_some(),
"cron service must be registered at boot"
);
}
#[tokio::test]
async fn boot_registers_container_service() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.services().get("containers").is_some(),
"container service must be registered at boot"
);
}
#[cfg(feature = "exochain")]
fn test_kernel_config_exochain() -> KernelConfig {
use clawft_types::config::{ChainConfig, ResourceTreeConfig};
KernelConfig {
enabled: true,
max_processes: 16,
health_check_interval_secs: 5,
cluster: None,
chain: Some(ChainConfig {
enabled: true,
checkpoint_interval: 10_000,
chain_id: 0,
checkpoint_path: None,
}),
resource_tree: Some(ResourceTreeConfig {
enabled: true,
checkpoint_path: None,
}),
}
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn boot_exochain_creates_chain_manager() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
assert!(
kernel.chain_manager().is_some(),
"chain manager must be present with exochain feature and enabled chain config"
);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn boot_exochain_creates_tree_manager() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
assert!(
kernel.tree_manager().is_some(),
"tree manager must be present with exochain feature and enabled config"
);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn boot_exochain_chain_has_boot_events() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
let chain = kernel.chain_manager().unwrap();
assert!(
chain.sequence() >= 6,
"chain should have at least 6 boot events, got {}",
chain.sequence()
);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn boot_exochain_governance_gate_present() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
assert!(
kernel.governance_gate().is_some(),
"governance gate should be present when chain is enabled"
);
}
#[cfg(feature = "ecc")]
#[tokio::test]
async fn boot_ecc_registers_hnsw_service() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.ecc_hnsw().is_some(),
"HNSW service must be present with ecc feature"
);
}
#[cfg(feature = "ecc")]
#[tokio::test]
async fn boot_ecc_registers_cognitive_tick() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.ecc_tick().is_some(),
"cognitive tick must be present with ecc feature"
);
}
#[cfg(feature = "ecc")]
#[tokio::test]
async fn boot_ecc_calibration_has_valid_results() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let cal = kernel.ecc_calibration().unwrap();
assert!(cal.tick_interval_ms > 0, "tick interval must be positive");
assert!(cal.compute_p50_us > 0, "p50 latency must be measured");
assert!(cal.compute_p95_us >= cal.compute_p50_us, "p95 >= p50");
}
#[cfg(feature = "ecc")]
#[tokio::test]
async fn boot_ecc_causal_graph_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(kernel.ecc_causal().is_some(), "causal graph must be accessible");
assert_eq!(kernel.ecc_causal().unwrap().node_count(), 0, "causal graph starts empty");
}
#[cfg(feature = "ecc")]
#[tokio::test]
async fn boot_ecc_crossrefs_and_impulses_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(kernel.ecc_crossrefs().is_some(), "cross-ref store must be accessible");
assert!(kernel.ecc_impulses().is_some(), "impulse queue must be accessible");
}
#[tokio::test]
async fn boot_cluster_membership_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let cm = kernel.cluster_membership();
assert!(!cm.local_node_id().is_empty(), "cluster membership should have a node ID");
}
#[tokio::test]
async fn shutdown_transitions_to_halted() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert_eq!(*kernel.state(), KernelState::Running);
kernel.shutdown().await.unwrap();
assert_eq!(*kernel.state(), KernelState::Halted);
}
#[tokio::test]
async fn shutdown_from_halted_fails() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
kernel.shutdown().await.unwrap();
let err = kernel.shutdown().await.unwrap_err();
match err {
KernelError::WrongState { expected, actual } => {
assert_eq!(expected, "Running");
assert_eq!(actual, "halted");
}
other => panic!("expected WrongState, got: {other:?}"),
}
}
#[tokio::test]
async fn process_table_has_kernel_pid_zero() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let entry = kernel.process_table().get(0).expect("PID 0 should exist");
assert_eq!(entry.agent_id, "kernel");
assert_eq!(entry.state, ProcessState::Running);
assert_eq!(entry.pid, 0);
}
#[tokio::test]
async fn a2a_router_accessible_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let a2a = kernel.a2a_router();
let _inbox = a2a.create_inbox(0); }
#[tokio::test]
async fn boot_log_contains_expected_phases() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let formatted = kernel.boot_log().format_all();
assert!(formatted.contains("WeftOS v0.1.0"), "should contain version");
assert!(formatted.contains("Service registry ready"), "should have service phase");
assert!(formatted.contains("A2A router ready"), "should have A2A phase");
assert!(formatted.contains("Boot complete"), "should have ready phase");
}
#[tokio::test]
async fn event_log_populated_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let events = kernel.event_log();
assert!(!events.is_empty(), "event log should have entries after boot");
}
#[tokio::test]
async fn health_system_accessible() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _health = kernel.health();
}
#[tokio::test]
async fn uptime_is_positive_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let uptime = kernel.uptime();
assert!(uptime.as_nanos() > 0, "uptime should be positive");
}
#[tokio::test]
async fn cron_service_accessible_and_empty() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert_eq!(kernel.cron_service().job_count(), 0, "no cron jobs at boot");
}
#[tokio::test]
async fn max_processes_matches_config() {
let platform = Arc::new(NativePlatform::new());
let mut kconfig = test_kernel_config();
kconfig.max_processes = 42;
let kernel = Kernel::boot(test_config(), kconfig, platform)
.await
.unwrap();
assert_eq!(kernel.process_table().max_processes(), 42);
}
#[test]
fn kernel_state_serde_roundtrip_all_variants() {
for state in [
KernelState::Booting,
KernelState::Running,
KernelState::ShuttingDown,
KernelState::Halted,
] {
let json = serde_json::to_string(&state).unwrap();
let restored: KernelState = serde_json::from_str(&json).unwrap();
assert_eq!(restored, state);
}
}
#[test]
fn kernel_state_equality() {
assert_eq!(KernelState::Running, KernelState::Running);
assert_ne!(KernelState::Running, KernelState::Halted);
}
#[tokio::test]
async fn kernel_state_is_running_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert_eq!(kernel.state(), &KernelState::Running);
}
#[tokio::test]
async fn boot_log_has_entries() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let log = kernel.boot_log();
assert!(log.events().len() > 0, "boot log should have events");
}
#[tokio::test]
async fn bus_is_accessible_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _bus = kernel.bus();
}
#[tokio::test]
async fn a2a_router_has_inboxes_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _router = kernel.a2a_router();
}
#[tokio::test]
async fn ipc_accessible_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _ipc = kernel.ipc();
}
#[tokio::test]
async fn supervisor_accessible_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _supervisor = kernel.supervisor();
}
#[tokio::test]
async fn kernel_config_accessor_returns_boot_config() {
let platform = Arc::new(NativePlatform::new());
let mut kconfig = test_kernel_config();
kconfig.max_processes = 99;
kconfig.health_check_interval_secs = 42;
let kernel = Kernel::boot(test_config(), kconfig, platform)
.await
.unwrap();
assert_eq!(kernel.kernel_config().max_processes, 99);
assert_eq!(kernel.kernel_config().health_check_interval_secs, 42);
}
#[tokio::test]
async fn shutdown_exits_all_agent_processes() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let spawn1 = kernel.supervisor().spawn(crate::supervisor::SpawnRequest {
agent_id: "test-agent-1".into(),
capabilities: None,
parent_pid: None,
env: std::collections::HashMap::new(),
backend: None,
});
let spawn2 = kernel.supervisor().spawn(crate::supervisor::SpawnRequest {
agent_id: "test-agent-2".into(),
capabilities: None,
parent_pid: None,
env: std::collections::HashMap::new(),
backend: None,
});
assert!(spawn1.is_ok());
assert!(spawn2.is_ok());
let pid1 = spawn1.unwrap().pid;
let pid2 = spawn2.unwrap().pid;
assert!(kernel.process_table().get(pid1).is_some());
assert!(kernel.process_table().get(pid2).is_some());
kernel.shutdown().await.unwrap();
if let Some(entry1) = kernel.process_table().get(pid1) {
assert!(
!matches!(entry1.state, ProcessState::Running),
"agent 1 should not be Running after shutdown, got: {}",
entry1.state
);
}
if let Some(entry2) = kernel.process_table().get(pid2) {
assert!(
!matches!(entry2.state, ProcessState::Running),
"agent 2 should not be Running after shutdown, got: {}",
entry2.state
);
}
}
#[tokio::test]
async fn shutdown_stops_services() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let svc_count_before = kernel.services().len();
assert!(svc_count_before > 0, "should have services before shutdown");
kernel.shutdown().await.unwrap();
assert_eq!(*kernel.state(), KernelState::Halted);
}
#[tokio::test]
async fn boot_with_custom_max_processes() {
let platform = Arc::new(NativePlatform::new());
let mut kconfig = test_kernel_config();
kconfig.max_processes = 128;
let kernel = Kernel::boot(test_config(), kconfig, platform)
.await
.unwrap();
assert_eq!(kernel.process_table().max_processes(), 128);
}
#[tokio::test]
async fn boot_creates_a2a_router_with_topic_router() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let _topic_router = kernel.a2a_router().topic_router();
}
#[tokio::test]
async fn event_log_has_shutdown_events() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let events_before = kernel.event_log().len();
kernel.shutdown().await.unwrap();
let events_after = kernel.event_log().len();
assert!(
events_after > events_before,
"event log should grow during shutdown (before={events_before}, after={events_after})"
);
}
#[tokio::test]
async fn boot_log_contains_max_processes() {
let platform = Arc::new(NativePlatform::new());
let mut kconfig = test_kernel_config();
kconfig.max_processes = 64;
let kernel = Kernel::boot(test_config(), kconfig, platform)
.await
.unwrap();
let formatted = kernel.boot_log().format_all();
assert!(
formatted.contains("Max processes: 64"),
"boot log should contain max_processes config value"
);
}
#[tokio::test]
async fn boot_log_contains_health_interval() {
let platform = Arc::new(NativePlatform::new());
let mut kconfig = test_kernel_config();
kconfig.health_check_interval_secs = 15;
let kernel = Kernel::boot(test_config(), kconfig, platform)
.await
.unwrap();
let formatted = kernel.boot_log().format_all();
assert!(
formatted.contains("Health check interval: 15s"),
"boot log should contain health check interval"
);
}
#[tokio::test]
async fn take_app_context_returns_some_then_none() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(kernel.take_app_context().is_some(), "first take should succeed");
assert!(kernel.take_app_context().is_none(), "second take should return None");
assert!(kernel.take_app_context().is_none(), "third take should also return None");
}
#[tokio::test]
async fn kernel_pid_zero_has_default_capabilities() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let entry = kernel.process_table().get(0).unwrap();
assert_eq!(entry.agent_id, "kernel");
assert!(entry.parent_pid.is_none(), "kernel should have no parent");
assert!(entry.capabilities.can_ipc, "kernel should be able to IPC");
}
#[tokio::test]
async fn cluster_membership_has_nonempty_node_id() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let node_id = kernel.cluster_membership().local_node_id();
assert!(!node_id.is_empty(), "cluster node ID should not be empty");
assert_eq!(node_id.len(), 36, "node ID should be UUID format");
}
#[tokio::test]
async fn boot_log_contains_cluster_info() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let formatted = kernel.boot_log().format_all();
assert!(
formatted.contains("Cluster membership ready"),
"boot log should mention cluster membership"
);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn shutdown_with_exochain_completes() {
let platform = Arc::new(NativePlatform::new());
let mut kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
assert!(kernel.chain_manager().is_some());
kernel.shutdown().await.unwrap();
assert_eq!(*kernel.state(), KernelState::Halted);
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn exochain_chain_integrity_valid_after_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
let chain = kernel.chain_manager().unwrap();
let result = chain.verify_integrity();
assert!(result.valid, "chain integrity should be valid after boot");
assert!(result.errors.is_empty(), "no integrity errors expected");
}
#[cfg(feature = "exochain")]
#[tokio::test]
async fn exochain_boot_manifest_event_present() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config_exochain(), platform)
.await
.unwrap();
let chain = kernel.chain_manager().unwrap();
let all_events = chain.tail(0);
let manifest_events: Vec<_> = all_events
.iter()
.filter(|e| e.kind == "boot.manifest")
.collect();
assert!(
!manifest_events.is_empty(),
"boot.manifest event should be on chain"
);
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn boot_creates_metrics_registry() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let registry = kernel.metrics_registry();
assert!(registry.is_some(), "MetricsRegistry should be created at boot");
let r = registry.unwrap();
let val = r.gauge_get("kernel.process.count");
assert!(val >= 1, "kernel.process.count gauge should be seeded");
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn boot_creates_log_service() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.log_service().is_some(),
"LogService should be created at boot"
);
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn boot_creates_timer_service() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
assert!(
kernel.timer_service().is_some(),
"TimerService should be created at boot"
);
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn boot_creates_dead_letter_queue() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let dlq = kernel.dead_letter_queue();
assert!(dlq.is_some(), "DeadLetterQueue should be created at boot");
assert!(dlq.unwrap().is_empty(), "DLQ should start empty");
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn kernel_metrics_accessor_returns_registry() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let registry = kernel.metrics_registry().unwrap();
registry.counter_inc(crate::metrics::METRIC_MESSAGES_SENT);
assert_eq!(registry.counter_get(crate::metrics::METRIC_MESSAGES_SENT), 1);
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn dlq_accessible_via_kernel_accessor() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let dlq = kernel.dead_letter_queue().unwrap();
assert_eq!(dlq.capacity(), crate::dead_letter::DEFAULT_DLQ_CAPACITY);
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn failed_a2a_send_routes_to_dlq() {
use crate::ipc::{KernelMessage, MessageTarget};
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let sender_entry = ProcessEntry {
pid: 0,
agent_id: "dlq-sender".to_owned(),
state: ProcessState::Running,
capabilities: AgentCapabilities::default(),
resource_usage: ResourceUsage::default(),
cancel_token: crate::process::CancellationToken::new(),
parent_pid: None,
};
let sender_pid = kernel.process_table().insert(sender_entry).unwrap();
kernel.a2a_router().create_inbox(sender_pid);
let msg = KernelMessage::text(
sender_pid,
MessageTarget::Process(9999),
"dead letter test",
);
let result = kernel.a2a_router().send(msg).await;
assert!(result.is_err(), "send to unknown PID should fail");
let dlq = kernel.dead_letter_queue().unwrap();
assert_eq!(dlq.len(), 1, "DLQ should contain 1 dead letter");
}
#[tokio::test]
#[cfg(feature = "os-patterns")]
async fn kernel_metrics_gauge_seeded_at_boot() {
let platform = Arc::new(NativePlatform::new());
let kernel = Kernel::boot(test_config(), test_kernel_config(), platform)
.await
.unwrap();
let registry = kernel.metrics_registry().unwrap();
assert!(
registry.gauge_get("kernel.process.count") >= 1,
"kernel.process.count should be >= 1 at boot"
);
assert_eq!(
registry.gauge_get("kernel.uptime_secs"),
0,
"kernel.uptime_secs gauge should exist and be 0 at boot"
);
assert_eq!(
registry.gauge_get("kernel.chain.height"),
0,
"kernel.chain.height gauge should exist and be 0 at boot"
);
}
}