From d1a2d51351e815dec1d65322c518a2ffa9a6c483 Mon Sep 17 00:00:00 2001
From: t8y2 <1156263951@qq.com>
Date: Thu, 17 Sep 2026 22:32:47 +0800
Subject: [PATCH] fix(plugins): block updates while connections are active

---
 .../plugins/PluginContributionsPanel.vue      |  11 +-
 .../PluginContributionsPanelBatch.spec.ts     |  31 ++-
 .../src/i18n/__tests__/backendErrors.spec.ts  |  16 ++
 apps/desktop/src/i18n/backend-errors.ts       |   4 +
 apps/desktop/src/i18n/locales/en.ts           |   3 +
 apps/desktop/src/i18n/locales/zh-CN.ts        |   3 +
 apps/desktop/src/i18n/locales/zh-TW.ts        |   3 +
 crates/dbx-core/src/connection.rs             |   2 +-
 crates/dbx-core/src/plugins.rs                |  66 +++++--
 crates/dbx-core/src/plugins/host.rs           |  72 +++++++
 crates/dbx-core/src/plugins/installer.rs      | 109 ++++++++++-
 crates/dbx-core/src/plugins/lifecycle.rs      | 183 ++++++++++++++++++
 crates/dbx-core/src/plugins/marketplace.rs    |  79 +++++++-
 crates/dbx-web/src/routes/plugins.rs          |  13 +-
 src-tauri/src/commands/plugins.rs             |  16 +-
 15 files changed, 571 insertions(+), 40 deletions(-)
 create mode 100644 crates/dbx-core/src/plugins/lifecycle.rs

diff --git a/apps/desktop/src/components/plugins/PluginContributionsPanel.vue b/apps/desktop/src/components/plugins/PluginContributionsPanel.vue
index 0e6074e380..1094760bb4 100644
--- a/apps/desktop/src/components/plugins/PluginContributionsPanel.vue
+++ b/apps/desktop/src/components/plugins/PluginContributionsPanel.vue
@@ -204,7 +204,7 @@ async function installMarketplaceListing(listing: MarketplacePluginListing) {
     installedPlugins.value = await api.listPlugins();
     selectPlugin(result.plugin.manifest.id);
   } catch (cause) {
-    toast(cause instanceof Error ? cause.message : String(cause), 8000);
+    toast(translateBackendError(t, cause), 8000);
     // A failed install can still have mutated the store (a partially replaced version directory, for
     // instance), so re-read the installed list instead of leaving the card on state it may no longer
     // describe.
@@ -218,8 +218,9 @@ function batchSummaryKey(outcome: { succeeded: unknown[]; failed: { name: string
   return outcome.failed.length ? "pluginPlatform.batchSummaryWithFailures" : "pluginPlatform.batchSummary";
 }
 
-function reportBatchSummary(outcome: { succeeded: unknown[]; failed: { name: string }[] }) {
+function reportBatchSummary(outcome: { succeeded: unknown[]; failed: { name: string; error: string }[] }) {
   const failedNames = outcome.failed.map((failure) => failure.name).join("、");
+  error.value = outcome.failed.map((failure) => `${failure.name}: ${translateBackendError(t, failure.error)}`).join("\n");
   toast(t(batchSummaryKey(outcome), { success: outcome.succeeded.length, failed: outcome.failed.length, names: failedNames }), outcome.failed.length ? 8000 : 4000);
 }
 
@@ -227,7 +228,7 @@ async function refreshAfterBatch() {
   try {
     installedPlugins.value = await api.listPlugins();
   } catch (cause) {
-    error.value = t("pluginPlatform.batchRefreshFailed", { error: cause instanceof Error ? cause.message : String(cause) });
+    error.value = [error.value, t("pluginPlatform.batchRefreshFailed", { error: cause instanceof Error ? cause.message : String(cause) })].filter(Boolean).join("\n");
   }
 }
 
@@ -661,7 +662,7 @@ async function rollbackSelectedPlugin() {
     installedPlugins.value = await api.listPlugins();
     selectPlugin(result.plugin.manifest.id);
   } catch (cause) {
-    toast(cause instanceof Error ? cause.message : String(cause), 5000);
+    toast(translateBackendError(t, cause), 8000);
   } finally {
     operating.value = false;
   }
@@ -732,7 +733,7 @@ onBeforeUnmount(() => {
 <template>
   <div ref="panelRootRef" class="plugin-center-view relative mx-auto flex h-full w-full max-w-6xl flex-col gap-4 overflow-hidden px-6 py-6" @dragenter="onWebDragEnter" @dragover="onWebDragOver" @dragleave="onWebDragLeave" @drop="onWebDrop">
     <input ref="webFileInput" type="file" accept=".dbxp" class="hidden" @change="handleWebPackage" />
-    <div v-if="error" class="shrink-0 rounded-lg border border-destructive/30 bg-destructive/5 px-3 py-2 text-xs text-destructive">{{ error }}</div>
+    <div v-if="error" class="shrink-0 whitespace-pre-wrap rounded-lg border border-destructive/30 bg-destructive/5 px-3 py-2 text-xs text-destructive">{{ error }}</div>
 
     <Tabs v-model="activeSection" class="min-h-0 flex-1 gap-3">
       <TabsList class="grid h-9 w-full grid-cols-3">
diff --git a/apps/desktop/src/components/plugins/__tests__/PluginContributionsPanelBatch.spec.ts b/apps/desktop/src/components/plugins/__tests__/PluginContributionsPanelBatch.spec.ts
index d036a341d6..99888c36d9 100644
--- a/apps/desktop/src/components/plugins/__tests__/PluginContributionsPanelBatch.spec.ts
+++ b/apps/desktop/src/components/plugins/__tests__/PluginContributionsPanelBatch.spec.ts
@@ -366,6 +366,33 @@ describe("PluginContributionsPanel mutation exclusion", () => {
 });
 
 describe("PluginContributionsPanel completed batch outcomes", () => {
+  it.each(["marketplace", "package", "url", "rollback"] as const)("shows related connection names when %s is blocked", async (entry) => {
+    const blocked = new Error("Plugin update blocked by active connections: Production S3");
+    const mutations = {
+      marketplace: mocks.installMarketplacePlugin,
+      package: mocks.installPluginPackage,
+      url: mocks.installPluginPackageFromUrl,
+      rollback: mocks.rollbackPlugin,
+    };
+    mutations[entry].mockRejectedValueOnce(blocked);
+    if (entry === "marketplace") await state.installMarketplaceListing(state.marketplaceListings[0]);
+    else if (entry === "package") await state.installPlugin("plugin.dbxp");
+    else if (entry === "url") await state.installPluginFromUrl();
+    else await state.rollbackSelectedPlugin();
+    expect(mocks.toast).toHaveBeenLastCalledWith('pluginPlatform.updateBlockedByConnections:{"labels":"Production S3"}', 8000);
+    expect(mocks.listPlugins).toHaveBeenCalledTimes(entry === "marketplace" ? 1 : 0);
+  });
+
+  it("shows per-plugin blockers while continuing other batch updates", async () => {
+    state.selectAllUpdatable();
+    mocks.installMarketplacePlugin.mockRejectedValueOnce(new Error("Plugin update blocked by active connections: Production S3"));
+    await state.runBatchInstallUpdate();
+    expect(mocks.installMarketplacePlugin).toHaveBeenCalledTimes(3);
+    expect(state.error).toBe('a: pluginPlatform.updateBlockedByConnections:{"labels":"Production S3"}');
+    await nextTick();
+    expect(host.textContent).toContain("Production S3");
+  });
+
   it.each(batches.flatMap((batch) => [0, 1, 3].map((failures) => ({ batch, failures }))))("preserves batch $batch outcomes with $failures failures when refresh rejects", async ({ batch, failures }) => {
     state.selectAllUpdatable();
     state.selectedInstalledIds = new Set(["a", "b", "c"]);
@@ -390,7 +417,7 @@ describe("PluginContributionsPanel completed batch outcomes", () => {
     expect(mutation).toHaveBeenCalledTimes(3);
     expect(mocks.toast).toHaveBeenCalledTimes(1);
     expect(mocks.toast).toHaveBeenLastCalledWith(summary, failures ? 8000 : 4000);
-    expect(state.error).toBe('pluginPlatform.batchRefreshFailed:{"error":"refresh offline"}');
+    expect(state.error).toBe([...["a", "b", "c"].slice(0, failures).map((name) => `${name}: denied`), 'pluginPlatform.batchRefreshFailed:{"error":"refresh offline"}'].join("\n"));
     await nextTick();
     expect(host.textContent).toContain(state.error);
     expect(state.batchRunning).toBe(false);
@@ -414,7 +441,7 @@ describe("PluginContributionsPanel completed batch outcomes", () => {
     expect(mutation.mock.calls.map(([request]) => (batch === "install" ? request.pluginId : request))).toEqual(["a", "b", "c"]);
     expect(mocks.listPlugins).toHaveBeenCalledTimes(1);
     expect(mocks.toast).toHaveBeenLastCalledWith('pluginPlatform.batchSummaryWithFailures:{"success":2,"failed":1,"names":"b"}', 8000);
-    expect(state.error).toBe("");
+    expect(state.error).toBe("b: denied");
     expect(state.batchRunning).toBe(false);
     expect(state.selectedListingKeys.size).toBe(0);
     expect(state.selectedInstalledIds.size).toBe(0);
diff --git a/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts b/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts
index 1fd255b40c..e099d5dafb 100644
--- a/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts
+++ b/apps/desktop/src/i18n/__tests__/backendErrors.spec.ts
@@ -60,6 +60,22 @@ const WINDOWS_JRE_REMOVE_ERROR = [
 // Every backend message changed away from hardcoded Chinese, paired with the
 // key and params it must resolve to.
 const CASES: { name: string; message: string; key: string; params?: Record<string, string> }[] = [
+  {
+    name: "plugin update requires closing related connections",
+    message: "Plugin update blocked by active connections: Production S3, Local storage",
+    key: "pluginPlatform.updateBlockedByConnections",
+    params: { labels: "Production S3, Local storage" },
+  },
+  {
+    name: "plugin update waits for active operations",
+    message: "Plugin update blocked by active operations. Please wait for them to finish.",
+    key: "pluginPlatform.updateBlockedByOperations",
+  },
+  {
+    name: "connection admission waits for plugin update",
+    message: "Plugin update is in progress. Please try again after it finishes.",
+    key: "pluginPlatform.updateInProgress",
+  },
   {
     name: "Nacos ordinary user must configure managed namespaces when namespace discovery is forbidden",
     message: "Failed to list Nacos namespaces: NACOS_ERROR[v3ManagedNamespacesRequired]: access denied",
diff --git a/apps/desktop/src/i18n/backend-errors.ts b/apps/desktop/src/i18n/backend-errors.ts
index 2fe8b1c135..1be5b5739d 100644
--- a/apps/desktop/src/i18n/backend-errors.ts
+++ b/apps/desktop/src/i18n/backend-errors.ts
@@ -82,11 +82,14 @@ const taggedAiCliErrorKeys: Record<string, string> = {
 };
 
 const exactMessageKeys: Record<string, string> = {
+  "Plugin update is in progress. Please try again after it finishes.": "pluginPlatform.updateInProgress",
+  "Plugin update blocked by active operations. Please wait for them to finish.": "pluginPlatform.updateBlockedByOperations",
   [PHOENIX_DRIVER_NOT_INSTALLED_ERROR]: "connection.phoenixDriverNotInstalled",
   [PHOENIX_JDBC_PLUGIN_NOT_INSTALLED_ERROR]: "connection.phoenixDriverNotInstalled",
 };
 
 const patterns: [RegExp, string][] = [
+  [/^Plugin update blocked by active connections: ([\s\S]+)$/, "pluginPlatform.updateBlockedByConnections"],
   [/^(.+?) driver is not installed\. Please install it from the Driver Manager\.$/, "connection.driverNotInstalled"],
   [/^JRE (.+?) runtime is not installed\. Please install it from the Driver Manager\.$/, "connection.jreNotInstalled"],
   [/^System Java runtime was not found on PATH\. Please install Java or choose a custom Java executable\.$/, "connection.systemJavaNotFound"],
@@ -164,6 +167,7 @@ const patterns: [RegExp, string][] = [
 // Named placeholders for each pattern's capture groups, in capture order.
 // A bare string is shorthand for a single capture group.
 const paramNames: Record<string, string | string[]> = {
+  "pluginPlatform.updateBlockedByConnections": "labels",
   "connection.driverNotInstalled": "driver",
   "connection.jreNotInstalled": "jre",
   "ai.configNameExists": "name",
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index f11444d20b..9088a5e6e3 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -195,6 +195,9 @@ export default {
       unsupported: "Unsupported",
     },
     updateSuccess: "Updated {name} to {version}",
+    updateBlockedByConnections: "Close these connections before updating or rolling back the plugin: {labels}",
+    updateBlockedByOperations: "Wait for this plugin's active operations to finish before updating or rolling back.",
+    updateInProgress: "This plugin is being updated. Try again after the update finishes.",
     noInstalledPlugins: "No plugins installed",
     browseMarketplace: "Browse Marketplace",
     localInstallTitle: "Install a plugin package",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index f87fcaf7d9..8a6f540dd6 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -115,6 +115,9 @@ export default withEnglishFallback({
     pluginHomepage: "主页",
     marketplaceStatus: { install: "安装", installed: "已安装", update: "更新", unsupported: "不支持" },
     updateSuccess: "已将 {name} 更新到 {version}",
+    updateBlockedByConnections: "请先关闭以下连接后再更新或回滚插件：{labels}",
+    updateBlockedByOperations: "该插件仍有操作正在执行，请等待完成后再更新或回滚。",
+    updateInProgress: "该插件正在更新，请等待完成后重试。",
     noInstalledPlugins: "尚未安装插件",
     browseMarketplace: "浏览插件商店",
     localInstallTitle: "安装插件包",
diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts
index 25f1c771ce..5bea667ee5 100644
--- a/apps/desktop/src/i18n/locales/zh-TW.ts
+++ b/apps/desktop/src/i18n/locales/zh-TW.ts
@@ -193,6 +193,9 @@ export default withEnglishFallback({
     pluginHomepage: "主頁",
     marketplaceStatus: { install: "安裝", installed: "已安裝", update: "更新", unsupported: "不支援" },
     updateSuccess: "已將 {name} 更新至 {version}",
+    updateBlockedByConnections: "請先關閉以下連線後再更新或回復外掛：{labels}",
+    updateBlockedByOperations: "此外掛仍有操作正在執行，請等待完成後再更新或回復。",
+    updateInProgress: "此外掛正在更新，請等待完成後重試。",
     noInstalledPlugins: "尚未安裝外掛",
     browseMarketplace: "瀏覽外掛商店",
     localInstallTitle: "安裝外掛套件",
diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs
index 16804b62d6..64cc2f4153 100644
--- a/crates/dbx-core/src/connection.rs
+++ b/crates/dbx-core/src/connection.rs
@@ -1560,7 +1560,7 @@ impl AppState {
 
     pub async fn external_driver_pool(&self, driver_id: &str, config: &ConnectionConfig) -> Result<PoolKind, String> {
         let env = self.external_driver_runtime_env(driver_id)?;
-        let session = self.plugins.start_driver_session_with_env(driver_id, env).await?;
+        let session = self.plugins.start_driver_session_for_connection(driver_id, env, &config.name).await?;
         let params = serde_json::json!({ "connection": config });
         session
             .invoke_with_timeout::<serde_json::Value>("connect", params, Some(external_driver_connect_timeout(config)))
diff --git a/crates/dbx-core/src/plugins.rs b/crates/dbx-core/src/plugins.rs
index b676f5aa42..7c42f05bdb 100644
--- a/crates/dbx-core/src/plugins.rs
+++ b/crates/dbx-core/src/plugins.rs
@@ -10,6 +10,7 @@ mod assets;
 mod filesystem;
 mod host;
 mod installer;
+mod lifecycle;
 mod manifest;
 mod marketplace;
 mod runtime;
@@ -28,6 +29,7 @@ pub use installer::{
     PluginRollbackResult, PluginSignatureStatus, PluginTrustStore, PluginTrustedKey, DBXP_EXTENSION,
     MAX_PLUGIN_PACKAGE_BYTES, PLUGIN_CHECKSUMS_FILE, PLUGIN_SIGNATURE_FILE,
 };
+pub use lifecycle::PluginLifecycle;
 pub use marketplace::{
     url_install_trust_store, PluginMarketplace, PluginMarketplaceArtifact, PluginMarketplaceCatalog,
     PluginMarketplaceInstallRequest, PluginMarketplaceLocalization, PluginMarketplacePlugin,
@@ -139,6 +141,7 @@ impl PluginRuntimeEnv {
 pub struct PluginRegistry {
     root_dir: PathBuf,
     app_version: String,
+    lifecycle: PluginLifecycle,
 }
 
 impl PluginRegistry {
@@ -147,7 +150,11 @@ impl PluginRegistry {
     }
 
     pub fn new_with_app_version(root_dir: PathBuf, app_version: impl Into<String>) -> Self {
-        Self { root_dir, app_version: app_version.into() }
+        Self { root_dir, app_version: app_version.into(), lifecycle: PluginLifecycle::default() }
+    }
+
+    pub fn lifecycle(&self) -> PluginLifecycle {
+        self.lifecycle.clone()
     }
 
     pub fn root_dir(&self) -> &Path {
@@ -282,12 +289,30 @@ impl PluginRegistry {
         &self,
         driver_id: &str,
         env: PluginRuntimeEnv,
+    ) -> Result<Arc<PluginDriverSession>, String> {
+        self.start_driver_session_for_connection(driver_id, env, driver_id).await
+    }
+
+    pub async fn start_driver_session_for_connection(
+        &self,
+        driver_id: &str,
+        env: PluginRuntimeEnv,
+        connection_label: &str,
     ) -> Result<Arc<PluginDriverSession>, String> {
         let plugin =
             self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
+        let activity = self.lifecycle.begin_connection(&plugin.manifest.id, connection_label)?;
+        let current =
+            self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
+        if current.manifest.id != plugin.manifest.id {
+            return Err(format!("Plugin driver '{driver_id}' changed. Please reconnect."));
+        }
+        let plugin = current;
         ensure_plugin_compatible(&plugin)?;
         let env = env.with_plugin_data_dir(&self.plugin_data_dir(&plugin.manifest.id));
-        PluginDriverSession::start(plugin, driver_id.to_string(), self.app_version.clone(), env).await.map(Arc::new)
+        PluginDriverSession::start(plugin, driver_id.to_string(), self.app_version.clone(), env, activity)
+            .await
+            .map(Arc::new)
     }
 }
 
@@ -326,6 +351,7 @@ fn ensure_plugin_compatible(plugin: &InstalledPlugin) -> Result<(), String> {
 pub struct PluginDriverSession {
     sidecar: Arc<PluginSidecarSession>,
     driver_id: String,
+    _activity: lifecycle::PluginUsageGuard,
 }
 
 impl PluginDriverSession {
@@ -334,9 +360,10 @@ impl PluginDriverSession {
         driver_id: String,
         app_version: String,
         env: PluginRuntimeEnv,
+        activity: lifecycle::PluginUsageGuard,
     ) -> Result<Self, String> {
         let sidecar = PluginSidecarSession::start(plugin, app_version, env).await?;
-        Ok(Self { sidecar, driver_id })
+        Ok(Self { sidecar, driver_id, _activity: activity })
     }
 
     pub async fn invoke<T>(&self, method: &str, params: serde_json::Value) -> Result<T, String>
@@ -384,7 +411,8 @@ impl PluginDriverSession {
         driver_id: String,
         env: PluginRuntimeEnv,
     ) -> Result<Self, String> {
-        Self::start(plugin, driver_id, env!("CARGO_PKG_VERSION").to_string(), env).await
+        let activity = PluginLifecycle::default().begin_connection(&plugin.manifest.id, &driver_id)?;
+        Self::start(plugin, driver_id, env!("CARGO_PKG_VERSION").to_string(), env, activity).await
     }
 }
 
@@ -472,6 +500,7 @@ sleep 30
         .unwrap();
 
         let registry = PluginRegistry::new_with_app_version(data_dir.path().join("plugins"), "0.5.67");
+        let lifecycle = registry.lifecycle();
         let host = PluginHost::new(registry);
         let reported: String =
             host.invoke("sample.sidecar", "sample/dataDir", serde_json::Value::Null, None, None).await.unwrap();
@@ -480,6 +509,24 @@ sleep 30
             data_dir.path().join("plugin-data").join("sample.sidecar"),
             "sidecar must see <data dir>/plugin-data/<id> in DBX_PLUGIN_DATA_DIR"
         );
+        assert!(lifecycle.begin_update("sample.sidecar").is_ok());
+        let request_host = host.clone();
+        let pending = tokio::spawn(async move {
+            request_host
+                .invoke::<serde_json::Value>("sample.sidecar", "sample/wait", serde_json::Value::Null, None, None)
+                .await
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while lifecycle.check_update("sample.sidecar").is_ok() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .unwrap();
+        assert!(lifecycle.begin_update("sample.sidecar").unwrap_err().contains("active operations"));
+        pending.abort();
+        assert!(pending.await.unwrap_err().is_cancelled());
+        assert!(lifecycle.begin_update("sample.sidecar").is_ok());
         host.stop_all().await;
     }
 
@@ -518,14 +565,9 @@ sleep 30
             env!("CARGO_PKG_VERSION"),
         );
 
-        let session = PluginDriverSession::start(
-            plugin,
-            "jdbc".to_string(),
-            env!("CARGO_PKG_VERSION").to_string(),
-            PluginRuntimeEnv::default(),
-        )
-        .await
-        .expect("session should start");
+        let session = PluginDriverSession::start_for_test(plugin, "jdbc".to_string(), PluginRuntimeEnv::default())
+            .await
+            .expect("session should start");
         let pid = session.pid().await.expect("child should have a pid");
 
         session.shutdown().await;
diff --git a/crates/dbx-core/src/plugins/host.rs b/crates/dbx-core/src/plugins/host.rs
index 1710a65c55..ee341b718e 100644
--- a/crates/dbx-core/src/plugins/host.rs
+++ b/crates/dbx-core/src/plugins/host.rs
@@ -45,6 +45,7 @@ pub struct PluginConnectionHandle {
     session: Option<Arc<PluginSidecarSession>>,
     disconnect: bool,
     params: serde_json::Value,
+    _activity: super::lifecycle::PluginUsageGuard,
 }
 
 impl PluginConnectionHandle {
@@ -115,6 +116,7 @@ impl PluginHost {
         plugin_id: &str,
         env: PluginRuntimeEnv,
     ) -> Result<Arc<PluginSidecarSession>, String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         if let Some(session) = self.running_session(plugin_id).await {
             return Ok(session);
         }
@@ -151,6 +153,7 @@ impl PluginHost {
     where
         T: DeserializeOwned,
     {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.invoke_with_timeout(method, params, None, timeout).await
@@ -163,6 +166,7 @@ impl PluginHost {
         params: serde_json::Value,
         required_permission: Option<&str>,
     ) -> Result<(), String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.notify(method, params, None).await
@@ -175,6 +179,7 @@ impl PluginHost {
         data: &[u8],
         required_permission: Option<&str>,
     ) -> Result<(), String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.send_binary(channel, data).await
@@ -195,6 +200,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<ConnectionTestResult, String> {
+        let _activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (plugin, provider) = self.resolve_connection_provider(config)?;
         validate_plugin_connection_values(config, &provider)?;
         let provider_label = provider.label.as_deref().unwrap_or(&plugin.manifest.name);
@@ -219,6 +229,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<PluginConnectionHandle, String> {
+        let activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (_, provider) = self.resolve_connection_provider(config)?;
         validate_plugin_connection_values(config, &provider)?;
         let params = plugin_connection_params(config, &provider, runtime_host, runtime_port)?;
@@ -250,6 +265,7 @@ impl PluginHost {
             session,
             disconnect,
             params,
+            _activity: activity,
         })
     }
 
@@ -260,6 +276,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<PluginConnectionActionResult, String> {
+        let _activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (_, provider) = self.resolve_connection_provider(config)?;
         let action = plugin_invoke_connection_action(&provider, action_id)?;
         validate_plugin_connection_values_for_action(config, &provider, action.requires_valid_form)?;
@@ -738,6 +759,57 @@ mod tests {
     use crate::models::connection::ConnectionConfig;
     use crate::plugins::PluginConnectionProviderContribution;
 
+    #[tokio::test]
+    async fn ui_only_connections_hold_update_guards_but_saved_configs_do_not() {
+        let root = tempfile::tempdir().unwrap();
+        let plugin_dir = root.path().join("sample.ui");
+        std::fs::create_dir(&plugin_dir).unwrap();
+        std::fs::write(plugin_dir.join("index.html"), "<html></html>").unwrap();
+        std::fs::write(
+            plugin_dir.join("manifest.json"),
+            serde_json::json!({
+                "manifest_version": 1,
+                "id": "sample.ui",
+                "name": "Sample UI",
+                "version": "1.0.0",
+                "publisher": "sample",
+                "engines": { "dbx": ">=0.5.0", "host_api": "^1.0" },
+                "entrypoints": { "ui": { "entry": "index.html" } },
+                "contributions": [{
+                    "type": "connection-provider",
+                    "id": "sample.connection",
+                    "database_type": "sample"
+                }]
+            })
+            .to_string(),
+        )
+        .unwrap();
+        let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
+            "id": "saved-connection",
+            "name": "Saved UI connection",
+            "db_type": "plugin",
+            "host": "localhost",
+            "port": 0,
+            "username": "",
+            "password": "",
+            "plugin_id": "sample.ui",
+            "plugin_connection_provider": "sample.connection",
+            "plugin_connection_type": "sample"
+        }))
+        .unwrap();
+        let registry = crate::plugins::PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let host = super::PluginHost::new(registry);
+        let update = lifecycle.begin_update("sample.ui").unwrap();
+        assert!(host.connect_connection(&config, "localhost", 0).await.is_err());
+        drop(update);
+        let connection = host.connect_connection(&config, "localhost", 0).await.unwrap();
+        assert!(lifecycle.begin_update("sample.ui").unwrap_err().contains("Saved UI connection"));
+        connection.disconnect().await.unwrap();
+        drop(connection);
+        assert!(lifecycle.begin_update("sample.ui").is_ok());
+    }
+
     #[test]
     fn rejects_zero_port_for_port_bound_connection_field() {
         let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
diff --git a/crates/dbx-core/src/plugins/installer.rs b/crates/dbx-core/src/plugins/installer.rs
index c73e286801..a94b5f4bd0 100644
--- a/crates/dbx-core/src/plugins/installer.rs
+++ b/crates/dbx-core/src/plugins/installer.rs
@@ -12,7 +12,8 @@ use semver::Version;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 
-use super::{InstalledPlugin, InstalledPluginInfo, PluginManifest};
+use super::lifecycle::PluginUsageGuard;
+use super::{InstalledPlugin, InstalledPluginInfo, PluginLifecycle, PluginManifest};
 
 pub const DBXP_EXTENSION: &str = "dbxp";
 pub const PLUGIN_CHECKSUMS_FILE: &str = "checksums.json";
@@ -48,12 +49,14 @@ pub struct PluginInstallResult {
     pub previous_version: Option<String>,
     pub package_sha256: String,
     pub signature: PluginSignatureStatus,
+    _update_guard: Option<PluginUsageGuard>,
 }
 
 #[derive(Debug, Clone)]
 pub struct PluginRollbackResult {
     pub plugin: InstalledPlugin,
     pub previous_version: String,
+    _update_guard: Option<PluginUsageGuard>,
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -135,6 +138,7 @@ pub struct PluginPackageInstaller {
     root_dir: PathBuf,
     app_version: String,
     trust_store: Arc<PluginTrustStore>,
+    lifecycle: Option<PluginLifecycle>,
 }
 
 #[derive(Debug, Clone)]
@@ -256,11 +260,16 @@ fn read_trust_document(root_dir: &Path) -> Result<PluginTrustDocument, String> {
 impl PluginPackageInstaller {
     pub fn new(root_dir: PathBuf, app_version: impl Into<String>) -> Result<Self, String> {
         let trust_store = super::marketplace::package_install_trust_store(&root_dir)?;
-        Ok(Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store) })
+        Ok(Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store), lifecycle: None })
     }
 
     pub fn with_trust_store(root_dir: PathBuf, app_version: impl Into<String>, trust_store: PluginTrustStore) -> Self {
-        Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store) }
+        Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store), lifecycle: None }
+    }
+
+    pub fn with_lifecycle(mut self, lifecycle: PluginLifecycle) -> Self {
+        self.lifecycle = Some(lifecycle);
+        self
     }
 
     pub fn install_file(
@@ -374,6 +383,7 @@ impl PluginPackageInstaller {
         }
         make_backend_executable(&compatibility.backend_executable)?;
 
+        let update_guard = self.lifecycle.as_ref().map(|lifecycle| lifecycle.begin_update(&manifest.id)).transpose()?;
         let container_dir = self.root_dir.join(&manifest.id);
         migrate_legacy_container(&container_dir)?;
         let versions_dir = container_dir.join(VERSIONS_DIR);
@@ -460,10 +470,11 @@ impl PluginPackageInstaller {
             log::warn!("Failed to prune plugin '{}' install history: {error}", manifest.id);
         }
         let plugin = InstalledPlugin::new(manifest, version_dir, &self.app_version);
-        Ok(PluginInstallResult { plugin, previous_version, package_sha256, signature })
+        Ok(PluginInstallResult { plugin, previous_version, package_sha256, signature, _update_guard: update_guard })
     }
 
     fn rollback_locked(&self, plugin_id: &str) -> Result<PluginRollbackResult, String> {
+        let update_guard = self.lifecycle.as_ref().map(|lifecycle| lifecycle.begin_update(plugin_id)).transpose()?;
         let container_dir = self.root_dir.join(plugin_id);
         let current = read_latest_activation(&container_dir)?
             .ok_or_else(|| format!("Plugin '{plugin_id}' does not have an active version"))?;
@@ -501,7 +512,7 @@ impl PluginPackageInstaller {
         if let Err(error) = prune_plugin_history(&container_dir, &activation) {
             log::warn!("Failed to prune plugin '{plugin_id}' rollback history: {error}");
         }
-        Ok(PluginRollbackResult { plugin, previous_version })
+        Ok(PluginRollbackResult { plugin, previous_version, _update_guard: update_guard })
     }
 }
 
@@ -1019,6 +1030,94 @@ mod tests {
     };
     use crate::plugins::{PluginManifest, PluginRegistry};
 
+    #[test]
+    fn runtime_updates_and_rollbacks_preserve_active_connections_and_versions() {
+        let root = tempfile::tempdir().unwrap();
+        let registry = PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let installer =
+            PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", PluginTrustStore::default())
+                .with_lifecycle(lifecycle.clone());
+        let first = package("1.0.0", None, false);
+        let second = package("1.1.0", None, false);
+        installer.install_bytes(&first, PluginInstallPolicy::LocalDevelopment).unwrap();
+        let connection = lifecycle.begin_connection("sample.hello", "My connection").unwrap();
+        for bytes in [&first, &second] {
+            let error = installer.install_bytes(bytes, PluginInstallPolicy::LocalDevelopment).unwrap_err();
+            assert!(error.contains("My connection"));
+            assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.0.0");
+        }
+        assert!(!root.path().join("sample.hello/versions/1.1.0").exists());
+        drop(connection);
+
+        let unrelated = lifecycle.begin_connection("other.plugin", "Unrelated connection").unwrap();
+        let installed = installer.install_bytes(&second, PluginInstallPolicy::LocalDevelopment).unwrap();
+        assert!(lifecycle.begin_connection("sample.hello", "New connection").is_err());
+        assert!(lifecycle.begin_operation("sample.hello").is_err());
+        drop(installed);
+        let connection = lifecycle.begin_connection("sample.hello", "My connection").unwrap();
+        assert!(installer.rollback("sample.hello").unwrap_err().contains("My connection"));
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.1.0");
+        drop(connection);
+        let rollback = installer.rollback("sample.hello").unwrap();
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.0.0");
+        assert!(lifecycle.begin_operation("sample.hello").is_err());
+        drop(rollback);
+        assert!(lifecycle.begin_connection("sample.hello", "Reconnected").is_ok());
+        drop(unrelated);
+    }
+
+    #[test]
+    fn runtime_install_file_and_signed_packages_share_the_update_guard() {
+        let root = tempfile::tempdir().unwrap();
+        let registry = PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let signer = SigningKey::from_bytes(&[23; 32]);
+        let trust = PluginTrustStore::from_base64_keys(BTreeMap::from([(
+            "test".to_string(),
+            base64::engine::general_purpose::STANDARD.encode(signer.verifying_key().as_bytes()),
+        )]))
+        .unwrap();
+        let installer = PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", trust)
+            .with_lifecycle(lifecycle.clone());
+        installer
+            .install_bytes(&package("1.0.0", Some((&signer, "test")), false), PluginInstallPolicy::LocalSigned)
+            .unwrap();
+        let bytes = package("1.1.0", Some((&signer, "test")), false);
+        let package_path = root.path().join("update.dbxp");
+        std::fs::write(&package_path, &bytes).unwrap();
+        let operation = lifecycle.begin_operation("sample.hello").unwrap();
+        assert!(installer
+            .install_file(&package_path, PluginInstallPolicy::LocalSigned)
+            .unwrap_err()
+            .contains("active operations"));
+        let expectation = PluginPackageExpectation {
+            id: "sample.hello".to_string(),
+            version: "1.1.0".to_string(),
+            publisher: "sample".to_string(),
+            permissions: BTreeSet::from(["host.events".to_string()]),
+            signing_key_id: "test".to_string(),
+        };
+        assert!(installer.install_marketplace_bytes(&bytes, &expectation).unwrap_err().contains("active operations"));
+        drop(operation);
+        installer.install_marketplace_bytes(&bytes, &expectation).unwrap();
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.1.0");
+    }
+
+    #[test]
+    fn failed_runtime_update_releases_the_admission_guard() {
+        let root = tempfile::tempdir().unwrap();
+        let lifecycle = crate::plugins::PluginLifecycle::default();
+        let installer =
+            PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", PluginTrustStore::default())
+                .with_lifecycle(lifecycle.clone());
+        installer.install_bytes(&package("1.0.0", None, false), PluginInstallPolicy::LocalDevelopment).unwrap();
+        assert!(installer.rollback("sample.hello").is_err());
+        assert!(lifecycle.begin_connection("sample.hello", "Retry").is_ok());
+        assert!(installer.install_bytes(&package("1.1.0", None, true), PluginInstallPolicy::LocalDevelopment).is_err());
+        assert!(lifecycle.begin_update("sample.hello").is_ok());
+    }
+
     #[test]
     fn installs_versioned_package_and_rolls_back_atomically() {
         let root = tempfile::tempdir().unwrap();
diff --git a/crates/dbx-core/src/plugins/lifecycle.rs b/crates/dbx-core/src/plugins/lifecycle.rs
new file mode 100644
index 0000000000..00b592cafb
--- /dev/null
+++ b/crates/dbx-core/src/plugins/lifecycle.rs
@@ -0,0 +1,183 @@
+use std::collections::{BTreeMap, HashMap};
+use std::sync::{Arc, Mutex};
+
+const UPDATE_IN_PROGRESS: &str = "Plugin update is in progress. Please try again after it finishes.";
+const OPERATIONS_ACTIVE: &str = "Plugin update blocked by active operations. Please wait for them to finish.";
+
+#[derive(Debug, Clone, Default)]
+pub struct PluginLifecycle {
+    usage: Arc<Mutex<HashMap<String, PluginUsage>>>,
+}
+
+#[derive(Debug, Default)]
+struct PluginUsage {
+    connections: BTreeMap<String, usize>,
+    operations: usize,
+    updating: bool,
+}
+
+impl PluginUsage {
+    fn ensure_updatable(&self) -> Result<(), String> {
+        if self.updating {
+            return Err(UPDATE_IN_PROGRESS.to_string());
+        }
+        if !self.connections.is_empty() {
+            return Err(format!(
+                "Plugin update blocked by active connections: {}",
+                self.connections.keys().cloned().collect::<Vec<_>>().join(", ")
+            ));
+        }
+        if self.operations > 0 {
+            return Err(OPERATIONS_ACTIVE.to_string());
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct PluginUsageGuard {
+    _lease: Arc<PluginUsageLease>,
+}
+
+#[derive(Debug)]
+struct PluginUsageLease {
+    lifecycle: PluginLifecycle,
+    plugin_id: String,
+    kind: PluginUsageKind,
+}
+
+#[derive(Debug)]
+enum PluginUsageKind {
+    Connection(String),
+    Operation,
+    Update,
+}
+
+impl PluginLifecycle {
+    pub(crate) fn check_update(&self, plugin_id: &str) -> Result<(), String> {
+        let usage = self.usage.lock().unwrap_or_else(|error| error.into_inner());
+        usage.get(plugin_id).map_or(Ok(()), PluginUsage::ensure_updatable)
+    }
+
+    pub(crate) fn begin_connection(&self, plugin_id: &str, label: &str) -> Result<PluginUsageGuard, String> {
+        let label = if label.trim().is_empty() { plugin_id } else { label };
+        self.acquire(plugin_id, PluginUsageKind::Connection(label.to_string()))
+    }
+
+    pub(crate) fn begin_operation(&self, plugin_id: &str) -> Result<PluginUsageGuard, String> {
+        self.acquire(plugin_id, PluginUsageKind::Operation)
+    }
+
+    pub(crate) fn begin_update(&self, plugin_id: &str) -> Result<PluginUsageGuard, String> {
+        self.acquire(plugin_id, PluginUsageKind::Update)
+    }
+
+    fn acquire(&self, plugin_id: &str, kind: PluginUsageKind) -> Result<PluginUsageGuard, String> {
+        let mut plugins = self.usage.lock().unwrap_or_else(|error| error.into_inner());
+        let usage = plugins.entry(plugin_id.to_string()).or_default();
+        if usage.updating {
+            return Err(UPDATE_IN_PROGRESS.to_string());
+        }
+        match &kind {
+            PluginUsageKind::Connection(label) => *usage.connections.entry(label.clone()).or_default() += 1,
+            PluginUsageKind::Operation => usage.operations += 1,
+            PluginUsageKind::Update => {
+                usage.ensure_updatable()?;
+                usage.updating = true;
+            }
+        }
+        Ok(PluginUsageGuard {
+            _lease: Arc::new(PluginUsageLease { lifecycle: self.clone(), plugin_id: plugin_id.to_string(), kind }),
+        })
+    }
+}
+
+impl Drop for PluginUsageLease {
+    fn drop(&mut self) {
+        let mut plugins = self.lifecycle.usage.lock().unwrap_or_else(|error| error.into_inner());
+        let Some(usage) = plugins.get_mut(&self.plugin_id) else {
+            return;
+        };
+        match &self.kind {
+            PluginUsageKind::Connection(label) => {
+                if let Some(count) = usage.connections.get_mut(label) {
+                    *count -= 1;
+                    if *count == 0 {
+                        usage.connections.remove(label);
+                    }
+                }
+            }
+            PluginUsageKind::Operation => usage.operations -= 1,
+            PluginUsageKind::Update => usage.updating = false,
+        }
+        if usage.connections.is_empty() && usage.operations == 0 && !usage.updating {
+            plugins.remove(&self.plugin_id);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn connections_block_only_their_plugin_until_all_handles_are_dropped() {
+        let lifecycle = PluginLifecycle::default();
+        let connection = lifecycle.begin_connection("s3", "Production S3").unwrap();
+        let clone = connection.clone();
+        let second = lifecycle.begin_connection("s3", "Production S3").unwrap();
+        assert_eq!(
+            lifecycle.begin_update("s3").unwrap_err(),
+            "Plugin update blocked by active connections: Production S3"
+        );
+        assert!(lifecycle.begin_update("other").is_ok());
+        drop(connection);
+        drop(second);
+        assert!(lifecycle.begin_update("s3").is_err());
+        drop(clone);
+        assert!(lifecycle.begin_update("s3").is_ok());
+    }
+
+    #[test]
+    fn operations_block_updates_without_connections() {
+        let lifecycle = PluginLifecycle::default();
+        let operation = lifecycle.begin_operation("tool").unwrap();
+        assert_eq!(lifecycle.begin_update("tool").unwrap_err(), OPERATIONS_ACTIVE);
+        drop(operation);
+        assert!(lifecycle.begin_update("tool").is_ok());
+    }
+
+    #[test]
+    fn updates_block_new_connections_operations_and_updates() {
+        let lifecycle = PluginLifecycle::default();
+        let update = lifecycle.begin_update("s3").unwrap();
+        assert_eq!(lifecycle.begin_connection("s3", "S3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert_eq!(lifecycle.begin_operation("s3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert_eq!(lifecycle.begin_update("s3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert!(lifecycle.begin_connection("other", "Other").is_ok());
+        drop(update);
+        assert!(lifecycle.begin_connection("s3", "S3").is_ok());
+        assert!(lifecycle.usage.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn connection_and_update_admission_are_atomic() {
+        for _ in 0..64 {
+            let lifecycle = PluginLifecycle::default();
+            let barrier = Arc::new(std::sync::Barrier::new(2));
+            let other_lifecycle = lifecycle.clone();
+            let other_barrier = barrier.clone();
+            let connection = std::thread::spawn(move || {
+                other_barrier.wait();
+                let result = other_lifecycle.begin_connection("s3", "S3");
+                other_barrier.wait();
+                result
+            });
+            barrier.wait();
+            let update = lifecycle.begin_update("s3");
+            barrier.wait();
+            let connection = connection.join().unwrap();
+            assert_ne!(connection.is_ok(), update.is_ok());
+        }
+    }
+}
diff --git a/crates/dbx-core/src/plugins/marketplace.rs b/crates/dbx-core/src/plugins/marketplace.rs
index 5428ce8c51..47352df59c 100644
--- a/crates/dbx-core/src/plugins/marketplace.rs
+++ b/crates/dbx-core/src/plugins/marketplace.rs
@@ -284,6 +284,7 @@ pub struct PluginMarketplace {
     app_version: String,
     repositories: PluginRepositoryStore,
     client: Client,
+    lifecycle: Option<super::PluginLifecycle>,
 }
 
 impl PluginMarketplace {
@@ -300,6 +301,7 @@ impl PluginMarketplace {
             root_dir,
             app_version: app_version.into(),
             client,
+            lifecycle: None,
         })
     }
 
@@ -307,6 +309,18 @@ impl PluginMarketplace {
         &self.repositories
     }
 
+    pub fn with_lifecycle(mut self, lifecycle: super::PluginLifecycle) -> Self {
+        self.lifecycle = Some(lifecycle);
+        self
+    }
+
+    fn guard_installer(&self, installer: PluginPackageInstaller) -> PluginPackageInstaller {
+        match &self.lifecycle {
+            Some(lifecycle) => installer.with_lifecycle(lifecycle.clone()),
+            None => installer,
+        }
+    }
+
     pub async fn fetch_catalogs(&self) -> Vec<PluginRepositoryCatalogResult> {
         let repositories = match self.repositories.list() {
             Ok(repositories) => repositories,
@@ -372,6 +386,9 @@ impl PluginMarketplace {
     }
 
     pub async fn install(&self, request: PluginMarketplaceInstallRequest) -> Result<PluginInstallResult, String> {
+        if let Some(lifecycle) = &self.lifecycle {
+            lifecycle.check_update(&request.plugin_id)?;
+        }
         let repository = self.repositories.find(&request.repository_id)?;
         if !repository.enabled {
             return Err(format!("Plugin repository '{}' is disabled", repository.id));
@@ -403,8 +420,12 @@ impl PluginMarketplace {
             permissions: plugin.permissions.iter().cloned().collect(),
             signing_key_id: artifact.signing_key_id.clone(),
         };
-        PluginPackageInstaller::with_trust_store(self.root_dir.clone(), self.app_version.clone(), trust_store)
-            .install_marketplace_bytes(&package, &expectation)
+        self.guard_installer(PluginPackageInstaller::with_trust_store(
+            self.root_dir.clone(),
+            self.app_version.clone(),
+            trust_store,
+        ))
+        .install_marketplace_bytes(&package, &expectation)
     }
 
     /// Downloads a .dbxp package from a direct http(s) URL and installs it with
@@ -445,7 +466,8 @@ impl PluginMarketplace {
             bytes.extend_from_slice(&chunk);
             on_progress(bytes.len() as u64, total);
         }
-        PluginPackageInstaller::new(self.root_dir.clone(), self.app_version.clone())?.install_bytes(&bytes, policy)
+        self.guard_installer(PluginPackageInstaller::new(self.root_dir.clone(), self.app_version.clone())?)
+            .install_bytes(&bytes, policy)
     }
 
     async fn download_limited(&self, url: Url, max_bytes: usize, label: &str) -> Result<Vec<u8>, String> {
@@ -1269,6 +1291,57 @@ mod tests {
         assert_eq!(last_progress, (package.len() as u64, Some(package.len() as u64)));
     }
 
+    #[tokio::test]
+    async fn direct_url_rechecks_runtime_usage_after_download() {
+        let root = tempfile::tempdir().unwrap();
+        let signing_key = SigningKey::from_bytes(&[7u8; 32]);
+        let key_id = "direct-url-release";
+        PluginTrustStore::save_base64_key(
+            root.path(),
+            key_id,
+            &base64::engine::general_purpose::STANDARD.encode(signing_key.verifying_key().as_bytes()),
+        )
+        .unwrap();
+        let registry = crate::plugins::PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.68");
+        let lifecycle = registry.lifecycle();
+        let marketplace =
+            PluginMarketplace::new(root.path().to_path_buf(), "0.5.68").unwrap().with_lifecycle(lifecycle.clone());
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let server = serve_package_once(listener, signed_package(&signing_key, key_id));
+        let mut connection = None;
+        let result = marketplace
+            .install_url_package(&format!("http://{address}/plugin.dbxp"), PluginInstallPolicy::LocalSigned, |_, _| {
+                connection.get_or_insert_with(|| {
+                    lifecycle.begin_connection("marketplace.install", "Started during download").unwrap()
+                });
+            })
+            .await;
+        server.await.unwrap();
+        assert!(result.unwrap_err().contains("Started during download"));
+        assert!(registry.find_plugin("marketplace.install").unwrap().is_none());
+        drop(connection);
+        assert!(lifecycle.begin_update("marketplace.install").is_ok());
+    }
+
+    #[tokio::test]
+    async fn marketplace_reports_active_connections_before_fetching_catalogs() {
+        let root = tempfile::tempdir().unwrap();
+        let lifecycle = crate::plugins::PluginLifecycle::default();
+        let marketplace =
+            PluginMarketplace::new(root.path().to_path_buf(), "0.5.68").unwrap().with_lifecycle(lifecycle.clone());
+        let _connection = lifecycle.begin_connection("marketplace.install", "Production").unwrap();
+        let error = marketplace
+            .install(super::PluginMarketplaceInstallRequest {
+                repository_id: "not-configured".to_string(),
+                plugin_id: "marketplace.install".to_string(),
+                version: None,
+            })
+            .await
+            .unwrap_err();
+        assert!(error.contains("Production"));
+    }
+
     #[tokio::test]
     async fn rejects_url_package_signed_by_an_untrusted_key() {
         let root = tempfile::tempdir().unwrap();
diff --git a/crates/dbx-web/src/routes/plugins.rs b/crates/dbx-web/src/routes/plugins.rs
index 2601cd3454..88870e8f01 100644
--- a/crates/dbx-web/src/routes/plugins.rs
+++ b/crates/dbx-web/src/routes/plugins.rs
@@ -228,7 +228,8 @@ pub async fn install_marketplace_plugin(
 ) -> Result<Json<PluginInstallResponse>, AppError> {
     let marketplace =
         PluginMarketplace::new(state.app.plugins.root_dir().to_path_buf(), state.app.plugins.app_version().to_string())
-            .map_err(AppError::from)?;
+            .map_err(AppError::from)?
+            .with_lifecycle(state.app.plugins.lifecycle());
     let result = marketplace.install(request).await.map_err(AppError::bad_request)?;
     state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await;
     stop_external_driver_pools(&state, &result.plugin).await;
@@ -288,6 +289,7 @@ pub async fn install_plugin(
         if query.allow_unsigned { PluginInstallPolicy::LocalDevelopment } else { PluginInstallPolicy::LocalSigned };
     let root_dir = state.app.plugins.root_dir().to_path_buf();
     let app_version = state.app.plugins.app_version().to_string();
+    let lifecycle = state.app.plugins.lifecycle();
     let result = tokio::task::spawn_blocking(move || {
         let installer = if query.from_url {
             let trust_store = dbx_core::plugins::url_install_trust_store(&root_dir)?;
@@ -295,7 +297,7 @@ pub async fn install_plugin(
         } else {
             PluginPackageInstaller::new(root_dir.clone(), app_version)?
         };
-        installer.install_bytes(&package, policy)
+        installer.with_lifecycle(lifecycle).install_bytes(&package, policy)
     })
     .await
     .map_err(|error| AppError::internal(error.to_string()))?
@@ -310,17 +312,18 @@ pub async fn rollback_plugin(
     State(state): State<Arc<WebState>>,
     Json(request): Json<PluginIdRequest>,
 ) -> Result<Json<PluginRollbackResponse>, AppError> {
-    state.app.remove_plugin_connection_pools(&request.plugin_id).await;
-    state.app.plugin_host.stop(&request.plugin_id).await;
     let root_dir = state.app.plugins.root_dir().to_path_buf();
     let app_version = state.app.plugins.app_version().to_string();
+    let lifecycle = state.app.plugins.lifecycle();
     let result = tokio::task::spawn_blocking(move || {
-        PluginPackageInstaller::new(root_dir, app_version)?.rollback(&request.plugin_id)
+        PluginPackageInstaller::new(root_dir, app_version)?.with_lifecycle(lifecycle).rollback(&request.plugin_id)
     })
     .await
     .map_err(|error| AppError::internal(error.to_string()))?
     .map_err(AppError::bad_request)?;
+    state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await;
     stop_external_driver_pools(&state, &result.plugin).await;
+    state.app.plugin_host.stop(&result.plugin.manifest.id).await;
     Ok(Json(result.response()))
 }
 
diff --git a/src-tauri/src/commands/plugins.rs b/src-tauri/src/commands/plugins.rs
index dd510b8360..c20f290025 100644
--- a/src-tauri/src/commands/plugins.rs
+++ b/src-tauri/src/commands/plugins.rs
@@ -119,7 +119,8 @@ pub async fn install_marketplace_plugin(
     request: PluginMarketplaceInstallRequest,
 ) -> Result<PluginInstallResponse, String> {
     let marketplace =
-        PluginMarketplace::new(state.plugins.root_dir().to_path_buf(), state.plugins.app_version().to_string())?;
+        PluginMarketplace::new(state.plugins.root_dir().to_path_buf(), state.plugins.app_version().to_string())?
+            .with_lifecycle(state.plugins.lifecycle());
     let result = marketplace.install(request).await?;
     stop_replaced_plugin_runtime(&state, &result.plugin).await;
     Ok(result.response())
@@ -190,8 +191,9 @@ pub async fn install_plugin_package(
     let app_version = state.plugins.app_version().to_string();
     let policy = if allow_unsigned { PluginInstallPolicy::LocalDevelopment } else { PluginInstallPolicy::LocalSigned };
     let path = std::path::PathBuf::from(path);
+    let lifecycle = state.plugins.lifecycle();
     let result = tauri::async_runtime::spawn_blocking(move || {
-        PluginPackageInstaller::new(root_dir, app_version)?.install_file(&path, policy)
+        PluginPackageInstaller::new(root_dir, app_version)?.with_lifecycle(lifecycle).install_file(&path, policy)
     })
     .await
     .map_err(|error| error.to_string())??;
@@ -207,7 +209,8 @@ pub async fn install_plugin_package_from_url(
     app: AppHandle,
 ) -> Result<PluginInstallResponse, String> {
     let marketplace =
-        PluginMarketplace::new(state.plugins.root_dir().to_path_buf(), state.plugins.app_version().to_string())?;
+        PluginMarketplace::new(state.plugins.root_dir().to_path_buf(), state.plugins.app_version().to_string())?
+            .with_lifecycle(state.plugins.lifecycle());
     let policy = if allow_unsigned { PluginInstallPolicy::LocalDevelopment } else { PluginInstallPolicy::LocalSigned };
     let result = marketplace
         .install_url_package(&url, policy, move |downloaded, total| {
@@ -229,16 +232,15 @@ pub async fn rollback_plugin(
     state: State<'_, Arc<AppState>>,
     plugin_id: String,
 ) -> Result<PluginRollbackResponse, String> {
-    state.remove_plugin_connection_pools(&plugin_id).await;
-    state.plugin_host.stop(&plugin_id).await;
     let root_dir = state.plugins.root_dir().to_path_buf();
     let app_version = state.plugins.app_version().to_string();
+    let lifecycle = state.plugins.lifecycle();
     let result = tauri::async_runtime::spawn_blocking(move || {
-        PluginPackageInstaller::new(root_dir, app_version)?.rollback(&plugin_id)
+        PluginPackageInstaller::new(root_dir, app_version)?.with_lifecycle(lifecycle).rollback(&plugin_id)
     })
     .await
     .map_err(|error| error.to_string())??;
-    stop_external_driver_pools(&state, &result.plugin).await;
+    stop_replaced_plugin_runtime(&state, &result.plugin).await;
     Ok(result.response())
 }