feat(plugins): async plugin loading (#2327)

* work

* refactor(plugins): break down start plugin async function

* work

* loading messages

* nice ui

* floating panes and error handling

* cleanups and conflicting plugin/direction

* find exact pane when relayouting

* fix plugin pane titles

* kill loading tasks on exit

* refactor: move stuff around

* style(fmt): rustfmt

* various fixes and refactors

26 files changed, 1454 insertions, 309 deletions

diff --git a/src/main.rs b/src/main.rs
index 54ca71cb..e3737b19 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -32,6 +32,7 @@ fn main() {
         {
             let command_cli_action = CliAction::NewPane {
                 command,
+                plugin: None,
                 direction,
                 cwd,
                 floating,
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__bracketed_paste.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__bracketed_paste.snap
index dcb76b9e..59bdf1ff 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__bracketed_paste.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__bracketed_paste.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 1671
+assertion_line: 1640
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT           
+ Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT  BASE   
  Tip: Alt + <n> => new pane. Alt + <←↓↑→> or Alt + <hjkl> => navigate. Alt + <+|-> => resize pane.                      
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__lock_mode.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__lock_mode.snap
index 5583430d..70fa42f4 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__lock_mode.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__lock_mode.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 803
+assertion_line: 804
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- Ctrl + <g> LOCK  <> PANE  <> TAB  <> RESIZE  <> MOVE  <> SEARCH  <> SESSION  <> QUIT                  
+ Ctrl + <g> LOCK  <> PANE  <> TAB  <> RESIZE  <> MOVE  <> SEARCH  <> SESSION  <> QUIT          BASE  
  -- INTERFACE LOCKED --                                                                                                 
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__starts_with_one_terminal.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__starts_with_one_terminal.snap
index 6f2822c5..b45f0114 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__starts_with_one_terminal.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__starts_with_one_terminal.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 107
+assertion_line: 108
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT           
+ Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT  BASE   
  Tip: Alt + <n> => new pane. Alt + <←↓↑→> or Alt + <hjkl> => navigate. Alt + <+|-> => resize pane.                      
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__status_bar_loads_custom_keybindings.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__status_bar_loads_custom_keybindings.snap
index 5a88e98a..065bbfe5 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__status_bar_loads_custom_keybindings.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__status_bar_loads_custom_keybindings.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 398
+assertion_line: 987
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- <F1> LOCK  <F2> PANE  <F3> TAB  <F4> RESIZE  <F5> MOVE  <F6> SEARCH  <F7> SESSION  <F8> QUIT            
+ <F1> LOCK  <F2> PANE  <F3> TAB  <F4> RESIZE  <F5> MOVE  <F6> SEARCH  <F7> SESSION  <F8> QUIT    BASE  
  Tip: UNBOUND => open new pane. UNBOUND => navigate between panes. UNBOUND => increase/decrease pane size.              
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_pane.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_pane.snap
index e8c27717..76d6ee12 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_pane.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_pane.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 1949
+assertion_line: 1881
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT           
+ Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT  BASE   
  Tip: Alt + <n> => new pane. Alt + <←↓↑→> or Alt + <hjkl> => navigate. Alt + <+|-> => resize pane.                      
diff --git a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_tab.snap b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_tab.snap
index 56b417d7..6f84144f 100644
--- a/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_tab.snap
+++ b/src/tests/e2e/snapshots/zellij__tests__e2e__cases__undo_rename_tab.snap
@@ -1,6 +1,6 @@
 ---
 source: src/tests/e2e/cases.rs
-assertion_line: 1900
+assertion_line: 1832
 expression: last_snapshot
 ---
  Zellij (e2e-test)  Tab #1                                                                                            
@@ -25,5 +25,5 @@ expression: last_snapshot
 │                                                                                                                      │
 │                                                                                                                      │
 └──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
- Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT           
+ Ctrl + <g> LOCK  <p> PANE  <t> TAB  <n> RESIZE  <h> MOVE  <s> SEARCH  <o> SESSION  <q> QUIT  BASE   
  Tip: Alt + <n> => new pane. Alt + <←↓↑→> or Alt + <hjkl> => navigate. Alt + <+|-> => resize pane.                      
diff --git a/zellij-server/src/background_jobs.rs b/zellij-server/src/background_jobs.rs
index c612c9d5..8061b4a4 100644
--- a/zellij-server/src/background_jobs.rs
+++ b/zellij-server/src/background_jobs.rs
@@ -2,6 +2,10 @@ use zellij_utils::async_std::task;
 use zellij_utils::errors::{prelude::*, BackgroundJobContext, ContextType};
 
 use std::collections::HashMap;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 use std::time::{Duration, Instant};
 
 use crate::panes::PaneId;
@@ -11,6 +15,8 @@ use crate::thread_bus::Bus;
 #[derive(Debug, Clone, Eq, PartialEq, Hash)]
 pub enum BackgroundJob {
     DisplayPaneError(Vec<PaneId>, String),
+    AnimatePluginLoading(u32),       // u32 - plugin_id
+    StopPluginLoadingAnimation(u32), // u32 - plugin_id
     Exit,
 }
 
@@ -18,16 +24,22 @@ impl From<&BackgroundJob> for BackgroundJobContext {
     fn from(background_job: &BackgroundJob) -> Self {
         match *background_job {
             BackgroundJob::DisplayPaneError(..) => BackgroundJobContext::DisplayPaneError,
+            BackgroundJob::AnimatePluginLoading(..) => BackgroundJobContext::AnimatePluginLoading,
+            BackgroundJob::StopPluginLoadingAnimation(..) => {
+                BackgroundJobContext::StopPluginLoadingAnimation
+            },
             BackgroundJob::Exit => BackgroundJobContext::Exit,
         }
     }
 }
 
 static FLASH_DURATION_MS: u64 = 1000;
+static PLUGIN_ANIMATION_OFFSET_DURATION_MD: u64 = 500;
 
 pub(crate) fn background_jobs_main(bus: Bus<BackgroundJob>) -> Result<()> {
     let err_context = || "failed to write to pty".to_string();
     let mut running_jobs: HashMap<BackgroundJob, Instant> = HashMap::new();
+    let mut loading_plugins: HashMap<u32, Arc<AtomicBool>> = HashMap::new(); // u32 - plugin_id
 
     loop {
         let (event, mut err_ctx) = bus.recv().with_context(err_context)?;
@@ -54,7 +66,37 @@ pub(crate) fn background_jobs_main(bus: Bus<BackgroundJob>) -> Result<()> {
                     }
                 });
             },
+            BackgroundJob::AnimatePluginLoading(pid) => {
+                let loading_plugin = Arc::new(AtomicBool::new(true));
+                if job_already_running(job, &mut running_jobs) {
+                    continue;
+                }
+                task::spawn({
+                    let senders = bus.senders.clone();
+                    let loading_plugin = loading_plugin.clone();
+                    async move {
+                        while loading_plugin.load(Ordering::SeqCst) {
+                            let _ = senders.send_to_screen(
+                                ScreenInstruction::ProgressPluginLoadingOffset(pid),
+                            );
+                            task::sleep(std::time::Duration::from_millis(
+                                PLUGIN_ANIMATION_OFFSET_DURATION_MD,
+                            ))
+                            .await;
+                        }
+                    }
+                });
+                loading_plugins.insert(pid, loading_plugin);
+            },
+            BackgroundJob::StopPluginLoadingAnimation(pid) => {
+                if let Some(loading_plugin) = loading_plugins.remove(&pid) {
+                    loading_plugin.store(false, Ordering::SeqCst);
+                }
+            },
             BackgroundJob::Exit => {
+                for loading_plugin in loading_plugins.values() {
+                    loading_plugin.store(false, Ordering::SeqCst);
+                }
                 return Ok(());
             },
         }
diff --git a/zellij-server/src/panes/plugin_pane.rs b/zellij-server/src/panes/plugin_pane.rs
index c6896f90..28828907 100644
--- a/zellij-server/src/panes/plugin_pane.rs
+++ b/zellij-server/src/panes/plugin_pane.rs
@@ -6,7 +6,10 @@ use crate::panes::{grid::Grid, sixel::SixelImageStore, LinkHandler, PaneId};
 use crate::plugins::PluginInstruction;
 use crate::pty::VteBytes;
 use crate::tab::Pane;
-use crate::ui::pane_boundaries_frame::{FrameParams, PaneFrame};
+use crate::ui::{
+    loading_indication::LoadingIndication,
+    pane_boundaries_frame::{FrameParams, PaneFrame},
+};
 use crate::ClientId;
 use std::cell::RefCell;
 use std::rc::Rc;
@@ -67,6 +70,7 @@ pub(crate) struct PluginPane {
     borderless: bool,
     pane_frame_color_override: Option<(PaletteColor, Option<String>)>,
     invoked_with: Option<Run>,
+    loading_indication: LoadingIndication,
 }
 
 impl PluginPane {
@@ -81,10 +85,13 @@ impl PluginPane {
         terminal_emulator_color_codes: Rc<RefCell<HashMap<usize, String>>>,
         link_handler: Rc<RefCell<LinkHandler>>,
         character_cell_size: Rc<RefCell<Option<SizeInPixels>>>,
+        currently_connected_clients: Vec<ClientId>,
         style: Style,
         invoked_with: Option<Run>,
     ) -> Self {
-        Self {
+        let loading_indication = LoadingIndication::new(title.clone()).with_colors(style.colors);
+        let initial_loading_message = loading_indication.to_string();
+        let mut plugin = PluginPane {
             pid,
             should_render: HashMap::new(),
             selectable: true,
@@ -108,7 +115,12 @@ impl PluginPane {
             style,
             pane_frame_color_override: None,
             invoked_with,
+            loading_indication,
+        };
+        for client_id in currently_connected_clients {
+            plugin.handle_plugin_bytes(client_id, initial_loading_message.as_bytes().to_vec());
         }
+        plugin
     }
 }
 
@@ -513,6 +525,24 @@ impl Pane for PluginPane {
     fn set_title(&mut self, title: String) {
         self.pane_title = title;
     }
+    fn update_loading_indication(&mut self, loading_indication: LoadingIndication) {
+        if self.loading_indication.ended {
+            return;
+        }
+        self.loading_indication.merge(loading_indication);
+        self.handle_plugin_bytes_for_all_clients(
+            self.loading_indication.to_string().as_bytes().to_vec(),
+        );
+    }
+    fn progress_animation_offset(&mut self) {
+        if self.loading_indication.ended {
+            return;
+        }
+        self.loading_indication.progress_animation_offset();
+        self.handle_plugin_bytes_for_all_clients(
+            self.loading_indication.to_string().as_bytes().to_vec(),
+        );
+    }
 }
 
 impl PluginPane {
@@ -527,4 +557,10 @@ impl PluginPane {
     fn set_client_should_render(&mut self, client_id: ClientId, should_render: bool) {
         self.should_render.insert(client_id, should_render);
     }
+    fn handle_plugin_bytes_for_all_clients(&mut self, bytes: VteBytes) {
+        let client_ids: Vec<ClientId> = self.grids.keys().copied().collect();
+        for client_id in client_ids {
+            self.handle_plugin_bytes(client_id, bytes.clone());
+        }
+    }
 }
diff --git a/zellij-server/src/plugins/mod.rs b/zellij-server/src/plugins/mod.rs
index 1cd03449..e1884b1e 100644
--- a/zellij-server/src/plugins/mod.rs
+++ b/zellij-server/src/plugins/mod.rs
@@ -1,8 +1,10 @@
+mod start_plugin;
 mod wasm_bridge;
 use log::info;
 use std::{collections::HashMap, fs, path::PathBuf};
 use wasmer::Store;
 
+use crate::screen::ScreenInstruction;
 use crate::{pty::PtyInstruction, thread_bus::Bus, ClientId};
 
 use wasm_bridge::WasmBridge;
@@ -20,7 +22,14 @@ use zellij_utils::{
 
 #[derive(Clone, Debug)]
 pub enum PluginInstruction {
-    Load(RunPlugin, usize, ClientId, Size), // plugin metadata, tab_index, client_ids
+    Load(
+        Option<bool>,   // should float
+        Option<String>, // pane title
+        RunPlugin,
+        usize, // tab index
+        ClientId,
+        Size,
+    ),
     Update(Vec<(Option<u32>, Option<ClientId>, Event)>), // Focused plugin / broadcast, client_id, event data
     Unload(u32),                                         // plugin_id
     Resize(u32, usize, usize),                           // plugin_id, columns, rows
@@ -33,6 +42,7 @@ pub enum PluginInstruction {
         usize, // tab_index
         ClientId,
     ),
+    ApplyCachedEvents(u32), // u32 is the plugin id
     Exit,
 }
 
@@ -47,6 +57,7 @@ impl From<&PluginInstruction> for PluginContext {
             PluginInstruction::AddClient(_) => PluginContext::AddClient,
             PluginInstruction::RemoveClient(_) => PluginContext::RemoveClient,
             PluginInstruction::NewTab(..) => PluginContext::NewTab,
+            PluginInstruction::ApplyCachedEvents(..) => PluginContext::ApplyCachedEvents,
         }
     }
 }
@@ -69,8 +80,21 @@ pub(crate) fn plugin_thread_main(
         let (event, mut err_ctx) = bus.recv().expect("failed to receive event on channel");
         err_ctx.add_call(ContextType::Plugin((&event).into()));
         match event {
-            PluginInstruction::Load(run, tab_index, client_id, size) => {
-                wasm_bridge.load_plugin(&run, tab_index, size, client_id)?;
+            PluginInstruction::Load(should_float, pane_title, run, tab_index, client_id, size) => {
+                match wasm_bridge.load_plugin(&run, tab_index, size, client_id) {
+                    Ok(plugin_id) => {
+                        drop(bus.senders.send_to_screen(ScreenInstruction::AddPlugin(
+                            should_float,
+                            run,
+                            pane_title,
+                            tab_index,
+                            plugin_id,
+                        )));
+                    },
+                    Err(e) => {
+                        log::error!("Failed to load plugin: {e}");
+                    },
+                }
             },
             PluginInstruction::Update(updates) => {
                 wasm_bridge.update_plugins(updates)?;
@@ -126,7 +150,13 @@ pub(crate) fn plugin_thread_main(
                     client_id,
                 )));
             },
-            PluginInstruction::Exit => break,
+            PluginInstruction::ApplyCachedEvents(plugin_id) => {
+                wasm_bridge.apply_cached_events(plugin_id)?;
+            },
+            PluginInstruction::Exit => {
+                wasm_bridge.cleanup();
+                break;
+            },
         }
     }
     info!("wasm main thread exits");
diff --git a/zellij-server/src/plugins/start_plugin.rs b/zellij-server/src/plugins/start_plugin.rs
new file mode 100644
index 00000000..28000bd1
--- /dev/null
+++ b/zellij-server/src/plugins/start_plugin.rs
@@ -0,0 +1,471 @@
+use crate::plugins::wasm_bridge::{wasi_read_string, zellij_exports, PluginEnv, PluginMap};
+use highway::{HighwayHash, PortableHash};
+use log::info;
+use semver::Version;
+use std::{
+    collections::{HashMap, HashSet},
+    fmt, fs,
+    path::PathBuf,
+    sync::{Arc, Mutex},
+    time::Instant,
+};
+use url::Url;
+use wasmer::{ChainableNamedResolver, Instance, Module, Store};
+use wasmer_wasi::{Pipe, WasiState};
+
+use crate::{
+    logging_pipe::LoggingPipe, screen::ScreenInstruction, thread_bus::ThreadSenders,
+    ui::loading_indication::LoadingIndication, ClientId,
+};
+
+use zellij_utils::{
+    consts::{VERSION, ZELLIJ_CACHE_DIR, ZELLIJ_TMP_DIR},
+    errors::prelude::*,
+    input::plugins::PluginConfig,
+    pane_size::Size,
+};
+
+/// Custom error for plugin version mismatch.
+///
+/// This is thrown when, during starting a plugin, it is detected that the plugin version doesn't
+/// match the zellij version. This is treated as a fatal error and leads to instantaneous
+/// termination.
+#[derive(Debug)]
+pub struct VersionMismatchError {
+    zellij_version: String,
+    plugin_version: String,
+    plugin_path: PathBuf,
+    // true for builtin plugins
+    builtin: bool,
+}
+
+impl std::error::Error for VersionMismatchError {}
+
+impl VersionMismatchError {
+    pub fn new(
+        zellij_version: &str,
+        plugin_version: &str,
+        plugin_path: &PathBuf,
+        builtin: bool,
+    ) -> Self {
+        VersionMismatchError {
+            zellij_version: zellij_version.to_owned(),
+            plugin_version: plugin_version.to_owned(),
+            plugin_path: plugin_path.to_owned(),
+            builtin,
+        }
+    }
+}
+
+impl fmt::Display for VersionMismatchError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let first_line = if self.builtin {
+            "It seems your version of zellij was built with outdated core plugins."
+        } else {
+            "If you're seeing this error a plugin version doesn't match the current
+zellij version."
+        };
+
+        write!(
+            f,
+            "{}
+Detected versions:
+
+- Plugin version: {}
+- Zellij version: {}
+- Offending plugin: {}
+
+If you're a user:
+    Please contact the distributor of your zellij version and report this error
+    to them.
+
+If you're a developer:
+    Please run zellij with updated plugins. The easiest way to achieve this
+    is to build zellij with `cargo xtask install`. Also refer to the docs:
+    https://github.com/zellij-org/zellij/blob/main/CONTRIBUTING.md#building
+",
+            first_line,
+            self.plugin_version.trim_end(),
+            self.zellij_version.trim_end(),
+            self.plugin_path.display()
+        )
+    }
+}
+
+// Returns `Ok` if the plugin version matches the zellij version.
+// Returns an `Err` otherwise.
+fn assert_plugin_version(instance: &Instance, plugin_env: &PluginEnv) -> Result<()> {
+    let err_context = || {
+        format!(
+            "failed to determine plugin version for plugin {}",
+            plugin_env.plugin.path.display()
+        )
+    };
+
+    let plugin_version_func = match instance.exports.get_function("plugin_version") {
+        Ok(val) => val,
+        Err(_) => {
+            return Err(anyError::new(VersionMismatchError::new(
+                VERSION,
+                "Unavailable",
+                &plugin_env.plugin.path,
+                plugin_env.plugin.is_builtin(),
+            )))
+        },
+    };
+
+    let plugin_version = plugin_version_func
+        .call(&[])
+        .map_err(anyError::new)
+        .and_then(|_| wasi_read_string(&plugin_env.wasi_env))
+        .and_then(|string| Version::parse(&string).context("failed to parse plugin version"))
+        .with_context(err_context)?;
+    let zellij_version = Version::parse(VERSION)
+        .context("failed to parse zellij version")
+        .with_context(err_context)?;
+    if plugin_version != zellij_version {
+        return Err(anyError::new(VersionMismatchError::new(
+            VERSION,
+            &plugin_version.to_string(),
+            &plugin_env.plugin.path,
+            plugin_env.plugin.is_builtin(),
+        )));
+    }
+
+    Ok(())
+}
+
+fn load_plugin_instance(instance: &mut Instance) -> Result<()> {
+    let err_context = || format!("failed to load plugin from instance {instance:#?}");
+
+    let load_function = instance
+        .exports
+        .get_function("_start")
+        .with_context(err_context)?;
+    // This eventually calls the `.load()` method
+    load_function.call(&[]).with_context(err_context)?;
+    Ok(())
+}
+
+pub fn start_plugin(
+    plugin_id: u32,
+    client_id: ClientId,
+    plugin: &PluginConfig,
+    tab_index: usize,
+    plugin_dir: PathBuf,
+    plugin_cache: Arc<Mutex<HashMap<PathBuf, Module>>>,
+    senders: ThreadSenders,
+    mut store: Store,
+    plugin_map: Arc<Mutex<PluginMap>>,
+    size: Size,
+    connected_clients: Arc<Mutex<Vec<ClientId>>>,
+    loading_indication: &mut LoadingIndication,
+) -> Result<()> {
+    let err_context = || format!("failed to start plugin {plugin:#?} for client {client_id}");
+    let plugin_own_data_dir = ZELLIJ_CACHE_DIR.join(Url::from(&plugin.location).to_string());
+    create_plugin_fs_entries(&plugin_own_data_dir)?;
+
+    loading_indication.indicate_loading_plugin_from_memory();
+    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+        plugin_id,
+        loading_indication.clone(),
+    ));
+    let (module, cache_hit) = {
+        let mut plugin_cache = plugin_cache.lock().unwrap();
+        let (module, cache_hit) = load_module_from_memory(&mut *plugin_cache, &plugin.path);
+        (module, cache_hit)
+    };
+
+    let module = match module {
+        Some(module) => {
+            loading_indication.indicate_loading_plugin_from_memory_success();
+            let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                plugin_id,
+                loading_indication.clone(),
+            ));
+            module
+        },
+        None => {
+            loading_indication.indicate_loading_plugin_from_memory_notfound();
+            loading_indication.indicate_loading_plugin_from_hd_cache();
+            let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                plugin_id,
+                loading_indication.clone(),
+            ));
+
+            let (wasm_bytes, cached_path) = plugin_bytes_and_cache_path(&plugin, &plugin_dir);
+            let timer = std::time::Instant::now();
+            match load_module_from_hd_cache(&mut store, &plugin.path, &timer, &cached_path) {
+                Ok(module) => {
+                    loading_indication.indicate_loading_plugin_from_hd_cache_success();
+                    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                        plugin_id,
+                        loading_indication.clone(),
+                    ));
+                    module
+                },
+                Err(_e) => {
+                    loading_indication.indicate_loading_plugin_from_hd_cache_notfound();
+                    loading_indication.indicate_compiling_plugin();
+                    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                        plugin_id,
+                        loading_indication.clone(),
+                    ));
+                    let module =
+                        compile_module(&mut store, &plugin.path, &timer, &cached_path, wasm_bytes)?;
+                    loading_indication.indicate_compiling_plugin_success();
+                    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                        plugin_id,
+                        loading_indication.clone(),
+                    ));
+                    module
+                },
+            }
+        },
+    };
+
+    let (instance, plugin_env) = create_plugin_instance_and_environment(
+        plugin_id,
+        client_id,
+        plugin,
+        &module,
+        tab_index,
+        plugin_own_data_dir,
+        senders.clone(),
+        &mut store,
+    )?;
+
+    if !cache_hit {
+        // Check plugin version
+        assert_plugin_version(&instance, &plugin_env).with_context(err_context)?;
+    }
+
+    // Only do an insert when everything went well!
+    let cloned_plugin = plugin.clone();
+    {
+        let mut plugin_cache = plugin_cache.lock().unwrap();
+        plugin_cache.insert(cloned_plugin.path, module);
+    }
+
+    let mut main_user_instance = instance.clone();
+    let main_user_env = plugin_env.clone();
+    loading_indication.indicate_starting_plugin();
+    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+        plugin_id,
+        loading_indication.clone(),
+    ));
+    load_plugin_instance(&mut main_user_instance).with_context(err_context)?;
+    loading_indication.indicate_starting_plugin_success();
+    loading_indication.indicate_writing_plugin_to_cache();
+    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+        plugin_id,
+        loading_indication.clone(),
+    ));
+
+    {
+        let mut plugin_map = plugin_map.lock().unwrap();
+        plugin_map.insert(
+            (plugin_id, client_id),
+            (main_user_instance, main_user_env, (size.rows, size.cols)),
+        );
+    }
+
+    loading_indication.indicate_writing_plugin_to_cache_success();
+    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+        plugin_id,
+        loading_indication.clone(),
+    ));
+
+    let connected_clients: Vec<ClientId> =
+        connected_clients.lock().unwrap().iter().copied().collect();
+    if !connected_clients.is_empty() {
+        loading_indication.indicate_cloning_plugin_for_other_clients();
+        let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+            plugin_id,
+            loading_indication.clone(),
+        ));
+        let mut plugin_map = plugin_map.lock().unwrap();
+        for client_id in connected_clients {
+            let (instance, new_plugin_env) =
+                clone_plugin_for_client(&plugin_env, client_id, &instance, &mut store)?;
+            plugin_map.insert(
+                (plugin_id, client_id),
+                (instance, new_plugin_env, (size.rows, size.cols)),
+            );
+        }
+        loading_indication.indicate_cloning_plugin_for_other_clients_success();
+        let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+            plugin_id,
+            loading_indication.clone(),
+        ));
+    }
+    loading_indication.end();
+    let _ = senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+        plugin_id,
+        loading_indication.clone(),
+    ));
+    Ok(())
+}
+
+fn create_plugin_fs_entries(plugin_own_data_dir: &PathBuf) -> Result<()> {
+    let err_context = || "failed to create plugin fs entries";
+    // Create filesystem entries mounted into WASM.
+    // We create them here to get expressive error messages in case they fail.
+    fs::create_dir_all(&plugin_own_data_dir)
+        .with_context(|| format!("failed to create datadir in {plugin_own_data_dir:?}"))
+        .with_context(err_context)?;
+    fs::create_dir_all(ZELLIJ_TMP_DIR.as_path())
+        .with_context(|| format!("failed to create tmpdir at {:?}", &ZELLIJ_TMP_DIR.as_path()))
+        .with_context(err_context)?;
+    Ok(())
+}
+
+fn compile_module(
+    store: &mut Store,
+    plugin_path: &PathBuf,
+    timer: &Instant,
+    cached_path: &PathBuf,
+    wasm_bytes: Vec<u8>,
+) -> Result<Module> {
+    let err_context = || "failed to recover cache dir";
+    fs::create_dir_all(ZELLIJ_CACHE_DIR.to_owned())
+        .map_err(anyError::new)
+        .and_then(|_| {
+            // compile module
+            Module::new(&*store, &wasm_bytes).map_err(anyError::new)
+        })
+        .map(|m| {
+            // serialize module to HD cache for faster loading in the future
+            m.serialize_to_file(&cached_path).map_err(anyError::new)?;
+            log::info!(
+                "Compiled plugin '{}' in {:?}",
+                plugin_path.display(),
+                timer.elapsed()
+            );
+            Ok(m)
+        })
+        .with_context(err_context)?
+}
+
+fn load_module_from_hd_cache(
+    store: &mut Store,
+    plugin_path: &PathBuf,
+    timer: &Instant,
+    cached_path: &PathBuf,
+) -> Result<Module> {
+    let module = unsafe { Module::deserialize_from_file(&*store, &cached_path)? };
+    log::info!(
+        "Loaded plugin '{}' from cache folder at '{}' in {:?}",
+        plugin_path.display(),
+        ZELLIJ_CACHE_DIR.display(),
+        timer.elapsed(),
+    );
+    Ok(module)
+}
+
+fn plugin_bytes_and_cache_path(plugin: &PluginConfig, plugin_dir: &PathBuf) -> (Vec<u8>, PathBuf) {
+    let err_context = || "failed to get plugin bytes and cached path";
+    // Populate plugin module cache for this plugin!
+    // Is it in the cache folder already?
+    if plugin._allow_exec_host_cmd {
+        info!(
+            "Plugin({:?}) is able to run any host command, this may lead to some security issues!",
+            plugin.path
+        );
+    }
+    // The plugins blob as stored on the filesystem
+    let wasm_bytes = plugin
+        .resolve_wasm_bytes(&plugin_dir)
+        .with_context(err_context)
+        .fatal();
+    let hash: String = PortableHash::default()
+        .hash256(&wasm_bytes)
+        .iter()
+        .map(ToString::to_string)
+        .collect();
+    let cached_path = ZELLIJ_CACHE_DIR.join(&hash);
+    (wasm_bytes, cached_path)
+}
+
+fn load_module_from_memory(
+    plugin_cache: &mut HashMap<PathBuf, Module>,
+    plugin_path: &PathBuf,
+) -> (Option<Module>, bool) {
+    let module = plugin_cache.remove(plugin_path);
+    let mut cache_hit = false;
+    if module.is_some() {
+        cache_hit = true;
+        log::debug!(
+            "Loaded plugin '{}' from plugin cache",
+            plugin_path.display()
+        );
+    }
+    (module, cache_hit)
+}
+
+fn create_plugin_instance_and_environment(
+    plugin_id: u32,
+    client_id: ClientId,
+    plugin: &PluginConfig,
+    module: &Module,
+    tab_index: usize,
+    plugin_own_data_dir: PathBuf,
+    senders: ThreadSenders,
+    store: &mut Store,
+) -> Result<(Instance, PluginEnv)> {
+    let err_context = || format!("Failed to create instance and plugin env for plugin {plugin_id}");
+    let mut wasi_env = WasiState::new("Zellij")
+        .env("CLICOLOR_FORCE", "1")
+        .map_dir("/host", ".")
+        .and_then(|wasi| wasi.map_dir("/data", &plugin_own_data_dir))
+        .and_then(|wasi| wasi.map_dir("/tmp", ZELLIJ_TMP_DIR.as_path()))
+        .and_then(|wasi| {
+            wasi.stdin(Box::new(Pipe::new()))
+                .stdout(Box::new(Pipe::new()))
+                .stderr(Box::new(LoggingPipe::new(
+                    &plugin.location.to_string(),
+                    plugin_id,
+                )))
+                .finalize()
+        })
+        .with_context(err_context)?;
+    let wasi = wasi_env.import_object(&module).with_context(err_context)?;
+
+    let mut mut_plugin = plugin.clone();
+    mut_plugin.set_tab_index(tab_index);
+    let plugin_env = PluginEnv {
+        plugin_id,
+        client_id,
+        plugin: mut_plugin,
+        senders: senders.clone(),
+        wasi_env,
+        subscriptions: Arc::new(Mutex::new(HashSet::new())),
+        plugin_own_data_dir,
+        tab_index,
+    };
+
+    let zellij = zellij_exports(&store, &plugin_env);
+    let instance = Instance::new(&module, &zellij.chain_back(wasi)).with_context(err_context)?;
+    Ok((instance, plugin_env))
+}
+
+fn clone_plugin_for_client(
+    plugin_env: &PluginEnv,
+    client_id: ClientId,
+    instance: &Instance,
+    store: &Store,
+) -> Result<(Instance, PluginEnv)> {
+    let err_context = || format!("Failed to clone plugin for client {client_id}");
+    let mut new_plugin_env = plugin_env.clone();
+    new_plugin_env.client_id = client_id;
+    let module = instance.module().clone();
+    let wasi = new_plugin_env
+        .wasi_env
+        .import_object(&module)
+        .with_context(err_context)?;
+    let zellij = zellij_exports(store, &new_plugin_env);
+    let mut instance =
+        Instance::new(&module, &zellij.chain_back(wasi)).with_context(err_context)?;
+    load_plugin_instance(&mut instance).with_context(err_context)?;
+    Ok((instance, new_plugin_env))
+}
diff --git a/zellij-server/src/plugins/wasm_bridge.rs b/zellij-server/src/plugins/wasm_bridge.rs
index 62e8cedb..68f8bc6d 100644
--- a/zellij-server/src/plugins/wasm_bridge.rs
+++ b/zellij-server/src/plugins/wasm_bridge.rs
@@ -1,11 +1,10 @@
 use super::PluginInstruction;
-use highway::{HighwayHash, PortableHash};
+use crate::plugins::start_plugin::start_plugin;
 use log::{debug, info, warn};
-use semver::Version;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{
     collections::{HashMap, HashSet},
-    fmt, fs,
+    fmt,
     path::PathBuf,
     process,
     str::FromStr,
@@ -13,24 +12,25 @@ use std::{
     thread,
     time::{Duration, Instant},
 };
-use url::Url;
 use wasmer::{
     imports, ChainableNamedResolver, Function, ImportObject, Instance, Module, Store, Value,
     WasmerEnv,
 };
-use wasmer_wasi::{Pipe, WasiEnv, WasiState};
+use wasmer_wasi::WasiEnv;
+use zellij_utils::async_std::task::{self, JoinHandle};
 
 use crate::{
-    logging_pipe::LoggingPipe,
+    background_jobs::BackgroundJob,
     panes::PaneId,
     pty::{ClientOrTabIndex, PtyInstruction},
     screen::ScreenInstruction,
     thread_bus::ThreadSenders,
+    ui::loading_indication::LoadingIndication,
     ClientId,
 };
 
 use zellij_utils::{
-    consts::{VERSION, ZELLIJ_CACHE_DIR, ZELLIJ_TMP_DIR},
+    consts::VERSION,
     data::{Event, EventType, PluginIds},
     errors::prelude::*,
     input::{
@@ -119,7 +119,7 @@ pub struct PluginEnv {
     pub tab_index: usize,
     pub client_id: ClientId,
     #[allow(dead_code)]
-    plugin_own_data_dir: PathBuf,
+    pub plugin_own_data_dir: PathBuf,
 }
 
 impl PluginEnv {
@@ -133,21 +133,24 @@ impl PluginEnv {
     }
 }
 
-type PluginMap = HashMap<(u32, ClientId), (Instance, PluginEnv, (usize, usize))>; // u32 =>
-                                                                                  // plugin_id,
-                                                                                  // (usize, usize)
-                                                                                  // => (rows,
-                                                                                  // columns)
+pub type PluginMap = HashMap<(u32, ClientId), (Instance, PluginEnv, (usize, usize))>; // u32 =>
+                                                                                      // plugin_id,
+                                                                                      // (usize, usize)
+                                                                                      // => (rows,
+                                                                                      // columns)
 
 pub struct WasmBridge {
-    connected_clients: Vec<ClientId>,
+    connected_clients: Arc<Mutex<Vec<ClientId>>>,
     plugins: PluginsConfig,
     senders: ThreadSenders,
     store: Store,
     plugin_dir: PathBuf,
-    plugin_cache: HashMap<PathBuf, Module>,
-    plugin_map: PluginMap,
+    plugin_cache: Arc<Mutex<HashMap<PathBuf, Module>>>,
+    plugin_map: Arc<Mutex<PluginMap>>,
     next_plugin_id: u32,
+    cached_events_for_pending_plugins: HashMap<u32, Vec<Event>>, // u32 is the plugin id
+    cached_resizes_for_pending_plugins: HashMap<u32, (usize, usize)>, // (rows, columns)
+    loading_plugins: HashMap<u32, JoinHandle<()>>,               // plugin_id to join-handle
 }
 
 impl WasmBridge {
@@ -157,9 +160,10 @@ impl WasmBridge {
         store: Store,
         plugin_dir: PathBuf,
     ) -> Self {
-        let plugin_map = HashMap::new();
-        let connected_clients: Vec<ClientId> = vec![];
-        let plugin_cache: HashMap<PathBuf, Module> = HashMap::new();
+        let plugin_map = Arc::new(Mutex::new(HashMap::new()));
+        let connected_clients: Arc<Mutex<Vec<ClientId>>> = Arc::new(Mutex::new(vec![]));
+        let plugin_cache: Arc<Mutex<HashMap<PathBuf, Module>>> =
+            Arc::new(Mutex::new(HashMap::new()));
         WasmBridge {
             connected_clients,
             plugins,
@@ -169,6 +173,9 @@ impl WasmBridge {
             plugin_cache,
             plugin_map,
             next_plugin_id: 0,
+            cached_events_for_pending_plugins: HashMap::new(),
+            cached_resizes_for_pending_plugins: HashMap::new(),
+            loading_plugins: HashMap::new(),
         }
     }
     pub fn load_plugin(
@@ -179,210 +186,96 @@ impl WasmBridge {
         client_id: ClientId,
     ) -> Result<u32> {
         // returns the plugin id
-        let err_context = || format!("failed to load plugin for client {client_id}");
+        let err_context = move || format!("failed to load plugin for client {client_id}");
         let plugin_id = self.next_plugin_id;
 
         let plugin = self
             .plugins
             .get(run)
             .with_context(|| format!("failed to resolve plugin {run:?}"))
-            .with_context(err_context)
-            .fatal();
-
-        let (instance, plugin_env) = self
-            .start_plugin(plugin_id, client_id, &plugin, tab_index)
             .with_context(err_context)?;
+        let plugin_name = run.location.to_string();
 
-        let mut main_user_instance = instance.clone();
-        let main_user_env = plugin_env.clone();
-        load_plugin_instance(&mut main_user_instance).with_context(err_context)?;
+        self.next_plugin_id += 1;
 
-        self.plugin_map.insert(
-            (plugin_id, client_id),
-            (main_user_instance, main_user_env, (size.rows, size.cols)),
-        );
+        self.cached_events_for_pending_plugins
+            .insert(plugin_id, vec![]);
+        self.cached_resizes_for_pending_plugins
+            .insert(plugin_id, (0, 0));
 
-        // clone plugins for the rest of the client ids if they exist
-        for client_id in self.connected_clients.iter() {
-            let mut new_plugin_env = plugin_env.clone();
-            new_plugin_env.client_id = *client_id;
-            let module = instance.module().clone();
-            let wasi = new_plugin_env
-                .wasi_env
-                .import_object(&module)
-                .with_context(err_context)?;
-            let zellij = zellij_exports(&self.store, &new_plugin_env);
-            let mut instance =
-                Instance::new(&module, &zellij.chain_back(wasi)).with_context(err_context)?;
-            load_plugin_instance(&mut instance).with_context(err_context)?;
-            self.plugin_map.insert(
-                (plugin_id, *client_id),
-                (instance, new_plugin_env, (size.rows, size.cols)),
-            );
-        }
+        let load_plugin_task = task::spawn({
+            let plugin_dir = self.plugin_dir.clone();
+            let plugin_cache = self.plugin_cache.clone();
+            let senders = self.senders.clone();
+            let store = self.store.clone();
+            let plugin_map = self.plugin_map.clone();
+            let connected_clients = self.connected_clients.clone();
+            async move {
+                let _ =
+                    senders.send_to_background_jobs(BackgroundJob::AnimatePluginLoading(plugin_id));
+                let mut loading_indication = LoadingIndication::new(plugin_name.clone());
+                match start_plugin(
+                    plugin_id,
+                    client_id,
+                    &plugin,
+                    tab_index,
+                    plugin_dir,
+                    plugin_cache,
+                    senders.clone(),
+                    store,
+                    plugin_map,
+                    size,
+                    connected_clients.clone(),
+                    &mut loading_indication,
+                ) {
+                    Ok(_) => {
+                        let _ = senders.send_to_background_jobs(
+                            BackgroundJob::StopPluginLoadingAnimation(plugin_id),
+                        );
+                        let _ =
+                            senders.send_to_plugin(PluginInstruction::ApplyCachedEvents(plugin_id));
+                    },
+                    Err(e) => {
+                        let _ = senders.send_to_background_jobs(
+                            BackgroundJob::StopPluginLoadingAnimation(plugin_id),
+                        );
+                        let _ =
+                            senders.send_to_plugin(PluginInstruction::ApplyCachedEvents(plugin_id));
+                        loading_indication.indicate_loading_error(e.to_string());
+                        let _ =
+                            senders.send_to_screen(ScreenInstruction::UpdatePluginLoadingStage(
+                                plugin_id,
+                                loading_indication.clone(),
+                            ));
+                    },
+                }
+            }
+        });
+        self.loading_plugins.insert(plugin_id, load_plugin_task);
         self.next_plugin_id += 1;
         Ok(plugin_id)
     }
     pub fn unload_plugin(&mut self, pid: u32) -> Result<()> {
         info!("Bye from plugin {}", &pid);
         // TODO: remove plugin's own data directory
-        let ids_in_plugin_map: Vec<(u32, ClientId)> = self.plugin_map.keys().copied().collect();
+        let mut plugin_map = self.plugin_map.lock().unwrap();
+        let ids_in_plugin_map: Vec<(u32, ClientId)> = plugin_map.keys().copied().collect();
         for (plugin_id, client_id) in ids_in_plugin_map {
             if pid == plugin_id {
-                drop(self.plugin_map.remove(&(plugin_id, client_id)));
+                drop(plugin_map.remove(&(plugin_id, client_id)));
             }
         }
         Ok(())
     }
-    #[allow(clippy::too_many_arguments)]
-    pub fn start_plugin(
-        &mut self,
-        plugin_id: u32,
-        client_id: ClientId,
-        plugin: &PluginConfig,
-        tab_index: usize,
-    ) -> Result<(Instance, PluginEnv)> {
-        let err_context = || format!("failed to start plugin {plugin:#?} for client {client_id}");
-
-        let plugin_own_data_dir = ZELLIJ_CACHE_DIR.join(Url::from(&plugin.location).to_string());
-        let cache_hit = self.plugin_cache.contains_key(&plugin.path);
-
-        // Create filesystem entries mounted into WASM.
-        // We create them here to get expressive error messages in case they fail.
-        fs::create_dir_all(&plugin_own_data_dir)
-            .with_context(|| format!("failed to create datadir in {plugin_own_data_dir:?}"))
-            .with_context(err_context)?;
-        fs::create_dir_all(ZELLIJ_TMP_DIR.as_path())
-            .with_context(|| format!("failed to create tmpdir at {:?}", &ZELLIJ_TMP_DIR.as_path()))
-            .with_context(err_context)?;
-
-        // We remove the entry here and repopulate it at the very bottom, if everything went well.
-        // We must do that because a `get` will only give us a borrow of the Module. This suffices for
-        // the purpose of setting everything up, but we cannot return a &Module from the "None" match
-        // arm, because we create the Module from scratch there. Any reference passed outside would
-        // outlive the Module we create there. Hence, we remove the plugin here and reinsert it
-        // below...
-        let module = match self.plugin_cache.remove(&plugin.path) {
-            Some(module) => {
-                log::debug!(
-                    "Loaded plugin '{}' from plugin cache",
-                    plugin.path.display()
-                );
-                module
-            },
-            None => {
-                // Populate plugin module cache for this plugin!
-                // Is it in the cache folder already?
-                if plugin._allow_exec_host_cmd {
-                    info!(
-                        "Plugin({:?}) is able to run any host command, this may lead to some security issues!",
-                        plugin.path
-                    );
-                }
-
-                // The plugins blob as stored on the filesystem
-                let wasm_bytes = plugin
-                    .resolve_wasm_bytes(&self.plugin_dir)
-                    .with_context(err_context)
-                    .fatal();
-
-                let hash: String = PortableHash::default()
-                    .hash256(&wasm_bytes)
-                    .iter()
-                    .map(ToString::to_string)
-                    .collect();
-                let cached_path = ZELLIJ_CACHE_DIR.join(&hash);
-
-                let timer = std::time::Instant::now();
-                unsafe {
-                    match Module::deserialize_from_file(&self.store, &cached_path) {
-                        Ok(m) => {
-                            log::info!(
-                                "Loaded plugin '{}' from cache folder at '{}' in {:?}",
-                                plugin.path.display(),
-                                ZELLIJ_CACHE_DIR.display(),
-                                timer.elapsed(),
-                            );
-                            m
-                        },
-                        Err(e) => {
-                            let inner_context = || format!("failed to recover from {e:?}");
-
-                            fs::create_dir_all(ZELLIJ_CACHE_DIR.to_owned())
-                                .map_err(anyError::new)
-                                .and_then(|_| {
-                                    Module::new(&self.store, &wasm_bytes).map_err(anyError::new)
-                                })
-                                .and_then(|m| {
-                                    m.serialize_to_file(&cached_path).map_err(anyError::new)?;
-                                    log::info!(
-                                        "Compiled plugin '{}' in {:?}",
-                                        plugin.path.display(),
-                                        timer.elapsed()
-                                    );
-                                    Ok(m)
-                                })
-                                .with_context(inner_context)
-                                .with_context(err_context)?
-                        },
-                    }
-                }
-            },
-        };
-
-        let mut wasi_env = WasiState::new("Zellij")
-            .env("CLICOLOR_FORCE", "1")
-            .map_dir("/host", ".")
-            .and_then(|wasi| wasi.map_dir("/data", &plugin_own_data_dir))
-            .and_then(|wasi| wasi.map_dir("/tmp", ZELLIJ_TMP_DIR.as_path()))
-            .and_then(|wasi| {
-                wasi.stdin(Box::new(Pipe::new()))
-                    .stdout(Box::new(Pipe::new()))
-                    .stderr(Box::new(LoggingPipe::new(
-                        &plugin.location.to_string(),
-                        plugin_id,
-                    )))
-                    .finalize()
-            })
-            .with_context(err_context)?;
-        let wasi = wasi_env.import_object(&module).with_context(err_context)?;
-
-        let mut mut_plugin = plugin.clone();
-        mut_plugin.set_tab_index(tab_index);
-        let plugin_env = PluginEnv {
-            plugin_id,
-            client_id,
-            plugin: mut_plugin,
-            senders: self.senders.clone(),
-            wasi_env,
-            subscriptions: Arc::new(Mutex::new(HashSet::new())),
-            plugin_own_data_dir,
-            tab_index,
-        };
-
-        let zellij = zellij_exports(&self.store, &plugin_env);
-        let instance =
-            Instance::new(&module, &zellij.chain_back(wasi)).with_context(err_context)?;
-
-        if !cache_hit {
-            // Check plugin version
-            assert_plugin_version(&instance, &plugin_env).with_context(err_context)?;
-        }
-
-        // Only do an insert when everything went well!
-        let cloned_plugin = plugin.clone();
-        self.plugin_cache.insert(cloned_plugin.path, module);
-
-        Ok((instance, plugin_env))
-    }
     pub fn add_client(&mut self, client_id: ClientId) -> Result<()> {
         let err_context = || format!("failed to add plugins for client {client_id}");
 
-        self.connected_clients.push(client_id);
+        self.connected_clients.lock().unwrap().push(client_id);
 
         let mut seen = HashSet::new();
         let mut new_plugins = HashMap::new();
-        for (&(plugin_id, _), (instance, plugin_env, (rows, columns))) in &self.plugin_map {
+        let mut plugin_map = self.plugin_map.lock().unwrap();
+        for (&(plugin_id, _), (instance, plugin_env, (rows, columns))) in &*plugin_map {
             if seen.contains(&plugin_id) {
                 continue;
             }
@@ -404,7 +297,7 @@ impl WasmBridge {
             let mut instance =
                 Instance::new(&module, &zellij.chain_back(wasi)).with_context(err_context)?;
             load_plugin_instance(&mut instance).with_context(err_context)?;
-            self.plugin_map.insert(
+            plugin_map.insert(
                 (plugin_id, client_id),
                 (instance, new_plugin_env, (rows, columns)),
             );
@@ -414,8 +307,9 @@ impl WasmBridge {
     pub fn resize_plugin(&mut self, pid: u32, new_columns: usize, new_rows: usize) -> Result<()> {
         let err_context = || format!("failed to resize plugin {pid}");
         let mut plugin_bytes = vec![];
+        let mut plugin_map = self.plugin_map.lock().unwrap();
         for ((plugin_id, client_id), (instance, plugin_env, (current_rows, current_columns))) in
-            self.plugin_map.iter_mut()
+            plugin_map.iter_mut()
         {
             if *plugin_id == pid {
                 *current_rows = new_rows;
@@ -440,6 +334,14 @@ impl WasmBridge {
                 plugin_bytes.push((*plugin_id, *client_id, rendered_bytes.as_bytes().to_vec()));
             }
         }
+        for (plugin_id, (current_rows, current_columns)) in
+            self.cached_resizes_for_pending_plugins.iter_mut()
+        {
+            if *plugin_id == pid {
+                *current_rows = new_rows;
+                *current_columns = new_columns;
+            }
+        }
         let _ = self
             .senders
             .send_to_screen(ScreenInstruction::PluginBytes(plugin_bytes));
@@ -451,11 +353,10 @@ impl WasmBridge {
     ) -> Result<()> {
         let err_context = || "failed to update plugin state".to_string();
 
+        let plugin_map = self.plugin_map.lock().unwrap();
         let mut plugin_bytes = vec![];
         for (pid, cid, event) in updates.drain(..) {
-            for (&(plugin_id, client_id), (instance, plugin_env, (rows, columns))) in
-                &self.plugin_map
-            {
+            for (&(plugin_id, client_id), (instance, plugin_env, (rows, columns))) in &*plugin_map {
                 let subs = plugin_env
                     .subscriptions
                     .lock()
@@ -470,102 +371,90 @@ impl WasmBridge {
                         || (cid.is_none() && pid == Some(plugin_id))
                         || (cid == Some(client_id) && pid == Some(plugin_id)))
                 {
-                    let update = instance
-                        .exports
-                        .get_function("update")
+                    apply_event_to_plugin(
+                        plugin_id,
+                        client_id,
+                        &instance,
+                        &plugin_env,
+                        &event,
+                        *rows,
+                        *columns,
+                        &mut plugin_bytes,
+                    )?;
+                }
+            }
+            for (plugin_id, cached_events) in self.cached_events_for_pending_plugins.iter_mut() {
+                if pid.is_none() || pid.as_ref() == Some(plugin_id) {
+                    cached_events.push(event.clone());
+                }
+            }
+        }
+        let _ = self
+            .senders
+            .send_to_screen(ScreenInstruction::PluginBytes(plugin_bytes));
+        Ok(())
+    }
+    pub fn apply_cached_events(&mut self, plugin_id: u32) -> Result<()> {
+        let err_context = || format!("Failed to apply cached events to plugin {plugin_id}");
+        if let Some(events) = self.cached_events_for_pending_plugins.remove(&plugin_id) {
+            let mut plugin_map = self.plugin_map.lock().unwrap();
+            let all_connected_clients: Vec<ClientId> = self
+                .connected_clients
+                .lock()
+                .unwrap()
+                .iter()
+                .copied()
+                .collect();
+            for client_id in all_connected_clients {
+                let mut plugin_bytes = vec![];
+                if let Some((instance, plugin_env, (rows, columns))) =
+                    plugin_map.get_mut(&(plugin_id, client_id))
+                {
+                    let subs = plugin_env
+                        .subscriptions
+                        .lock()
+                        .to_anyhow()
                         .with_context(err_context)?;
-                    wasi_write_object(&plugin_env.wasi_env, &event).with_context(err_context)?;
-                    let update_return = update.call(&[]).or_else::<anyError, _>(|e| {
-                        match e.downcast::<serde_json::Error>() {
-                            Ok(_) => panic!(
-                                "{}",
-                                anyError::new(VersionMismatchError::new(
-                                    VERSION,
-                                    "Unavailable",
-                                    &plugin_env.plugin.path,
-                                    plugin_env.plugin.is_builtin(),
-                                ))
-                            ),
-                            Err(e) => Err(e).with_context(err_context),
+                    for event in events.clone() {
+                        let event_type =
+                            EventType::from_str(&event.to_string()).with_context(err_context)?;
+                        if !subs.contains(&event_type) {
+                            continue;
                         }
-                    })?;
-                    let should_render = match update_return.get(0) {
-                        Some(Value::I32(n)) => *n == 1,
-                        _ => false,
-                    };
-
-                    if *rows > 0 && *columns > 0 && should_render {
-                        let rendered_bytes = instance
-                            .exports
-                            .get_function("render")
-                            .map_err(anyError::new)
-                            .and_then(|render| {
-                                render
-                                    .call(&[Value::I32(*rows as i32), Value::I32(*columns as i32)])
-                                    .map_err(anyError::new)
-                            })
-                            .and_then(|_| wasi_read_string(&plugin_env.wasi_env))
-                            .with_context(err_context)?;
-                        plugin_bytes.push((
+                        apply_event_to_plugin(
                             plugin_id,
                             client_id,
-                            rendered_bytes.as_bytes().to_vec(),
-                        ));
+                            &instance,
+                            &plugin_env,
+                            &event,
+                            *rows,
+                            *columns,
+                            &mut plugin_bytes,
+                        )?;
                     }
+                    let _ = self
+                        .senders
+                        .send_to_screen(ScreenInstruction::PluginBytes(plugin_bytes));
                 }
             }
         }
-        let _ = self
-            .senders
-            .send_to_screen(ScreenInstruction::PluginBytes(plugin_bytes));
+        if let Some((rows, columns)) = self.cached_resizes_for_pending_plugins.remove(&plugin_id) {
+            self.resize_plugin(plugin_id, columns, rows)?;
+        }
+        self.loading_plugins.remove(&plugin_id);
         Ok(())
     }
     pub fn remove_client(&mut self, client_id: ClientId) {
-        self.connected_clients.retain(|c| c != &client_id);
+        self.connected_clients
+            .lock()
+            .unwrap()
+            .retain(|c| c != &client_id);
     }
-}
-
-// Returns `Ok` if the plugin version matches the zellij version.
-// Returns an `Err` otherwise.
-fn assert_plugin_version(instance: &Instance, plugin_env: &PluginEnv) -> Result<()> {
-    let err_context = || {
-        format!(
-            "failed to determine plugin version for plugin {}",
-            plugin_env.plugin.path.display()
-        )
-    };
-
-    let plugin_version_func = match instance.exports.get_function("plugin_version") {
-        Ok(val) => val,
-        Err(_) => {
-            return Err(anyError::new(VersionMismatchError::new(
-                VERSION,
-                "Unavailable",
-                &plugin_env.plugin.path,
-                plugin_env.plugin.is_builtin(),
-            )))
-        },
-    };
-
-    let plugin_version = plugin_version_func
-        .call(&[])
-        .map_err(anyError::new)
-        .and_then(|_| wasi_read_string(&plugin_env.wasi_env))
-        .and_then(|string| Version::parse(&string).context("failed to parse plugin version"))
-        .with_context(err_context)?;
-    let zellij_version = Version::parse(VERSION)
-        .context("failed to parse zellij version")
-        .with_context(err_context)?;
-    if plugin_version != zellij_version {
-        return Err(anyError::new(VersionMismatchError::new(
-            VERSION,
-            &plugin_version.to_string(),
-            &plugin_env.plugin.path,
-            plugin_env.plugin.is_builtin(),
-        )));
+    pub fn cleanup(&mut self) {
+        for (_plugin_id, loading_plugin_task) in self.loading_plugins.drain() {
+            drop(loading_plugin_task.cancel());
+        }
     }
-
-    Ok(())
 }
 
 fn load_plugin_instance(instance: &mut Instance) -> Result<()> {
@@ -853,3 +742,56 @@ pub fn wasi_read_object<T: DeserializeOwned>(wasi_env: &WasiEnv) -> Result<T> {
         .and_then(|string| serde_json::from_str(&string).map_err(anyError::new))
         .with_context(|| format!("failed to deserialize object from WASI env '{wasi_env:?}'"))
 }
+
+pub fn apply_event_to_plugin(
+    plugin_id: u32,
+    client_id: ClientId,
+    instance: &Instance,
+    plugin_env: &PluginEnv,
+    event: &Event,
+    rows: usize,
+    columns: usize,
+    plugin_bytes: &mut Vec<(u32, ClientId, Vec<u8>)>,
+) -> Result<()> {
+    let err_context = || format!("Failed to apply event to plugin {plugin_id}");
+    let update = instance
+        .exports
+        .get_function("update")
+        .with_context(err_context)?;
+    wasi_write_object(&plugin_env.wasi_env, &event).with_context(err_context)?;
+    let update_return =
+        update
+            .call(&[])
+            .or_else::<anyError, _>(|e| match e.downcast::<serde_json::Error>() {
+                Ok(_) => panic!(
+                    "{}",
+                    anyError::new(VersionMismatchError::new(
+                        VERSION,
+                        "Unavailable",
+                        &plugin_env.plugin.path,
+                        plugin_env.plugin.is_builtin(),
+                    ))
+                ),
+                Err(e) => Err(e).with_context(err_context),
+            })?;
+    let should_render = match update_return.get(0) {
+        Some(Value::I32(n)) => *n == 1,
+        _ => false,
+    };
+
+    if rows > 0 && columns > 0 && should_render {
+        let rendered_bytes = instance
+            .exports
+            .get_function("render")
+            .map_err(anyError::new)
+            .and_then(|render| {
+                render
+                    .call(&[Value::I32(rows as i32), Value::I32(columns as i32)])
+                    .map_err(anyError::new)
+            })
+            .and_then(|_| wasi_read_string(&plugin_env.wasi_env))
+            .with_context(err_context)?;
+        plugin_bytes.push((plugin_id, client_id, rendered_bytes.as_bytes().to_vec()));
+    }
+    Ok(())
+}
diff --git a/zellij-server/src/route.rs b/zellij-server/src/route.rs
index 53b43ecd..f63da2d2 100644
--- a/zellij-server/src/route.rs
+++ b/zellij-server/src/route.rs
@@ -667,6 +667,22 @@ pub(crate) fn route_action(
                 .send_to_screen(ScreenInstruction::QueryTabNames(client_id))
                 .with_context(err_context)?;
         },
+        Action::NewTiledPluginPane(run_plugin, name) => {
+            session
+                .senders
+                .send_to_screen(ScreenInstruction::NewTiledPluginPane(
+                    run_plugin, name, client_id,
+                ))
+                .with_context(err_context)?;
+        },
+        Action::NewFloatingPluginPane(run_plugin, name) => {
+            session
+                .senders
+                .send_to_screen(ScreenInstruction::NewFloatingPluginPane(
+                    run_plugin, name, client_id,
+                ))
+                .with_context(err_context)?;
+        },
     }
     Ok(should_break)
 }
diff --git a/zellij-server/src/screen.rs b/zellij-server/src/screen.rs
index 5a422314..f1dd3874 100644
--- a/zellij-server/src/screen.rs
+++ b/zellij-server/src/screen.rs
@@ -13,7 +13,8 @@ use zellij_utils::pane_size::{Size, SizeInPixels};
 use zellij_utils::{
     input::command::TerminalAction,
     input::layout::{
-        FloatingPaneLayout, RunPluginLocation, SwapFloatingLayout, SwapTiledLayout, TiledPaneLayout,
+        FloatingPaneLayout, Run, RunPlugin, RunPluginLocation, SwapFloatingLayout, SwapTiledLayout,
+        TiledPaneLayout,
     },
     position::Position,
 };
@@ -29,7 +30,10 @@ use crate::{
     pty::{ClientOrTabIndex, PtyInstruction, VteBytes},
     tab::Tab,
     thread_bus::Bus,
-    ui::overlay::{Overlay, OverlayWindow, Overlayable},
+    ui::{
+        loading_indication::LoadingIndication,
+        overlay::{Overlay, OverlayWindow, Overlayable},
+    },
     ClientId, ServerInstruction,
 };
 use zellij_utils::{
@@ -250,6 +254,18 @@ pub enum ScreenInstruction {
     PreviousSwapLayout(ClientId),
     NextSwapLayout(ClientId),
     QueryTabNames(ClientId),
+    NewTiledPluginPane(RunPluginLocation, Option<String>, ClientId), // Option<String> is
+    NewFloatingPluginPane(RunPluginLocation, Option<String>, ClientId), // Option<String> is an
+    // optional pane title
+    AddPlugin(
+        Option<bool>, // should_float
+        RunPlugin,
+        Option<String>, // pane title
+        usize,          // tab index
+        u32,            // plugin id
+    ),
+    UpdatePluginLoadingStage(u32, LoadingIndication), // u32 - plugin_id
+    ProgressPluginLoadingOffset(u32),                 // u32 - plugin id
 }
 
 impl From<&ScreenInstruction> for ScreenContext {
@@ -393,6 +409,15 @@ impl From<&ScreenInstruction> for ScreenContext {
             ScreenInstruction::PreviousSwapLayout(..) => ScreenContext::PreviousSwapLayout,
             ScreenInstruction::NextSwapLayout(..) => ScreenContext::NextSwapLayout,
             ScreenInstruction::QueryTabNames(..) => ScreenContext::QueryTabNames,
+            ScreenInstruction::NewTiledPluginPane(..) => ScreenContext::NewTiledPluginPane,
+            ScreenInstruction::NewFloatingPluginPane(..) => ScreenContext::NewFloatingPluginPane,
+            ScreenInstruction::AddPlugin(..) => ScreenContext::AddPlugin,
+            ScreenInstruction::UpdatePluginLoadingStage(..) => {
+                ScreenContext::UpdatePluginLoadingStage
+            },
+            ScreenInstruction::ProgressPluginLoadingOffset(..) => {
+                ScreenContext::ProgressPluginLoadingOffset
+            },
         }
     }
 }
@@ -2426,6 +2451,89 @@ pub(crate) fn screen_thread_main(
                     .senders
                     .send_to_server(ServerInstruction::Log(tab_names, client_id))?;
             },
+            ScreenInstruction::NewTiledPluginPane(run_plugin_location, pane_title, client_id) => {
+                let tab_index = screen.active_tab_indices.values().next().unwrap_or(&1);
+                let size = Size::default();
+                let should_float = Some(false);
+                let run_plugin = RunPlugin {
+                    _allow_exec_host_cmd: false,
+                    location: run_plugin_location,
+                };
+                screen.bus.senders.send_to_plugin(PluginInstruction::Load(
+                    should_float,
+                    pane_title,
+                    run_plugin,
+                    *tab_index,
+                    client_id,
+                    size,
+                ))?;
+            },
+            ScreenInstruction::NewFloatingPluginPane(
+                run_plugin_location,
+                pane_title,
+                client_id,
+            ) => {
+                let tab_index = screen.active_tab_indices.values().next().unwrap(); // TODO: no
+                                                                                    // unwrap and
+                                                                                    // better
+                let size = Size::default(); // TODO: ???
+                let should_float = Some(true);
+                let run_plugin = RunPlugin {
+                    _allow_exec_host_cmd: false,
+                    location: run_plugin_location,
+                };
+                screen.bus.senders.send_to_plugin(PluginInstruction::Load(
+                    should_float,
+                    pane_title,
+                    run_plugin,
+                    *tab_index,
+                    client_id,
+                    size,
+                ))?;
+            },
+            ScreenInstruction::AddPlugin(
+                should_float,
+                run_plugin_location,
+                pane_title,
+                tab_index,
+                plugin_id,
+            ) => {
+                let pane_title =
+                    pane_title.unwrap_or_else(|| run_plugin_location.location.to_string());
+                let run_plugin = Run::Plugin(run_plugin_location);
+                if let Some(active_tab) = screen.tabs.get_mut(&tab_index) {
+                    active_tab.new_plugin_pane(
+                        PaneId::Plugin(plugin_id),
+                        pane_title,
+                        should_float,
+                        run_plugin,
+                        None,
+                    )?;
+                } else {
+                    log::error!("Tab index not found: {:?}", tab_index);
+                }
+                screen.unblock_input()?;
+            },
+            ScreenInstruction::UpdatePluginLoadingStage(pid, loading_indication) => {
+                let all_tabs = screen.get_tabs_mut();
+                for tab in all_tabs.values_mut() {
+                    if tab.has_plugin(pid) {
+                        tab.update_plugin_loading_stage(pid, loading_indication);
+                        break;
+                    }
+                }
+                screen.render()?;
+            },
+            ScreenInstruction::ProgressPluginLoadingOffset(pid) => {
+                let all_tabs = screen.get_tabs_mut();
+                for tab in all_tabs.values_mut() {
+                    if tab.has_plugin(pid) {
+                        tab.progress_plugin_loading_offset(pid);
+                        break;
+                    }
+                }
+                screen.render()?;
+            },
         }
     }
     Ok(())
diff --git a/zellij-server/src/tab/layout_applier.rs b/zellij-server/src/tab/layout_applier.rs
index 7964b698..04c22fe9 100644
--- a/zellij-server/src/tab/layout_applier.rs
+++ b/zellij-server/src/tab/layout_applier.rs
@@ -14,7 +14,7 @@ use crate::{
     ClientId,
 };
 use std::cell::RefCell;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::rc::Rc;
 use zellij_utils::{
     data::{Palette, Style},
@@ -30,6 +30,7 @@ pub struct LayoutApplier<'a> {
     terminal_emulator_colors: Rc<RefCell<Palette>>,
     terminal_emulator_color_codes: Rc<RefCell<HashMap<usize, String>>>,
     character_cell_size: Rc<RefCell<Option<SizeInPixels>>>,
+    connected_clients: Rc<RefCell<HashSet<ClientId>>>,
     style: Style,
     display_area: Rc<RefCell<Size>>, // includes all panes (including eg. the status bar and tab bar in the default layout)
     tiled_panes: &'a mut TiledPanes,
@@ -48,6 +49,7 @@ impl<'a> LayoutApplier<'a> {
         terminal_emulator_colors: &Rc<RefCell<Palette>>,
         terminal_emulator_color_codes: &Rc<RefCell<HashMap<usize, String>>>,
         character_cell_size: &Rc<RefCell<Option<SizeInPixels>>>,
+        connected_clients: &Rc<RefCell<HashSet<ClientId>>>,
         style: &Style,
         display_area: &Rc<RefCell<Size>>, // includes all panes (including eg. the status bar and tab bar in the default layout)
         tiled_panes: &'a mut TiledPanes,
@@ -63,6 +65,7 @@ impl<'a> LayoutApplier<'a> {
         let terminal_emulator_colors = terminal_emulator_colors.clone();
         let terminal_emulator_color_codes = terminal_emulator_color_codes.clone();
         let character_cell_size = character_cell_size.clone();
+        let connected_clients = connected_clients.clone();
         let style = style.clone();
         let display_area = display_area.clone();
         let os_api = os_api.clone();
@@ -74,6 +77,7 @@ impl<'a> LayoutApplier<'a> {
             terminal_emulator_colors,
             terminal_emulator_color_codes,
             character_cell_size,
+            connected_clients,
             style,
             display_area,
             tiled_panes,
@@ -119,7 +123,32 @@ impl<'a> LayoutApplier<'a> {
                 let mut existing_tab_state =
                     ExistingTabState::new(self.tiled_panes.drain(), currently_focused_pane_id);
                 let mut pane_focuser = PaneFocuser::new(refocus_pane);
+                let mut positions_left = vec![];
                 for (layout, position_and_size) in positions_in_layout {
+                    // first try to find panes with contents matching the layout exactly
+                    match existing_tab_state.find_and_extract_exact_match_pane(
+                        &layout.run,
+                        &position_and_size,
+                        true,
+                    ) {
+                        Some(mut pane) => {
+                            self.apply_layout_properties_to_pane(
+                                &mut pane,
+                                &layout,
+                                Some(position_and_size),
+                            );
+                            pane_focuser.set_pane_id_in_focused_location(layout.focus, &pane);
+                            resize_pty!(pane, self.os_api, self.senders, self.character_cell_size)?;
+                            self.tiled_panes
+                                .add_pane_with_existing_geom(pane.pid(), pane);
+                        },
+                        None => {
+                            positions_left.push((layout, position_and_size));
+                        },
+                    }
+                }
+                for (layout, position_and_size) in positions_left {
+                    // now let's try to find panes on a best-effort basis
                     if let Some(mut pane) = existing_tab_state.find_and_extract_pane(
                         &layout.run,
                         &position_and_size,
@@ -198,6 +227,7 @@ impl<'a> LayoutApplier<'a> {
                             self.terminal_emulator_color_codes.clone(),
                             self.link_handler.clone(),
                             self.character_cell_size.clone(),
+                            self.connected_clients.borrow().iter().copied().collect(),
                             self.style,
                             layout.run.clone(),
                         );
@@ -300,6 +330,7 @@ impl<'a> LayoutApplier<'a> {
                     self.terminal_emulator_color_codes.clone(),
                     self.link_handler.clone(),
                     self.character_cell_size.clone(),
+                    self.connected_clients.borrow().iter().copied().collect(),
                     self.style,
                     floating_pane_layout.run.clone(),
                 );
@@ -574,6 +605,22 @@ impl ExistingTabState {
             currently_focused_pane_id,
         }
     }
+    pub fn find_and_extract_exact_match_pane(
+        &mut self,
+        run: &Option<Run>,
+        position_and_size: &PaneGeom,
+        default_to_closest_position: bool,
+    ) -> Option<Box<dyn Pane>> {
+        let candidates = self.pane_candidates(run, position_and_size, default_to_closest_position);
+        if let Some(current_pane_id_with_same_contents) =
+            self.find_pane_id_with_same_contents_and_location(&candidates, run, position_and_size)
+        {
+            return self
+                .existing_panes
+                .remove(&current_pane_id_with_same_contents);
+        }
+        None
+    }
     pub fn find_and_extract_pane(
         &mut self,
         run: &Option<Run>,
@@ -679,6 +726,18 @@ impl ExistingTabState {
             .map(|(pid, _p)| *pid)
             .copied()
     }
+    fn find_pane_id_with_same_contents_and_location(
+        &self,
+        candidates: &Vec<(&PaneId, &Box<dyn Pane>)>,
+        run: &Option<Run>,
+        position: &PaneGeom,
+    ) -> Option<PaneId> {
+        candidates
+            .iter()
+            .find(|(_pid, p)| p.invoked_with() == run && p.position_and_size() == *position)
+            .map(|(pid, _p)| *pid)
+            .copied()
+    }
 }
 
 #[derive(Default, Debug)]
diff --git a/zellij-server/src/tab/mod.rs b/zellij-server/src/tab/mod.rs
index a2eb7920..1c1a77ac 100644
--- a/zellij-server/src/tab/mod.rs
+++ b/zellij-server/src/tab/mod.rs
@@ -18,7 +18,7 @@ use zellij_utils::{position::Position, serde};
 use crate::background_jobs::BackgroundJob;
 use crate::pty_writer::PtyWriteInstruction;
 use crate::screen::CopyOptions;
-use crate::ui::pane_boundaries_frame::FrameParams;
+use crate::ui::{loading_indication::LoadingIndication, pane_boundaries_frame::FrameParams};
 use layout_applier::LayoutApplier;
 use swap_layouts::SwapLayouts;
 
@@ -28,7 +28,7 @@ use crate::{
     output::{CharacterChunk, Output, SixelImageChunk},
     panes::sixel::SixelImageStore,
     panes::{FloatingPanes, TiledPanes},
-    panes::{LinkHandler, PaneId, TerminalPane},
+    panes::{LinkHandler, PaneId, PluginPane, TerminalPane},
     plugins::PluginInstruction,
     pty::{ClientOrTabIndex, PtyInstruction, VteBytes},
     thread_bus::ThreadSenders,
@@ -438,6 +438,8 @@ pub trait Pane {
     fn frame_color_override(&self) -> Option<PaletteColor>;
     fn invoked_with(&self) -> &Option<Run>;
     fn set_title(&mut self, title: String);
+    fn update_loading_indication(&mut self, _loading_indication: LoadingIndication) {} // only relevant for plugins
+    fn progress_animation_offset(&mut self) {} // only relevant for plugins
 }
 
 #[derive(Clone, Debug)]
@@ -599,6 +601,7 @@ impl Tab {
             &self.terminal_emulator_colors,
             &self.terminal_emulator_color_codes,
             &self.character_cell_size,
+            &self.connected_clients,
             &self.style,
             &self.display_area,
             &mut self.tiled_panes,
@@ -657,6 +660,7 @@ impl Tab {
                 &self.terminal_emulator_colors,
                 &self.terminal_emulator_color_codes,
                 &self.character_cell_size,
+                &self.connected_clients,
                 &self.style,
                 &self.display_area,
                 &mut self.tiled_panes,
@@ -709,6 +713,7 @@ impl Tab {
                 &self.terminal_emulator_colors,
                 &self.terminal_emulator_color_codes,
                 &self.character_cell_size,
+                &self.connected_clients,
                 &self.style,
                 &self.display_area,
                 &mut self.tiled_panes,
@@ -1104,6 +1109,114 @@ impl Tab {
         }
         Ok(())
     }
+    pub fn new_plugin_pane(
+        &mut self,
+        pid: PaneId,
+        initial_pane_title: String,
+        should_float: Option<bool>,
+        run_plugin: Run,
+        client_id: Option<ClientId>,
+    ) -> Result<()> {
+        let err_context = || format!("failed to create new pane with id {pid:?}");
+
+        match should_float {
+            Some(true) => self.show_floating_panes(),
+            Some(false) => self.hide_floating_panes(),
+            None => {},
+        };
+        if self.floating_panes.panes_are_visible() {
+            if let Some(new_pane_geom) = self.floating_panes.find_room_for_new_pane() {
+                if let PaneId::Plugin(plugin_pid) = pid {
+                    let mut new_pane = PluginPane::new(
+                        plugin_pid,
+                        new_pane_geom,
+                        self.senders
+                            .to_plugin
+                            .as_ref()
+                            .with_context(err_context)?
+                            .clone(),
+                        initial_pane_title,
+                        String::new(),
+                        self.sixel_image_store.clone(),
+                        self.terminal_emulator_colors.clone(),
+                        self.terminal_emulator_color_codes.clone(),
+                        self.link_handler.clone(),
+                        self.character_cell_size.clone(),
+                        self.connected_clients.borrow().iter().copied().collect(),
+                        self.style,
+                        Some(run_plugin),
+                    );
+                    new_pane.set_active_at(Instant::now());
+                    new_pane.set_content_offset(Offset::frame(1)); // floating panes always have a frame
+                    resize_pty!(
+                        new_pane,
+                        self.os_api,
+                        self.senders,
+                        self.character_cell_size
+                    )
+                    .with_context(err_context)?;
+                    self.floating_panes.add_pane(pid, Box::new(new_pane));
+                    self.floating_panes.focus_pane_for_all_clients(pid);
+                }
+                if self.auto_layout && !self.swap_layouts.is_floating_damaged() {
+                    // only do this if we're already in this layout, otherwise it might be
+                    // confusing and not what the user intends
+                    self.swap_layouts.set_is_floating_damaged(); // we do this so that we won't skip to the
+                                                                 // next layout
+                    self.next_swap_layout(client_id, true)?;
+                }
+            }
+        } else {
+            if self.tiled_panes.fullscreen_is_active() {
+                self.tiled_panes.unset_fullscreen();
+            }
+            let should_auto_layout = self.auto_layout && !self.swap_layouts.is_tiled_damaged();
+            if self.tiled_panes.has_room_for_new_pane() {
+                if let PaneId::Plugin(plugin_pid) = pid {
+                    let mut new_pane = PluginPane::new(
+                        plugin_pid,
+                        PaneGeom::default(), // the initial size will be set later
+                        self.senders
+                            .to_plugin
+                            .as_ref()
+                            .with_context(err_context)?
+                            .clone(),
+                        initial_pane_title,
+                        String::new(),
+                        self.sixel_image_store.clone(),
+                        self.terminal_emulator_colors.clone(),
+                        self.terminal_emulator_color_codes.clone(),
+                        self.link_handler.clone(),
+                        self.character_cell_size.clone(),
+                        self.connected_clients.borrow().iter().copied().collect(),
+                        self.style,
+                        Some(run_plugin),
+                    );
+                    new_pane.set_active_at(Instant::now());
+                    if should_auto_layout {
+                        // no need to relayout here, we'll do it when reapplying the swap layout
+                        // below
+                        self.tiled_panes
+                            .insert_pane_without_relayout(pid, Box::new(new_pane));
+                    } else {
+                        self.tiled_panes.insert_pane(pid, Box::new(new_pane));
+                    }
+                    self.should_clear_display_before_rendering = true;
+                    if let Some(client_id) = client_id {
+                        self.tiled_panes.focus_pane(pid, client_id);
+                    }
+                }
+            }
+            if should_auto_layout {
+                // only do this if we're already in this layout, otherwise it might be
+                // confusing and not what the user intends
+                self.swap_layouts.set_is_tiled_damaged(); // we do this so that we won't skip to the
+                                                          // next layout
+                self.next_swap_layout(client_id, true)?;
+            }
+        }
+        Ok(())
+    }
     pub fn suppress_active_pane(&mut self, pid: PaneId, client_id: ClientId) -> Result<()> {
         // this method creates a new pane from pid and replaces it with the active pane
         // the active pane is then suppressed (hidden and not rendered) until the current
@@ -3220,7 +3333,34 @@ impl Tab {
             pane.clear_pane_frame_color_override();
         }
     }
-
+    pub fn update_plugin_loading_stage(&mut self, pid: u32, loading_indication: LoadingIndication) {
+        if let Some(plugin_pane) = self
+            .tiled_panes
+            .get_pane_mut(PaneId::Plugin(pid))
+            .or_else(|| self.floating_panes.get_pane_mut(PaneId::Plugin(pid)))
+            .or_else(|| {
+                self.suppressed_panes
+                    .values_mut()
+                    .find(|s_p| s_p.pid() == PaneId::Plugin(pid))
+            })
+        {
+            plugin_pane.update_loading_indication(loading_indication);
+        }
+    }
+    pub fn progress_plugin_loading_offset(&mut self, pid: u32) {
+        if let Some(plugin_pane) = self
+            .tiled_panes
+            .get_pane_mut(PaneId::Plugin(pid))
+            .or_else(|| self.floating_panes.get_pane_mut(PaneId::Plugin(pid)))
+            .or_else(|| {
+                self.suppressed_panes
+                    .values_mut()
+                    .find(|s_p| s_p.pid() == PaneId::Plugin(pid))
+            })
+        {
+            plugin_pane.progress_animation_offset();
+        }
+    }
     fn show_floating_panes(&mut self) {
         // this function is to be preferred to directly invoking floating_panes.toggle_show_panes(true)
         self.floating_panes.toggle_show_panes(true);
diff --git a/zellij-server/src/ui/loading_indication.rs b/zellij-server/src/ui/loading_indication.rs
new file mode 100644
index 00000000..26bf299d
--- /dev/null
+++ b/zellij-server/src/ui/loading_indication.rs
@@ -0,0 +1,260 @@
+use std::fmt::{Display, Error, Formatter};
+
+use zellij_utils::{
+    data::{Palette, PaletteColor},
+    errors::prelude::*,
+};
+
+#[derive(Debug, Clone)]
+pub enum LoadingStatus {
+    InProgress,
+    Success,
+    NotFound,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct LoadingIndication {
+    pub ended: bool,
+    loading_from_memory: Option<LoadingStatus>,
+    loading_from_hd_cache: Option<LoadingStatus>,
+    compiling: Option<LoadingStatus>,
+    starting_plugin: Option<LoadingStatus>,
+    writing_plugin_to_cache: Option<LoadingStatus>,
+    cloning_plugin_for_other_clients: Option<LoadingStatus>,
+    error: Option<String>,
+    animation_offset: usize,
+    plugin_name: String,
+    terminal_emulator_colors: Option<Palette>,
+}
+
+impl LoadingIndication {
+    pub fn new(plugin_name: String) -> Self {
+        LoadingIndication {
+            plugin_name,
+            animation_offset: 0,
+            ..Default::default()
+        }
+    }
+    pub fn with_colors(mut self, terminal_emulator_colors: Palette) -> Self {
+        self.terminal_emulator_colors = Some(terminal_emulator_colors);
+        self
+    }
+    pub fn merge(&mut self, other: LoadingIndication) {
+        let current_animation_offset = self.animation_offset;
+        let current_terminal_emulator_colors = self.terminal_emulator_colors.take();
+        drop(std::mem::replace(self, other));
+        self.animation_offset = current_animation_offset;
+        self.terminal_emulator_colors = current_terminal_emulator_colors;
+    }
+    pub fn indicate_loading_plugin_from_memory(&mut self) {
+        self.loading_from_memory = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_loading_plugin_from_memory_success(&mut self) {
+        self.loading_from_memory = Some(LoadingStatus::Success);
+    }
+    pub fn indicate_loading_plugin_from_memory_notfound(&mut self) {
+        self.loading_from_memory = Some(LoadingStatus::NotFound);
+    }
+    pub fn indicate_loading_plugin_from_hd_cache(&mut self) {
+        self.loading_from_hd_cache = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_loading_plugin_from_hd_cache_success(&mut self) {
+        self.loading_from_hd_cache = Some(LoadingStatus::Success);
+    }
+    pub fn indicate_loading_plugin_from_hd_cache_notfound(&mut self) {
+        self.loading_from_hd_cache = Some(LoadingStatus::NotFound);
+    }
+    pub fn indicate_compiling_plugin(&mut self) {
+        self.compiling = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_compiling_plugin_success(&mut self) {
+        self.compiling = Some(LoadingStatus::Success);
+    }
+    pub fn indicate_starting_plugin(&mut self) {
+        self.starting_plugin = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_starting_plugin_success(&mut self) {
+        self.starting_plugin = Some(LoadingStatus::Success);
+    }
+    pub fn indicate_writing_plugin_to_cache(&mut self) {
+        self.writing_plugin_to_cache = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_writing_plugin_to_cache_success(&mut self) {
+        self.writing_plugin_to_cache = Some(LoadingStatus::Success);
+    }
+    pub fn indicate_cloning_plugin_for_other_clients(&mut self) {
+        self.cloning_plugin_for_other_clients = Some(LoadingStatus::InProgress);
+    }
+    pub fn indicate_cloning_plugin_for_other_clients_success(&mut self) {
+        self.cloning_plugin_for_other_clients = Some(LoadingStatus::Success);
+    }
+    pub fn end(&mut self) {
+        self.ended = true;
+    }
+    pub fn progress_animation_offset(&mut self) {
+        if self.animation_offset == 3 {
+            self.animation_offset = 0;
+        } else {
+            self.animation_offset += 1;
+        }
+    }
+    pub fn indicate_loading_error(&mut self, error_text: String) {
+        self.error = Some(error_text);
+    }
+    fn started_loading(&self) -> bool {
+        self.loading_from_memory.is_some()
+            || self.loading_from_hd_cache.is_some()
+            || self.compiling.is_some()
+            || self.starting_plugin.is_some()
+            || self.writing_plugin_to_cache.is_some()
+            || self.cloning_plugin_for_other_clients.is_some()
+    }
+}
+
+macro_rules! style {
+    ($fg:expr) => {
+        ansi_term::Style::new().fg(match $fg {
+            PaletteColor::Rgb((r, g, b)) => ansi_term::Color::RGB(r, g, b),
+            PaletteColor::EightBit(color) => ansi_term::Color::Fixed(color),
+        })
+    };
+}
+
+impl Display for LoadingIndication {
+    fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
+        let cyan = match self.terminal_emulator_colors {
+            Some(terminal_emulator_colors) => style!(terminal_emulator_colors.cyan).bold(),
+            None => ansi_term::Style::new(),
+        };
+        let green = match self.terminal_emulator_colors {
+            Some(terminal_emulator_colors) => style!(terminal_emulator_colors.green).bold(),
+            None => ansi_term::Style::new(),
+        };
+        let yellow = match self.terminal_emulator_colors {
+            Some(terminal_emulator_colors) => style!(terminal_emulator_colors.yellow).bold(),
+            None => ansi_term::Style::new(),
+        };
+        let red = match self.terminal_emulator_colors {
+            Some(terminal_emulator_colors) => style!(terminal_emulator_colors.red).bold(),
+            None => ansi_term::Style::new(),
+        };
+        let bold = ansi_term::Style::new().bold().italic();
+        let plugin_name = &self.plugin_name;
+        let success = green.paint("SUCCESS");
+        let failure = red.paint("FAILED");
+        let not_found = yellow.paint("NOT FOUND");
+        let add_dots = |stringified: &mut String| {
+            for _ in 0..self.animation_offset {
+                stringified.push('.');
+            }
+            stringified.push(' ');
+        };
+        let mut stringified = String::new();
+        let loading_text = "Loading";
+        let loading_from_memory_text = "Attempting to load from memory";
+        let loading_from_hd_cache_text = "Attempting to load from cache";
+        let compiling_text = "Compiling WASM";
+        let starting_plugin_text = "Starting";
+        let writing_plugin_to_cache_text = "Writing to cache";
+        let cloning_plugin_for_other_clients_text = "Cloning for other clients";
+        if self.started_loading() {
+            stringified.push_str(&format!("{} {}...", loading_text, cyan.paint(plugin_name)));
+        } else {
+            stringified.push_str(&format!(
+                "{} {}",
+                bold.paint(loading_text),
+                cyan.italic().paint(plugin_name)
+            ));
+            add_dots(&mut stringified);
+        }
+        match self.loading_from_memory {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!("\n\r{}", bold.paint(loading_from_memory_text)));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!("\n\r{loading_from_memory_text}... {success}"));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!("\n\r{loading_from_memory_text}... {not_found}"));
+            },
+            None => {},
+        }
+        match self.loading_from_hd_cache {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!("\n\r{}", bold.paint(loading_from_hd_cache_text)));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!("\n\r{loading_from_hd_cache_text}... {success}"));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!("\n\r{loading_from_hd_cache_text}... {not_found}"));
+            },
+            None => {},
+        }
+        match self.compiling {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!("\n\r{}", bold.paint(compiling_text)));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!("\n\r{compiling_text}... {success}"));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!("\n\r{compiling_text}... {failure}"));
+            },
+            None => {},
+        }
+        match self.starting_plugin {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!("\n\r{}", bold.paint(starting_plugin_text)));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!("\n\r{starting_plugin_text}... {success}"));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!("\n\r{starting_plugin_text}... {failure}"));
+            },
+            None => {},
+        }
+        match self.writing_plugin_to_cache {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!("\n\r{}", bold.paint(writing_plugin_to_cache_text)));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!("\n\r{writing_plugin_to_cache_text}... {success}"));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!("\n\r{writing_plugin_to_cache_text}... {failure}"));
+            },
+            None => {},
+        }
+        match self.cloning_plugin_for_other_clients {
+            Some(LoadingStatus::InProgress) => {
+                stringified.push_str(&format!(
+                    "\n\r{}",
+                    bold.paint(cloning_plugin_for_other_clients_text)
+                ));
+                add_dots(&mut stringified);
+            },
+            Some(LoadingStatus::Success) => {
+                stringified.push_str(&format!(
+                    "\n\r{cloning_plugin_for_other_clients_text}... {success}"
+                ));
+            },
+            Some(LoadingStatus::NotFound) => {
+                stringified.push_str(&format!(
+                    "\n\r{cloning_plugin_for_other_clients_text}... {failure}"
+                ));
+            },
+            None => {},
+        }
+        if let Some(error_text) = &self.error {
+            stringified.push_str(&format!("\n\r{} {error_text}", red.bold().paint("ERROR:")));
+        }
+        write!(f, "{}", stringified)
+    }
+}
diff --git a/zellij-server/src/ui/mod.rs b/zellij-server/src/ui/mod.rs
index 71cdaab5..60cb90bf 100644
--- a/zellij-server/src/ui/mod.rs
+++ b/zellij-server/src/ui/mod.rs
@@ -1,4 +1,5 @@
 pub mod boundaries;
+pub mod loading_indication;
 pub mod overlay;
 pub mod pane_boundaries_frame;
 pub mod pane_contents_and_ui;
diff --git a/zellij-server/src/unit/screen_tests.rs b/zellij-server/src/unit/screen_tests.rs
index 516c8630..adc7b044 100644
--- a/zellij-server/src/unit/screen_tests.rs
+++ b/zellij-server/src/unit/screen_tests.rs
@@ -1969,6 +1969,7 @@ pub fn send_cli_new_pane_action_with_default_parameters() {
     let cli_new_pane_action = CliAction::NewPane {
         direction: None,
         command: vec![],
+        plugin: None,
         cwd: None,
         floating: false,
         name: None,
@@ -2009,6 +2010,7 @@ pub fn send_cli_new_pane_action_with_split_direction() {
     let cli_new_pane_action = CliAction::NewPane {
         direction: Some(Direction::Right),
         command: vec![],
+        plugin: None,
         cwd: None,
         floating: false,
         name: None,
@@ -2049,6 +2051,7 @@ pub fn send_cli_new_pane_action_with_command_and_cwd() {
     let cli_new_pane_action = CliAction::NewPane {
         direction: Some(Direction::Right),
         command: vec!["htop".into()],
+        plugin: None,
         cwd: Some("/some/folder".into()),
         floating: false,
         name: None,
diff --git a/zellij-utils/assets/plugins/compact-bar.wasm b/zellij-utils/assets/plugins/compact-bar.wasm
index 9c2b0642..c69ee88e 100755
--- a/zellij-utils/assets/plugins/compact-bar.wasm
+++ b/zellij-utils/assets/plugins/compact-bar.wasm
diff --git a/zellij-utils/assets/plugins/status-bar.wasm b/zellij-utils/assets/plugins/status-bar.wasm
index b6fee839..a9ae7691 100755
--- a/zellij-utils/assets/plugins/status-bar.wasm
+++ b/zellij-utils/assets/plugins/status-bar.wasm
diff --git a/zellij-utils/assets/plugins/strider.wasm b/zellij-utils/assets/plugins/strider.wasm
index b5dc726a..5e4a54f9 100755
--- a/zellij-utils/assets/plugins/strider.wasm
+++ b/zellij-utils/assets/plugins/strider.wasm
diff --git a/zellij-utils/assets/plugins/tab-bar.wasm b/zellij-utils/assets/plugins/tab-bar.wasm
index 2df1fe6a..8ea487d8 100755
--- a/zellij-utils/assets/plugins/tab-bar.wasm
+++ b/zellij-utils/assets/plugins/tab-bar.wasm
diff --git a/zellij-utils/src/cli.rs b/zellij-utils/src/cli.rs
index 3947dfad..f232438b 100644
--- a/zellij-utils/src/cli.rs
+++ b/zellij-utils/src/cli.rs
@@ -254,6 +254,9 @@ pub enum CliAction {
         #[clap(last(true))]
         command: Vec<String>,
 
+        #[clap(short, long, conflicts_with("command"), conflicts_with("direction"))]
+        plugin: Option<String>,
+
         /// Change the working directory of the new pane
         #[clap(long, value_parser)]
         cwd: Option<PathBuf>,
diff --git a/zellij-utils/src/errors.rs b/zellij-utils/src/errors.rs
index f1522c1d..8a0f6331 100644
--- a/zellij-utils/src/errors.rs
+++ b/zellij-utils/src/errors.rs
@@ -324,6 +324,11 @@ pub enum ScreenContext {
     PreviousSwapLayout,
     NextSwapLayout,
     QueryTabNames,
+    NewTiledPluginPane,
+    NewFloatingPluginPane,
+    AddPlugin,
+    UpdatePluginLoadingStage,
+    ProgressPluginLoadingOffset,
 }
 
 /// Stack call representations corresponding to the different types of [`PtyInstruction`]s.
@@ -354,6 +359,7 @@ pub enum PluginContext {
     AddClient,
     RemoveClient,
     NewTab,
+    ApplyCachedEvents,
 }
 
 /// Stack call representations corresponding to the different types of [`ClientInstruction`]s.
@@ -399,6 +405,8 @@ pub enum PtyWriteContext {
 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 pub enum BackgroundJobContext {
     DisplayPaneError,
+    AnimatePluginLoading,
+    StopPluginLoadingAnimation,
     Exit,
 }
 
diff --git a/zellij-utils/src/input/actions.rs b/zellij-utils/src/input/actions.rs
index b2ec8d82..8e614d6d 100644
--- a/zellij-utils/src/input/actions.rs
+++ b/zellij-utils/src/input/actions.rs
@@ -2,7 +2,8 @@
 
 use super::command::RunCommandAction;
 use super::layout::{
-    FloatingPaneLayout, Layout, SwapFloatingLayout, SwapTiledLayout, TiledPaneLayout,
+    FloatingPaneLayout, Layout, RunPluginLocation, SwapFloatingLayout, SwapTiledLayout,
+    TiledPaneLayout,
 };
 use crate::cli::CliAction;
 use crate::data::InputMode;
@@ -225,6 +226,9 @@ pub enum Action {
     NextSwapLayout,
     /// Query all tab names
     QueryTabNames,
+    /// Open a new tiled (embedded, non-floating) plugin pane
+    NewTiledPluginPane(RunPluginLocation, Option<String>), // String is an optional name
+    NewFloatingPluginPane(RunPluginLocation, Option<String>), // String is an optional name
 }
 
 impl Action {
@@ -270,13 +274,34 @@ impl Action {
             CliAction::NewPane {
                 direction,
                 command,
+                plugin,
                 cwd,
                 floating,
                 name,
                 close_on_exit,
                 start_suspended,
             } => {
-                if !command.is_empty() {
+                if let Some(plugin) = plugin {
+                    if floating {
+                        let plugin = RunPluginLocation::parse(&plugin).map_err(|e| {
+                            format!("Failed to parse plugin loction {plugin}: {}", e)
+                        })?;
+                        Ok(vec![Action::NewFloatingPluginPane(plugin, name)])
+                    } else {
+                        let plugin = RunPluginLocation::parse(&plugin).map_err(|e| {
+                            format!("Failed to parse plugin location {plugin}: {}", e)
+                        })?;
+                        // it is intentional that a new tiled plugin pane cannot include a
+                        // direction
+                        // this is because the cli client opening a tiled plugin pane is a
+                        // different client than the one opening the pane, and this can potentially
+                        // create very confusing races if the client changes focus while the plugin
+                        // is being loaded
+                        // this is not the case with terminal panes for historical reasons of
+                        // backwards compatibility to a time before we had auto layouts
+                        Ok(vec![Action::NewTiledPluginPane(plugin, name)])
+                    }
+                } else if !command.is_empty() {
                     let mut command = command.clone();
                     let (command, args) = (PathBuf::from(command.remove(0)), command);
                     let current_dir = get_current_dir();