crates/fluent-agent/src/config.rs (1)

64-71: Placeholder implementation needs attention.

The get_base_engine method returns None with a comment acknowledging the type mismatch (Arc<Box<dyn Engine>> to Arc<dyn Engine>). This placeholder implementation may cause runtime issues if called.

Consider one of the following solutions:

1. Restructure to avoid double-boxing:

 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>,

2. If double-boxing is required, unwrap correctly:

 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
+    // Clone the inner Box and wrap in a new Arc
+    Some(Arc::from(self.reasoning_engine.as_ref().clone_box()))
 }

Note: Option 2 requires Engine: dyn_clone::DynClone or similar mechanism to clone trait objects.

crates/fluent-agent/src/production_mcp/client.rs (4)

110-150: Release the client map lock before awaiting tool calls

execute_tool_with_failover keeps the RwLockReadGuard from self.clients.read().await alive while you await both find_servers_with_tool and each client.execute_tool(...). If any of those calls block (network hiccups, retries, slow tools), every writer wanting to add/remove clients is stalled behind that read guard, and the read‑guard itself is held across .await, violating Tokio’s lock guidelines and risking starvation/deadlock scenarios under load. Snapshot the Arc<ProductionMcpClient> handles first, drop the guard, and then drive the async work.

-        let clients = self.clients.read().await;
-        let available_servers = self.find_servers_with_tool(&clients, tool_name).await;
+        let clients_snapshot: Vec<(String, Arc<ProductionMcpClient>)> = {
+            let guard = self.clients.read().await;
+            guard
+                .iter()
+                .map(|(name, client)| (name.clone(), Arc::clone(client)))
+                .collect()
+        };
+        let available_servers =
+            Self::find_servers_with_tool(&clients_snapshot, tool_name).await;
 
-        for server_name in available_servers {
-            if let Some(client) = clients.get(&server_name) {
-                match client.execute_tool(tool_name, parameters.clone()).await {
+        for (server_name, client) in available_servers {
+            match client.execute_tool(tool_name, parameters.clone()).await {
                 Ok(result) => {
                     self.metrics_collector
-                        .record_tool_execution_success(tool_name, &server_name)
+                        .record_tool_execution_success(tool_name, &server_name)
                         .await;
                     return Ok(result);
                 }
                 Err(error) => {
                     self.metrics_collector
                         .record_tool_execution_failure(tool_name, &server_name, &error)
                         .await;

Adjust find_servers_with_tool to operate on the snapshot (returning (String, Arc<ProductionMcpClient>) pairs) so the lock is never held across awaits.

---

196-215: Avoid holding the clients read-lock inside find_servers_with_tool

find_servers_with_tool loops over the shared HashMap while awaiting client.has_tool(...), so the same RwLockReadGuard from the caller stays live across each .await. That blocks writers for the duration of every probe. With the snapshot approach above, refactor this helper to consume the pre-collected (String, Arc<...>) list and only await on cloned Arcs.

-    async fn find_servers_with_tool(
-        &self,
-        clients: &HashMap<String, Arc<ProductionMcpClient>>,
-        tool_name: &str,
-    ) -> Vec<String> {
-        let mut servers = Vec::new();
-
-        for (server_name, client) in clients {
-            if client.has_tool(tool_name).await {
-                servers.push(server_name.clone());
-            }
-        }
-
-        // Sort by server health and performance metrics
-        servers.sort_by(|a, b| {
-            // Implementation would sort by health score, latency, etc.
-            a.cmp(b)
-        });
-
-        servers
-    }
+    async fn find_servers_with_tool(
+        clients: &[(String, Arc<ProductionMcpClient>)],
+        tool_name: &str,
+    ) -> Vec<(String, Arc<ProductionMcpClient>)> {
+        let mut servers = Vec::new();
+
+        for (server_name, client) in clients {
+            if client.has_tool(tool_name).await {
+                servers.push((server_name.clone(), Arc::clone(client)));
+            }
+        }
+
+        servers.sort_by(|a, b| a.0.cmp(&b.0));
+        servers
+    }

This keeps all awaits outside the lock’s lifetime.

---

223-236: Spawned health loop holds the clients lock across awaits

Inside start_health_monitoring, the spawned task keeps clients.read().await alive while awaiting client.check_health() and health_monitor.update_client_health(...). That means the read lock survives across each network call or health update, choking writers (connect/disconnect/config changes). Snapshot the client handles first, drop the guard, then iterate.

-                let clients_guard = clients.read().await;
-
-                for (name, client) in clients_guard.iter() {
-                    let health_status = client.check_health().await;
-                    health_monitor
-                        .update_client_health(name, health_status)
-                        .await;
-                }
+                let entries: Vec<(String, Arc<ProductionMcpClient>)> = {
+                    let guard = clients.read().await;
+                    guard
+                        .iter()
+                        .map(|(name, client)| (name.clone(), Arc::clone(client)))
+                        .collect()
+                };
+
+                for (name, client) in entries {
+                    let health_status = client.check_health().await;
+                    health_monitor
+                        .update_client_health(&name, health_status)
+                        .await;
+                }

---

246-258: Connection maintenance loop needs the same lock-release treatment

The maintenance task mirrors the health loop: it keeps the read guard from clients.read().await alive while it awaits client.maintain_connection(). Apply the same snapshot/drop pattern so we never suspend with the lock held.

-                let clients_guard = clients.read().await;
-
-                for client in clients_guard.values() {
-                    if let Err(e) = client.maintain_connection().await {
-                        log::warn!("Connection maintenance failed: {}", e);
-                    }
-                }
+                let entries: Vec<Arc<ProductionMcpClient>> = {
+                    let guard = clients.read().await;
+                    guard.values().map(Arc::clone).collect()
+                };
+
+                for client in entries {
+                    if let Err(e) = client.maintain_connection().await {
+                        log::warn!("Connection maintenance failed: {}", e);
+                    }
+                }

crates/fluent-agent/src/reasoning/enhanced_multi_modal.rs (2)

398-425: Drop the working-memory write lock before awaiting

analyze_problem_context keeps the RwLockWriteGuard from self.working_memory.write().await alive while awaiting self.update_attention_weights(...). If any other task tries to lock working_memory, it’s blocked until the attention update completes, and if another code path takes the locks in the opposite order you risk deadlock. Clone the deque (or move the updated data out) so you can drop the guard before awaiting.

-        let mut memory = self.working_memory.write().await;
+        let mut memory = self.working_memory.write().await;
         // ... mutate memory ...
-        self.update_attention_weights(&memory.active_concepts)
-            .await?;
+        let active_snapshot = memory.active_concepts.clone();
+        drop(memory);
+
+        self.update_attention_weights(&active_snapshot).await?;

This preserves the updates while ensuring no .await occurs with the write guard held.

---

850-868: Guard against zero-capacity working memory

capacity_utilization = total_items as f64 / memory.capacity as f64; will panic if a caller supplies EnhancedReasoningConfig { working_memory_capacity: 0, .. } (which is possible because the config is public). Never divide by zero; clamp the capacity to at least 1 (or return an error when constructing the engine).

-        let total_items = memory.active_concepts.len();
-        let capacity_utilization = total_items as f64 / memory.capacity as f64;
+        let total_items = memory.active_concepts.len();
+        let capacity = memory.capacity.max(1);
+        let capacity_utilization = total_items as f64 / capacity as f64;

That avoids runtime panics when misconfigured.

crates/fluent-agent/src/production_mcp/metrics.rs (1)

35-46: Stop spawns leaving runaway background tasks

start() spawns an infinite-loop metrics task but never keeps the JoinHandle, so every call leaks a live task and stop() can’t cancel it. Repeated start() calls pile up loops, and even a single run keeps ticking after shutdown, which is a reliability bug. Please track the handle and abort it on stop() (and guard against double-start). Example fix:

-use tokio::sync::RwLock;
+use tokio::sync::{Mutex, RwLock};

 pub struct MetricsCollector {
     client_metrics: Arc<RwLock<ClientMetrics>>,
     server_metrics: Arc<RwLock<ServerMetrics>>,
     transport_metrics: Arc<RwLock<TransportMetrics>>,
     tool_metrics: Arc<RwLock<ToolMetrics>>,
     resource_metrics: Arc<RwLock<ResourceMetrics>>,
     system_metrics: Arc<RwLock<SystemMetrics>>,
+    system_metrics_task: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
     start_time: Instant,
 }

 impl MetricsCollector {
     pub fn new() -> Self {
         Self {
             client_metrics: Arc::new(RwLock::new(ClientMetrics::new())),
             server_metrics: Arc::new(RwLock::new(ServerMetrics::new())),
             transport_metrics: Arc::new(RwLock::new(TransportMetrics::new())),
             tool_metrics: Arc::new(RwLock::new(ToolMetrics::new())),
             resource_metrics: Arc::new(RwLock::new(ResourceMetrics::new())),
             system_metrics: Arc::new(RwLock::new(SystemMetrics::new())),
+            system_metrics_task: Arc::new(Mutex::new(None)),
             start_time: Instant::now(),
         }
     }

     pub async fn start(&self) -> Result<(), McpError> {
-        self.start_system_metrics_collection().await;
+        self.start_system_metrics_collection().await?;
         Ok(())
     }

     pub async fn stop(&self) -> Result<(), McpError> {
-        // Stop background tasks
+        if let Some(handle) = self.system_metrics_task.lock().await.take() {
+            handle.abort();
+        }
         Ok(())
     }

-    async fn start_system_metrics_collection(&self) {
-        let system_metrics = self.system_metrics.clone();
-
-        tokio::spawn(async move {
+    async fn start_system_metrics_collection(&self) -> Result<(), McpError> {
+        let mut guard = self.system_metrics_task.lock().await;
+        if guard.is_some() {
+            return Ok(());
+        }
+
+        let system_metrics = self.system_metrics.clone();
+        *guard = Some(tokio::spawn(async move {
             let mut interval = tokio::time::interval(Duration::from_secs(10));
             loop {
                 interval.tick().await;
                 let mut metrics = system_metrics.write().await;
                 metrics.update_system_metrics().await;
             }
-        });
+        }));
+        Ok(())
     }
 }

Also applies to: 101-113

crates/fluent-agent/src/planning/dependency_analyzer.rs (1)

376-399: Dependency inference misses reverse edges

The pairwise scan only tests task_a → task_b. Because the inner iterator skips earlier tasks, any dependency where the later task depends on an earlier one (e.g., “Task B depends on Task A”) is never detected, so the edge is missing and the schedule/topology is wrong. Please check both directions before advancing the loop.

         for task_b in tasks.iter().skip(i + 1) {
             if self.has_dependency(task_a, task_b).await? {
                 // task_a depends on task_b
                 graph
                     .dependencies
                     .entry(task_a.task_id.clone())
                     .or_default()
                     .insert(task_b.task_id.clone());

                 graph
                     .dependents
                     .entry(task_b.task_id.clone())
                     .or_default()
                     .insert(task_a.task_id.clone());

                 if let Some(node_a) = graph.nodes.get_mut(&task_a.task_id) {
                     node_a.dependency_count += 1;
                 }
                 if let Some(node_b) = graph.nodes.get_mut(&task_b.task_id) {
                     node_b.dependent_count += 1;
                 }
+            } else if self.has_dependency(task_b, task_a).await? {
+                graph
+                    .dependencies
+                    .entry(task_b.task_id.clone())
+                    .or_default()
+                    .insert(task_a.task_id.clone());
+
+                graph
+                    .dependents
+                    .entry(task_a.task_id.clone())
+                    .or_default()
+                    .insert(task_b.task_id.clone());
+
+                if let Some(node_a) = graph.nodes.get_mut(&task_a.task_id) {
+                    node_a.dependent_count += 1;
+                }
+                if let Some(node_b) = graph.nodes.get_mut(&task_b.task_id) {
+                    node_b.dependency_count += 1;
+                }
             }
         }

crates/fluent-agent/src/performance/cache.rs (1)

241-280: Redis fallback still returns Err, breaking the new graceful-degradation contract.

RedisCache::get/set/remove/clear continue to exit with Err whenever available is false. The updated tests (and the surrounding changes) now expect these paths to succeed with Ok(None)/Ok(()) so higher tiers can continue working. As written, every call in fallback mode still surfaces an error, so the new tests fail and the multi-tier cache never reaches the intended graceful behavior. Please return success in these branches instead.

@@
     async fn get(&self, _key: &K) -> Result<Option<V>> {
-        if !self.available {
-            return Err(anyhow::anyhow!("Redis cache not available: {}", self.url));
-        }
+        if !self.available {
+            return Ok(None);
+        }
@@
     async fn set(&self, _key: &K, _value: &V, ttl: Duration) -> Result<()> {
-        if !self.available {
-            return Err(anyhow::anyhow!("Redis cache not available: {}", self.url));
-        }
+        if !self.available {
+            return Ok(());
+        }
@@
     async fn remove(&self, _key: &K) -> Result<()> {
-        if !self.available {
-            return Err(anyhow::anyhow!("Redis cache not available: {}", self.url));
-        }
+        if !self.available {
+            return Ok(());
+        }
@@
     async fn clear(&self) -> Result<()> {
-        if !self.available {
-            return Err(anyhow::anyhow!("Redis cache not available: {}", self.url));
-        }
+        if !self.available {
+            return Ok(());
+        }

crates/fluent-agent/src/lib.rs (1)

229-251: Don't break on first EOF; read both stdout and stderr until both are done.

The loop exits when either pipe hits EOF, truncating the other stream and risking hangs. Track both pipes and only break when both are closed or buffers hit limits.

Apply this diff:

-        let read_fut = async {
-            let mut tmp_out = [0u8; 8192];
-            let mut tmp_err = [0u8; 8192];
-            loop {
-                tokio::select! {
-                    read = stdout.read(&mut tmp_out) => {
-                        let n = read?;
-                        if n == 0 { break; }
-                        let to_take = n.min(max_output_bytes.saturating_sub(out_buf.len()));
-                        out_buf.extend_from_slice(&tmp_out[..to_take]);
-                        if out_buf.len() >= max_output_bytes { break; }
-                    }
-                    read = stderr.read(&mut tmp_err) => {
-                        let n = read?;
-                        if n == 0 { break; }
-                        let to_take = n.min(max_output_bytes.saturating_sub(err_buf.len()));
-                        err_buf.extend_from_slice(&tmp_err[..to_take]);
-                        if err_buf.len() >= max_output_bytes { break; }
-                    }
-                }
-            }
-            Ok::<(), anyhow::Error>(())
-        };
+        let read_fut = async {
+            let mut tmp_out = [0u8; 8192];
+            let mut tmp_err = [0u8; 8192];
+            let mut out_done = false;
+            let mut err_done = false;
+            loop {
+                tokio::select! {
+                    read = stdout.read(&mut tmp_out), if !out_done => {
+                        let n = read?;
+                        if n == 0 {
+                            out_done = true;
+                        } else {
+                            let to_take = n.min(max_output_bytes.saturating_sub(out_buf.len()));
+                            out_buf.extend_from_slice(&tmp_out[..to_take]);
+                        }
+                    }
+                    read = stderr.read(&mut tmp_err), if !err_done => {
+                        let n = read?;
+                        if n == 0 {
+                            err_done = true;
+                        } else {
+                            let to_take = n.min(max_output_bytes.saturating_sub(err_buf.len()));
+                            err_buf.extend_from_slice(&tmp_err[..to_take]);
+                        }
+                    }
+                }
+                let out_full = out_buf.len() >= max_output_bytes;
+                let err_full = err_buf.len() >= max_output_bytes;
+                if (out_done || out_full) && (err_done || err_full) {
+                    break;
+                }
+            }
+            Ok::<(), anyhow::Error>(())
+        };

crates/fluent-agent/src/adapters.rs (2)

74-111: Preserve registry tool names when planning searches.

Lowercasing every tool name means we now schedule tool_name like "websearch" even when the registry only knows "WebSearch" (or any other case-sensitive identifier). The follow-up MCP execution will fail at runtime because the lookup misses. Keep the original name for execution and only lowercase inside the predicate.

     fn available_tool_names(&self) -> Vec<String> {
         self.tool_registry
             .get_all_available_tools()
             .into_iter()
-            .map(|t| t.name.to_lowercase())
+            .map(|t| t.name)
             .collect()
     }
@@
-                let maybe_search = names
-                    .into_iter()
-                    .find(|n| n.contains("search") || n.contains("browse") || n.contains("web"));
+                let maybe_search = names.into_iter().find(|n| {
+                    let n_lc = n.to_lowercase();
+                    n_lc.contains("search") || n_lc.contains("browse") || n_lc.contains("web")
+                });

---

397-418: Fix success criteria to track the actual chapter file.

When persisting chapters beyond Chapter 1 we still report success against docs/book/ch_01.md, so subsequent chapters never satisfy the criteria. That stalls long-form runs. Tie the success check to the path we just computed.

                 return Ok(act::ActionPlan {
                     action_id: uuid::Uuid::new_v4().to_string(),
                     action_type: ActionType::FileOperation,
                     description: format!("Persist Chapter {}", chapter_num),
                     parameters: params,
                     expected_outcome: "Chapter saved".to_string(),
                     confidence_score: 0.8,
                     estimated_duration: None,
                     risk_level: act::RiskLevel::Low,
                     alternatives: Vec::new(),
                     prerequisites: Vec::new(),
-                    success_criteria: vec![format!("non_empty_file:{}/ch_01.md", base)],
+                    success_criteria: vec![format!("non_empty_file:{}", path)],
                 });