Fix settings schema missing extension-provided LSP settings on remote (#62355)

## Objective

Fixes an issue where language servers provided by extensions are not
included in the settings schemas served by remote servers. When
connecting to a remote server via the
`open_remote_project_with_existing_connection` flow (dev servers,
multi-workspace reopen, git worktree picker), setup-N servers never
received any extension sync, so their schemas omitted extension-provided
language servers, breaking language support for remote development.

## Solution

Three changes ensure that language servers provided by extensions are
included in the settings schemas served by remote servers:

1. **Register connections opened via
`open_remote_project_with_existing_connection` with the extension
store**: dev servers, multi-workspace reopen, and the git worktree
picker now register their connections with the extension store so that
extensions are synced to those servers. Previously only the
`open_remote_project` flow did this, so setup-N servers never received
any extension sync and their schemas omitted extension language servers.

2. **Emit `ExtensionsInstalledChanged` from `HeadlessExtensionStore`
after sync and install complete**: this allows the remote process to
invalidate its cached settings schemas. The local `ExtensionStore`
already emitted this event; without it, the remote schema cache stayed
stale for the lifetime of the server process.

3. **Include available LSP adapters in the project settings schema**:
this matches the user settings schema (this was fixed for `settings` in
#46766 but never applied to `project_settings`).

## Testing

- Manually verified that after connecting to a remote server via the
`open_remote_project_with_existing_connection` flow (including dev
servers and the git worktree picker), extension language servers appear
in the schema served by the remote server.
- Verified that `ExtensionsInstalledChanged` is emitted after sync and
install complete, and that the remote schema cache is correctly
invalidated and rebuilt.
- Verified that the `project_settings` schema includes the available LSP
adapters, matching the behavior of the user settings schema.
- Reviewers should focus on the multi-workspace reopen and git worktree
picker flows (the two previously unregistered flows) to confirm that
extension language servers are correctly synced and appear in the
schema.

## Self-Review Checklist:

- [x] I've reviewed my own diff for quality, security, and reliability
- [ ] Unsafe blocks (if any) have justifying comments
- [ ] The content adheres to Zed's UI standards
([UX/UI](https://github.com/zed-industries/zed/blob/main/CONTRIBUTING.md#uiux-checklist)
and
[icon](https://github.com/zed-industries/zed/blob/main/crates/icons/README.md)
guidelines)
- [x] Tests cover the new/changed behavior
- [x] Performance impact has been considered and is acceptable

## Showcase
**Server / Project** — before the fix
<img width="1061" height="749" alt="image"
src="https://github.com/user-attachments/assets/d6c5459f-58fb-4769-a701-f6fb317adc37"
/>
**Local**
<img width="1119" height="449" alt="image"
src="https://github.com/user-attachments/assets/fb30a1e2-74f0-4439-a25f-1022ed164074"
/>

---

Release Notes:

- Fixed an issue where language servers provided by extensions were
missing from settings schemas served by remote servers.

---------

Co-authored-by: vancez <vancez@users.noreply.github.com>
Co-authored-by: Kirill Bulatov <kirill@zed.dev>

Cargo.lock

@@ -9242,6 +9242,7 @@ dependencies = [
  "dap",
  "extension",
  "feature_flags",
+ "fs",
  "gpui",
  "language",
  "parking_lot",

crates/extension/src/extension_host_proxy.rs

@@ -235,7 +235,9 @@ pub trait ExtensionLanguageProxy: Send + Sync + 'static {
         matcher: Arc<LanguageMatcher>,
         hidden: bool,
         load: LanguageLoader,
-    );
+    ) -> bool;
+
+    fn is_language_registered(&self, language: &LanguageName) -> bool;
 
     fn remove_languages(
         &self,
@@ -253,14 +255,22 @@ impl ExtensionLanguageProxy for ExtensionHostProxy {
         matcher: Arc<LanguageMatcher>,
         hidden: bool,
         load: LanguageLoader,
-    ) {
+    ) -> bool {
         let Some(proxy) = self.language_proxy.read().clone() else {
-            return;
+            return false;
         };
 
         proxy.register_language(language, grammar, matcher, hidden, load)
     }
 
+    fn is_language_registered(&self, language: &LanguageName) -> bool {
+        let Some(proxy) = self.language_proxy.read().clone() else {
+            return false;
+        };
+
+        proxy.is_language_registered(language)
+    }
+
     fn remove_languages(
         &self,
         languages_to_remove: &[LanguageName],

crates/extension_host/Cargo.toml

@@ -65,6 +65,7 @@ language = { workspace = true, features = ["test-support"] }
 language_extension.workspace = true
 parking_lot.workspace = true
 project = { workspace = true, features = ["test-support"] }
+remote = { workspace = true, features = ["test-support"] }
 reqwest_client.workspace = true
 theme = { workspace = true, features = ["test-support"] }
 theme_settings.workspace = true

crates/extension_host/src/extension_host.rs

@@ -20,19 +20,19 @@ use extension::{
     ExtensionLanguageServerProxy, ExtensionSnippetProxy, ExtensionThemeProxy,
 };
 use fs::{Fs, RemoveOptions, RenameOptions};
-use futures::future::join_all;
+use futures::future::{Shared, join_all};
 use futures::{
     AsyncReadExt as _, Future, FutureExt as _, StreamExt as _,
     channel::{
-        mpsc::{UnboundedSender, unbounded},
+        mpsc::{UnboundedReceiver, UnboundedSender, unbounded},
         oneshot,
     },
     io::BufReader,
     select_biased,
 };
 use gpui::{
-    App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, Global, Task, TaskExt,
-    UpdateGlobal as _, WeakEntity, actions,
+    App, AppContext as _, AsyncApp, Context, Entity, EntityId, EventEmitter, Global, Subscription,
+    Task, TaskExt, UpdateGlobal as _, WeakEntity, actions,
 };
 use http_client::{AsyncBody, HttpClient, HttpClientWithUrl};
 use language::{
@@ -40,15 +40,16 @@ use language::{
     QueryFileContents, QueryFiles, Rope,
 };
 use node_runtime::NodeRuntime;
-use project::ContextProviderWithTasks;
+use project::{ContextProviderWithTasks, Project};
 use release_channel::ReleaseChannel;
-use remote::RemoteClient;
+use remote::{ConnectionState, RemoteClient, RemoteClientEvent};
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use settings::{SemanticTokenRules, Settings, SettingsStore};
 use std::ops::RangeInclusive;
 use std::str::FromStr;
 use std::sync::LazyLock;
+use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
 use std::{
     borrow::Cow,
     cmp::Ordering,
@@ -58,7 +59,10 @@ use std::{
 };
 use task::TaskTemplates;
 use url::Url;
-use util::{PathExt, ResultExt, paths::RemotePathBuf};
+use util::{
+    PathExt, ResultExt,
+    paths::{PathStyle, RemotePathBuf},
+};
 use wasm_host::{
     WasmExtension, WasmHost,
     wit::{is_supported_wasm_api_version, wasm_api_version_range},
@@ -69,8 +73,34 @@ pub use extension::{
 };
 pub use extension_settings::ExtensionSettings;
 
+use crate::headless_host::hash_directory_contents;
+
 pub const RELOAD_DEBOUNCE_DURATION: Duration = Duration::from_millis(200);
 const FS_WATCH_LATENCY: Duration = Duration::from_millis(100);
+pub(crate) const REMOTE_SYNC_RETRY_DELAY: Duration = Duration::from_secs(1);
+pub(crate) const MAX_REMOTE_SYNC_RETRY_DELAY: Duration = Duration::from_secs(60);
+pub(crate) const MAX_REMOTE_SYNC_ATTEMPTS: usize = 10;
+pub(crate) const REMOTE_SYNC_TIMEOUT: Duration = Duration::from_secs(60 * 60);
+
+pub(crate) fn remote_sync_retry_delay(attempts: usize) -> Duration {
+    let exponential = REMOTE_SYNC_RETRY_DELAY * 2u32.saturating_pow(attempts.min(30) as u32);
+    exponential.min(MAX_REMOTE_SYNC_RETRY_DELAY)
+}
+
+async fn with_remote_sync_timeout<T>(
+    cx: &AsyncApp,
+    timeout: Duration,
+    description: &str,
+    future: impl Future<Output = Result<T>>,
+) -> Result<T> {
+    let timer = cx.background_executor().timer(timeout).fuse();
+    let future = future.fuse();
+    futures::pin_mut!(timer, future);
+    select_biased! {
+        result = future => result,
+        _ = timer => anyhow::bail!("timed out after {timeout:?} while {description}"),
+    }
+}
 
 /// The current extension [`SchemaVersion`] supported by Zed.
 const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(1);
@@ -144,8 +174,20 @@ pub struct ExtensionStore {
     pub wasm_host: Arc<WasmHost>,
     pub wasm_extensions: Vec<(Arc<ExtensionManifest>, WasmExtension)>,
     pub tasks: Vec<Task<()>>,
-    pub remote_clients: Vec<WeakEntity<RemoteClient>>,
-    pub ssh_registered_tx: UnboundedSender<()>,
+    pub(crate) remote_clients: HashMap<EntityId, RemoteClientState>,
+    pub(crate) initial_index_load: Shared<Task<()>>,
+}
+
+pub(crate) struct RemoteClientState {
+    dirty_tx: UnboundedSender<RemoteSyncSignal>,
+    _task: Task<()>,
+    _subscriptions: Subscription,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum RemoteSyncSignal {
+    IndexChanged,
+    Reconnected,
 }
 
 #[derive(Clone, Copy)]
@@ -226,6 +268,10 @@ impl RemoteSyncExtensions {
     fn into_entries(self) -> impl Iterator<Item = (Arc<str>, ExtensionIndexEntry)> {
         self.0.into_iter()
     }
+
+    fn contains(&self, id: &str) -> bool {
+        self.0.contains_key(id)
+    }
 }
 
 #[derive(Clone, PartialEq, Eq, Debug, Deserialize, Serialize)]
@@ -292,6 +338,16 @@ pub fn init(
     });
 
     cx.set_global(GlobalExtensionStore(store));
+
+    cx.observe_new::<Project>(|project, _window, cx| {
+        let Some(client) = project.remote_client() else {
+            return;
+        };
+        if let Some(store) = ExtensionStore::try_global(cx) {
+            store.update(cx, |store, cx| store.register_remote_client(client, cx));
+        }
+    })
+    .detach();
 }
 
 impl ExtensionStore {
@@ -322,7 +378,6 @@ impl ExtensionStore {
         let index_path = extensions_dir.join("index.json");
 
         let (reload_tx, mut reload_rx) = unbounded();
-        let (connection_registered_tx, mut connection_registered_rx) = unbounded();
         let mut this = Self {
             proxy: extension_host_proxy.clone(),
             extension_index: Default::default(),
@@ -348,8 +403,8 @@ impl ExtensionStore {
             reload_tx,
             tasks: Vec::new(),
 
-            remote_clients: Default::default(),
-            ssh_registered_tx: connection_registered_tx,
+            remote_clients: HashMap::default(),
+            initial_index_load: Task::ready(()).shared(),
         };
 
         // The extensions store maintains an index file, which contains a complete
@@ -391,10 +446,17 @@ impl ExtensionStore {
             reload_future = Some(this.reload(None, cx));
         }
 
+        let initial_index_load = cx
+            .spawn(async move |_, _| {
+                if let Some(future) = reload_future {
+                    future.await;
+                }
+            })
+            .shared();
+        this.initial_index_load = initial_index_load.clone();
+
         cx.spawn(async move |this, cx| {
-            if let Some(future) = reload_future {
-                future.await;
-            }
+            initial_index_load.await;
             this.update(cx, |this, cx| this.auto_install_extensions(cx))
                 .ok();
             this.update(cx, |this, cx| this.check_for_updates(cx)).ok();
@@ -422,11 +484,6 @@ impl ExtensionStore {
                                     .await;
                                 index_changed = false;
                             }
-
-                            Self::update_remote_clients(&this, cx).await?;
-                        }
-                        _ = connection_registered_rx.next() => {
-                            debounce_timer = cx.background_executor().timer(RELOAD_DEBOUNCE_DURATION).fuse()
                         }
                         extension_id = reload_rx.next() => {
                             let Some(extension_id) = extension_id else { break; };
@@ -1254,6 +1311,15 @@ impl ExtensionStore {
             extensions_to_unload.len() - reload_count
         );
 
+        let old_remote_sync_extensions = old_index.extensions_to_sync_to_remote();
+        let new_remote_sync_extensions = new_index.extensions_to_sync_to_remote();
+        let remote_sync_changed = extensions_to_unload
+            .iter()
+            .any(|id| old_remote_sync_extensions.contains(id.as_ref()))
+            || extensions_to_load
+                .iter()
+                .any(|id| new_remote_sync_extensions.contains(id.as_ref()));
+
         let extension_ids = extensions_to_load
             .iter()
             .filter_map(|id| {
@@ -1288,17 +1354,19 @@ impl ExtensionStore {
                 }
             })
             .collect::<Vec<_>>();
-        let languages_to_remove = old_index
-            .languages
-            .iter()
-            .filter_map(|(name, entry)| {
-                if extensions_to_unload.contains(&entry.extension) {
-                    Some(name.clone())
-                } else {
-                    None
+        let mut languages_to_remove = Vec::new();
+        let mut languages_to_readd = Vec::new();
+        for (name, entry) in &old_index.languages {
+            if !extensions_to_unload.contains(&entry.extension) {
+                continue;
+            }
+            match new_index.languages.get(name) {
+                Some(new_entry) if !extensions_to_load.contains(&new_entry.extension) => {
+                    languages_to_readd.push((name.clone(), new_entry.clone()));
                 }
-            })
-            .collect::<Vec<_>>();
+                _ => languages_to_remove.push(name.clone()),
+            }
+        }
         let mut grammars_to_remove = Vec::new();
         let mut server_removal_tasks = Vec::with_capacity(extensions_to_unload.len());
         for extension_id in &extensions_to_unload {
@@ -1335,9 +1403,14 @@ impl ExtensionStore {
             .remove_languages(&languages_to_remove, &grammars_to_remove);
 
         // Remove semantic token rules for languages being unloaded.
-        if !languages_to_remove.is_empty() {
+        let semantic_token_rules_to_remove = languages_to_remove
+            .iter()
+            .filter(|language| !self.proxy.is_language_registered(language))
+            .chain(languages_to_readd.iter().map(|(name, _)| name))
+            .collect::<Vec<_>>();
+        if !semantic_token_rules_to_remove.is_empty() {
             SettingsStore::update_global(cx, |store, cx| {
-                for language in &languages_to_remove {
+                for language in semantic_token_rules_to_remove {
                     store.remove_language_semantic_token_rules(language.as_ref(), cx);
                 }
             });
@@ -1387,29 +1460,54 @@ impl ExtensionStore {
             }));
         }
 
+        for (name, entry) in &languages_to_readd {
+            let Some(grammar_name) = entry.grammar.clone() else {
+                continue;
+            };
+            if !grammars_to_remove.contains(&grammar_name) {
+                continue;
+            }
+            let owner = std::iter::once(&entry.extension)
+                .chain(new_index.extensions.keys())
+                .find(|id| {
+                    new_index
+                        .extensions
+                        .get(id.as_ref())
+                        .is_some_and(|extension| {
+                            extension.manifest.grammars.contains_key(&grammar_name)
+                        })
+                });
+            let Some(owner) = owner else {
+                log::warn!(
+                    "not re-registering grammar {grammar_name} for language {name}: no installed extension provides it"
+                );
+                continue;
+            };
+            let mut grammar_path = self.installed_dir.clone();
+            grammar_path.extend([owner.as_ref(), "grammars"]);
+            grammar_path.push(grammar_name.as_ref());
+            grammar_path.set_extension("wasm");
+            grammars_to_add.push((grammar_name, grammar_path));
+        }
+
         self.proxy.register_grammars(grammars_to_add);
         let languages_to_add = new_index
             .languages
             .iter()
             .filter(|(_, entry)| extensions_to_load.contains(&entry.extension))
+            .map(|(name, entry)| (name.clone(), entry.clone()))
+            .chain(languages_to_readd)
             .collect::<Vec<_>>();
-        let mut semantic_token_rules_to_add: Vec<(LanguageName, SemanticTokenRules)> = Vec::new();
+        let mut semantic_token_rules_paths: Vec<(LanguageName, PathBuf)> = Vec::new();
         for (language_name, language) in languages_to_add {
             let mut language_path = self.installed_dir.clone();
             language_path.extend([
                 Path::new(language.extension.as_ref()),
                 language.path.as_path(),
             ]);
-
-            // Load semantic token rules if present in the language directory.
             let rules_path = language_path.join(SemanticTokenRules::FILE_NAME);
-            if std::fs::exists(&rules_path).is_ok_and(|exists| exists)
-                && let Some(rules) = SemanticTokenRules::load(&rules_path).log_err()
-            {
-                semantic_token_rules_to_add.push((language_name.clone(), rules));
-            }
 
-            self.proxy.register_language(
+            let registered = self.proxy.register_language(
                 language_name.clone(),
                 language.grammar.clone(),
                 language.matcher.clone(),
@@ -1425,15 +1523,11 @@ impl ExtensionStore {
                     }
                 }),
             );
-        }
+            if !registered {
+                continue;
+            }
 
-        // Register semantic token rules for newly loaded extension languages.
-        if !semantic_token_rules_to_add.is_empty() {
-            SettingsStore::update_global(cx, |store, cx| {
-                for (language_name, rules) in semantic_token_rules_to_add {
-                    store.set_language_semantic_token_rules(language_name.0.clone(), rules, cx);
-                }
-            });
+            semantic_token_rules_paths.push((language_name, rules_path));
         }
 
         let fs = self.fs.clone();
@@ -1447,43 +1541,79 @@ impl ExtensionStore {
         self.extension_index = new_index;
         cx.notify();
         cx.emit(Event::ExtensionsUpdated);
+        if remote_sync_changed {
+            self.sync_remote_clients();
+        }
 
         cx.spawn(async move |this, cx| {
-            cx.background_spawn({
-                let fs = fs.clone();
-                async move {
-                    let _ = join_all(server_removal_tasks).await;
-                    for theme_path in themes_to_add {
-                        proxy
-                            .load_user_theme(theme_path, fs.clone())
-                            .await
-                            .log_err();
-                    }
-
-                    for (icon_theme_path, icons_root_path) in icon_themes_to_add {
-                        proxy
-                            .load_icon_theme(icon_theme_path, icons_root_path, fs.clone())
-                            .await
-                            .log_err();
-                    }
-
-                    for snippets_path in &snippets_to_add {
-                        match fs
-                            .load(snippets_path)
-                            .await
-                            .with_context(|| format!("Loading snippets from {snippets_path:?}"))
-                        {
-                            Ok(snippets_contents) => {
-                                proxy
-                                    .register_snippet(snippets_path, &snippets_contents)
-                                    .log_err();
-                            }
-                            Err(e) => log::error!("Cannot load snippets: {e:#}"),
+            let semantic_token_rules_to_add = cx
+                .background_spawn({
+                    let fs = fs.clone();
+                    async move {
+                        let _ = join_all(server_removal_tasks).await;
+                        for theme_path in themes_to_add {
+                            proxy
+                                .load_user_theme(theme_path, fs.clone())
+                                .await
+                                .log_err();
                         }
+
+                        for (icon_theme_path, icons_root_path) in icon_themes_to_add {
+                            proxy
+                                .load_icon_theme(icon_theme_path, icons_root_path, fs.clone())
+                                .await
+                                .log_err();
+                        }
+
+                        for snippets_path in &snippets_to_add {
+                            match fs
+                                .load(snippets_path)
+                                .await
+                                .with_context(|| format!("Loading snippets from {snippets_path:?}"))
+                            {
+                                Ok(snippets_contents) => {
+                                    proxy
+                                        .register_snippet(snippets_path, &snippets_contents)
+                                        .log_err();
+                                }
+                                Err(e) => log::error!("Cannot load snippets: {e:#}"),
+                            }
+                        }
+
+                        // Load semantic token rules if present in the language directory.
+                        let mut semantic_token_rules_to_add = Vec::new();
+                        for (language_name, rules_path) in semantic_token_rules_paths {
+                            if !fs.is_file(&rules_path).await {
+                                continue;
+                            }
+                            let rules = fs
+                                .load(&rules_path)
+                                .await
+                                .and_then(|content| SemanticTokenRules::parse(&content));
+                            if let Some(rules) = rules.log_err() {
+                                semantic_token_rules_to_add.push((language_name, rules));
+                            }
+                        }
+                        semantic_token_rules_to_add
                     }
-                }
-            })
-            .await;
+                })
+                .await;
+
+            // Register semantic token rules for newly loaded extension languages.
+            if !semantic_token_rules_to_add.is_empty() {
+                this.update(cx, |_, cx| {
+                    SettingsStore::update_global(cx, |store, cx| {
+                        for (language_name, rules) in semantic_token_rules_to_add {
+                            store.set_language_semantic_token_rules(
+                                language_name.0.clone(),
+                                rules,
+                                cx,
+                            );
+                        }
+                    })
+                })
+                .ok();
+            }
 
             let mut wasm_extensions = Vec::new();
             for extension in extension_entries {
@@ -1858,114 +1988,340 @@ impl ExtensionStore {
         })
     }
 
-    async fn sync_extensions_to_remotes(
+    fn sync_remote_clients(&mut self) {
+        for state in self.remote_clients.values() {
+            state
+                .dirty_tx
+                .unbounded_send(RemoteSyncSignal::IndexChanged)
+                .ok();
+        }
+    }
+
+    async fn reconcile_remote_client(
+        this: WeakEntity<Self>,
+        client: WeakEntity<RemoteClient>,
+        mut dirty_rx: UnboundedReceiver<RemoteSyncSignal>,
+        cx: &mut AsyncApp,
+    ) {
+        let mut failed_attempts = 0_usize;
+        loop {
+            while let Ok(signal) = dirty_rx.try_recv() {
+                if signal == RemoteSyncSignal::Reconnected {
+                    failed_attempts = 0;
+                }
+            }
+
+            let Ok(connection_state) =
+                client.read_with(cx, |client, _cx| client.connection_state())
+            else {
+                return;
+            };
+            if connection_state == ConnectionState::Disconnected
+                || connection_state == ConnectionState::Reconnecting
+            {
+                failed_attempts = 0;
+                if dirty_rx.next().await.is_none() {
+                    return;
+                }
+                continue;
+            }
+
+            match Self::sync_extensions_to_remote(&this, client.clone(), cx).await {
+                Ok(()) => {
+                    failed_attempts = 0;
+                    if dirty_rx.next().await.is_none() {
+                        return;
+                    }
+                }
+                Err(error) => {
+                    failed_attempts += 1;
+                    if failed_attempts >= MAX_REMOTE_SYNC_ATTEMPTS {
+                        log::error!(
+                            "Failed to sync extensions to a remote client {failed_attempts} times, waiting for an extension or connection change before retrying: {error:#}"
+                        );
+                        match dirty_rx.next().await {
+                            None => return,
+                            Some(RemoteSyncSignal::Reconnected) => failed_attempts = 0,
+                            Some(RemoteSyncSignal::IndexChanged) => {}
+                        }
+                        continue;
+                    }
+                    let delay = remote_sync_retry_delay(failed_attempts - 1);
+                    log::error!(
+                        "Failed to sync extensions to a remote client (attempt {failed_attempts}), will retry in {delay:?}: {error:#}"
+                    );
+                    let timer = cx.background_executor().timer(delay).fuse();
+                    futures::pin_mut!(timer);
+                    loop {
+                        select_biased! {
+                            signal = dirty_rx.next() => {
+                                match signal {
+                                    None => return,
+                                    Some(RemoteSyncSignal::Reconnected) => {
+                                        failed_attempts = 0;
+                                        break;
+                                    }
+                                    Some(RemoteSyncSignal::IndexChanged) => {}
+                                }
+                            }
+                            _ = timer => break,
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    async fn sync_extensions_to_remote(
         this: &WeakEntity<Self>,
         client: WeakEntity<RemoteClient>,
         cx: &mut AsyncApp,
     ) -> Result<()> {
-        let extensions = this.update(cx, |this, _cx| {
+        let entries = this.update(cx, |this, _cx| {
             this.extension_index
                 .extensions_to_sync_to_remote()
                 .into_entries()
-                .map(|(id, entry)| proto::Extension {
-                    id: id.to_string(),
-                    version: entry.manifest.version.to_string(),
-                    dev: entry.dev,
-                })
-                .collect()
+                .collect::<Vec<_>>()
         })?;
+        let mut prepared_dev_payloads = HashMap::default();
+        let mut extensions = Vec::new();
+        for (id, entry) in entries {
+            let mut content_fingerprint = None;
+            if entry.dev {
+                match Self::prepare_dev_extension_payload(this, &id, cx).await {
+                    Ok((payload_dir, fingerprint)) => {
+                        content_fingerprint = Some(fingerprint);
+                        prepared_dev_payloads.insert(id.to_string(), payload_dir);
+                    }
+                    Err(error) => {
+                        log::warn!(
+                            "failed to prepare dev extension {id} for a remote sync: {error:#}"
+                        );
+                    }
+                }
+            }
+            extensions.push(proto::Extension {
+                id: id.to_string(),
+                version: entry.manifest.version.to_string(),
+                dev: entry.dev,
+                content_fingerprint,
+            });
+        }
 
-        let response = client
-            .update(cx, |client, _cx| {
-                client
-                    .proto_client()
-                    .request(proto::SyncExtensions { extensions })
-            })?
-            .await?;
+        let request = client.update(cx, |client, _cx| {
+            client
+                .proto_client()
+                .request(proto::SyncExtensions { extensions })
+        })?;
+        let response = with_remote_sync_timeout(
+            cx,
+            REMOTE_SYNC_TIMEOUT,
+            "requesting the remote extension list",
+            request,
+        )
+        .await?;
         let path_style = client.read_with(cx, |client, _| client.path_style())?;
 
+        let mut failed_installs = Vec::new();
         for missing_extension in response.missing_extensions.into_iter() {
-            let tmp_dir = tempfile::tempdir()?;
+            let prepared_payload = prepared_dev_payloads.remove(&missing_extension.id);
+            if let Err(error) = Self::install_extension_on_remote(
+                this,
+                &client,
+                &missing_extension,
+                &response.tmp_dir,
+                path_style,
+                prepared_payload,
+                cx,
+            )
+            .await
+            {
+                log::error!(
+                    "Failed to install extension {} on the remote: {error:#}",
+                    missing_extension.id
+                );
+                failed_installs.push(missing_extension.id);
+            }
+        }
+        if !prepared_dev_payloads.is_empty() {
+            cx.background_executor()
+                .spawn(async move { drop(prepared_dev_payloads) })
+                .detach();
+        }
+
+        anyhow::ensure!(
+            failed_installs.is_empty(),
+            "failed to install extensions on the remote: {failed_installs:?}"
+        );
+        anyhow::Ok(())
+    }
+
+    async fn prepare_dev_extension_payload(
+        this: &WeakEntity<Self>,
+        id: &Arc<str>,
+        cx: &mut AsyncApp,
+    ) -> Result<(tempfile::TempDir, u64)> {
+        let payload_dir = cx
+            .background_executor()
+            .spawn(async move { tempfile::tempdir() })
+            .await?;
+        this.update(cx, |this, cx| {
+            this.prepare_remote_extension(id.clone(), true, payload_dir.path().to_owned(), cx)
+        })?
+        .await?;
+        let fs = this.read_with(cx, |this, _cx| this.fs.clone())?;
+        let fingerprint = cx
+            .background_executor()
+            .spawn({
+                let path = payload_dir.path().to_owned();
+                async move { hash_directory_contents(&fs, &path).await }
+            })
+            .await?;
+        Ok((payload_dir, fingerprint))
+    }
+
+    async fn install_extension_on_remote(
+        this: &WeakEntity<Self>,
+        client: &WeakEntity<RemoteClient>,
+        missing_extension: &proto::Extension,
+        remote_tmp_dir: &str,
+        path_style: PathStyle,
+        prepared_payload: Option<tempfile::TempDir>,
+        cx: &mut AsyncApp,
+    ) -> Result<()> {
+        let already_prepared = prepared_payload.is_some();
+        let tmp_dir = match prepared_payload {
+            Some(payload_dir) => payload_dir,
+            None => {
+                cx.background_executor()
+                    .spawn(async move { tempfile::tempdir() })
+                    .await?
+            }
+        };
+        let result = Self::upload_extension_to_remote(
+            this,
+            client,
+            missing_extension,
+            remote_tmp_dir,
+            path_style,
+            tmp_dir.path().to_owned(),
+            already_prepared,
+            cx,
+        )
+        .await;
+        cx.background_executor()
+            .spawn(async move { drop(tmp_dir) })
+            .detach();
+        result
+    }
+
+    async fn upload_extension_to_remote(
+        this: &WeakEntity<Self>,
+        client: &WeakEntity<RemoteClient>,
+        missing_extension: &proto::Extension,
+        remote_tmp_dir: &str,
+        path_style: PathStyle,
+        local_dir: PathBuf,
+        already_prepared: bool,
+        cx: &mut AsyncApp,
+    ) -> Result<()> {
+        static UPLOAD_NONCE: AtomicU64 = AtomicU64::new(0);
+
+        if !already_prepared {
             this.update(cx, |this, cx| {
                 this.prepare_remote_extension(
                     missing_extension.id.clone().into(),
                     missing_extension.dev,
-                    tmp_dir.path().to_owned(),
+                    local_dir.clone(),
                     cx,
                 )
             })?
             .await?;
-            let dest_dir = RemotePathBuf::new(
-                path_style
-                    .join(&response.tmp_dir, &missing_extension.id)
-                    .with_context(|| {
-                        format!(
-                            "failed to construct destination path: {:?}, {:?}",
-                            response.tmp_dir, missing_extension.id,
-                        )
-                    })?,
-                path_style,
-            );
-            log::info!(
-                "Uploading extension {} to {:?}",
-                missing_extension.clone().id,
-                dest_dir
-            );
-
-            client
-                .update(cx, |client, cx| {
-                    client.upload_directory(tmp_dir.path().to_owned(), dest_dir.clone(), cx)
-                })?
-                .await?;
-
-            log::info!(
-                "Finished uploading extension {}",
-                missing_extension.clone().id
-            );
-
-            let result = client
-                .update(cx, |client, _cx| {
-                    client.proto_client().request(proto::InstallExtension {
-                        tmp_dir: dest_dir.to_proto(),
-                        extension: Some(missing_extension.clone()),
-                    })
-                })?
-                .await;
-
-            if let Err(e) = result {
-                log::error!(
-                    "Failed to install extension {}: {}",
-                    missing_extension.id,
-                    e
-                );
-            }
         }
+        let timestamp = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |duration| duration.as_nanos());
+        let upload_name = format!(
+            "{}-{}-{}-{}",
+            missing_extension.id,
+            std::process::id(),
+            UPLOAD_NONCE.fetch_add(1, AtomicOrdering::Relaxed),
+            timestamp,
+        );
+        let dest_dir = RemotePathBuf::new(
+            path_style
+                .join(remote_tmp_dir, &upload_name)
+                .with_context(|| {
+                    format!(
+                        "failed to construct destination path: {remote_tmp_dir:?}, {upload_name:?}"
+                    )
+                })?,
+            path_style,
+        );
+        log::info!(
+            "Uploading extension {} to {:?}",
+            missing_extension.id,
+            dest_dir
+        );
 
-        anyhow::Ok(())
-    }
-
-    pub async fn update_remote_clients(this: &WeakEntity<Self>, cx: &mut AsyncApp) -> Result<()> {
-        let clients = this.update(cx, |this, _cx| {
-            this.remote_clients.retain(|v| v.upgrade().is_some());
-            this.remote_clients.clone()
+        let upload = client.update(cx, |client, cx| {
+            client.upload_directory(local_dir, dest_dir.clone(), cx)
         })?;
+        with_remote_sync_timeout(cx, REMOTE_SYNC_TIMEOUT, "uploading an extension", upload).await?;
 
-        for client in clients {
-            Self::sync_extensions_to_remotes(this, client, cx)
-                .await
-                .log_err();
-        }
+        log::info!("Finished uploading extension {}", missing_extension.id);
 
-        anyhow::Ok(())
+        let install = client.update(cx, |client, _cx| {
+            client.proto_client().request(proto::InstallExtension {
+                tmp_dir: dest_dir.to_proto(),
+                extension: Some(missing_extension.clone()),
+            })
+        })?;
+        with_remote_sync_timeout(cx, REMOTE_SYNC_TIMEOUT, "installing an extension", install)
+            .await?;
+        Ok(())
     }
 
-    pub fn register_remote_client(
-        &mut self,
-        client: Entity<RemoteClient>,
-        _cx: &mut Context<Self>,
-    ) {
-        self.remote_clients.push(client.downgrade());
-        self.ssh_registered_tx.unbounded_send(()).ok();
+    pub fn register_remote_client(&mut self, client: Entity<RemoteClient>, cx: &mut Context<Self>) {
+        let entity_id = client.entity_id();
+        if self.remote_clients.contains_key(&entity_id) {
+            return;
+        }
+
+        let (dirty_tx, dirty_rx) = unbounded();
+
+        let event_subscription = cx.subscribe(&client, |store, client, event, _cx| match event {
+            RemoteClientEvent::Reconnected => {
+                if let Some(state) = store.remote_clients.get(&client.entity_id()) {
+                    state
+                        .dirty_tx
+                        .unbounded_send(RemoteSyncSignal::Reconnected)
+                        .ok();
+                }
+            }
+            RemoteClientEvent::Disconnected { .. } => {}
+        });
+        let release_subscription = cx.observe_release(&client, move |store, _client, _cx| {
+            store.remote_clients.remove(&entity_id);
+        });
+
+        let task = cx.spawn({
+            let client = client.downgrade();
+            let initial_index_load = self.initial_index_load.clone();
+            async move |this, cx| {
+                initial_index_load.await;
+                Self::reconcile_remote_client(this, client, dirty_rx, cx).await;
+            }
+        });
+
+        self.remote_clients.insert(
+            entity_id,
+            RemoteClientState {
+                dirty_tx,
+                _task: task,
+                _subscriptions: Subscription::join(event_subscription, release_subscription),
+            },
+        );
     }
 }
 

crates/extension_host/src/headless_host.rs

@@ -1,19 +1,28 @@
-use std::{path::PathBuf, sync::Arc};
+use std::{
+    hash::{Hash, Hasher},
+    path::{Path, PathBuf},
+    sync::Arc,
+    time::{Duration, SystemTime},
+};
 
 use anyhow::{Context as _, Result};
 use client::{TypedEnvelope, proto};
-use collections::{HashMap, HashSet};
+use collections::{BTreeMap, FxHasher, HashSet};
 use extension::{
-    Extension, ExtensionDebugAdapterProviderProxy, ExtensionHostProxy, ExtensionLanguageProxy,
-    ExtensionLanguageServerProxy, ExtensionManifest,
+    Event, Extension, ExtensionDebugAdapterProviderProxy, ExtensionEvents, ExtensionHostProxy,
+    ExtensionLanguageProxy, ExtensionLanguageServerProxy, ExtensionManifest,
 };
 use fs::{Fs, RemoveOptions, RenameOptions};
-use futures::future::{FutureExt as _, join_all};
+use futures::{
+    StreamExt as _,
+    future::{FutureExt as _, join_all},
+};
 use gpui::{App, AppContext as _, AsyncApp, Context, Entity, Task, WeakEntity};
 use http_client::HttpClient;
 use language::{LanguageConfig, LanguageName, LanguageQueries, LoadedLanguage};
 use lsp::LanguageServerName;
 use node_runtime::NodeRuntime;
+use util::ResultExt as _;
 
 use crate::wasm_host::{WasmExtension, WasmHost};
 
@@ -22,16 +31,32 @@ pub struct ExtensionVersion {
     pub id: String,
     pub version: String,
     pub dev: bool,
+    pub content_fingerprint: Option<u64>,
 }
 
+pub(crate) const STALE_UPLOAD_TTL: Duration =
+    Duration::from_secs(crate::REMOTE_SYNC_TIMEOUT.as_secs() * 3);
+
 pub struct HeadlessExtensionStore {
     pub fs: Arc<dyn Fs>,
     pub extension_dir: PathBuf,
     pub proxy: Arc<ExtensionHostProxy>,
     pub wasm_host: Arc<WasmHost>,
-    pub loaded_extensions: HashMap<Arc<str>, Arc<str>>,
-    pub loaded_languages: HashMap<Arc<str>, Vec<LanguageName>>,
-    pub loaded_language_servers: HashMap<Arc<str>, Vec<(LanguageServerName, LanguageName)>>,
+    pub(crate) loaded_extensions: BTreeMap<Arc<str>, LoadedExtension>,
+    failed_removals: HashSet<Arc<str>>,
+    operation_lock: Arc<futures::lock::Mutex<()>>,
+    _stale_uploads_sweep: Task<()>,
+}
+
+#[derive(Clone)]
+pub(crate) struct LoadedExtension {
+    pub version: Arc<str>,
+    pub languages: Vec<(LanguageName, LanguageConfig)>,
+    pub language_servers: Vec<(LanguageServerName, LanguageName)>,
+    pub debug_adapters: Vec<(Arc<str>, PathBuf)>,
+    pub debug_locators: Vec<Arc<str>>,
+    pub wasm_extension: Option<Arc<dyn Extension>>,
+    pub content_fingerprint: Option<u64>,
 }
 
 impl HeadlessExtensionStore {
@@ -43,21 +68,34 @@ impl HeadlessExtensionStore {
         node_runtime: NodeRuntime,
         cx: &mut App,
     ) -> Entity<Self> {
-        cx.new(|cx| Self {
-            fs: fs.clone(),
-            wasm_host: WasmHost::new(
-                fs.clone(),
-                http_client.clone(),
-                node_runtime,
-                extension_host_proxy.clone(),
-                extension_dir.join("work"),
-                cx,
-            ),
-            extension_dir,
-            proxy: extension_host_proxy,
-            loaded_extensions: Default::default(),
-            loaded_languages: Default::default(),
-            loaded_language_servers: Default::default(),
+        cx.new(|cx| {
+            let stale_uploads_sweep = cx.spawn({
+                let fs = fs.clone();
+                async move |_, cx| {
+                    loop {
+                        remove_stale_uploads(&fs, paths::remote_extensions_uploads_dir()).await;
+                        cx.background_executor().timer(STALE_UPLOAD_TTL).await;
+                    }
+                }
+            });
+
+            Self {
+                fs: fs.clone(),
+                wasm_host: WasmHost::new(
+                    fs.clone(),
+                    http_client.clone(),
+                    node_runtime,
+                    extension_host_proxy.clone(),
+                    extension_dir.join("work"),
+                    cx,
+                ),
+                extension_dir,
+                proxy: extension_host_proxy,
+                loaded_extensions: BTreeMap::default(),
+                failed_removals: HashSet::default(),
+                operation_lock: Arc::default(),
+                _stale_uploads_sweep: stale_uploads_sweep,
+            }
         })
     }
 
@@ -66,159 +104,383 @@ impl HeadlessExtensionStore {
         extensions: Vec<ExtensionVersion>,
         cx: &Context<Self>,
     ) -> Task<Result<Vec<ExtensionVersion>>> {
-        let on_client = HashSet::from_iter(extensions.iter().map(|e| e.id.as_str()));
-        let to_remove: Vec<Arc<str>> = self
-            .loaded_extensions
-            .keys()
-            .filter(|id| !on_client.contains(id.as_ref()))
-            .cloned()
-            .collect();
-        let to_load: Vec<ExtensionVersion> = extensions
-            .into_iter()
-            .filter(|e| {
-                if e.dev {
-                    return true;
+        let operation_lock = self.operation_lock.clone();
+        cx.spawn(async move |store, cx| {
+            let _operation_guard = operation_lock.lock().await;
+            let (to_remove, to_load) = store.update(cx, |store, _cx| {
+                let on_client = HashSet::from_iter(extensions.iter().map(|e| e.id.as_str()));
+                store
+                    .failed_removals
+                    .retain(|id| !on_client.contains(id.as_ref()));
+                let mut to_remove = store
+                    .loaded_extensions
+                    .keys()
+                    .filter(|id| !on_client.contains(id.as_ref()))
+                    .cloned()
+                    .collect::<Vec<Arc<str>>>();
+                to_remove.extend(store.failed_removals.drain());
+                let to_load = extensions
+                    .into_iter()
+                    .filter(|extension| {
+                        let is_new = store
+                            .loaded_extensions
+                            .get(extension.id.as_str())
+                            .is_none_or(|loaded| {
+                                loaded.version.as_ref() != extension.version.as_str()
+                            });
+                        extension.dev || is_new
+                    })
+                    .collect::<Vec<ExtensionVersion>>();
+                (to_remove, to_load)
+            })?;
+
+            let mut extensions_changed = false;
+            let result = async {
+                let mut missing = Vec::new();
+
+                for extension_id in to_remove {
+                    log::info!("removing extension: {extension_id}");
+                    let (was_loaded, removal) = store.update(cx, |store, cx| {
+                        (
+                            store.loaded_extensions.contains_key(&extension_id),
+                            store.uninstall_extension(&extension_id, cx),
+                        )
+                    })?;
+                    if was_loaded {
+                        extensions_changed = true;
+                    }
+                    if let Err(error) = removal.await {
+                        log::error!("failed to remove extension: {extension_id}, {error:#}");
+                        store.update(cx, |store, _cx| {
+                            store.failed_removals.insert(extension_id.clone());
+                        })?;
+                    }
                 }
-                self.loaded_extensions
-                    .get(e.id.as_str())
-                    .is_none_or(|loaded| loaded.as_ref() != e.version.as_str())
-            })
-            .collect();
 
-        cx.spawn(async move |this, cx| {
-            let mut missing = Vec::new();
+                for extension in to_load {
+                    match Self::load_extension(store.clone(), &extension, cx).await {
+                        Ok(changed) => {
+                            if changed {
+                                extensions_changed = true;
+                            }
+                            if extension.dev
+                                && !Self::is_loaded_content_up_to_date(&store, &extension, cx)?
+                            {
+                                missing.push(extension)
+                            }
+                        }
+                        Err(error) => {
+                            log::info!("failed to load extension: {}, {:#}", extension.id, error);
+                            missing.push(extension)
+                        }
+                    }
+                }
 
-            for extension_id in to_remove {
-                log::info!("removing extension: {}", extension_id);
-                this.update(cx, |this, cx| this.uninstall_extension(&extension_id, cx))?
-                    .await?;
+                anyhow::Ok(missing)
+            }
+            .await;
+
+            if extensions_changed {
+                store.update(cx, |_, cx| notify_extensions_changed(cx)).ok();
             }
 
-            for extension in to_load {
-                if let Err(e) = Self::load_extension(this.clone(), extension.clone(), cx).await {
-                    log::info!("failed to load extension: {}, {:#}", extension.id, e);
-                    missing.push(extension)
-                } else if extension.dev {
-                    missing.push(extension)
-                }
-            }
-
-            Ok(missing)
+            result
         })
     }
 
-    pub async fn load_extension(
-        this: WeakEntity<Self>,
-        extension: ExtensionVersion,
-        cx: &mut AsyncApp,
-    ) -> Result<()> {
-        let (fs, wasm_host, extension_dir) = this.update(cx, |this, _cx| {
-            this.loaded_extensions.insert(
-                extension.id.clone().into(),
-                extension.version.clone().into(),
-            );
-            (
-                this.fs.clone(),
-                this.wasm_host.clone(),
-                this.extension_dir.join(&extension.id),
-            )
+    fn is_loaded_content_up_to_date(
+        store: &WeakEntity<Self>,
+        extension: &ExtensionVersion,
+        cx: &AsyncApp,
+    ) -> Result<bool> {
+        let loaded_fingerprint = store.read_with(cx, |store, _cx| {
+            store
+                .loaded_extensions
+                .get(extension.id.as_str())
+                .and_then(|loaded| loaded.content_fingerprint)
         })?;
+        Ok(extension.content_fingerprint.is_some()
+            && extension.content_fingerprint == loaded_fingerprint)
+    }
 
-        let manifest = Arc::new(ExtensionManifest::load(fs.clone(), &extension_dir).await?);
+    async fn load_extension(
+        store: WeakEntity<Self>,
+        extension: &ExtensionVersion,
+        cx: &mut AsyncApp,
+    ) -> Result<bool> {
+        let (fs, wasm_host, extension_dir, loaded_fingerprint) =
+            store.read_with(cx, |store, _cx| {
+                (
+                    store.fs.clone(),
+                    store.wasm_host.clone(),
+                    store.extension_dir.join(&extension.id),
+                    store
+                        .loaded_extensions
+                        .get(extension.id.as_str())
+                        .filter(|loaded| loaded.version.as_ref() == extension.version.as_str())
+                        .and_then(|loaded| loaded.content_fingerprint),
+                )
+            })?;
+        let content_fingerprint = if extension.dev {
+            fingerprint_directory(&fs, &extension_dir, cx).await
+        } else {
+            None
+        };
+        if content_fingerprint.is_some() && content_fingerprint == loaded_fingerprint {
+            return Ok(false);
+        }
+        let loaded = Self::prepare_extension(
+            fs,
+            wasm_host,
+            extension_dir.clone(),
+            extension_dir,
+            extension,
+            content_fingerprint,
+            cx,
+        )
+        .await?;
+        let removal_tasks = store.update(cx, |store, cx| {
+            store.commit_extension(extension.id.as_str().into(), Some(loaded), cx)
+        })?;
+        for removal in join_all(removal_tasks).await {
+            removal.log_err();
+        }
+        Ok(true)
+    }
+
+    async fn prepare_extension(
+        fs: Arc<dyn Fs>,
+        wasm_host: Arc<WasmHost>,
+        load_dir: PathBuf,
+        installed_dir: PathBuf,
+        extension: &ExtensionVersion,
+        content_fingerprint: Option<u64>,
+        cx: &mut AsyncApp,
+    ) -> Result<LoadedExtension> {
+        let manifest = Arc::new(ExtensionManifest::load(fs.clone(), &load_dir).await?);
 
         debug_assert!(!manifest.languages.is_empty() || manifest.allow_remote_load());
 
-        if manifest.version.as_ref() != extension.version.as_str() {
-            anyhow::bail!(
-                "mismatched versions: ({}) != ({})",
-                manifest.version,
-                extension.version
-            )
-        }
+        anyhow::ensure!(
+            manifest.version.as_ref() == extension.version.as_str(),
+            "mismatched versions: ({}) != ({})",
+            manifest.version,
+            extension.version,
+        );
 
+        let mut languages = Vec::new();
         for language_path in &manifest.languages {
-            let language_path = extension_dir.join(language_path);
-            let config = fs
-                .load(&language_path.join(LanguageConfig::FILE_NAME))
-                .await?;
+            let config_path = load_dir.join(language_path).join(LanguageConfig::FILE_NAME);
+            let config = fs.load(&config_path).await?;
             let mut config = ::toml::from_str::<LanguageConfig>(&config)?;
-
-            this.update(cx, |this, _cx| {
-                this.loaded_languages
-                    .entry(manifest.id.clone())
-                    .or_default()
-                    .push(config.name.clone());
-
-                config.grammar = None;
-
-                this.proxy.register_language(
-                    config.name.clone(),
-                    None,
-                    config.matcher.clone(),
-                    config.hidden,
-                    Arc::new(move || {
-                        let config = config.clone();
-                        async move {
-                            Ok(LoadedLanguage {
-                                config,
-                                queries: LanguageQueries::default(),
-                                context_provider: None,
-                                toolchain_provider: None,
-                                manifest_name: None,
-                            })
-                        }
-                        .boxed()
-                    }),
-                );
-            })?;
+            config.grammar = None;
+            languages.push((config.name.clone(), config));
         }
 
-        if !manifest.allow_remote_load() {
-            return Ok(());
+        let mut language_servers = Vec::new();
+        let mut debug_adapters = Vec::new();
+        let mut debug_locators = Vec::new();
+        let mut wasm_extension: Option<Arc<dyn Extension>> = None;
+        if manifest.allow_remote_load() {
+            wasm_extension = Some(Arc::new(
+                WasmExtension::load(&load_dir, &manifest, wasm_host, cx).await?,
+            ));
+
+            for (language_server_id, language_server_config) in &manifest.language_servers {
+                for language in language_server_config.languages() {
+                    language_servers.push((language_server_id.clone(), language));
+                }
+            }
+
+            for (debug_adapter, meta) in &manifest.debug_adapters {
+                let schema_path = extension::build_debug_adapter_schema_path(debug_adapter, meta)?;
+                debug_adapters.push((debug_adapter.clone(), installed_dir.join(schema_path)));
+            }
+
+            debug_locators = manifest.debug_locators.keys().cloned().collect();
         }
 
-        let wasm_extension: Arc<dyn Extension> =
-            Arc::new(WasmExtension::load(&extension_dir, &manifest, wasm_host.clone(), cx).await?);
+        Ok(LoadedExtension {
+            version: extension.version.as_str().into(),
+            languages,
+            language_servers,
+            debug_adapters,
+            debug_locators,
+            wasm_extension,
+            content_fingerprint,
+        })
+    }
 
-        for (language_server_id, language_server_config) in &manifest.language_servers {
-            for language in language_server_config.languages() {
-                this.update(cx, |this, _cx| {
-                    this.loaded_language_servers
-                        .entry(manifest.id.clone())
-                        .or_default()
-                        .push((language_server_id.clone(), language.clone()));
-                    this.proxy.register_language_server(
-                        wasm_extension.clone(),
-                        language_server_id.clone(),
+    pub(crate) fn commit_extension(
+        &mut self,
+        extension_id: Arc<str>,
+        loaded: Option<LoadedExtension>,
+        cx: &mut App,
+    ) -> Vec<Task<Result<()>>> {
+        let previous = match loaded {
+            Some(loaded) => self.loaded_extensions.insert(extension_id.clone(), loaded),
+            None => self.loaded_extensions.remove(&extension_id),
+        };
+        let current = self.loaded_extensions.get(&extension_id).cloned();
+
+        let mut removal_tasks = Vec::new();
+        if let Some(previous) = previous {
+            let mut languages_to_remove = Vec::new();
+            for (language, _) in &previous.languages {
+                if current.as_ref().is_some_and(|current| {
+                    current.languages.iter().any(|(name, _)| name == language)
+                }) {
+                    continue;
+                }
+                match self.surviving_language_config(language) {
+                    Some(config) => register_language_from_config(&self.proxy, config),
+                    None => languages_to_remove.push(language.clone()),
+                }
+            }
+            self.proxy.remove_languages(&languages_to_remove, &[]);
+
+            for (server_name, language) in &previous.language_servers {
+                removal_tasks.push(self.proxy.remove_language_server(language, server_name, cx));
+                let in_current = current.as_ref().is_some_and(|current| {
+                    current
+                        .language_servers
+                        .iter()
+                        .any(|(name, language_name)| {
+                            name == server_name && language_name == language
+                        })
+                });
+                if in_current {
+                    continue;
+                }
+                if let Some(extension) = self.surviving_language_server(server_name, language) {
+                    self.proxy.register_language_server(
+                        extension,
+                        server_name.clone(),
                         language.clone(),
                     );
-                })?;
+                }
+            }
+
+            for (adapter_name, _) in &previous.debug_adapters {
+                if current.as_ref().is_some_and(|current| {
+                    current
+                        .debug_adapters
+                        .iter()
+                        .any(|(name, _)| name == adapter_name)
+                }) {
+                    continue;
+                }
+                match self.surviving_debug_adapter(adapter_name) {
+                    Some((schema_path, extension)) => {
+                        self.proxy.register_debug_adapter(
+                            extension,
+                            adapter_name.clone(),
+                            &schema_path,
+                        );
+                    }
+                    None => self.proxy.unregister_debug_adapter(adapter_name.clone()),
+                }
+            }
+
+            for locator_name in &previous.debug_locators {
+                if current.as_ref().is_some_and(|current| {
+                    current
+                        .debug_locators
+                        .iter()
+                        .any(|name| name == locator_name)
+                }) {
+                    continue;
+                }
+                match self.surviving_debug_locator(locator_name) {
+                    Some(extension) => {
+                        self.proxy
+                            .register_debug_locator(extension, locator_name.clone());
+                    }
+                    None => self.proxy.unregister_debug_locator(locator_name.clone()),
+                }
             }
-            log::info!("Loaded language server: {}", language_server_id);
         }
 
-        for (debug_adapter, meta) in &manifest.debug_adapters {
-            let schema_path = extension::build_debug_adapter_schema_path(debug_adapter, meta)?;
-
-            this.update(cx, |this, _cx| {
-                this.proxy.register_debug_adapter(
-                    wasm_extension.clone(),
-                    debug_adapter.clone(),
-                    &extension_dir.join(schema_path),
-                );
-            })?;
-            log::info!("Loaded debug adapter: {}", debug_adapter);
+        if let Some(current) = &current {
+            for (_, config) in &current.languages {
+                register_language_from_config(&self.proxy, config.clone());
+            }
+            if let Some(wasm_extension) = &current.wasm_extension {
+                for (server_name, language) in &current.language_servers {
+                    self.proxy.register_language_server(
+                        wasm_extension.clone(),
+                        server_name.clone(),
+                        language.clone(),
+                    );
+                    log::info!("Loaded language server: {server_name}");
+                }
+                for (adapter_name, schema_path) in &current.debug_adapters {
+                    self.proxy.register_debug_adapter(
+                        wasm_extension.clone(),
+                        adapter_name.clone(),
+                        schema_path,
+                    );
+                    log::info!("Loaded debug adapter: {adapter_name}");
+                }
+                for locator_name in &current.debug_locators {
+                    self.proxy
+                        .register_debug_locator(wasm_extension.clone(), locator_name.clone());
+                    log::info!("Loaded debug locator: {locator_name}");
+                }
+            }
         }
 
-        for debug_locator in manifest.debug_locators.keys() {
-            this.update(cx, |this, _cx| {
-                this.proxy
-                    .register_debug_locator(wasm_extension.clone(), debug_locator.clone());
-            })?;
-            log::info!("Loaded debug locator: {}", debug_locator);
-        }
+        removal_tasks
+    }
 
-        Ok(())
+    fn surviving_language_config(&self, language: &LanguageName) -> Option<LanguageConfig> {
+        self.loaded_extensions.values().find_map(|extension| {
+            extension
+                .languages
+                .iter()
+                .find(|(name, _)| name == language)
+                .map(|(_, config)| config.clone())
+        })
+    }
+
+    fn surviving_language_server(
+        &self,
+        server_name: &LanguageServerName,
+        language: &LanguageName,
+    ) -> Option<Arc<dyn Extension>> {
+        self.loaded_extensions.values().find_map(|extension| {
+            let wasm_extension = extension.wasm_extension.clone()?;
+            extension
+                .language_servers
+                .iter()
+                .any(|(name, language_name)| name == server_name && language_name == language)
+                .then_some(wasm_extension)
+        })
+    }
+
+    fn surviving_debug_adapter(
+        &self,
+        adapter_name: &Arc<str>,
+    ) -> Option<(PathBuf, Arc<dyn Extension>)> {
+        self.loaded_extensions.values().find_map(|extension| {
+            let wasm_extension = extension.wasm_extension.clone()?;
+            extension
+                .debug_adapters
+                .iter()
+                .find(|(name, _)| name == adapter_name)
+                .map(|(_, schema_path)| (schema_path.clone(), wasm_extension))
+        })
+    }
+
+    fn surviving_debug_locator(&self, locator_name: &Arc<str>) -> Option<Arc<dyn Extension>> {
+        self.loaded_extensions.values().find_map(|extension| {
+            let wasm_extension = extension.wasm_extension.clone()?;
+            extension
+                .debug_locators
+                .iter()
+                .any(|name| name == locator_name)
+                .then_some(wasm_extension)
+        })
     }
 
     fn uninstall_extension(
@@ -226,33 +488,14 @@ impl HeadlessExtensionStore {
         extension_id: &Arc<str>,
         cx: &mut Context<Self>,
     ) -> Task<Result<()>> {
-        self.loaded_extensions.remove(extension_id);
+        let removal_tasks = self.commit_extension(extension_id.clone(), None, cx);
 
-        let languages_to_remove = self
-            .loaded_languages
-            .remove(extension_id)
-            .unwrap_or_default();
-        self.proxy.remove_languages(&languages_to_remove, &[]);
-
-        let servers_to_remove = self
-            .loaded_language_servers
-            .remove(extension_id)
-            .unwrap_or_default();
-        let proxy = self.proxy.clone();
-        let path = self.extension_dir.join(&extension_id.to_string());
+        let path = self.extension_dir.join(extension_id.as_ref());
         let fs = self.fs.clone();
-        cx.spawn(async move |_, cx| {
-            let mut removal_tasks = Vec::with_capacity(servers_to_remove.len());
-            cx.update(|cx| {
-                for (language_server_name, language) in servers_to_remove {
-                    removal_tasks.push(proxy.remove_language_server(
-                        &language,
-                        &language_server_name,
-                        cx,
-                    ));
-                }
-            });
-            let _ = join_all(removal_tasks).await;
+        cx.spawn(async move |_, _cx| {
+            for removal in join_all(removal_tasks).await {
+                removal.log_err();
+            }
 
             fs.remove_dir(
                 &path,
@@ -274,20 +517,104 @@ impl HeadlessExtensionStore {
     ) -> Task<Result<()>> {
         let path = self.extension_dir.join(&extension.id);
         let fs = self.fs.clone();
+        let wasm_host = self.wasm_host.clone();
+        let operation_lock = self.operation_lock.clone();
 
-        cx.spawn(async move |this, cx| {
-            if fs.is_dir(&path).await {
-                this.update(cx, |this, cx| {
-                    this.uninstall_extension(&extension.id.clone().into(), cx)
-                })?
-                .await?;
+        cx.spawn(async move |store, cx| {
+            let _operation_guard = operation_lock.lock().await;
+
+            let loaded_fingerprint = store.read_with(cx, |store, _cx| {
+                store
+                    .loaded_extensions
+                    .get(extension.id.as_str())
+                    .filter(|loaded| loaded.version.as_ref() == extension.version.as_str())
+                    .and_then(|loaded| loaded.content_fingerprint)
+            })?;
+            let content_fingerprint = if extension.dev {
+                fingerprint_directory(&fs, &tmp_path, cx).await
+            } else {
+                None
+            };
+            if content_fingerprint.is_some() && content_fingerprint == loaded_fingerprint {
+                fs.remove_dir(
+                    &tmp_path,
+                    RemoveOptions {
+                        recursive: true,
+                        ignore_if_not_exists: true,
+                    },
+                )
+                .await
+                .log_err();
+                return Ok(());
             }
 
-            fs.rename(&tmp_path, &path, RenameOptions::default())
-                .await
-                .with_context(|| format!("Failed to rename {tmp_path:?} to {path:?}"))?;
+            let mut installed_dir_touched = false;
+            let result = async {
+                let loaded = Self::prepare_extension(
+                    fs.clone(),
+                    wasm_host,
+                    tmp_path.clone(),
+                    path.clone(),
+                    &extension,
+                    content_fingerprint,
+                    cx,
+                )
+                .await?;
 
-            Self::load_extension(this, extension, cx).await
+                installed_dir_touched = true;
+                fs.remove_dir(
+                    &path,
+                    RemoveOptions {
+                        recursive: true,
+                        ignore_if_not_exists: true,
+                    },
+                )
+                .await
+                .with_context(|| format!("Removing existing directory {path:?}"))?;
+                fs.rename(&tmp_path, &path, RenameOptions::default())
+                    .await
+                    .with_context(|| format!("Failed to rename {tmp_path:?} to {path:?}"))?;
+
+                let removal_tasks = store.update(cx, |store, cx| {
+                    store.commit_extension(extension.id.as_str().into(), Some(loaded), cx)
+                })?;
+                installed_dir_touched = false;
+                for removal in join_all(removal_tasks).await {
+                    removal.log_err();
+                }
+
+                store.update(cx, |_, cx| notify_extensions_changed(cx))?;
+                anyhow::Ok(())
+            }
+            .await;
+
+            if result.is_err() {
+                fs.remove_dir(
+                    &tmp_path,
+                    RemoveOptions {
+                        recursive: true,
+                        ignore_if_not_exists: true,
+                    },
+                )
+                .await
+                .log_err();
+
+                if installed_dir_touched
+                    && let Ok(removal_tasks) = store.update(cx, |store, cx| {
+                        store
+                            .failed_removals
+                            .insert(Arc::from(extension.id.as_str()));
+                        store.commit_extension(extension.id.as_str().into(), None, cx)
+                    })
+                {
+                    for removal in join_all(removal_tasks).await {
+                        removal.log_err();
+                    }
+                    store.update(cx, |_, cx| notify_extensions_changed(cx)).ok();
+                }
+            }
+
+            result
         })
     }
 
@@ -305,6 +632,7 @@ impl HeadlessExtensionStore {
                     id: p.id,
                     version: p.version,
                     dev: p.dev,
+                    content_fingerprint: p.content_fingerprint,
                 });
         let missing_extensions = extension_store
             .update(&mut cx, |extension_store, cx| {
@@ -312,6 +640,9 @@ impl HeadlessExtensionStore {
             })
             .await?;
 
+        let fs = extension_store.read_with(&cx, |extension_store, _cx| extension_store.fs.clone());
+        remove_stale_uploads(&fs, paths::remote_extensions_uploads_dir()).await;
+
         Ok(proto::SyncExtensionsResponse {
             missing_extensions: missing_extensions
                 .into_iter()
@@ -319,6 +650,7 @@ impl HeadlessExtensionStore {
                     id: e.id,
                     version: e.version,
                     dev: e.dev,
+                    content_fingerprint: e.content_fingerprint,
                 })
                 .collect(),
             tmp_dir: paths::remote_extensions_uploads_dir()
@@ -344,6 +676,7 @@ impl HeadlessExtensionStore {
                         id: extension.id,
                         version: extension.version,
                         dev: extension.dev,
+                        content_fingerprint: extension.content_fingerprint,
                     },
                     PathBuf::from(envelope.payload.tmp_dir),
                     cx,
@@ -354,3 +687,121 @@ impl HeadlessExtensionStore {
         Ok(proto::Ack {})
     }
 }
+
+pub(crate) async fn remove_stale_uploads(fs: &Arc<dyn Fs>, uploads_dir: &Path) {
+    let Ok(mut entries) = fs.read_dir(uploads_dir).await else {
+        return;
+    };
+    let now = SystemTime::now();
+    while let Some(entry) = entries.next().await {
+        let Some(path) = entry.log_err() else {
+            continue;
+        };
+        let Ok(Some(metadata)) = fs.metadata(&path).await else {
+            continue;
+        };
+        let is_stale = match now.duration_since(metadata.mtime.timestamp_for_user()) {
+            Ok(age) => age > STALE_UPLOAD_TTL,
+            Err(error) => error.duration() > STALE_UPLOAD_TTL,
+        };
+        if !is_stale {
+            continue;
+        }
+        if metadata.is_dir {
+            fs.remove_dir(
+                &path,
+                RemoveOptions {
+                    recursive: true,
+                    ignore_if_not_exists: true,
+                },
+            )
+            .await
+            .log_err();
+        } else {
+            fs.remove_file(
+                &path,
+                RemoveOptions {
+                    recursive: false,
+                    ignore_if_not_exists: true,
+                },
+            )
+            .await
+            .log_err();
+        }
+    }
+}
+
+async fn fingerprint_directory(fs: &Arc<dyn Fs>, path: &Path, cx: &AsyncApp) -> Option<u64> {
+    let fs = fs.clone();
+    let path = path.to_path_buf();
+    cx.background_executor()
+        .spawn(async move { hash_directory_contents(&fs, &path).await })
+        .await
+        .log_err()
+}
+
+pub(crate) async fn hash_directory_contents(fs: &Arc<dyn Fs>, root: &Path) -> Result<u64> {
+    let mut hasher = FxHasher::default();
+    let mut directories = vec![root.to_path_buf()];
+    while let Some(directory) = directories.pop() {
+        let mut paths = Vec::new();
+        let mut entries = fs.read_dir(&directory).await?;
+        while let Some(entry) = entries.next().await {
+            paths.push(entry?);
+        }
+        paths.sort();
+        for path in paths {
+            let metadata = fs
+                .metadata(&path)
+                .await?
+                .with_context(|| format!("missing metadata for {path:?}"))?;
+            for component in path.strip_prefix(root)?.components() {
+                component.as_os_str().to_string_lossy().hash(&mut hasher);
+            }
+            if metadata.is_symlink {
+                2_u8.hash(&mut hasher);
+                fs.read_link(&path)
+                    .await?
+                    .to_string_lossy()
+                    .hash(&mut hasher);
+            } else if metadata.is_dir {
+                1_u8.hash(&mut hasher);
+                directories.push(path);
+            } else {
+                0_u8.hash(&mut hasher);
+                fs.load_bytes(&path).await?.hash(&mut hasher);
+            }
+        }
+    }
+    Ok(hasher.finish())
+}
+
+fn notify_extensions_changed(cx: &mut App) {
+    if let Some(events) = ExtensionEvents::try_global(cx) {
+        events.update(cx, |events, cx| {
+            events.emit(Event::ExtensionsInstalledChanged, cx)
+        });
+    }
+}
+
+fn register_language_from_config(proxy: &ExtensionHostProxy, config: LanguageConfig) {
+    proxy.register_language(
+        config.name.clone(),
+        None,
+        config.matcher.clone(),
+        config.hidden,
+        Arc::new(move || {
+            let config = config.clone();
+            async move {
+                Ok(LoadedLanguage {
+                    config,
+                    queries: LanguageQueries::default(),
+                    context_provider: None,
+                    toolchain_provider: None,
+                    manifest_name: None,
+                })
+            }
+            .boxed()
+        }),
+    );
+}

crates/fs/src/fs.rs

@@ -2527,6 +2527,13 @@ impl FakeFs {
             .insert(Self::remove_dir_error_key(path.as_ref()), message);
     }
 
+    pub fn clear_remove_dir_error(&self, path: impl AsRef<Path>) {
+        self.state
+            .lock()
+            .remove_dir_errors
+            .remove(&Self::remove_dir_error_key(path.as_ref()));
+    }
+
     /// Entry resolution in `try_entry` ignores drive prefixes, so the error
     /// injection map must too.
     /// Otherwise, on Windows, a key like `C:\workspace\dir` would never match a

crates/json_schema_store/Cargo.toml

@@ -15,7 +15,10 @@ path = "src/json_schema_store.rs"
 default = []
 
 [dev-dependencies]
+fs = { workspace = true, features = ["test-support"] }
 gpui = { workspace = true, features = ["test-support"] }
+language = { workspace = true, features = ["test-support"] }
+project = { workspace = true, features = ["test-support"] }
 settings = { workspace = true, features = ["test-support"] }
 
 [dependencies]

crates/json_schema_store/src/json_schema_store.rs

@@ -309,26 +309,7 @@ async fn resolve_dynamic_schema(
             })
         }
         "settings" => {
-            let mut lsp_adapter_names: Vec<String> = languages
-                .all_lsp_adapters()
-                .into_iter()
-                .map(|adapter| adapter.name())
-                .chain(languages.available_lsp_adapter_names())
-                .map(|name| name.to_string())
-                .collect();
-
-            let mut i = 0;
-            while i < lsp_adapter_names.len() {
-                let mut j = i + 1;
-                while j < lsp_adapter_names.len() {
-                    if lsp_adapter_names[i] == lsp_adapter_names[j] {
-                        lsp_adapter_names.swap_remove(j);
-                    } else {
-                        j += 1;
-                    }
-                }
-                i += 1;
-            }
+            let lsp_adapter_names = all_lsp_adapter_names(&languages);
 
             cx.update(|cx| {
                 let font_names = &cx.text_system().all_font_names();
@@ -374,11 +355,7 @@ async fn resolve_dynamic_schema(
             })
         }
         "project_settings" => {
-            let lsp_adapter_names = languages
-                .all_lsp_adapters()
-                .into_iter()
-                .map(|adapter| adapter.name().to_string())
-                .collect::<Vec<_>>();
+            let lsp_adapter_names = all_lsp_adapter_names(&languages);
 
             let language_names = &languages
                 .language_names()
@@ -619,6 +596,19 @@ fn root_schema_from_action_schema(
     schema
 }
 
+fn all_lsp_adapter_names(languages: &LanguageRegistry) -> Vec<String> {
+    let mut names = languages
+        .all_lsp_adapters()
+        .into_iter()
+        .map(|adapter| adapter.name())
+        .chain(languages.available_lsp_adapter_names())
+        .map(|name| name.to_string())
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+    names.dedup();
+    names
+}
+
 #[inline]
 fn schema_file_match(path: &std::path::Path) -> String {
     path.strip_prefix(path.parent().unwrap().parent().unwrap())
@@ -627,3 +617,46 @@ fn schema_file_match(path: &std::path::Path) -> String {
         .to_string()
         .replace('\\', "/")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use fs::FakeFs;
+    use gpui::TestAppContext;
+    use language::FakeLspAdapter;
+    use project::Project;
+    use settings::SettingsStore;
+
+    #[gpui::test]
+    async fn test_project_settings_schema_includes_available_lsp_adapters(cx: &mut TestAppContext) {
+        cx.update(|cx| {
+            let settings_store = SettingsStore::test(cx);
+            cx.set_global(settings_store);
+        });
+
+        let fs = FakeFs::new(cx.executor());
+        let project = Project::test(fs, [], cx).await;
+        let (lsp_store, languages) = project.read_with(cx, |project, _| {
+            (project.lsp_store(), project.languages().clone())
+        });
+
+        languages.register_available_lsp_adapter(
+            LanguageServerName("extension-provided-lsp".into()),
+            Arc::new(FakeLspAdapter::default()),
+        );
+
+        let schema = resolve_dynamic_schema(lsp_store, "project_settings", &mut cx.to_async())
+            .await
+            .unwrap();
+
+        let lsp_properties = schema
+            .pointer("/$defs/LspSettingsMap/properties")
+            .expect("LspSettingsMap should have properties")
+            .as_object()
+            .unwrap();
+        assert_eq!(
+            lsp_properties.keys().cloned().collect::<Vec<_>>(),
+            vec!["extension-provided-lsp".to_string()],
+        );
+    }
+}

crates/language/src/available_languages.rs

@@ -17,6 +17,13 @@ pub struct AvailableLanguage {
     pub(super) load: LanguageLoader,
     pub(super) loaded: bool,
     pub(super) manifest_name: Option<ManifestName>,
+    pub(super) origin: LanguageOrigin,
+}
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+pub enum LanguageOrigin {
+    Native,
+    Extension,
 }
 
 impl AvailableLanguage {
@@ -57,17 +64,28 @@ impl AvailableLanguages {
         hidden: bool,
         manifest_name: Option<ManifestName>,
         load: LanguageLoader,
-    ) -> bool {
+        origin: LanguageOrigin,
+    ) -> Option<bool> {
         if let Some(existing_language) = self
             .0
             .iter_mut()
             .find(|existing_language| existing_language.name == name)
         {
+            if origin == LanguageOrigin::Extension
+                && existing_language.origin == LanguageOrigin::Native
+            {
+                return None;
+            }
+            existing_language.id = LanguageId::new();
             existing_language.grammar = grammar;
             existing_language.matcher = matcher;
+            existing_language.hidden = hidden;
             existing_language.load = load;
             existing_language.manifest_name = manifest_name;
-            false
+            existing_language.origin = origin;
+            let was_loaded = existing_language.loaded;
+            existing_language.loaded = false;
+            Some(was_loaded)
         } else {
             self.add(AvailableLanguage {
                 id: LanguageId::new(),
@@ -78,8 +96,9 @@ impl AvailableLanguages {
                 load,
                 loaded: false,
                 manifest_name,
+                origin,
             });
-            true
+            Some(false)
         }
     }
 
@@ -159,8 +178,20 @@ impl AvailableLanguages {
         }
     }
 
-    pub(super) fn remove(&mut self, names: &[LanguageName]) {
-        self.0.retain(|language| !names.contains(&language.name));
+    pub(super) fn remove_extension_languages(
+        &mut self,
+        names: &[LanguageName],
+    ) -> Vec<LanguageName> {
+        let mut removed = Vec::new();
+        self.0.retain(|language| {
+            let should_remove =
+                language.origin == LanguageOrigin::Extension && names.contains(&language.name);
+            if should_remove {
+                removed.push(language.name.clone());
+            }
+            !should_remove
+        });
+        removed
     }
 
     pub(super) fn mark_loaded(&mut self, id: LanguageId) {

crates/language/src/buffer_tests.rs

@@ -397,6 +397,162 @@ async fn test_language_for_file_with_custom_file_types(cx: &mut TestAppContext)
     assert_eq!(language_name(language), "Dockerfile");
 }
 
+#[gpui::test]
+async fn test_reregistering_language_during_load_yields_current_language(cx: &mut TestAppContext) {
+    let registry = Arc::new(LanguageRegistry::test(cx.executor()));
+    let (unblock_stale_load_tx, unblock_stale_load_rx) = futures::channel::oneshot::channel::<()>();
+    let unblock_stale_load_rx = std::sync::Mutex::new(Some(unblock_stale_load_rx));
+
+    let stale_config = LanguageConfig {
+        name: LanguageName::new_static("TheLanguage"),
+        matcher: Arc::new(LanguageMatcher {
+            path_suffixes: vec!["stale".to_string()],
+            ..LanguageMatcher::default()
+        }),
+        ..LanguageConfig::default()
+    };
+    registry.register_language(
+        stale_config.name.clone(),
+        None,
+        stale_config.matcher.clone(),
+        false,
+        None,
+        Arc::new(move || {
+            let unblock_stale_load_rx = unblock_stale_load_rx
+                .lock()
+                .expect("the stale loader mutex should not be poisoned")
+                .take();
+            let stale_config = stale_config.clone();
+            async move {
+                if let Some(unblock_stale_load_rx) = unblock_stale_load_rx {
+                    unblock_stale_load_rx.await.ok();
+                }
+                Ok(LoadedLanguage {
+                    config: stale_config,
+                    queries: LanguageQueries::default(),
+                    context_provider: None,
+                    toolchain_provider: None,
+                    manifest_name: None,
+                })
+            }
+            .boxed()
+        }),
+    );
+
+    let pending_language = registry.language_for_name("TheLanguage");
+    cx.executor().run_until_parked();
+
+    registry.register_test_language(LanguageConfig {
+        name: LanguageName::new_static("TheLanguage"),
+        matcher: Arc::new(LanguageMatcher {
+            path_suffixes: vec!["fresh".to_string()],
+            ..LanguageMatcher::default()
+        }),
+        ..LanguageConfig::default()
+    });
+
+    unblock_stale_load_tx.send(()).unwrap();
+    let language = pending_language.await.unwrap();
+    assert_eq!(
+        language.config.matcher.path_suffixes,
+        vec!["fresh".to_string()],
+        "a load that races with a re-registration should resolve to the re-registered language"
+    );
+}
+
+#[gpui::test]
+async fn test_reregistering_language_during_failed_load_yields_current_language(
+    cx: &mut TestAppContext,
+) {
+    let registry = Arc::new(LanguageRegistry::test(cx.executor()));
+    let (unblock_stale_load_tx, unblock_stale_load_rx) = futures::channel::oneshot::channel::<()>();
+    let unblock_stale_load_rx = std::sync::Mutex::new(Some(unblock_stale_load_rx));
+
+    registry.register_language(
+        LanguageName::new_static("TheLanguage"),
+        None,
+        Arc::new(LanguageMatcher {
+            path_suffixes: vec!["stale".to_string()],
+            ..LanguageMatcher::default()
+        }),
+        false,
+        None,
+        Arc::new(move || {
+            let unblock_stale_load_rx = unblock_stale_load_rx
+                .lock()
+                .expect("the stale loader mutex should not be poisoned")
+                .take();
+            async move {
+                if let Some(unblock_stale_load_rx) = unblock_stale_load_rx {
+                    unblock_stale_load_rx.await.ok();
+                }
+                Err(anyhow::anyhow!("simulated load failure"))
+            }
+            .boxed()
+        }),
+    );
+
+    let pending_language = registry.language_for_name("TheLanguage");
+    cx.executor().run_until_parked();
+
+    registry.register_test_language(LanguageConfig {
+        name: LanguageName::new_static("TheLanguage"),
+        matcher: Arc::new(LanguageMatcher {
+            path_suffixes: vec!["fresh".to_string()],
+            ..LanguageMatcher::default()
+        }),
+        ..LanguageConfig::default()
+    });
+
+    unblock_stale_load_tx.send(()).unwrap();
+    let language = pending_language.await.unwrap();
+    assert_eq!(
+        language.config.matcher.path_suffixes,
+        vec!["fresh".to_string()],
+        "a failed load that races with a re-registration should resolve to the re-registered language"
+    );
+}
+
+#[gpui::test]
+async fn test_extension_grammar_cannot_shadow_native_grammar(cx: &mut TestAppContext) {
+    let registry = Arc::new(LanguageRegistry::test(cx.executor()));
+    registry.register_native_grammars([("rust", tree_sitter_rust::LANGUAGE)]);
+    registry.register_wasm_grammars(vec![(
+        Arc::from("rust"),
+        PathBuf::from("/extensions/bogus/grammars/rust.wasm"),
+    )]);
+
+    registry.register_test_language(LanguageConfig {
+        name: LanguageName::new_static("TheLanguage"),
+        grammar: Some(Arc::from("rust")),
+        matcher: Arc::new(LanguageMatcher {
+            path_suffixes: vec!["the".to_string()],
+            ..LanguageMatcher::default()
+        }),
+        ..LanguageConfig::default()
+    });
+    let language = registry.language_for_name("TheLanguage").await.unwrap();
+    assert!(
+        language.grammar().is_some(),
+        "an extension grammar must not replace a native grammar with the same name"
+    );
+
+    registry.remove_languages(&[], &[Arc::from("rust")]);
+    registry.register_test_language(LanguageConfig {
+        name: LanguageName::new_static("TheOtherLanguage"),
+        grammar: Some(Arc::from("rust")),
+        ..LanguageConfig::default()
+    });
+    let language = registry
+        .language_for_name("TheOtherLanguage")
+        .await
+        .unwrap();
+    assert!(
+        language.grammar().is_some(),
+        "removing an extension grammar must not remove the native grammar it failed to shadow"
+    );
+}
+
 fn file(path: &str) -> Arc<dyn File> {
     Arc::new(TestFile {
         path: Arc::from(rel_path(path)),

crates/language/src/language_registry.rs

@@ -1,4 +1,4 @@
-use crate::available_languages::AvailableLanguage;
+use crate::available_languages::{AvailableLanguage, LanguageOrigin};
 use crate::{
     CachedLspAdapter, File, Language, LanguageConfig, LanguageId, LanguageMatcher,
     LanguageServerName, LspAdapter, ManifestName, PLAIN_TEXT, ToolchainLister,
@@ -384,23 +384,71 @@ impl LanguageRegistry {
         manifest_name: Option<ManifestName>,
         load: LanguageLoader,
     ) {
-        let state = &mut *self.state.write();
-
-        let was_added = state.available_languages.register(
+        self.register_language_with_origin(
             name,
             grammar_name,
             matcher,
             hidden,
             manifest_name,
             load,
+            LanguageOrigin::Native,
         );
-        if !was_added {
-            return;
+    }
+
+    pub fn register_extension_language(
+        &self,
+        name: LanguageName,
+        grammar_name: Option<Arc<str>>,
+        matcher: Arc<LanguageMatcher>,
+        hidden: bool,
+        manifest_name: Option<ManifestName>,
+        load: LanguageLoader,
+    ) -> bool {
+        self.register_language_with_origin(
+            name,
+            grammar_name,
+            matcher,
+            hidden,
+            manifest_name,
+            load,
+            LanguageOrigin::Extension,
+        )
+    }
+
+    fn register_language_with_origin(
+        &self,
+        name: LanguageName,
+        grammar_name: Option<Arc<str>>,
+        matcher: Arc<LanguageMatcher>,
+        hidden: bool,
+        manifest_name: Option<ManifestName>,
+        load: LanguageLoader,
+        origin: LanguageOrigin,
+    ) -> bool {
+        let state = &mut *self.state.write();
+
+        let Some(was_loaded) = state.available_languages.register(
+            name.clone(),
+            grammar_name,
+            matcher,
+            hidden,
+            manifest_name,
+            load,
+            origin,
+        ) else {
+            log::warn!(
+                "not registering extension language {name}: a language with this name is already registered outside of extensions"
+            );
+            return false;
+        };
+        if was_loaded {
+            state.languages.retain(|language| language.name() != name);
         }
 
         state.version += 1;
         state.reload_count += 1;
         *state.subscription.0.borrow_mut() = ();
+        true
     }
 
     /// Adds grammars to the registry. Language configurations reference a grammar by name. The
@@ -423,11 +471,17 @@ impl LanguageRegistry {
         }
 
         let mut state = self.state.write();
-        state.grammars.extend(
-            grammars
-                .into_iter()
-                .map(|(name, path)| (name, AvailableGrammar::Unloaded(path))),
-        );
+        for (name, path) in grammars {
+            if let Some(AvailableGrammar::Native(_)) = state.grammars.get(&name) {
+                log::warn!(
+                    "not registering extension grammar {name}: a native grammar with this name is already registered"
+                );
+                continue;
+            }
+            state
+                .grammars
+                .insert(name, AvailableGrammar::Unloaded(path));
+        }
         state.version += 1;
         state.reload_count += 1;
         *state.subscription.0.borrow_mut() = ();
@@ -469,6 +523,7 @@ impl LanguageRegistry {
             manifest_name: None,
             load: Arc::new(|| async { Err(anyhow!("already loaded")) }.boxed()),
             loaded: true,
+            origin: LanguageOrigin::Native,
         });
         state.add(language);
     }
@@ -665,34 +720,76 @@ impl LanguageRegistry {
                         }
                         .await;
 
-                        match language {
-                            Ok(language) => {
-                                let language = Arc::new(language);
-                                let mut state = this.state.write();
-
-                                state.add(language.clone());
+                        let language = language.map(Arc::new);
+                        if let Err(error) = &language {
+                            log::error!("failed to load language {language_name}:\n{error:?}");
+                        }
+                        let stale_txs = {
+                            let mut state = this.state.write();
+                            let is_current = state
+                                .available_languages
+                                .get_language(language_id)
+                                .is_some();
+                            if is_current {
+                                if let Ok(language) = &language {
+                                    state.add(language.clone());
+                                }
                                 state.mark_language_loaded(language_id);
-                                if let Some(mut txs) = state.loading_languages.remove(&language_id)
-                                {
-                                    for tx in txs.drain(..) {
-                                        let _ = tx.send(Ok(language.clone()));
-                                    }
-                                }
-                            }
-                            Err(e) => {
-                                log::error!("failed to load language {language_name}:\n{e:?}");
-                                let mut state = this.state.write();
-                                state.mark_language_loaded(language_id);
-                                if let Some(mut txs) = state.loading_languages.remove(&language_id)
-                                {
-                                    for tx in txs.drain(..) {
-                                        let _ = tx.send(Err(anyhow!(
-                                            "failed to load language {language_name}: {e}",
-                                        )));
+                                if let Some(txs) = state.loading_languages.remove(&language_id) {
+                                    for tx in txs {
+                                        let _ = tx.send(match &language {
+                                            Ok(language) => Ok(language.clone()),
+                                            Err(error) => Err(anyhow!(
+                                                "failed to load language {language_name}: {error}"
+                                            )),
+                                        });
                                     }
                                 }
+                                None
+                            } else {
+                                let txs = state
+                                    .loading_languages
+                                    .remove(&language_id)
+                                    .unwrap_or_default();
+                                let replacement_id = state
+                                    .available_languages
+                                    .find_by_exact_name(language_name.0.as_ref())
+                                    .map(|language| language.id());
+                                Some((txs, replacement_id))
                             }
                         };
+                        if let Some((txs, replacement_id)) = stale_txs
+                            && !txs.is_empty()
+                        {
+                            match replacement_id {
+                                Some(replacement_id) => {
+                                    let result = match this.load_language(replacement_id).await {
+                                        Ok(result) => result,
+                                        Err(_) => Err(anyhow!(LanguageNotFound)),
+                                    };
+                                    match result {
+                                        Ok(replacement_language) => {
+                                            for tx in txs {
+                                                let _ = tx.send(Ok(replacement_language.clone()));
+                                            }
+                                        }
+                                        Err(error) => {
+                                            let message = format!("{error:#}");
+                                            for tx in txs {
+                                                let _ = tx.send(Err(anyhow!(
+                                                    "failed to load language {language_name}: {message}"
+                                                )));
+                                            }
+                                        }
+                                    }
+                                }
+                                None => {
+                                    for tx in txs {
+                                        let _ = tx.send(Err(anyhow!(LanguageNotFound)));
+                                    }
+                                }
+                            }
+                        }
                     })
                     .detach();
 
@@ -933,11 +1030,14 @@ impl LanguageRegistryState {
             return;
         }
 
+        let removed_languages = self
+            .available_languages
+            .remove_extension_languages(languages_to_remove);
         self.languages
-            .retain(|language| !languages_to_remove.contains(&language.name()));
-        self.available_languages.remove(languages_to_remove);
-        self.grammars
-            .retain(|name, _| !grammars_to_remove.contains(name));
+            .retain(|language| !removed_languages.contains(&language.name()));
+        self.grammars.retain(|name, grammar| {
+            !grammars_to_remove.contains(name) || matches!(grammar, AvailableGrammar::Native(_))
+        });
         self.version += 1;
         self.reload_count += 1;
         *self.subscription.0.borrow_mut() = ();

crates/language_extension/src/language_extension.rs

@@ -51,9 +51,15 @@ impl ExtensionLanguageProxy for LanguageServerRegistryProxy {
         matcher: Arc<LanguageMatcher>,
         hidden: bool,
         load: LanguageLoader,
-    ) {
+    ) -> bool {
         self.language_registry
-            .register_language(language, grammar, matcher, hidden, None, load);
+            .register_extension_language(language, grammar, matcher, hidden, None, load)
+    }
+
+    fn is_language_registered(&self, language: &LanguageName) -> bool {
+        self.language_registry
+            .available_language_for_name(language.0.as_ref())
+            .is_some()
     }
 
     fn remove_languages(

crates/project/src/project.rs

@@ -3862,6 +3862,7 @@ impl Project {
                 });
                 cx.emit(Event::DisconnectedFromRemote { server_not_running });
             }
+            &remote::RemoteClientEvent::Reconnected => {}
         }
     }
 

crates/proto/proto/app.proto

@@ -44,6 +44,7 @@ message Extension {
   string id = 1;
   string version = 2;
   bool dev = 3;
+  optional uint64 content_fingerprint = 4;
 }
 
 message SyncExtensions {

crates/recent_projects/src/remote_connections.rs

@@ -6,7 +6,6 @@ use std::{
 use anyhow::{Context as _, Result};
 use askpass::EncryptedPassword;
 use editor::Editor;
-use extension_host::ExtensionStore;
 use futures::{FutureExt as _, channel::oneshot, select};
 use gpui::{AppContext, AsyncApp, PromptLevel, WindowHandle};
 
@@ -246,7 +245,6 @@ pub async fn open_remote_project(
         (window, workspace)
     };
 
-    let mut remote_workspace = None;
     loop {
         let (cancel_tx, mut cancel_rx) = oneshot::channel();
         let delegate = window.update(cx, {
@@ -413,8 +411,7 @@ pub async fn open_remote_project(
                 });
             }
 
-            Ok((workspace, items)) => {
-                remote_workspace = workspace;
+            Ok((_, items)) => {
                 navigate_to_positions(&window, items, &paths_with_positions, cx);
             }
         }
@@ -422,15 +419,6 @@ pub async fn open_remote_project(
         break;
     }
 
-    if let Some(remote_workspace) = remote_workspace {
-        remote_workspace.update(cx, |workspace, cx| {
-            if let Some(client) = workspace.project().read(cx).remote_client()
-                && let Some(extension_store) = ExtensionStore::try_global(cx)
-            {
-                extension_store.update(cx, |store, cx| store.register_remote_client(client, cx));
-            }
-        });
-    }
     Ok(window)
 }
 

crates/remote/src/remote_client.rs

@@ -339,6 +339,7 @@ pub struct RemoteClient {
 #[derive(Debug)]
 pub enum RemoteClientEvent {
     Disconnected { server_not_running: bool },
+    Reconnected,
 }
 
 impl EventEmitter<RemoteClientEvent> for RemoteClient {}
@@ -725,6 +726,8 @@ impl RemoteClient {
         cx.spawn(async move |this, cx| {
             let new_state = reconnect_task.await;
             this.update(cx, |this, cx| {
+                let reconnected = this.state_is(State::is_reconnecting)
+                    && matches!(&new_state, State::Connected { .. });
                 this.try_set_state(cx, |old_state| {
                     if old_state.is_reconnecting() {
                         match &new_state {
@@ -754,6 +757,10 @@ impl RemoteClient {
                     }
                 });
 
+                if reconnected {
+                    cx.emit(RemoteClientEvent::Reconnected);
+                }
+
                 if this.state_is(State::is_reconnect_failed) {
                     this.reconnect(cx)
                 } else if this.state_is(State::is_reconnect_exhausted) {

crates/remote/src/transport/docker.rs

@@ -436,6 +436,7 @@ impl DockerExecConnection {
         dst_path: String,
     ) -> Result<()> {
         let mut command = util::command::new_command(&docker_cli);
+        command.kill_on_drop(true);
         command.arg("cp");
         command.arg("-a");
         command.arg(&src_path);
@@ -455,6 +456,7 @@ impl DockerExecConnection {
         }
 
         let mut chown_command = util::command::new_command(&docker_cli);
+        chown_command.kill_on_drop(true);
         chown_command.arg("exec");
         chown_command.arg(connection_options.container_id);
         chown_command.arg("chown");

crates/remote/src/transport/ssh.rs

@@ -412,8 +412,10 @@ impl RemoteConnection for SshRemoteConnection {
         let src_path_display = src_path.display().to_string();
 
         let mut sftp_command = self.build_sftp_command();
+        sftp_command.kill_on_drop(true);
         let mut scp_command =
             self.build_scp_command(&src_path, &dest_path_str, Some(&["-C", "-r"]));
+        scp_command.kill_on_drop(true);
 
         cx.background_spawn(async move {
             // We will try SFTP first, and if that fails, we will fall back to SCP.

crates/remote/src/transport/wsl.rs

@@ -491,12 +491,13 @@ impl RemoteConnection for WslRemoteConnection {
             let options = self.connection_options.clone();
             async move {
                 let wsl_src = windows_path_to_wsl_path_impl(&options, &src_path).await?;
-                let command = wsl_command_impl(
+                let mut command = wsl_command_impl(
                     &options,
                     "cp",
                     &["-r", &wsl_src, &dest_path.to_string()],
                     true,
                 );
+                command.kill_on_drop(true);
                 run_wsl_command_impl(command).await.map_err(|e| {
                     anyhow!(
                         "failed to upload directory {} -> {}: {}",

crates/remote_server/src/remote_editing_tests.rs

@@ -45,7 +45,7 @@ use project::{
     image_store,
     search::{SearchQuery, SearchResult},
 };
-use remote::RemoteClient;
+use remote::{ConnectionState, RemoteClient, RemoteClientEvent};
 use rpc::proto;
 use serde_json::json;
 use settings::{
@@ -58,7 +58,7 @@ use std::{
     str::FromStr,
     sync::{
         Arc,
-        atomic::{AtomicUsize, Ordering},
+        atomic::{AtomicBool, AtomicUsize, Ordering},
     },
 };
 use unindent::Unindent as _;
@@ -2778,6 +2778,15 @@ async fn test_reconnect(cx: &mut TestAppContext, server_cx: &mut TestAppContext)
     });
 
     let client = cx.read(|cx| project.read(cx).remote_client().unwrap());
+    let reconnected = Arc::new(AtomicBool::new(false));
+    let _subscription = cx.update(|cx| {
+        let reconnected = reconnected.clone();
+        cx.subscribe(&client, move |_client, event, _cx| {
+            if matches!(event, RemoteClientEvent::Reconnected) {
+                reconnected.store(true, Ordering::SeqCst);
+            }
+        })
+    });
     client
         .update(cx, |client, cx| client.simulate_disconnect(cx))
         .detach();
@@ -2793,6 +2802,12 @@ async fn test_reconnect(cx: &mut TestAppContext, server_cx: &mut TestAppContext)
             .unwrap(),
         "fn one() -> usize { 100 }"
     );
+
+    cx.run_until_parked();
+    assert!(
+        reconnected.load(Ordering::SeqCst),
+        "a successful reconnect should emit RemoteClientEvent::Reconnected"
+    );
 }
 
 #[gpui::test]
@@ -4633,6 +4648,50 @@ async fn test_remote_trash_restore(cx: &mut TestAppContext, server_cx: &mut Test
     });
 }
 
+#[gpui::test]
+async fn test_remote_project_creation_notifies_new_entity_observers(
+    cx: &mut TestAppContext,
+    server_cx: &mut TestAppContext,
+) {
+    let server_fs = Arc::new(FakeFs::new(server_cx.executor()));
+    server_fs
+        .insert_tree(
+            path!("/project"),
+            json!({
+                "src": {
+                    "main.rs": "fn main() {}",
+                },
+                "README.md": "# Test Project",
+            }),
+        )
+        .await;
+
+    let observer_invocations = Arc::new(AtomicUsize::new(0));
+    cx.update(|cx| {
+        let observer_invocations = observer_invocations.clone();
+        cx.observe_new::<Project>(move |project, _window, cx| {
+            let Some(client) = project.remote_client() else {
+                return;
+            };
+            assert_eq!(
+                client.read(cx).connection_state(),
+                ConnectionState::Connected
+            );
+            observer_invocations.fetch_add(1, Ordering::SeqCst);
+        })
+        .detach();
+    });
+
+    let (project, _headless) = init_test(&server_fs, cx, server_cx).await;
+
+    assert_eq!(
+        observer_invocations.load(Ordering::SeqCst),
+        1,
+        "creating a remote project should notify new-entity observers with a connected remote client exactly once"
+    );
+    assert!(project.read_with(cx, |project, _| project.is_remote()));
+}
+
 pub async fn init_test(
     server_fs: &Arc<FakeFs>,
     cx: &mut TestAppContext,

crates/settings_content/src/project.rs

@@ -289,20 +289,25 @@ impl SemanticTokenRules {
     pub const FILE_NAME: &'static str = "semantic_token_rules.json";
 
     pub fn load(file_path: &Path) -> anyhow::Result<Self> {
-        let rules_content = std::fs::read(file_path).with_context(|| {
+        let rules_content = std::fs::read_to_string(file_path).with_context(|| {
             anyhow::anyhow!(
                 "Could not read semantic token rules from {}",
                 file_path.display()
             )
         })?;
 
-        serde_json_lenient::from_slice::<SemanticTokenRules>(&rules_content).with_context(|| {
+        Self::parse(&rules_content).with_context(|| {
             anyhow::anyhow!(
                 "Failed to parse semantic token rules from {}",
                 file_path.display()
             )
         })
     }
+
+    pub fn parse(file_content: &str) -> anyhow::Result<Self> {
+        serde_json_lenient::from_str::<SemanticTokenRules>(file_content)
+            .context("failed to parse semantic token rules")
+    }
 }
 
 impl crate::merge_from::MergeFrom for SemanticTokenRules {