use std::sync::Arc;
use tracing::{debug, info};
use clawft_platform::Platform;
use clawft_types::config::Config;
use crate::agent::context::ContextBuilder;
use crate::agent::loop_core::{AgentLoop, AutoDelegation};
use crate::agent::memory::MemoryStore;
use crate::agent::skills::SkillsLoader;
use crate::bus::MessageBus;
use crate::pipeline::assembler::TokenBudgetAssembler;
use crate::pipeline::classifier::KeywordClassifier;
use crate::pipeline::cost_tracker::CostTracker;
use crate::pipeline::rate_limiter::RateLimiter;
use crate::pipeline::router::StaticRouter;
use crate::pipeline::tiered_router::TieredRouter;
use crate::pipeline::traits::{ModelRouter, Pipeline, PipelineRegistry};
use crate::pipeline::transport::OpenAiCompatTransport;
use crate::session::SessionManager;
use crate::tools::registry::ToolRegistry;
pub struct AppContext<P: Platform> {
config: Config,
platform: Arc<P>,
bus: Arc<MessageBus>,
sessions: Arc<SessionManager<P>>,
tools: Arc<ToolRegistry>,
pipeline: PipelineRegistry,
context: ContextBuilder<P>,
memory: Arc<MemoryStore<P>>,
skills: Arc<SkillsLoader<P>>,
auto_delegation: Option<Arc<dyn AutoDelegation>>,
}
impl<P: Platform> AppContext<P> {
pub async fn new(config: Config, platform: Arc<P>) -> clawft_types::Result<Self> {
info!("bootstrapping application context");
let bus = Arc::new(MessageBus::new());
debug!("message bus created");
let sessions = SessionManager::new(platform.clone()).await?;
debug!("session manager initialized");
let memory = Arc::new(MemoryStore::new(platform.clone())?);
debug!(
memory_path = %memory.memory_path().display(),
"memory store initialized"
);
let mut skills_loader = SkillsLoader::new(platform.clone())?;
for candidate in [
config.workspace_path().join("skills"),
std::env::current_dir().unwrap_or_default().join("skills"),
] {
if candidate.is_dir() {
skills_loader.add_extra_dir(candidate);
}
}
let skills = Arc::new(skills_loader);
debug!(
skills_dir = %skills.skills_dir().display(),
"skills loader initialized"
);
let context = ContextBuilder::new(
config.agents.clone(),
memory.clone(),
skills.clone(),
platform.clone(),
);
debug!("context builder created");
let tools = Arc::new(ToolRegistry::new());
let pipeline = build_default_pipeline(&config);
debug!("default pipeline wired");
info!("bootstrap complete");
Ok(Self {
config,
platform,
bus,
sessions: Arc::new(sessions),
tools,
pipeline,
context,
memory,
skills,
auto_delegation: None,
})
}
pub fn into_agent_loop(self) -> AgentLoop<P> {
let resolver = crate::pipeline::permissions::PermissionResolver::new(
&self.config.routing,
None, );
let mut agent = AgentLoop::new(
self.config.agents,
self.platform,
self.bus,
self.pipeline,
self.tools.clone(),
self.context,
self.sessions.clone(),
resolver,
);
if let Some(delegation) = self.auto_delegation {
agent = agent.with_auto_delegation(delegation);
}
agent
}
pub fn config(&self) -> &Config {
&self.config
}
pub fn platform(&self) -> &Arc<P> {
&self.platform
}
pub fn bus(&self) -> &Arc<MessageBus> {
&self.bus
}
pub fn tools_mut(&mut self) -> &mut ToolRegistry {
Arc::get_mut(&mut self.tools).expect("tools already shared -- register tools before cloning Arc")
}
pub fn tools(&self) -> &Arc<ToolRegistry> {
&self.tools
}
pub fn tools_arc(&self) -> Arc<ToolRegistry> {
self.tools.clone()
}
pub fn sessions(&self) -> &Arc<SessionManager<P>> {
&self.sessions
}
pub fn memory(&self) -> &Arc<MemoryStore<P>> {
&self.memory
}
pub fn skills(&self) -> &Arc<SkillsLoader<P>> {
&self.skills
}
pub fn set_auto_delegation(&mut self, delegation: Arc<dyn AutoDelegation>) {
self.auto_delegation = Some(delegation);
}
pub fn set_pipeline(&mut self, pipeline: PipelineRegistry) {
self.pipeline = pipeline;
}
#[cfg(feature = "native")]
pub fn enable_live_llm(&mut self) {
let pipeline = build_live_pipeline(&self.config);
self.set_pipeline(pipeline);
}
}
#[cfg(feature = "native")]
pub fn build_live_pipeline(config: &Config) -> PipelineRegistry {
crate::pipeline::llm_adapter::build_live_pipeline(config)
}
fn build_default_pipeline(config: &Config) -> PipelineRegistry {
let classifier = Arc::new(KeywordClassifier::new());
let router: Arc<dyn ModelRouter> = build_router_from_config(config);
let assembler = Arc::new(TokenBudgetAssembler::new(
config.agents.defaults.max_tokens.max(1) as usize,
));
let transport = Arc::new(OpenAiCompatTransport::new());
let scorer = crate::pipeline::build_scorer(&config.pipeline);
let learner = crate::pipeline::build_learner(&config.pipeline);
let pipeline = Pipeline {
classifier,
router,
assembler,
transport,
scorer,
learner,
};
PipelineRegistry::new(pipeline)
}
fn build_router_from_config(config: &Config) -> Arc<dyn ModelRouter> {
if config.routing.mode == "tiered" {
let routing = config.routing.clone();
let cost_tracker = Arc::new(
CostTracker::new(routing.cost_budgets.reset_hour_utc),
);
let rate_limiter = Arc::new(
RateLimiter::new(routing.rate_limiting.window_seconds, routing.rate_limiting.global_rate_limit_rpm),
);
Arc::new(
TieredRouter::new(routing)
.with_cost_tracker(cost_tracker)
.with_rate_limiter(rate_limiter),
)
} else {
Arc::new(StaticRouter::from_config(&config.agents))
}
}
#[cfg(test)]
mod tests {
use super::*;
use clawft_platform::NativePlatform;
use clawft_types::config::{AgentDefaults, AgentsConfig};
use std::sync::Arc;
fn test_config() -> Config {
Config {
agents: AgentsConfig {
defaults: AgentDefaults {
workspace: "~/.clawft/workspace".into(),
model: "deepseek/deepseek-chat".into(),
max_tokens: 4096,
temperature: 0.7,
max_tool_iterations: 10,
memory_window: 50,
},
},
..Config::default()
}
}
#[tokio::test]
async fn new_creates_app_context() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await;
assert!(ctx.is_ok());
}
#[tokio::test]
async fn config_accessor() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
assert_eq!(
ctx.config().agents.defaults.model,
"deepseek/deepseek-chat"
);
assert_eq!(ctx.config().agents.defaults.max_tokens, 4096);
}
#[tokio::test]
async fn platform_accessor() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let _p = ctx.platform();
}
#[tokio::test]
async fn bus_accessor() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let bus = ctx.bus();
let _tx = bus.inbound_sender();
}
#[tokio::test]
async fn tools_starts_empty() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
assert!(ctx.tools().is_empty());
}
#[tokio::test]
async fn tools_mut_allows_registration() {
let platform = Arc::new(NativePlatform::new());
let mut ctx = AppContext::new(test_config(), platform).await.unwrap();
use crate::tools::registry::Tool;
use async_trait::async_trait;
struct DummyTool;
#[async_trait]
impl Tool for DummyTool {
fn name(&self) -> &str {
"dummy"
}
fn description(&self) -> &str {
"A dummy tool"
}
fn parameters(&self) -> serde_json::Value {
serde_json::json!({"type": "object", "properties": {}})
}
async fn execute(
&self,
_args: serde_json::Value,
) -> Result<serde_json::Value, crate::tools::registry::ToolError> {
Ok(serde_json::json!({}))
}
}
ctx.tools_mut().register(Arc::new(DummyTool));
assert_eq!(ctx.tools().len(), 1);
assert_eq!(ctx.tools().list(), vec!["dummy"]);
}
#[tokio::test]
async fn memory_accessor() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let memory = ctx.memory();
assert!(memory.memory_path().is_absolute());
}
#[tokio::test]
async fn skills_accessor() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let skills = ctx.skills();
assert!(skills.skills_dir().is_absolute());
}
#[tokio::test]
async fn into_agent_loop_produces_valid_loop() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let agent = ctx.into_agent_loop();
assert_eq!(agent.config().defaults.model, "deepseek/deepseek-chat");
assert_eq!(agent.config().defaults.max_tokens, 4096);
}
#[tokio::test]
async fn set_pipeline_replaces_default() {
let platform = Arc::new(NativePlatform::new());
let mut ctx = AppContext::new(test_config(), platform).await.unwrap();
let custom_pipeline = build_default_pipeline(&test_config());
ctx.set_pipeline(custom_pipeline);
let agent = ctx.into_agent_loop();
assert_eq!(agent.config().defaults.model, "deepseek/deepseek-chat");
}
#[tokio::test]
async fn default_config_bootstrap() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(Config::default(), platform).await;
assert!(ctx.is_ok());
}
#[test]
fn build_default_pipeline_creates_registry() {
let config = test_config();
let _registry = build_default_pipeline(&config);
}
#[test]
fn build_default_pipeline_with_defaults() {
let config = Config::default();
let _registry = build_default_pipeline(&config);
}
#[tokio::test]
async fn bus_clone_survives_into_agent_loop() {
let platform = Arc::new(NativePlatform::new());
let ctx = AppContext::new(test_config(), platform).await.unwrap();
let tx = ctx.bus().inbound_sender();
let _agent = ctx.into_agent_loop();
use chrono::Utc;
use clawft_types::event::InboundMessage;
use std::collections::HashMap;
let msg = InboundMessage {
channel: "test".into(),
sender_id: "user1".into(),
chat_id: "chat1".into(),
content: "hello".into(),
timestamp: Utc::now(),
media: vec![],
metadata: HashMap::new(),
};
assert!(tx.send(msg).await.is_ok());
}
#[tokio::test]
async fn app_context_is_send() {
fn assert_send<T: Send>() {}
assert_send::<AppContext<NativePlatform>>();
}
#[tokio::test]
async fn build_live_pipeline_transport_is_configured() {
use crate::pipeline::traits::{LlmMessage, TaskType, TransportRequest};
let config = test_config();
let registry = build_live_pipeline(&config);
let pipeline = registry.get(&TaskType::Unknown);
let transport_req = TransportRequest {
provider: "anthropic".into(),
model: "claude-opus-4-5".into(),
messages: vec![LlmMessage {
role: "user".into(),
content: "test".into(),
tool_call_id: None,
tool_calls: None,
}],
tools: vec![],
max_tokens: Some(10),
temperature: Some(0.0),
};
let result = pipeline.transport.complete(&transport_req).await;
match result {
Ok(_) => {} Err(e) => {
let msg = e.to_string();
assert!(
!msg.contains("transport not configured"),
"live pipeline should not use stub transport, but got: {msg}"
);
}
}
}
#[tokio::test]
async fn app_context_enable_live_llm() {
let platform = Arc::new(NativePlatform::new());
let mut ctx = AppContext::new(test_config(), platform).await.unwrap();
ctx.enable_live_llm();
let agent = ctx.into_agent_loop();
assert_eq!(agent.config().defaults.model, "deepseek/deepseek-chat");
}
#[tokio::test]
async fn default_pipeline_transport_is_stub() {
use crate::pipeline::traits::{LlmMessage, TaskType, TransportRequest};
let config = test_config();
let registry = build_default_pipeline(&config);
let pipeline = registry.get(&TaskType::Unknown);
let transport_req = TransportRequest {
provider: "test".into(),
model: "test".into(),
messages: vec![LlmMessage {
role: "user".into(),
content: "hello".into(),
tool_call_id: None,
tool_calls: None,
}],
tools: vec![],
max_tokens: None,
temperature: None,
};
let result = pipeline.transport.complete(&transport_req).await;
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(
err_msg.contains("transport not configured"),
"default pipeline should use stub transport, got: {err_msg}"
);
}
#[test]
fn build_live_pipeline_creates_registry() {
let config = test_config();
let registry = build_live_pipeline(&config);
use crate::pipeline::traits::TaskType;
let _pipeline = registry.get(&TaskType::Chat);
let _pipeline = registry.get(&TaskType::CodeGeneration);
let _pipeline = registry.get(&TaskType::Unknown);
}
}