fix(voice): atomic install-start guard for Whisper/Piper install RPCs (tinyhumansai#1787)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: sanil-23

src/openhuman/composio/client.rs

@@ -150,6 +150,38 @@ impl ComposioClient {
 
     // ── Execute ─────────────────────────────────────────────────────
 
+    /// `POST /agent-integrations/composio/execute` — single, non-retrying
+    /// HTTP round-trip. Use this when the caller owns the retry loop
+    /// (e.g. `auth_retry`) to avoid double-retry.
+    pub(crate) async fn execute_tool_once(
+        &self,
+        tool: &str,
+        arguments: Option<serde_json::Value>,
+    ) -> Result<ComposioExecuteResponse> {
+        let tool = tool.trim();
+        if tool.is_empty() {
+            anyhow::bail!("composio.execute_tool_once: tool slug must not be empty");
+        }
+        let arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default()));
+        tracing::debug!(tool = %tool, "[composio] execute_tool_once (no built-in retry)");
+        let body = json!({ "tool": tool, "arguments": arguments });
+        let result = self.post_execute_tool(&body).await;
+        match &result {
+            Ok(resp) => tracing::debug!(
+                tool = %tool,
+                successful = resp.successful,
+                has_error = resp.error.is_some(),
+                "[composio] execute_tool_once completed"
+            ),
+            Err(err) => tracing::error!(
+                tool = %tool,
+                error = %err,
+                "[composio] execute_tool_once failed"
+            ),
+        }
+        result
+    }
+
     /// `POST /agent-integrations/composio/execute` — run a Composio
     /// action and return the provider result + cost.
     pub async fn execute_tool(
@@ -233,46 +265,6 @@ impl ComposioClient {
             .await
     }
 
-    /// Single-shot `execute_tool` — same body construction and slug validation
-    /// as [`Self::execute_tool`], but **without** the inner post-OAuth retry
-    /// that [`Self::execute_tool_with_post_oauth_retry`] performs. Reserved
-    /// for callers that already own a higher-level retry policy and would
-    /// otherwise stack two retry layers (4 hits to the gateway instead of 2).
-    /// In particular, [`super::auth_retry::execute_with_auth_retry`] uses
-    /// this entry point so its `must retry exactly once` contract still
-    /// holds after PR #1707 introduced the inner retry.
-    pub(crate) async fn execute_tool_once(
-        &self,
-        tool: &str,
-        arguments: Option<serde_json::Value>,
-    ) -> Result<ComposioExecuteResponse> {
-        let tool = tool.trim();
-        if tool.is_empty() {
-            anyhow::bail!("composio.execute_tool_once: tool slug must not be empty");
-        }
-        let arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default()));
-        tracing::debug!(
-            tool = %tool,
-            "[composio] execute_tool_once start"
-        );
-        let body = json!({ "tool": tool, "arguments": arguments });
-        let result = self.post_execute_tool(&body).await;
-        match &result {
-            Ok(resp) => tracing::debug!(
-                tool = %tool,
-                successful = resp.successful,
-                has_error = resp.error.is_some(),
-                "[composio] execute_tool_once completed"
-            ),
-            Err(err) => tracing::debug!(
-                tool = %tool,
-                error = %err,
-                "[composio] execute_tool_once failed"
-            ),
-        }
-        result
-    }
-
     /// `GET /agent-integrations/composio/github/repos` — list repositories
     /// available via the user's authorized GitHub connected account.
     pub async fn list_github_repos(

src/openhuman/local_ai/schemas.rs

@@ -1036,20 +1036,29 @@ fn handle_local_ai_install_whisper(params: Map<String, Value>) -> ControllerFutu
         let config = config_rpc::load_config_with_timeout().await?;
         let force = p.force.unwrap_or(false);
 
-        // Idempotency: a duplicate click while an install is already in
-        // flight should be a no-op, not a second concurrent download.
-        let current = crate::openhuman::local_ai::voice_install_common::read_status(
+        // Atomic install-start guard. A duplicate click while an install
+        // is already in flight (or a parallel auto-install firing
+        // alongside a manual click) must be a no-op — not a second
+        // concurrent download racing on the same `.part` file inside
+        // `download_to_file`. The previous read_status -> check ->
+        // write_status sequence was non-atomic and let two callers slip
+        // through; `try_acquire_install_slot` does the check-and-claim
+        // under a single mutex acquisition.
+        let slot = match crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot(
             crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER,
-        );
-        if current.state
-            == crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing
-        {
-            tracing::debug!(
-                "[voice-install:whisper] already installing — returning current status"
-            );
-            return serde_json::to_value(current)
-                .map_err(|e| format!("serialize whisper status: {e}"));
-        }
+        ) {
+            Some(slot) => slot,
+            None => {
+                tracing::debug!(
+                    "[voice-install:whisper] slot already held — returning current status"
+                );
+                let current = crate::openhuman::local_ai::voice_install_common::read_status(
+                    crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER,
+                );
+                return serde_json::to_value(current)
+                    .map_err(|e| format!("serialize whisper status: {e}"));
+            }
+        };
 
         // Mark "installing" before the spawn so the very next status poll
         // (≤ 2s away) reflects the new state without a stale read.
@@ -1073,7 +1082,12 @@ fn handle_local_ai_install_whisper(params: Map<String, Value>) -> ControllerFutu
             "[voice-install:whisper] spawning background install"
         );
         let model_size = p.model_size.clone();
+        // Move the slot into the spawned task so it lives for the actual
+        // install duration (download + extract + validate), not just the
+        // RPC handler's lifetime. The slot's Drop releases the
+        // single-writer guard on task exit, including via panic.
         tokio::spawn(async move {
+            let _slot = slot;
             if let Err(e) = crate::openhuman::local_ai::install_whisper::install_whisper(
                 &config, model_size, force,
             )
@@ -1096,16 +1110,23 @@ fn handle_local_ai_install_piper(params: Map<String, Value>) -> ControllerFuture
         let config = config_rpc::load_config_with_timeout().await?;
         let force = p.force.unwrap_or(false);
 
-        let current = crate::openhuman::local_ai::voice_install_common::read_status(
+        // See the whisper handler above for why this is an atomic slot
+        // acquisition rather than a read_status / write_status pair.
+        let slot = match crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot(
             crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER,
-        );
-        if current.state
-            == crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing
-        {
-            tracing::debug!("[voice-install:piper] already installing — returning current status");
-            return serde_json::to_value(current)
-                .map_err(|e| format!("serialize piper status: {e}"));
-        }
+        ) {
+            Some(slot) => slot,
+            None => {
+                tracing::debug!(
+                    "[voice-install:piper] slot already held — returning current status"
+                );
+                let current = crate::openhuman::local_ai::voice_install_common::read_status(
+                    crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER,
+                );
+                return serde_json::to_value(current)
+                    .map_err(|e| format!("serialize piper status: {e}"));
+            }
+        };
 
         crate::openhuman::local_ai::voice_install_common::write_status(
             crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus {
@@ -1126,7 +1147,10 @@ fn handle_local_ai_install_piper(params: Map<String, Value>) -> ControllerFuture
             "[voice-install:piper] spawning background install"
         );
         let voice_id = p.voice_id.clone();
+        // Move the slot into the spawned task — same rationale as the
+        // whisper handler.
         tokio::spawn(async move {
+            let _slot = slot;
             if let Err(e) =
                 crate::openhuman::local_ai::install_piper::install_piper(&config, voice_id, force)
                     .await

src/openhuman/local_ai/schemas_tests.rs

@@ -269,3 +269,130 @@ async fn handle_set_ollama_path_accepts_empty_string_to_clear() {
         std::env::remove_var("OPENHUMAN_WORKSPACE");
     }
 }
+
+/// Regression test for the CodeRabbit #7 race on PR #1755: when two
+/// concurrent RPC calls (e.g. a double-click, or the auto-install firing
+/// alongside a manual click) hit `handle_local_ai_install_whisper` at
+/// the same time, only one of them must spawn a real install task. The
+/// other must short-circuit and return the in-flight status without
+/// starting a second download that would race on the same `.part` file.
+///
+/// We exercise the actual handler — not just the slot primitive — so
+/// the wiring at the call site is also covered.
+#[tokio::test]
+async fn install_whisper_handler_serializes_concurrent_calls() {
+    let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    let tmp = TempDir::new().unwrap();
+    unsafe {
+        std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path());
+    }
+
+    // Pre-acquire the install slot from the test so we're guaranteed to
+    // observe the "already in flight" code path. Holding the slot here
+    // also means the handler under test will short-circuit immediately
+    // rather than spawning a real install task that would try to hit
+    // the network in CI.
+    let slot = crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot(
+        crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER,
+    )
+    .expect("test should be able to claim the slot first");
+
+    // Mark the status table as `Installing` so the handler's
+    // short-circuit branch (which reads current status to return) sees
+    // a coherent snapshot.
+    crate::openhuman::local_ai::voice_install_common::write_status(
+        crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus {
+            engine: crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER.to_string(),
+            state: crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing,
+            progress: Some(0),
+            downloaded_bytes: None,
+            total_bytes: None,
+            stage: Some("queued".to_string()),
+            error_detail: None,
+        },
+    );
+
+    // Fire two handler calls in parallel. Both must succeed and both
+    // must return the existing `Installing` status — neither must
+    // mutate or re-spawn. This is exactly the double-click / auto-fire
+    // shape described in CodeRabbit #7.
+    let (r1, r2) = tokio::join!(
+        handle_local_ai_install_whisper(Map::new()),
+        handle_local_ai_install_whisper(Map::new())
+    );
+
+    unsafe {
+        std::env::remove_var("OPENHUMAN_WORKSPACE");
+    }
+    drop(slot);
+    // Clean up so other tests see Missing.
+    crate::openhuman::local_ai::voice_install_common::reset_status(
+        crate::openhuman::local_ai::voice_install_common::ENGINE_WHISPER,
+    );
+
+    let v1 = r1.expect("first call ok");
+    let v2 = r2.expect("second call ok");
+    // Both calls must report the engine is already installing — proving
+    // the handler short-circuited rather than running the spawn path.
+    for (label, v) in [("first", &v1), ("second", &v2)] {
+        let state = v.get("state").and_then(|s| s.as_str());
+        assert_eq!(
+            state,
+            Some("installing"),
+            "{label} concurrent call should see Installing, got {v:?}"
+        );
+    }
+}
+
+/// Same regression for Piper. The two handlers share the slot
+/// infrastructure but live in separate code paths, so the wiring needs
+/// independent coverage.
+#[tokio::test]
+async fn install_piper_handler_serializes_concurrent_calls() {
+    let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    let tmp = TempDir::new().unwrap();
+    unsafe {
+        std::env::set_var("OPENHUMAN_WORKSPACE", tmp.path());
+    }
+
+    let slot = crate::openhuman::local_ai::voice_install_common::try_acquire_install_slot(
+        crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER,
+    )
+    .expect("test should be able to claim the slot first");
+
+    crate::openhuman::local_ai::voice_install_common::write_status(
+        crate::openhuman::local_ai::voice_install_common::VoiceInstallStatus {
+            engine: crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER.to_string(),
+            state: crate::openhuman::local_ai::voice_install_common::VoiceInstallState::Installing,
+            progress: Some(0),
+            downloaded_bytes: None,
+            total_bytes: None,
+            stage: Some("queued".to_string()),
+            error_detail: None,
+        },
+    );
+
+    let (r1, r2) = tokio::join!(
+        handle_local_ai_install_piper(Map::new()),
+        handle_local_ai_install_piper(Map::new())
+    );
+
+    unsafe {
+        std::env::remove_var("OPENHUMAN_WORKSPACE");
+    }
+    drop(slot);
+    crate::openhuman::local_ai::voice_install_common::reset_status(
+        crate::openhuman::local_ai::voice_install_common::ENGINE_PIPER,
+    );
+
+    let v1 = r1.expect("first call ok");
+    let v2 = r2.expect("second call ok");
+    for (label, v) in [("first", &v1), ("second", &v2)] {
+        let state = v.get("state").and_then(|s| s.as_str());
+        assert_eq!(
+            state,
+            Some("installing"),
+            "{label} concurrent call should see Installing, got {v:?}"
+        );
+    }
+}