fix(rap): treat servers that don't support /tool_call_status as having given up on the call

crates/infinity-agent-core/src/batch_processor.rs

@@ -537,6 +537,9 @@ mod tests {
         async fn post(&self, _: &str, _: &str) -> Result<u16, E> {
             Ok(200)
         }
+        async fn post_read(&self, _: &str, _: &str) -> Result<(u16, Vec<u8>), E> {
+            Ok((200, vec![]))
+        }
         async fn get(&self, _: &str) -> Result<(u16, Vec<u8>), E> {
             Ok((200, vec![]))
         }

crates/infinity-agent-lambda/src/tools/rap_http.rs

@@ -126,6 +126,49 @@ impl HttpClient for RapHttpClient {
         Ok(response.status().as_u16())
     }
 
+    async fn post_read(&self, url: &str, body: &str) -> Result<(u16, Vec<u8>), HttpError> {
+        let parsed = url::Url::parse(url).map_err(|e| HttpError(e.to_string()))?;
+        let host = parsed
+            .host_str()
+            .ok_or(HttpError("missing host".into()))?
+            .to_owned();
+
+        let signed_headers = self
+            .sign_request(
+                "POST",
+                url,
+                std::iter::once(("host", host.as_str()))
+                    .chain(std::iter::once(("content-type", "application/json"))),
+                SignableBody::Bytes(body.as_bytes()),
+            )
+            .await?;
+
+        let mut request = self
+            .http_client
+            .post(url)
+            .header("host", &host)
+            .header("content-type", "application/json");
+
+        for (name, value) in &signed_headers {
+            request = request.header(name.as_str(), value.as_str());
+        }
+
+        let response = request
+            .body(body.to_owned())
+            .send()
+            .await
+            .map_err(|e| HttpError(e.to_string()))?;
+
+        let status = response.status().as_u16();
+        let body_bytes = response
+            .bytes()
+            .await
+            .map_err(|e| HttpError(e.to_string()))?
+            .to_vec();
+
+        Ok((status, body_bytes))
+    }
+
     async fn get(&self, url: &str) -> Result<(u16, Vec<u8>), HttpError> {
         let parsed = url::Url::parse(url).map_err(|e| HttpError(e.to_string()))?;
         let host = parsed

crates/infinity-daemon/src/mcp_proxy.rs

@@ -9,7 +9,10 @@ use hyper::body::{Bytes, Incoming};
 use hyper::server::conn::http1;
 use hyper::{Request, Response, StatusCode};
 use hyper_util::rt::TokioIo;
-use rap_protocol::{DisplaySegment, RapCallback, RapInvocation, RapToolResult};
+use rap_protocol::{
+    DisplaySegment, RapCallback, RapInvocation, RapToolCallStatusRequest,
+    RapToolCallStatusResponse, RapToolResult,
+};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::convert::Infallible;
@@ -294,6 +297,26 @@ struct ProxyState {
     client_factory: McpClientFactory,
     client: Mutex<Option<Box<dyn McpTransport>>>,
     port: u16,
+    /// Tool call IDs currently being processed, for `/tool_call_status`
+    /// queries. Uses a std mutex so it can be updated from a drop guard.
+    in_flight: std::sync::Mutex<std::collections::HashSet<String>>,
+}
+
+/// RAII guard that removes a tool call ID from the proxy's in-flight set when
+/// the invocation task finishes (on any exit path).
+struct InFlightGuard {
+    id: String,
+    state: Arc<ProxyState>,
+}
+
+impl Drop for InFlightGuard {
+    fn drop(&mut self) {
+        self.state
+            .in_flight
+            .lock()
+            .expect("bug: in_flight mutex poisoned")
+            .remove(&self.id);
+    }
 }
 
 impl ProxyState {
@@ -457,19 +480,52 @@ async fn handle(req: Request<Incoming>, state: Arc<ProxyState>) -> Response<Full
         return text_response(StatusCode::METHOD_NOT_ALLOWED, "POST only");
     }
 
-    // Parse invocation
+    // Parse body
     let body = match req.into_body().collect().await {
         Ok(b) => b.to_bytes(),
         Err(_) => return text_response(StatusCode::BAD_REQUEST, "bad body"),
     };
+
+    // Tool call status query: report whether an invocation is still in flight.
+    // A freshly restarted proxy has an empty in-flight set, so calls from
+    // before the restart correctly report `alive: false`.
+    if path.ends_with("/tool_call_status") {
+        let request: RapToolCallStatusRequest = match serde_json::from_slice(&body) {
+            Ok(r) => r,
+            Err(e) => return text_response(StatusCode::BAD_REQUEST, &format!("bad json: {e}")),
+        };
+        let alive = state
+            .in_flight
+            .lock()
+            .expect("bug: in_flight mutex poisoned")
+            .contains(&request.tool_call_id);
+        let response = serde_json::to_string(&RapToolCallStatusResponse { alive })
+            .expect("bug: serialize RapToolCallStatusResponse");
+        return json_response(StatusCode::OK, &response);
+    }
+
+    // Parse invocation
     let inv: RapInvocation = match serde_json::from_slice(&body) {
         Ok(i) => i,
         Err(e) => return text_response(StatusCode::BAD_REQUEST, &format!("bad json: {e}")),
     };
 
+    // Track the invocation as in-flight for the duration of the async task,
+    // so `/tool_call_status` reports it alive until the callback is sent.
+    state
+        .in_flight
+        .lock()
+        .expect("bug: in_flight mutex poisoned")
+        .insert(inv.id.clone());
+    let in_flight_guard = InFlightGuard {
+        id: inv.id.clone(),
+        state: state.clone(),
+    };
+
     // Return immediately, process async
     let state = state.clone();
     tokio::spawn(rap_protocol::log_panic("mcp_proxy_invoke", async move {
+        let _in_flight_guard = in_flight_guard;
         let res = if inv.operation.ends_with("_list_tools") {
             state.list_tools().await
         } else if inv.operation.ends_with("_invoke_tool") {
@@ -580,6 +636,7 @@ pub async fn start_proxy_server(name: String, factory: McpClientFactory) -> Resu
         client_factory: factory,
         client: Mutex::new(None),
         port,
+        in_flight: std::sync::Mutex::new(std::collections::HashSet::new()),
     });
 
     tokio::spawn(rap_protocol::log_panic(

crates/infinity-daemon/src/rap_tools.rs

@@ -1,5 +1,7 @@
 //! RAP tool support for the CLI: loads tools from RAP servers using rap-client.
 
+use std::collections::HashMap;
+
 use infinity_agent_core::tools::Tool;
 use infinity_agent_core::tools::rap_tool::RapTool;
 use infinity_agent_core::traits::InputSender;
@@ -15,6 +17,10 @@ pub struct LoadedRapTools<M: InputSender + 'static> {
     pub tools: Vec<Box<dyn Tool<M>>>,
     /// Servers that declared needsMigration: true, as (config_id, url) pairs.
     pub migration_servers: Vec<(String, String)>,
+    /// Maps each loaded RAP tool name → the base URL of the server that
+    /// provides it. Used to route protocol messages (e.g. `/tool_call_status`
+    /// queries) to the server that originally received an invocation.
+    pub tool_servers: HashMap<String, String>,
 }
 
 pub async fn load_rap_tools<M: InputSender + 'static>(
@@ -29,8 +35,21 @@ pub async fn load_rap_tools<M: InputSender + 'static>(
 
     let mut tools: Vec<Box<dyn Tool<M>>> = Vec::new();
     let mut migration_servers = Vec::new();
+    let mut tool_servers = HashMap::new();
     for ts in loaded {
         let endpoint = ts.manifest.endpoint.clone();
+        // Resolve the configured base URL for this toolset's server. Falls
+        // back to the endpoint with any trailing `/invoke` path stripped.
+        let base_url = servers
+            .iter()
+            .find(|(u, _)| endpoint.starts_with(u.as_str()))
+            .map(|(u, _)| u.clone())
+            .unwrap_or_else(|| {
+                endpoint
+                    .trim_end_matches('/')
+                    .trim_end_matches("/invoke")
+                    .to_owned()
+            });
         if ts.manifest.needs_migration {
             // Find the (url, id) entry for this toolset
             if let Some((url, Some(id))) = servers
@@ -42,6 +61,7 @@ pub async fn load_rap_tools<M: InputSender + 'static>(
         }
         for def in ts.manifest.tools {
             tracing::info!("Loaded RAP tool: {} from {}", def.name, endpoint);
+            tool_servers.insert(def.name.clone(), base_url.clone());
             tools.push(Box::new(RapTool {
                 name: def.name,
                 description: def.description,
@@ -55,5 +75,6 @@ pub async fn load_rap_tools<M: InputSender + 'static>(
     Ok(LoadedRapTools {
         tools,
         migration_servers,
+        tool_servers,
     })
 }

crates/infinity-daemon/src/session/mod.rs

@@ -25,6 +25,7 @@ use crate::sleep_tools::{SleepTool, SleepUntilTool};
 
 pub mod agent_loop;
 pub mod display;
+pub mod reconcile;
 pub mod thread_worker;
 
 pub use agent_loop::agent_loop;
@@ -324,7 +325,9 @@ impl SessionManager {
         let spawned_servers = booted.spawned_servers;
         let urls = booted.urls;
 
-        let rap_tools: Vec<Box<dyn Tool<InMemoryMessageSender>>> = if !urls.is_empty() {
+        let rap_tools: Vec<Box<dyn Tool<InMemoryMessageSender>>>;
+        let rap_tool_servers: HashMap<String, String>;
+        if !urls.is_empty() {
             let servers_with_ids: Vec<(String, Option<String>)> = urls
                 .iter()
                 .map(|u| {
@@ -338,16 +341,19 @@ impl SessionManager {
 
                     emit(info(String::new())).await;
 
-                    loaded.tools
+                    rap_tools = loaded.tools;
+                    rap_tool_servers = loaded.tool_servers;
                 }
                 Err(e) => {
                     emit(info(format!("Warning: failed to load RAP tools: {e}"))).await;
-                    Vec::new()
+                    rap_tools = Vec::new();
+                    rap_tool_servers = HashMap::new();
                 }
             }
         } else {
-            Vec::new()
-        };
+            rap_tools = Vec::new();
+            rap_tool_servers = HashMap::new();
+        }
 
         let extra_system_prompt = Some(format!(
             "The user's current working directory is: {cwd:?}\n\n\
@@ -361,6 +367,28 @@ impl SessionManager {
 
         let (shutdown_tx, shutdown_rx) = oneshot::channel();
 
+        // Reconcile pending RAP tool calls / active subscriptions against
+        // their servers in the background: if the server gave up on any of
+        // them while the agent was down, inject failure messages so the
+        // affected threads don't hang forever. See `session::reconcile`.
+        if !rap_tool_servers.is_empty() {
+            let conversation_store = self.conversation_store.clone();
+            let reconcile_state_store = self.state_store.clone();
+            let reconcile_sender = sender.clone();
+            let reconcile_session_id = session_id.clone();
+            tokio::task::spawn_local(rap_protocol::log_panic("rap_reconcile", async move {
+                reconcile::reconcile_rap_state(
+                    &conversation_store,
+                    &reconcile_state_store,
+                    &reconcile_session_id,
+                    &rap_tool_servers,
+                    &rap_tools::SimpleHttpClient::new(),
+                    &reconcile_sender,
+                )
+                .await;
+            }));
+        }
+
         let (idle_tx, agent_handle, subscriber_map) = self.start_agent_loop(
             session_id.clone(),
             agent_rx,