Support long-running local inference with configurable timeouts and busy-agent queueing

Introduces configurable HTTP/tool/runtime timeouts, for inter-agent work, persistent queueing for messages sent while an agent is busy, and safer agent state cleanup so agents do not get stuck in a permanent busy state

Author: kyREclown

crates/openfang-api/src/channel_bridge.rs

@@ -57,6 +57,7 @@ use openfang_channels::wecom::WeComAdapter;
 use openfang_kernel::OpenFangKernel;
 use openfang_runtime::kernel_handle::KernelHandle;
 use openfang_types::agent::AgentId;
+use uuid::Uuid;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tracing::{error, info, warn};
@@ -111,6 +112,75 @@ impl ChannelBridgeHandle for KernelBridgeAdapter {
         Ok(result.response)
     }
 
+    fn queue_max_retries(&self) -> usize {
+        std::env::var("OPENFANG_QUEUE_MAX_RETRIES")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or_else(|| {
+                self.kernel.config.channels.queue_max_retries.unwrap_or(300) as usize
+            })
+    }
+
+    fn queue_sleep_secs(&self) -> u64 {
+        std::env::var("OPENFANG_QUEUE_SLEEP_SECS")
+            .ok()
+            .and_then(|s| s.parse().ok())
+            .unwrap_or_else(|| {
+                self.kernel.config.channels.queue_sleep_secs.unwrap_or(2)
+            })
+    }
+
+    async fn is_agent_busy(&self, agent_id: AgentId) -> bool {
+        self.kernel
+            .registry
+            .get(agent_id)
+            .map(|e| e.state == openfang_types::agent::AgentState::Thinking)
+            .unwrap_or(false)
+    }
+
+    async fn get_channel_queue(&self) -> Result<String, String> {
+        let nil_id = openfang_types::agent::AgentId(Uuid::nil());
+        let val = self
+            .kernel
+            .memory
+            .structured_get(nil_id, "channels_queue")
+            .map_err(|e| format!("{e}"))?;
+        match val {
+            Some(serde_json::Value::String(s)) => Ok(s),
+            _ => Ok(String::new()),
+        }
+    }
+
+    async fn save_channel_queue(&self, queue_json: &str) -> Result<(), String> {
+        let nil_id = openfang_types::agent::AgentId(Uuid::nil());
+        self.kernel
+            .memory
+            .structured_set(
+                nil_id,
+                "channels_queue",
+                serde_json::Value::String(queue_json.to_string()),
+            )
+            .map_err(|e| format!("{e}"))?;
+        Ok(())
+    }
+
+    fn queue_enabled(&self) -> bool {
+        self.kernel.config.channels.queue_enabled.unwrap_or(true)
+    }
+
+    fn queue_poll_secs(&self) -> u64 {
+        self.kernel
+            .config
+            .channels
+            .queue_poll_secs
+            .or_else(|| {
+                std::env::var("OPENFANG_QUEUE_POLL_SECS")
+                    .ok()
+                    .and_then(|s| s.parse().ok())
+            })
+            .unwrap_or(30)
+    }
+
     async fn find_agent_by_name(&self, name: &str) -> Result<Option<AgentId>, String> {
         Ok(self.kernel.registry.find_by_name(name).map(|e| e.id))
     }

crates/openfang-api/src/routes.rs

@@ -212,7 +212,7 @@ pub async fn list_agents(State(state): State<Arc<AppState>>) -> impl IntoRespons
                 })
                 .unwrap_or(("unknown".to_string(), "unknown".to_string()));
 
-            let ready = matches!(e.state, openfang_types::agent::AgentState::Running)
+            let ready = matches!(e.state, openfang_types::agent::AgentState::Running | openfang_types::agent::AgentState::Thinking)
                 && auth_status != "missing";
 
             // Issue #1026: surface which agents are currently calling the LLM
@@ -3546,7 +3546,7 @@ pub async fn prometheus_metrics(State(state): State<Arc<AppState>>) -> impl Into
     let agents = state.kernel.registry.list();
     let active = agents
         .iter()
-        .filter(|a| matches!(a.state, openfang_types::agent::AgentState::Running))
+        .filter(|a| matches!(a.state, openfang_types::agent::AgentState::Running | openfang_types::agent::AgentState::Thinking))
         .count();
     out.push_str("# HELP openfang_agents_active Number of active agents.\n");
     out.push_str("# TYPE openfang_agents_active gauge\n");
@@ -7780,6 +7780,7 @@ pub async fn set_provider_key(
                     api_key_env: env_var.clone(),
                     base_url: None,
                     subprocess_timeout_secs: None,
+                    http_timeout_secs: None,
                 };
                 let mut guard = state
                     .kernel
@@ -7948,6 +7949,7 @@ pub async fn test_provider(
         },
         skip_permissions: true,
         subprocess_timeout_secs: None,
+        http_timeout_secs: None,
     };
 
     match openfang_runtime::drivers::create_driver(&driver_config) {

crates/openfang-api/tests/api_integration_test.rs

@@ -62,6 +62,7 @@ async fn start_test_server_with_provider(
             api_key_env: api_key_env.to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };
@@ -907,6 +908,7 @@ async fn start_test_server_with_auth(api_key: &str) -> TestServer {
             api_key_env: "OLLAMA_API_KEY".to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };

crates/openfang-api/tests/daemon_lifecycle_test.rs

@@ -99,6 +99,7 @@ async fn test_full_daemon_lifecycle() {
             api_key_env: "OLLAMA_API_KEY".to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };
@@ -227,6 +228,7 @@ async fn test_server_immediate_responsiveness() {
             api_key_env: "OLLAMA_API_KEY".to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };

crates/openfang-api/tests/load_test.rs

@@ -43,6 +43,7 @@ async fn start_test_server() -> TestServer {
             api_key_env: "OLLAMA_API_KEY".to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };

crates/openfang-api/tests/skill_config_api_test.rs

@@ -77,6 +77,7 @@ async fn start_test_server() -> TestServer {
             api_key_env: "OLLAMA_API_KEY".to_string(),
             base_url: None,
             subprocess_timeout_secs: None,
+            http_timeout_secs: None,
         },
         ..KernelConfig::default()
     };