feat(agent): unify orchestrator loop and global sqlite memory

Author: njfio

Cargo.toml

@@ -147,6 +147,8 @@ lru = "0.12.4"
 lz4_flex = "0.11.3"
 # Hex encoding/decoding - pin to exact version
 hex = "0.4.3"
+# OS-specific data directories
+directories = "5.0.1"
 # Additional dependencies from fluent-agent
 # WebSocket support - pin to exact version
 tokio-tungstenite = "0.20.1"

crates/fluent-agent/Cargo.toml

@@ -44,6 +44,9 @@ prometheus = { workspace = true }
 # Additional utilities
 base64 = { workspace = true }
 thiserror = { workspace = true }
+sha2 = { workspace = true }
+hex = { workspace = true }
+directories = { workspace = true }
 urlencoding = "2.1"
 bincode = "1.3"
 toml = { workspace = true }
@@ -53,6 +56,12 @@ futures-util = { version = "0.3", features = ["sink", "std"] }
 # Async cancellation support
 tokio-util = "0.7"
 
+[features]
+# Deprecated integration suites kept for reference.
+deprecated_agent_tests = []
+deprecated_memory_tests = []
+deprecated_mcp_tests = []
+
 [dev-dependencies]
 tempfile = { workspace = true }
 tokio-stream = "0.1"

crates/fluent-agent/src/adapters.rs

@@ -573,12 +573,17 @@ impl act::ActionPlanner for LongFormWriterPlanner {
 
 pub struct McpRegistryExecutor {
     client_mgr: std::sync::Arc<ProductionMcpClientManager>,
+    policy: crate::tools::ToolExecutionConfig,
 }
 
 impl McpRegistryExecutor {
-    pub fn new(manager: std::sync::Arc<ProductionMcpManager>) -> Self {
+    pub fn new(
+        manager: std::sync::Arc<ProductionMcpManager>,
+        policy: crate::tools::ToolExecutionConfig,
+    ) -> Self {
         Self {
             client_mgr: manager.client_manager(),
+            policy,
         }
     }
 }
@@ -649,10 +654,31 @@ impl crate::tools::ToolExecutor for McpRegistryExecutor {
 
     fn validate_tool_request(
         &self,
-        _tool_name: &str,
-        _parameters: &std::collections::HashMap<String, serde_json::Value>,
+        tool_name: &str,
+        parameters: &std::collections::HashMap<String, serde_json::Value>,
     ) -> anyhow::Result<()> {
-        // Basic pass-through validation; MCP server handles schema
+        // Enforce the same basic policy checks as local tools.
+        // MCP servers may have their own validation, but we do not delegate safety.
+        if self.policy.read_only {
+            let lower = tool_name.to_lowercase();
+            if lower.contains("write") || lower.contains("create") || lower.contains("delete") {
+                return Err(anyhow::anyhow!(
+                    "MCP tool '{}' is blocked in read-only mode",
+                    tool_name
+                ));
+            }
+        }
+
+        for key in ["path", "file_path", "out_path", "dest", "directory", "dir"] {
+            if let Some(v) = parameters.get(key).and_then(|v| v.as_str()) {
+                let _ = validation::validate_path(v, &self.policy.allowed_paths)?;
+            }
+        }
+
+        if let Some(cmd) = parameters.get("command").and_then(|v| v.as_str()) {
+            validation::validate_command(cmd, &self.policy.allowed_commands)?;
+        }
+
         Ok(())
     }
 }
@@ -689,11 +715,11 @@ impl act::ToolExecutor for RegistryToolAdapter {
 
 /// Simple LLM-backed code generator
 pub struct LlmCodeGenerator {
-    engine: Arc<Box<dyn Engine>>,
+    engine: Arc<dyn Engine>,
 }
 
 impl LlmCodeGenerator {
-    pub fn new(engine: Arc<Box<dyn Engine>>) -> Self {
+    pub fn new(engine: Arc<dyn Engine>) -> Self {
         Self { engine }
     }
 }

crates/fluent-agent/src/advanced_tools.rs

@@ -249,7 +249,7 @@ impl AdvancedToolRegistry {
 
         self.tools_by_category
             .entry(category)
-            .or_insert_with(Vec::new)
+            .or_default()
             .push(tool.clone());
         self.tools_by_name.insert(name.clone(), tool);
 

crates/fluent-agent/src/autonomy/supervisor.rs

@@ -271,7 +271,7 @@ impl AutonomySupervisor {
         }
 
         if let Some(stderr) = action_result.metadata.get("stderr") {
-            if stderr.as_str().unwrap_or_default().len() > 0 {
+            if !stderr.as_str().unwrap_or_default().is_empty() {
                 score += 0.1;
                 triggers.push("stderr_present".to_string());
             }
@@ -287,7 +287,7 @@ impl AutonomySupervisor {
             stage: SupervisorStage::PostAction,
             risk_score: score,
             risk_level,
-            confidence: action_result.success.then(|| 0.8).unwrap_or(0.3),
+            confidence: if action_result.success { 0.8 } else { 0.3 },
             triggers,
             recommended_action: decision,
         })

crates/fluent-agent/src/benchmarks.rs

@@ -408,9 +408,8 @@ impl AutonomousBenchmarkSuite {
                 "Optimize algorithm performance for dataset size {}",
                 i * 1000
             );
-            match tot_engine.reason(&problem, &context).await {
-                Ok(_) => success_count += 1,
-                Err(_) => {}
+            if tot_engine.reason(&problem, &context).await.is_ok() {
+                success_count += 1;
             }
         }
 
@@ -456,9 +455,12 @@ impl AutonomousBenchmarkSuite {
                 GoalType::Analysis,
             );
 
-            match htn_planner.plan_decomposition(&goal, &context).await {
-                Ok(_) => success_count += 1,
-                Err(_) => {}
+            if htn_planner
+                .plan_decomposition(&goal, &context)
+                .await
+                .is_ok()
+            {
+                success_count += 1;
             }
         }
 
@@ -510,9 +512,8 @@ impl AutonomousBenchmarkSuite {
                 );
             }
 
-            match memory_system.update_context(&context).await {
-                Ok(_) => success_count += 1,
-                Err(_) => {}
+            if memory_system.update_context(&context).await.is_ok() {
+                success_count += 1;
             }
         }
 

crates/fluent-agent/src/config.rs

@@ -31,7 +31,7 @@ impl Default for RateLimitConfig {
         Self {
             enabled: true,
             reasoning_rps: 5.0,  // 5 requests per second
-            action_rps: 10.0,   // 10 requests per second
+            action_rps: 10.0,    // 10 requests per second
             reflection_rps: 3.0, // 3 requests per second
         }
     }
@@ -82,6 +82,8 @@ pub struct AgentEngineConfig {
     pub action_engine: String,
     pub reflection_engine: String,
     pub memory_database: String,
+    #[serde(default = "default_memory_enabled")]
+    pub memory_enabled: bool,
     pub tools: ToolConfig,
     pub config_path: Option<String>,
     pub max_iterations: Option<u32>,
@@ -109,12 +111,16 @@ fn default_web_browsing() -> bool {
     true
 }
 
+fn default_memory_enabled() -> bool {
+    true
+}
+
 /// Runtime configuration with loaded engines and credentials
 #[derive(Clone)]
 pub struct AgentRuntimeConfig {
-    pub reasoning_engine: Arc<Box<dyn Engine>>,
-    pub action_engine: Arc<Box<dyn Engine>>,
-    pub reflection_engine: Arc<Box<dyn Engine>>,
+    pub reasoning_engine: Arc<dyn Engine>,
+    pub action_engine: Arc<dyn Engine>,
+    pub reflection_engine: Arc<dyn Engine>,
     pub config: AgentEngineConfig,
     pub credentials: HashMap<String, String>,
     pub supervisor: Option<AutonomySupervisorConfig>,
@@ -129,11 +135,7 @@ pub struct AgentRuntimeConfig {
 impl AgentRuntimeConfig {
     /// Get the base engine for enhanced reasoning
     pub fn get_base_engine(&self) -> Option<Arc<dyn Engine>> {
-        // Return a clone of the reasoning engine for use as base engine
-        // We need to convert from Arc<Box<dyn Engine>> to Arc<dyn Engine>
-        // This is a workaround - we can't directly cast, so we'll return None for now
-        // In a real implementation, we'd need to restructure to avoid this type mismatch
-        None
+        Some(self.reasoning_engine.clone())
     }
 
     /// Acquire a rate limit token for reasoning operations
@@ -276,52 +278,42 @@ impl AgentEngineConfig {
         };
 
         // Create reflection engine (can be the same as reasoning)
-        let reflection_engine = if self.reflection_engine == self.reasoning_engine {
-            // Create a new instance of the same engine
-            self.create_engine(
-                &fluent_config_content,
-                &self.reflection_engine,
-                &credentials,
-                model_override,
-            )
-            .await?
-        } else if self.reflection_engine == self.action_engine {
-            // Create a new instance of the same engine
-            self.create_engine(
-                &fluent_config_content,
-                &self.reflection_engine,
-                &credentials,
-                model_override,
-            )
-            .await?
-        } else {
-            self.create_engine(
+        let reflection_engine = self
+            .create_engine(
                 &fluent_config_content,
                 &self.reflection_engine,
                 &credentials,
                 model_override,
             )
-            .await?
-        };
+            .await?;
 
         // Initialize rate limiters based on config
-        let rate_limit_config = self.rate_limit.clone().unwrap_or_else(RateLimitConfig::from_environment);
+        let rate_limit_config = self
+            .rate_limit
+            .clone()
+            .unwrap_or_else(RateLimitConfig::from_environment);
 
         let (reasoning_rate_limiter, action_rate_limiter, reflection_rate_limiter) =
             if rate_limit_config.enabled {
                 (
-                    Some(Arc::new(fluent_engines::RateLimiter::new(rate_limit_config.reasoning_rps))),
-                    Some(Arc::new(fluent_engines::RateLimiter::new(rate_limit_config.action_rps))),
-                    Some(Arc::new(fluent_engines::RateLimiter::new(rate_limit_config.reflection_rps))),
+                    Some(Arc::new(fluent_engines::RateLimiter::new(
+                        rate_limit_config.reasoning_rps,
+                    ))),
+                    Some(Arc::new(fluent_engines::RateLimiter::new(
+                        rate_limit_config.action_rps,
+                    ))),
+                    Some(Arc::new(fluent_engines::RateLimiter::new(
+                        rate_limit_config.reflection_rps,
+                    ))),
                 )
             } else {
                 (None, None, None)
             };
 
         Ok(AgentRuntimeConfig {
-            reasoning_engine: Arc::new(reasoning_engine),
-            action_engine: Arc::new(action_engine),
-            reflection_engine: Arc::new(reflection_engine),
+            reasoning_engine: Arc::from(reasoning_engine),
+            action_engine: Arc::from(action_engine),
+            reflection_engine: Arc::from(reflection_engine),
             config: self.clone(),
             credentials,
             supervisor: self.supervisor.clone(),
@@ -334,6 +326,31 @@ impl AgentEngineConfig {
         })
     }
 
+    /// Resolve the configured SQLite memory database to a filesystem path.
+    ///
+    /// Supported forms:
+    /// - `sqlite://global` (default)
+    /// - `sqlite://:memory:`
+    /// - `sqlite:///absolute/path/to/db`
+    /// - `sqlite://./relative/path/to/db`
+    pub fn resolve_memory_db_path(&self) -> std::path::PathBuf {
+        let url = self.memory_database.trim();
+        let Some(rest) = url.strip_prefix("sqlite://") else {
+            return crate::paths::global_agent_memory_db_path();
+        };
+
+        if rest.is_empty() || rest == "global" {
+            return crate::paths::global_agent_memory_db_path();
+        }
+        if rest == ":memory:" {
+            return std::path::PathBuf::from(":memory:");
+        }
+
+        // `sqlite:///abs/path` yields rest like "/abs/path" (keep it absolute)
+        // `sqlite://./rel/path` yields rest like "./rel/path"
+        std::path::PathBuf::from(rest)
+    }
+
     pub fn supervisor_config(&self) -> AutonomySupervisorConfig {
         self.supervisor.clone().unwrap_or_default()
     }
@@ -531,7 +548,8 @@ impl AgentEngineConfig {
             reasoning_engine: "sonnet3.5".to_string(),
             action_engine: "gpt-4o".to_string(),
             reflection_engine: "gemini-flash".to_string(),
-            memory_database: "sqlite://./agent_memory.db".to_string(),
+            memory_database: "sqlite://global".to_string(),
+            memory_enabled: true,
             tools: ToolConfig {
                 file_operations: true,
                 shell_commands: false, // Disabled by default for security
@@ -627,8 +645,7 @@ pub mod credentials {
 
         // Load CREDENTIAL_ prefixed variables (fluent_cli pattern)
         for (key, value) in env::vars() {
-            if key.starts_with("CREDENTIAL_") {
-                let credential_key = &key[11..]; // Remove CREDENTIAL_ prefix
+            if let Some(credential_key) = key.strip_prefix("CREDENTIAL_") {
                 credentials.insert(credential_key.to_string(), value);
             }
         }