feat(plugins): introduce 'pipes', allowing users to pipe data to and control plugins from the command line (#3066)

* prototype - working with message from the cli

* prototype - pipe from the CLI to plugins

* prototype - pipe from the CLI to plugins and back again

* prototype - working with better cli interface

* prototype - working after removing unused stuff

* prototype - working with launching plugin if it is not launched, also fixed event ordering

* refactor: change message to cli-message

* prototype - allow plugins to send messages to each other

* fix: allow cli messages to send plugin parameters (and implement backpressure)

* fix: use input_pipe_id to identify cli pipes instead of their message name

* fix: come cleanups and add skip_cache parameter

* fix: pipe/client-server communication robustness

* fix: leaking messages between plugins while loading

* feat: allow plugins to specify how a new plugin instance is launched when sending messages

* fix: add permissions

* refactor: adjust cli api

* fix: improve cli plugin loading error messages

* docs: cli pipe

* fix: take plugin configuration into account when messaging between plugins

* refactor: pipe message protobuf interface

* refactor: update(event) -> pipe

* refactor - rename CliMessage to CliPipe

* fix: add is_private to pipes and change some naming

* refactor - cli client

* refactor: various cleanups

* style(fmt): rustfmt

* fix(pipes): backpressure across multiple plugins

* style: some cleanups

* style(fmt): rustfmt

* style: fix merge conflict mistake

* style(wording): clarify pipe permission

48 files changed, 3071 insertions, 305 deletions

diff --git a/default-plugins/fixture-plugin-for-tests/src/main.rs b/default-plugins/fixture-plugin-for-tests/src/main.rs
index 79783a1b..fcdc363f 100644
--- a/default-plugins/fixture-plugin-for-tests/src/main.rs
+++ b/default-plugins/fixture-plugin-for-tests/src/main.rs
@@ -11,6 +11,7 @@ struct State {
     received_events: Vec<Event>,
     received_payload: Option<String>,
     configuration: BTreeMap<String, String>,
+    message_to_plugin_payload: Option<String>,
 }
 
 #[derive(Default, Serialize, Deserialize)]
@@ -34,9 +35,12 @@ impl<'de> ZellijWorker<'de> for TestWorker {
     }
 }
 
+#[cfg(target_family = "wasm")]
 register_plugin!(State);
+#[cfg(target_family = "wasm")]
 register_worker!(TestWorker, test_worker, TEST_WORKER);
 
+#[cfg(target_family = "wasm")]
 impl ZellijPlugin for State {
     fn load(&mut self, configuration: BTreeMap<String, String>) {
         request_permission(&[
@@ -49,6 +53,8 @@ impl ZellijPlugin for State {
             PermissionType::OpenTerminalsOrPlugins,
             PermissionType::WriteToStdin,
             PermissionType::WebAccess,
+            PermissionType::ReadCliPipes,
+            PermissionType::MessageAndLaunchOtherPlugins,
         ]);
         self.configuration = configuration;
         subscribe(&[
@@ -295,10 +301,35 @@ impl ZellijPlugin for State {
         self.received_events.push(event);
         should_render
     }
+    fn pipe(&mut self, pipe_message: PipeMessage) -> bool {
+        let input_pipe_id = match pipe_message.source {
+            PipeSource::Cli(id) => id.clone(),
+            PipeSource::Plugin(id) => format!("{}", id),
+        };
+        let name = pipe_message.name;
+        let payload = pipe_message.payload;
+        if name == "message_name" && payload == Some("message_payload".to_owned()) {
+            unblock_cli_pipe_input(&input_pipe_id);
+        } else if name == "message_name_block" {
+            block_cli_pipe_input(&input_pipe_id);
+        } else if name == "pipe_output" {
+            cli_pipe_output(&name, "this_is_my_output");
+        } else if name == "pipe_message_to_plugin" {
+            pipe_message_to_plugin(
+                MessageToPlugin::new("message_to_plugin").with_payload("my_cool_payload"),
+            );
+        } else if name == "message_to_plugin" {
+            self.message_to_plugin_payload = payload.clone();
+        }
+        let should_render = true;
+        should_render
+    }
 
     fn render(&mut self, rows: usize, cols: usize) {
         if let Some(payload) = self.received_payload.as_ref() {
             println!("Payload from worker: {:?}", payload);
+        } else if let Some(payload) = self.message_to_plugin_payload.take() {
+            println!("Payload from self: {:?}", payload);
         } else {
             println!(
                 "Rows: {:?}, Cols: {:?}, Received events: {:?}",
diff --git a/src/main.rs b/src/main.rs
index 955a982b..7481bfc5 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -111,6 +111,31 @@ fn main() {
             commands::convert_old_theme_file(old_theme_file);
             std::process::exit(0);
         }
+        if let Some(Command::Sessions(Sessions::Pipe {
+            name,
+            payload,
+            args,
+            plugin,
+            plugin_configuration,
+        })) = opts.command
+        {
+            let command_cli_action = CliAction::Pipe {
+                name,
+                payload,
+                args,
+                plugin,
+                plugin_configuration,
+
+                force_launch_plugin: false,
+                skip_plugin_cache: false,
+                floating_plugin: None,
+                in_place_plugin: None,
+                plugin_cwd: None,
+                plugin_title: None,
+            };
+            commands::send_action_to_session(command_cli_action, opts.session, config);
+            std::process::exit(0);
+        }
     }
 
     if let Some(Command::Sessions(Sessions::ListSessions {
diff --git a/zellij-client/src/cli_client.rs b/zellij-client/src/cli_client.rs
index ef9f9122..38d73a4c 100644
--- a/zellij-client/src/cli_client.rs
+++ b/zellij-client/src/cli_client.rs
@@ -1,15 +1,23 @@
 //! The `[cli_client]` is used to attach to a running server session
 //! and dispatch actions, that are specified through the command line.
+use std::collections::BTreeMap;
+use std::io::BufRead;
 use std::process;
 use std::{fs, path::PathBuf};
 
 use crate::os_input_output::ClientOsApi;
 use zellij_utils::{
+    errors::prelude::*,
     input::actions::Action,
-    ipc::{ClientToServerMsg, ServerToClientMsg},
+    ipc::{ClientToServerMsg, ExitReason, ServerToClientMsg},
+    uuid::Uuid,
 };
 
-pub fn start_cli_client(os_input: Box<dyn ClientOsApi>, session_name: &str, actions: Vec<Action>) {
+pub fn start_cli_client(
+    mut os_input: Box<dyn ClientOsApi>,
+    session_name: &str,
+    actions: Vec<Action>,
+) {
     let zellij_ipc_pipe: PathBuf = {
         let mut sock_dir = zellij_utils::consts::ZELLIJ_SOCK_DIR.clone();
         fs::create_dir_all(&sock_dir).unwrap();
@@ -21,10 +29,166 @@ pub fn start_cli_client(os_input: Box<dyn ClientOsApi>, session_name: &str, acti
     let pane_id = os_input
         .env_variable("ZELLIJ_PANE_ID")
         .and_then(|e| e.trim().parse().ok());
+
     for action in actions {
-        let msg = ClientToServerMsg::Action(action, pane_id, None);
-        os_input.send_to_server(msg);
+        match action {
+            Action::CliPipe {
+                pipe_id,
+                name,
+                payload,
+                plugin,
+                args,
+                configuration,
+                launch_new,
+                skip_cache,
+                floating,
+                in_place,
+                cwd,
+                pane_title,
+            } => {
+                pipe_client(
+                    &mut os_input,
+                    pipe_id,
+                    name,
+                    payload,
+                    plugin,
+                    args,
+                    configuration,
+                    launch_new,
+                    skip_cache,
+                    floating,
+                    in_place,
+                    pane_id,
+                    cwd,
+                    pane_title,
+                );
+            },
+            action => {
+                single_message_client(&mut os_input, action, pane_id);
+            },
+        }
     }
+}
+
+fn pipe_client(
+    os_input: &mut Box<dyn ClientOsApi>,
+    pipe_id: String,
+    mut name: Option<String>,
+    payload: Option<String>,
+    plugin: Option<String>,
+    args: Option<BTreeMap<String, String>>,
+    mut configuration: Option<BTreeMap<String, String>>,
+    launch_new: bool,
+    skip_cache: bool,
+    floating: Option<bool>,
+    in_place: Option<bool>,
+    pane_id: Option<u32>,
+    cwd: Option<PathBuf>,
+    pane_title: Option<String>,
+) {
+    let mut stdin = os_input.get_stdin_reader();
+    let name = name.take().or_else(|| Some(Uuid::new_v4().to_string()));
+    if launch_new {
+        // we do this to make sure the plugin is unique (has a unique configuration parameter) so
+        // that a new one would be launched, but we'll still send it to the same instance rather
+        // than launching a new one in every iteration of the loop
+        configuration
+            .get_or_insert_with(BTreeMap::new)
+            .insert("_zellij_id".to_owned(), Uuid::new_v4().to_string());
+    }
+    let create_msg = |payload: Option<String>| -> ClientToServerMsg {
+        ClientToServerMsg::Action(
+            Action::CliPipe {
+                pipe_id: pipe_id.clone(),
+                name: name.clone(),
+                payload,
+                args: args.clone(),
+                plugin: plugin.clone(),
+                configuration: configuration.clone(),
+                floating,
+                in_place,
+                launch_new,
+                skip_cache,
+                cwd: cwd.clone(),
+                pane_title: pane_title.clone(),
+            },
+            pane_id,
+            None,
+        )
+    };
+    loop {
+        if payload.is_some() {
+            // we got payload from the command line, we should use it and not wait for more
+            let msg = create_msg(payload);
+            os_input.send_to_server(msg);
+            break;
+        }
+        // we didn't get payload from the command line, meaning we listen on STDIN because this
+        // signifies the user is about to pipe more (eg. cat my-large-file | zellij pipe ...)
+        let mut buffer = String::new();
+        let _ = stdin.read_line(&mut buffer);
+        if buffer.is_empty() {
+            // end of pipe, send an empty message down the pipe
+            let msg = create_msg(None);
+            os_input.send_to_server(msg);
+            break;
+        } else {
+            // we've got data! send it down the pipe (most common)
+            let msg = create_msg(Some(buffer));
+            os_input.send_to_server(msg);
+        }
+        loop {
+            // wait for a response and act accordingly
+            match os_input.recv_from_server() {
+                Some((ServerToClientMsg::UnblockCliPipeInput(pipe_name), _)) => {
+                    // unblock this pipe, meaning we need to stop waiting for a response and read
+                    // once more from STDIN
+                    if pipe_name == pipe_id {
+                        break;
+                    }
+                },
+                Some((ServerToClientMsg::CliPipeOutput(pipe_name, output), _)) => {
+                    // send data to STDOUT, this *does not* mean we need to unblock the input
+                    let err_context = "Failed to write to stdout";
+                    if pipe_name == pipe_id {
+                        let mut stdout = os_input.get_stdout_writer();
+                        stdout
+                            .write_all(output.as_bytes())
+                            .context(err_context)
+                            .non_fatal();
+                        stdout.flush().context(err_context).non_fatal();
+                    }
+                },
+                Some((ServerToClientMsg::Log(log_lines), _)) => {
+                    log_lines.iter().for_each(|line| println!("{line}"));
+                    process::exit(0);
+                },
+                Some((ServerToClientMsg::LogError(log_lines), _)) => {
+                    log_lines.iter().for_each(|line| eprintln!("{line}"));
+                    process::exit(2);
+                },
+                Some((ServerToClientMsg::Exit(exit_reason), _)) => match exit_reason {
+                    ExitReason::Error(e) => {
+                        eprintln!("{}", e);
+                        process::exit(2);
+                    },
+                    _ => {
+                        process::exit(0);
+                    },
+                },
+                _ => {},
+            }
+        }
+    }
+}
+
+fn single_message_client(
+    os_input: &mut Box<dyn ClientOsApi>,
+    action: Action,
+    pane_id: Option<u32>,
+) {
+    let msg = ClientToServerMsg::Action(action, pane_id, None);
+    os_input.send_to_server(msg);
     loop {
         match os_input.recv_from_server() {
             Some((ServerToClientMsg::UnblockInputThread, _)) => {
@@ -39,6 +203,15 @@ pub fn start_cli_client(os_input: Box<dyn ClientOsApi>, session_name: &str, acti
                 log_lines.iter().for_each(|line| eprintln!("{line}"));
                 process::exit(2);
             },
+            Some((ServerToClientMsg::Exit(exit_reason), _)) => match exit_reason {
+                ExitReason::Error(e) => {
+                    eprintln!("{}", e);
+                    process::exit(2);
+                },
+                _ => {
+                    process::exit(0);
+                },
+            },
             _ => {},
         }
     }
diff --git a/zellij-client/src/lib.rs b/zellij-client/src/lib.rs
index c704ad84..640cbcd8 100644
--- a/zellij-client/src/lib.rs
+++ b/zellij-client/src/lib.rs
@@ -49,6 +49,8 @@ pub(crate) enum ClientInstruction {
     LogError(Vec<String>),
     SwitchSession(ConnectToSession),
     SetSynchronizedOutput(Option<SyncOutput>),
+    UnblockCliPipeInput(String),   // String -> pipe name
+    CliPipeOutput(String, String), // String -> pipe name, String -> output
 }
 
 impl From<ServerToClientMsg> for ClientInstruction {
@@ -67,6 +69,12 @@ impl From<ServerToClientMsg> for ClientInstruction {
             ServerToClientMsg::SwitchSession(connect_to_session) => {
                 ClientInstruction::SwitchSession(connect_to_session)
             },
+            ServerToClientMsg::UnblockCliPipeInput(pipe_name) => {
+                ClientInstruction::UnblockCliPipeInput(pipe_name)
+            },
+            ServerToClientMsg::CliPipeOutput(pipe_name, output) => {
+                ClientInstruction::CliPipeOutput(pipe_name, output)
+            },
         }
     }
 }
@@ -87,6 +95,8 @@ impl From<&ClientInstruction> for ClientContext {
             ClientInstruction::DoneParsingStdinQuery => ClientContext::DoneParsingStdinQuery,
             ClientInstruction::SwitchSession(..) => ClientContext::SwitchSession,
             ClientInstruction::SetSynchronizedOutput(..) => ClientContext::SetSynchronisedOutput,
+            ClientInstruction::UnblockCliPipeInput(..) => ClientContext::UnblockCliPipeInput,
+            ClientInstruction::CliPipeOutput(..) => ClientContext::CliPipeOutput,
         }
     }
 }
diff --git a/zellij-client/src/os_input_output.rs b/zellij-client/src/os_input_output.rs
index d6fb3b30..50e7e2eb 100644
--- a/zellij-client/src/os_input_output.rs
+++ b/zellij-client/src/os_input_output.rs
@@ -95,7 +95,8 @@ pub trait ClientOsApi: Send + Sync {
     fn unset_raw_mode(&self, fd: RawFd) -> Result<(), nix::Error>;
     /// Returns the writer that allows writing to standard output.
     fn get_stdout_writer(&self) -> Box<dyn io::Write>;
-    fn get_stdin_reader(&self) -> Box<dyn io::Read>;
+    /// Returns a BufReader that allows to read from STDIN line by line, also locks STDIN
+    fn get_stdin_reader(&self) -> Box<dyn io::BufRead>;
     fn update_session_name(&mut self, new_session_name: String);
     /// Returns the raw contents of standard input.
     fn read_from_stdin(&mut self) -> Result<Vec<u8>, &'static str>;
@@ -186,9 +187,10 @@ impl ClientOsApi for ClientOsInputOutput {
         let stdout = ::std::io::stdout();
         Box::new(stdout)
     }
-    fn get_stdin_reader(&self) -> Box<dyn io::Read> {
+
+    fn get_stdin_reader(&self) -> Box<dyn io::BufRead> {
         let stdin = ::std::io::stdin();
-        Box::new(stdin)
+        Box::new(stdin.lock())
     }
 
     fn send_to_server(&self, msg: ClientToServerMsg) {
diff --git a/zellij-client/src/unit/stdin_tests.rs b/zellij-client/src/unit/stdin_tests.rs
index 3091af50..8e104887 100644
--- a/zellij-client/src/unit/stdin_tests.rs
+++ b/zellij-client/src/unit/stdin_tests.rs
@@ -151,10 +151,10 @@ impl ClientOsApi for FakeClientOsApi {
         let fake_stdout_writer = FakeStdoutWriter::new(self.stdout_buffer.clone());
         Box::new(fake_stdout_writer)
     }
-    fn get_stdin_reader(&self) -> Box<dyn io::Read> {
+    fn get_stdin_reader(&self) -> Box<dyn io::BufRead> {
         unimplemented!()
     }
-    fn update_session_name(&mut self, new_session_name: String) {}
+    fn update_session_name(&mut self, _new_session_name: String) {}
     fn read_from_stdin(&mut self) -> Result<Vec<u8>, &'static str> {
         Ok(self.stdin_buffer.drain(..).collect())
     }
diff --git a/zellij-server/src/lib.rs b/zellij-server/src/lib.rs
index 51daeb0e..437f465b 100644
--- a/zellij-server/src/lib.rs
+++ b/zellij-server/src/lib.rs
@@ -85,14 +85,21 @@ pub enum ServerInstruction {
     ConnStatus(ClientId),
     ActiveClients(ClientId),
     Log(Vec<String>, ClientId),
+    LogError(Vec<String>, ClientId),
     SwitchSession(ConnectToSession, ClientId),
+    UnblockCliPipeInput(String),   // String -> Pipe name
+    CliPipeOutput(String, String), // String -> Pipe name, String -> Output
+    AssociatePipeWithClient {
+        pipe_id: String,
+        client_id: ClientId,
+    },
 }
 
 impl From<&ServerInstruction> for ServerContext {
     fn from(server_instruction: &ServerInstruction) -> Self {
         match *server_instruction {
             ServerInstruction::NewClient(..) => ServerContext::NewClient,
-            ServerInstruction::Render(_) => ServerContext::Render,
+            ServerInstruction::Render(..) => ServerContext::Render,
             ServerInstruction::UnblockInputThread => ServerContext::UnblockInputThread,
             ServerInstruction::ClientExit(..) => ServerContext::ClientExit,
             ServerInstruction::RemoveClient(..) => ServerContext::RemoveClient,
@@ -103,7 +110,13 @@ impl From<&ServerInstruction> for ServerContext {
             ServerInstruction::ConnStatus(..) => ServerContext::ConnStatus,
             ServerInstruction::ActiveClients(_) => ServerContext::ActiveClients,
             ServerInstruction::Log(..) => ServerContext::Log,
+            ServerInstruction::LogError(..) => ServerContext::LogError,
             ServerInstruction::SwitchSession(..) => ServerContext::SwitchSession,
+            ServerInstruction::UnblockCliPipeInput(..) => ServerContext::UnblockCliPipeInput,
+            ServerInstruction::CliPipeOutput(..) => ServerContext::CliPipeOutput,
+            ServerInstruction::AssociatePipeWithClient { .. } => {
+                ServerContext::AssociatePipeWithClient
+            },
         }
     }
 }
@@ -186,12 +199,14 @@ macro_rules! send_to_client {
 #[derive(Clone, Debug, PartialEq)]
 pub(crate) struct SessionState {
     clients: HashMap<ClientId, Option<Size>>,
+    pipes: HashMap<String, ClientId>, // String => pipe_id
 }
 
 impl SessionState {
     pub fn new() -> Self {
         SessionState {
             clients: HashMap::new(),
+            pipes: HashMap::new(),
         }
     }
     pub fn new_client(&mut self) -> ClientId {
@@ -207,8 +222,12 @@ impl SessionState {
         self.clients.insert(next_client_id, None);
         next_client_id
     }
+    pub fn associate_pipe_with_client(&mut self, pipe_id: String, client_id: ClientId) {
+        self.pipes.insert(pipe_id, client_id);
+    }
     pub fn remove_client(&mut self, client_id: ClientId) {
         self.clients.remove(&client_id);
+        self.pipes.retain(|_p_id, c_id| c_id != &client_id);
     }
     pub fn set_client_size(&mut self, client_id: ClientId, size: Size) {
         self.clients.insert(client_id, Some(size));
@@ -240,6 +259,17 @@ impl SessionState {
     pub fn client_ids(&self) -> Vec<ClientId> {
         self.clients.keys().copied().collect()
     }
+    pub fn active_clients_are_connected(&self) -> bool {
+        let ids_of_pipe_clients: HashSet<ClientId> = self.pipes.values().copied().collect();
+        let mut active_clients_connected = false;
+        for client_id in self.clients.keys() {
+            if ids_of_pipe_clients.contains(client_id) {
+                continue;
+            }
+            active_clients_connected = true;
+        }
+        active_clients_connected
+    }
 }
 
 pub fn start_server(mut os_input: Box<dyn ServerOsApi>, socket_path: PathBuf) {
@@ -490,6 +520,52 @@ pub fn start_server(mut os_input: Box<dyn ServerOsApi>, socket_path: PathBuf) {
                     );
                 }
             },
+            ServerInstruction::UnblockCliPipeInput(pipe_name) => {
+                match session_state.read().unwrap().pipes.get(&pipe_name) {
+                    Some(client_id) => {
+                        send_to_client!(
+                            *client_id,
+                            os_input,
+                            ServerToClientMsg::UnblockCliPipeInput(pipe_name.clone()),
+                            session_state
+                        );
+                    },
+                    None => {
+                        // send to all clients, this pipe might not have been associated yet
+                        for client_id in session_state.read().unwrap().clients.keys() {
+                            send_to_client!(
+                                *client_id,
+                                os_input,
+                                ServerToClientMsg::UnblockCliPipeInput(pipe_name.clone()),
+                                session_state
+                            );
+                        }
+                    },
+                }
+            },
+            ServerInstruction::CliPipeOutput(pipe_name, output) => {
+                match session_state.read().unwrap().pipes.get(&pipe_name) {
+                    Some(client_id) => {
+                        send_to_client!(
+                            *client_id,
+                            os_input,
+                            ServerToClientMsg::CliPipeOutput(pipe_name.clone(), output.clone()),
+                            session_state
+                        );
+                    },
+                    None => {
+                        // send to all clients, this pipe might not have been associated yet
+                        for client_id in session_state.read().unwrap().clients.keys() {
+                            send_to_client!(
+                                *client_id,
+                                os_input,
+                                ServerToClientMsg::CliPipeOutput(pipe_name.clone(), output.clone()),
+                                session_state
+                            );
+                        }
+                    },
+                }
+            },
             ServerInstruction::ClientExit(client_id) => {
                 let _ =
                     os_input.send_to_client(client_id, ServerToClientMsg::Exit(ExitReason::Normal));
@@ -520,8 +596,19 @@ pub fn start_server(mut os_input: Box<dyn ServerOsApi>, socket_path: PathBuf) {
                     .senders
                     .send_to_plugin(PluginInstruction::RemoveClient(client_id))
                     .unwrap();
-                if session_state.read().unwrap().clients.is_empty() {
+                if !session_state.read().unwrap().active_clients_are_connected() {
                     *session_data.write().unwrap() = None;
+                    let client_ids_to_cleanup: Vec<ClientId> = session_state
+                        .read()
+                        .unwrap()
+                        .clients
+                        .keys()
+                        .copied()
+                        .collect();
+                    // these are just the pipes
+                    for client_id in client_ids_to_cleanup {
+                        remove_client!(client_id, os_input, session_state);
+                    }
                     break;
                 }
             },
@@ -654,6 +741,14 @@ pub fn start_server(mut os_input: Box<dyn ServerOsApi>, socket_path: PathBuf) {
                     session_state
                 );
             },
+            ServerInstruction::LogError(lines_to_log, client_id) => {
+                send_to_client!(
+                    client_id,
+                    os_input,
+                    ServerToClientMsg::LogError(lines_to_log),
+                    session_state
+                );
+            },
             ServerInstruction::SwitchSession(connect_to_session, client_id) => {
                 if let Some(min_size) = session_state.read().unwrap().min_client_terminal_size() {
                     session_data
@@ -689,6 +784,12 @@ pub fn start_server(mut os_input: Box<dyn ServerOsApi>, socket_path: PathBuf) {
                 );
                 remove_client!(client_id, os_input, session_state);
             },
+            ServerInstruction::AssociatePipeWithClient { pipe_id, client_id } => {
+                session_state
+                    .write()
+                    .unwrap()
+                    .associate_pipe_with_client(pipe_id, client_id);
+            },
         }
     }
 
@@ -825,7 +926,7 @@ fn init_session(
         .spawn({
             let plugin_bus = Bus::new(
                 vec![plugin_receiver],
-                Some(&to_screen),
+                Some(&to_screen_bounded),
                 Some(&to_pty),
                 Some(&to_plugin),
                 Some(&to_server),
diff --git a/zellij-server/src/panes/terminal_pane.rs b/zellij-server/src/panes/terminal_pane.rs
index 1e550abd..bff78b98 100644
--- a/zellij-server/src/panes/terminal_pane.rs
+++ b/zellij-server/src/panes/terminal_pane.rs
@@ -16,7 +16,7 @@ use std::time::{self, Instant};
 use zellij_utils::input::command::RunCommand;
 use zellij_utils::pane_size::Offset;
 use zellij_utils::{
-    data::{InputMode, Palette, PaletteColor, Style},
+    data::{InputMode, Palette, PaletteColor, PaneId as ZellijUtilsPaneId, Style},
     errors::prelude::*,
     input::layout::Run,
     pane_size::PaneGeom,
@@ -85,6 +85,16 @@ pub enum PaneId {
     Plugin(u32), // FIXME: Drop the trait object, make this a wrapper for the struct?
 }
 
+// because crate architecture and reasons...
+impl From<ZellijUtilsPaneId> for PaneId {
+    fn from(zellij_utils_pane_id: ZellijUtilsPaneId) -> Self {
+        match zellij_utils_pane_id {
+            ZellijUtilsPaneId::Terminal(id) => PaneId::Terminal(id),
+            ZellijUtilsPaneId::Plugin(id) => PaneId::Plugin(id),
+        }
+    }
+}
+
 type IsFirstRun = bool;
 
 // FIXME: This should hold an os_api handle so that terminal panes can set their own size via FD in
diff --git a/zellij-server/src/plugins/mod.rs b/zellij-server/src/plugins/mod.rs
index 9dbacd5c..ebae02a3 100644
--- a/zellij-server/src/plugins/mod.rs
+++ b/zellij-server/src/plugins/mod.rs
@@ -1,3 +1,4 @@
+mod pipes;
 mod plugin_loader;
 mod plugin_map;
 mod plugin_worker;
@@ -6,7 +7,7 @@ mod watch_filesystem;
 mod zellij_exports;
 use log::info;
 use std::{
-    collections::{HashMap, HashSet},
+    collections::{BTreeMap, HashMap, HashSet},
     fs,
     path::PathBuf,
     sync::{Arc, Mutex},
@@ -19,11 +20,15 @@ use crate::screen::ScreenInstruction;
 use crate::session_layout_metadata::SessionLayoutMetadata;
 use crate::{pty::PtyInstruction, thread_bus::Bus, ClientId, ServerInstruction};
 
+pub use wasm_bridge::PluginRenderAsset;
 use wasm_bridge::WasmBridge;
 
 use zellij_utils::{
     async_std::{channel, future::timeout, task},
-    data::{Event, EventType, PermissionStatus, PermissionType, PluginCapabilities},
+    data::{
+        Event, EventType, MessageToPlugin, PermissionStatus, PermissionType, PipeMessage,
+        PipeSource, PluginCapabilities,
+    },
     errors::{prelude::*, ContextType, PluginContext},
     input::{
         command::TerminalAction,
@@ -73,7 +78,10 @@ pub enum PluginInstruction {
         usize, // tab_index
         ClientId,
     ),
-    ApplyCachedEvents(Vec<PluginId>),
+    ApplyCachedEvents {
+        plugin_ids: Vec<PluginId>,
+        done_receiving_permissions: bool,
+    },
     ApplyCachedWorkerMessages(PluginId),
     PostMessagesToPluginWorker(
         PluginId,
@@ -100,6 +108,28 @@ pub enum PluginInstruction {
     ),
     DumpLayout(SessionLayoutMetadata, ClientId),
     LogLayoutToHd(SessionLayoutMetadata),
+    CliPipe {
+        pipe_id: String,
+        name: String,
+        payload: Option<String>,
+        plugin: Option<String>,
+        args: Option<BTreeMap<String, String>>,
+        configuration: Option<BTreeMap<String, String>>,
+        floating: Option<bool>,
+        pane_id_to_replace: Option<PaneId>,
+        pane_title: Option<String>,
+        cwd: Option<PathBuf>,
+        skip_cache: bool,
+        cli_client_id: ClientId,
+    },
+    CachePluginEvents {
+        plugin_id: PluginId,
+    },
+    MessageFromPlugin {
+        source_plugin_id: u32,
+        message: MessageToPlugin,
+    },
+    UnblockCliPipes(Vec<PluginRenderAsset>),
     Exit,
 }
 
@@ -115,7 +145,7 @@ impl From<&PluginInstruction> for PluginContext {
             PluginInstruction::AddClient(_) => PluginContext::AddClient,
             PluginInstruction::RemoveClient(_) => PluginContext::RemoveClient,
             PluginInstruction::NewTab(..) => PluginContext::NewTab,
-            PluginInstruction::ApplyCachedEvents(..) => PluginContext::ApplyCachedEvents,
+            PluginInstruction::ApplyCachedEvents { .. } => PluginContext::ApplyCachedEvents,
             PluginInstruction::ApplyCachedWorkerMessages(..) => {
                 PluginContext::ApplyCachedWorkerMessages
             },
@@ -131,6 +161,10 @@ impl From<&PluginInstruction> for PluginContext {
             },
             PluginInstruction::DumpLayout(..) => PluginContext::DumpLayout,
             PluginInstruction::LogLayoutToHd(..) => PluginContext::LogLayoutToHd,
+            PluginInstruction::CliPipe { .. } => PluginContext::CliPipe,
+            PluginInstruction::CachePluginEvents { .. } => PluginContext::CachePluginEvents,
+            PluginInstruction::MessageFromPlugin { .. } => PluginContext::MessageFromPlugin,
+            PluginInstruction::UnblockCliPipes { .. } => PluginContext::UnblockCliPipes,
         }
     }
 }
@@ -188,19 +222,20 @@ pub(crate) fn plugin_thread_main(
                 skip_cache,
             ) => match wasm_bridge.load_plugin(
                 &run,
-                tab_index,
+                Some(tab_index),
                 size,
                 cwd.clone(),
                 skip_cache,
                 Some(client_id),
+                None,
             ) {
-                Ok(plugin_id) => {
+                Ok((plugin_id, client_id)) => {
                     drop(bus.senders.send_to_screen(ScreenInstruction::AddPlugin(
                         should_float,
                         should_be_open_in_place,
                         run,
                         pane_title,
-                        tab_index,
+                        Some(tab_index),
                         plugin_id,
                         pane_id_to_replace,
                         cwd,
@@ -230,17 +265,23 @@ pub(crate) fn plugin_thread_main(
                             // we intentionally do not provide the client_id here because it belongs to
                             // the cli who spawned the command and is not an existing client_id
                             let skip_cache = true; // when reloading we always skip cache
-                            match wasm_bridge
-                                .load_plugin(&run, tab_index, size, None, skip_cache, None)
-                            {
-                                Ok(plugin_id) => {
+                            match wasm_bridge.load_plugin(
+                                &run,
+                                Some(tab_index),
+                                size,
+                                None,
+                                skip_cache,
+                                None,
+                                None,
+                            ) {
+                                Ok((plugin_id, _client_id)) => {
                                     let should_be_open_in_place = false;
                                     drop(bus.senders.send_to_screen(ScreenInstruction::AddPlugin(
                                         should_float,
                                         should_be_open_in_place,
                                         run,
                                         pane_title,
-                                        tab_index,
+                                        Some(tab_index),
                                         plugin_id,
                                         None,
                                         None,
@@ -298,13 +339,14 @@ pub(crate) fn plugin_thread_main(
                 for run_instruction in extracted_run_instructions {
                     if let Some(Run::Plugin(run)) = run_instruction {
                         let skip_cache = false;
-                        let plugin_id = wasm_bridge.load_plugin(
+                        let (plugin_id, _client_id) = wasm_bridge.load_plugin(
                             &run,
-                            tab_index,
+                            Some(tab_index),
                             size,
                             None,
                             skip_cache,
                             Some(client_id),
+                            None,
                         )?;
                         plugin_ids
                             .entry((run.location, run.configuration))
@@ -322,8 +364,15 @@ pub(crate) fn plugin_thread_main(
                     client_id,
                 )));
             },
-            PluginInstruction::ApplyCachedEvents(plugin_id) => {
-                wasm_bridge.apply_cached_events(plugin_id, shutdown_send.clone())?;
+            PluginInstruction::ApplyCachedEvents {
+                plugin_ids,
+                done_receiving_permissions,
+            } => {
+                wasm_bridge.apply_cached_events(
+                    plugin_ids,
+                    done_receiving_permissions,
+                    shutdown_send.clone(),
+                )?;
             },
             PluginInstruction::ApplyCachedWorkerMessages(plugin_id) => {
                 wasm_bridge.apply_cached_worker_messages(plugin_id)?;
@@ -383,6 +432,12 @@ pub(crate) fn plugin_thread_main(
                     Event::PermissionRequestResult(status),
                 )];
                 wasm_bridge.update_plugins(updates, shutdown_send.clone())?;
+                let done_receiving_permissions = true;
+                wasm_bridge.apply_cached_events(
+                    vec![plugin_id],
+                    done_receiving_permissions,
+                    shutdown_send.clone(),
+                )?;
             },
             PluginInstruction::DumpLayout(mut session_layout_metadata, client_id) => {
                 populate_session_layout_metadata(&mut session_layout_metadata, &wasm_bridge);
@@ -398,6 +453,128 @@ pub(crate) fn plugin_thread_main(
                         .send_to_pty(PtyInstruction::LogLayoutToHd(session_layout_metadata)),
                 );
             },
+            PluginInstruction::CliPipe {
+                pipe_id,
+                name,
+                payload,
+                plugin,
+                args,
+                configuration,
+                floating,
+                pane_id_to_replace,
+                pane_title,
+                cwd,
+                skip_cache,
+                cli_client_id,
+            } => {
+                let should_float = floating.unwrap_or(true);
+                let mut pipe_messages = vec![];
+                match plugin {
+                    Some(plugin_url) => {
+                        // send to specific plugin(s)
+                        pipe_to_specific_plugins(
+                            PipeSource::Cli(pipe_id.clone()),
+                            &plugin_url,
+                            &configuration,
+                            &cwd,
+                            skip_cache,
+                            should_float,
+                            &pane_id_to_replace,
+                            &pane_title,
+                            Some(cli_client_id),
+                            &mut pipe_messages,
+                            &name,
+                            &payload,
+                            &args,
+                            &bus,
+                            &mut wasm_bridge,
+                        );
+                    },
+                    None => {
+                        // no specific destination, send to all plugins
+                        pipe_to_all_plugins(
+                            PipeSource::Cli(pipe_id.clone()),
+                            &name,
+                            &payload,
+                            &args,
+                            &mut wasm_bridge,
+                            &mut pipe_messages,
+                        );
+                    },
+                }
+                wasm_bridge.pipe_messages(pipe_messages, shutdown_send.clone())?;
+            },
+            PluginInstruction::CachePluginEvents { plugin_id } => {
+                wasm_bridge.cache_plugin_events(plugin_id);
+            },
+            PluginInstruction::MessageFromPlugin {
+                source_plugin_id,
+                message,
+            } => {
+                let cwd = message.new_plugin_args.as_ref().and_then(|n| n.cwd.clone());
+                let mut pipe_messages = vec![];
+                let skip_cache = message
+                    .new_plugin_args
+                    .as_ref()
+                    .map(|n| n.skip_cache)
+                    .unwrap_or(false);
+                let should_float = message
+                    .new_plugin_args
+                    .as_ref()
+                    .and_then(|n| n.should_float)
+                    .unwrap_or(true);
+                let pane_title = message
+                    .new_plugin_args
+                    .as_ref()
+                    .and_then(|n| n.pane_title.clone());
+                let pane_id_to_replace = message
+                    .new_plugin_args
+                    .as_ref()
+                    .and_then(|n| n.pane_id_to_replace);
+                match message.plugin_url {
+                    Some(plugin_url) => {
+                        // send to specific plugin(s)
+                        pipe_to_specific_plugins(
+                            PipeSource::Plugin(source_plugin_id),
+                            &plugin_url,
+                            &Some(message.plugin_config),
+                            &cwd,
+                            skip_cache,
+                            should_float,
+                            &pane_id_to_replace.map(|p| p.into()),
+                            &pane_title,
+                            None,
+                            &mut pipe_messages,
+                            &message.message_name,
+                            &message.message_payload,
+                            &Some(message.message_args),
+                            &bus,
+                            &mut wasm_bridge,
+                        );
+                    },
+                    None => {
+                        // send to all plugins
+                        pipe_to_all_plugins(
+                            PipeSource::Plugin(source_plugin_id),
+                            &message.message_name,
+                            &message.message_payload,
+                            &Some(message.message_args),
+                            &mut wasm_bridge,
+                            &mut pipe_messages,
+                        );
+                    },
+                }
+                wasm_bridge.pipe_messages(pipe_messages, shutdown_send.clone())?;
+            },
+            PluginInstruction::UnblockCliPipes(pipes_to_unblock) => {
+                let pipes_to_unblock = wasm_bridge.update_cli_pipe_state(pipes_to_unblock);
+                for pipe_name in pipes_to_unblock {
+                    let _ = bus
+                        .senders
+                        .send_to_server(ServerInstruction::UnblockCliPipeInput(pipe_name))
+                        .context("failed to unblock input pipe");
+                }
+            },
             PluginInstruction::Exit => {
                 break;
             },
@@ -448,6 +625,82 @@ fn populate_session_layout_metadata(
     session_layout_metadata.update_plugin_cmds(plugin_ids_to_cmds);
 }
 
+fn pipe_to_all_plugins(
+    pipe_source: PipeSource,
+    name: &str,
+    payload: &Option<String>,
+    args: &Option<BTreeMap<String, String>>,
+    wasm_bridge: &mut WasmBridge,
+    pipe_messages: &mut Vec<(Option<PluginId>, Option<ClientId>, PipeMessage)>,
+) {
+    let is_private = false;
+    let all_plugin_ids = wasm_bridge.all_plugin_ids();
+    for (plugin_id, client_id) in all_plugin_ids {
+        pipe_messages.push((
+            Some(plugin_id),
+            Some(client_id),
+            PipeMessage::new(pipe_source.clone(), name, payload, &args, is_private),
+        ));
+    }
+}
+
+fn pipe_to_specific_plugins(
+    pipe_source: PipeSource,
+    plugin_url: &str,
+    configuration: &Option<BTreeMap<String, String>>,
+    cwd: &Option<PathBuf>,
+    skip_cache: bool,
+    should_float: bool,
+    pane_id_to_replace: &Option<PaneId>,
+    pane_title: &Option<String>,
+    cli_client_id: Option<ClientId>,
+    pipe_messages: &mut Vec<(Option<PluginId>, Option<ClientId>, PipeMessage)>,
+    name: &str,
+    payload: &Option<String>,
+    args: &Option<BTreeMap<String, String>>,
+    bus: &Bus<PluginInstruction>,
+    wasm_bridge: &mut WasmBridge,
+) {
+    let is_private = true;
+    let size = Size::default();
+    match RunPlugin::from_url(&plugin_url) {
+        Ok(mut run_plugin) => {
+            if let Some(configuration) = configuration {
+                run_plugin.configuration = PluginUserConfiguration::new(configuration.clone());
+            }
+            let all_plugin_ids = wasm_bridge.get_or_load_plugins(
+                run_plugin,
+                size,
+                cwd.clone(),
+                skip_cache,
+                should_float,
+                pane_id_to_replace.is_some(),
+                pane_title.clone(),
+                pane_id_to_replace.clone(),
+                cli_client_id,
+            );
+            for (plugin_id, client_id) in all_plugin_ids {
+                pipe_messages.push((
+                    Some(plugin_id),
+                    client_id,
+                    PipeMessage::new(pipe_source.clone(), name, payload, args, is_private), // PipeMessage::new(PipeSource::Cli(pipe_id.clone()), &name, &payload, &args, is_private)
+                ));
+            }
+        },
+        Err(e) => match cli_client_id {
+            Some(cli_client_id) => {
+                let _ = bus.senders.send_to_server(ServerInstruction::LogError(
+                    vec![format!("Failed to parse plugin url: {}", e)],
+                    cli_client_id,
+                ));
+            },
+            None => {
+                log::error!("Failed to parse plugin url: {}", e);
+            },
+        },
+    }
+}
+
 const EXIT_TIMEOUT: Duration = Duration::from_secs(3);
 
 #[path = "./unit/plugin_tests.rs"]
diff --git a/zellij-server/src/plugins/pipes.rs b/zellij-server/src/plugins/pipes.rs
new file mode 100644
index 00000000..ba3fe99b
--- /dev/null
+++ b/zellij-server/src/plugins/pipes.rs
@@ -0,0 +1,257 @@
+use super::{PluginId, PluginInstruction};
+use crate::plugins::plugin_map::RunningPlugin;
+use crate::plugins::wasm_bridge::PluginRenderAsset;
+use crate::plugins::zellij_exports::{wasi_read_string, wasi_write_object};
+use std::collections::{HashMap, HashSet};
+use wasmer::Value;
+use zellij_utils::data::{PipeMessage, PipeSource};
+use zellij_utils::plugin_api::pipe_message::ProtobufPipeMessage;
+
+use zellij_utils::errors::prelude::*;
+use zellij_utils::prost::Message;
+
+use crate::{thread_bus::ThreadSenders, ClientId};
+
+#[derive(Debug, Clone)]
+pub enum PipeStateChange {
+    NoChange,
+    Block,
+    Unblock,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct PendingPipes {
+    pipes: HashMap<String, PendingPipeInfo>,
+}
+
+impl PendingPipes {
+    pub fn mark_being_processed(
+        &mut self,
+        pipe_id: &str,
+        plugin_id: &PluginId,
+        client_id: &ClientId,
+    ) {
+        if self.pipes.contains_key(pipe_id) {
+            self.pipes.get_mut(pipe_id).map(|pending_pipe_info| {
+                pending_pipe_info.add_processing_plugin(plugin_id, client_id)
+            });
+        } else {
+            self.pipes.insert(
+                pipe_id.to_owned(),
+                PendingPipeInfo::new(plugin_id, client_id),
+            );
+        }
+    }
+    // returns a list of pipes that are no longer pending and should be unblocked
+    pub fn update_pipe_state_change(
+        &mut self,
+        cli_pipe_name: &str,
+        pipe_state_change: PipeStateChange,
+        plugin_id: &PluginId,
+        client_id: &ClientId,
+    ) -> Vec<String> {
+        let mut pipe_names_to_unblock = vec![];
+        match self.pipes.get_mut(cli_pipe_name) {
+            Some(pending_pipe_info) => {
+                let should_unblock_this_pipe =
+                    pending_pipe_info.update_state_change(pipe_state_change, plugin_id, client_id);
+                if should_unblock_this_pipe {
+                    pipe_names_to_unblock.push(cli_pipe_name.to_owned());
+                }
+            },
+            None => {
+                // state somehow corrupted, let's recover...
+                pipe_names_to_unblock.push(cli_pipe_name.to_owned());
+            },
+        }
+        for pipe_name in &pipe_names_to_unblock {
+            self.pipes.remove(pipe_name);
+        }
+        pipe_names_to_unblock
+    }
+    // returns a list of pipes that are no longer pending and should be unblocked
+    pub fn unload_plugin(&mut self, plugin_id: &PluginId) -> Vec<String> {
+        let mut pipe_names_to_unblock = vec![];
+        for (pipe_name, pending_pipe_info) in self.pipes.iter_mut() {
+            let should_unblock_this_pipe = pending_pipe_info.unload_plugin(plugin_id);
+            if should_unblock_this_pipe {
+                pipe_names_to_unblock.push(pipe_name.to_owned());
+            }
+        }
+        for pipe_name in &pipe_names_to_unblock {
+            self.pipes.remove(pipe_name);
+        }
+        pipe_names_to_unblock
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct PendingPipeInfo {
+    is_explicitly_blocked: bool,
+    currently_being_processed_by: HashSet<(PluginId, ClientId)>,
+}
+
+impl PendingPipeInfo {
+    pub fn new(plugin_id: &PluginId, client_id: &ClientId) -> Self {
+        let mut currently_being_processed_by = HashSet::new();
+        currently_being_processed_by.insert((*plugin_id, *client_id));
+        PendingPipeInfo {
+            currently_being_processed_by,
+            ..Default::default()
+        }
+    }
+    pub fn add_processing_plugin(&mut self, plugin_id: &PluginId, client_id: &ClientId) {
+        self.currently_being_processed_by
+            .insert((*plugin_id, *client_id));
+    }
+    // returns true if this pipe should be unblocked
+    pub fn update_state_change(
+        &mut self,
+        pipe_state_change: PipeStateChange,
+        plugin_id: &PluginId,
+        client_id: &ClientId,
+    ) -> bool {
+        match pipe_state_change {
+            PipeStateChange::Block => {
+                self.is_explicitly_blocked = true;
+            },
+            PipeStateChange::Unblock => {
+                self.is_explicitly_blocked = false;
+            },
+            _ => {},
+        };
+        self.currently_being_processed_by
+            .remove(&(*plugin_id, *client_id));
+        let pipe_should_be_unblocked =
+            self.currently_being_processed_by.is_empty() && !self.is_explicitly_blocked;
+        pipe_should_be_unblocked
+    }
+    // returns true if this pipe should be unblocked
+    pub fn unload_plugin(&mut self, plugin_id_to_unload: &PluginId) -> bool {
+        self.currently_being_processed_by
+            .retain(|(plugin_id, _)| plugin_id != plugin_id_to_unload);
+        if self.currently_being_processed_by.is_empty() && !self.is_explicitly_blocked {
+            true
+        } else {
+            false
+        }
+    }
+}
+
+pub fn apply_pipe_message_to_plugin(
+    plugin_id: PluginId,
+    client_id: ClientId,
+    running_plugin: &mut RunningPlugin,
+    pipe_message: &PipeMessage,
+    plugin_render_assets: &mut Vec<PluginRenderAsset>,
+    senders: &ThreadSenders,
+) -> Result<()> {
+    let instance = &running_plugin.instance;
+    let plugin_env = &running_plugin.plugin_env;
+    let rows = running_plugin.rows;
+    let columns = running_plugin.columns;
+
+    let err_context = || format!("Failed to apply event to plugin {plugin_id}");
+    let protobuf_pipe_message: ProtobufPipeMessage = pipe_message
+        .clone()
+        .try_into()
+        .map_err(|e| anyhow!("Failed to convert to protobuf: {:?}", e))?;
+    match instance.exports.get_function("pipe") {
+        Ok(pipe) => {
+            wasi_write_object(&plugin_env.wasi_env, &protobuf_pipe_message.encode_to_vec())
+                .with_context(err_context)?;
+            let pipe_return = pipe
+                .call(&mut running_plugin.store, &[])
+                .with_context(err_context)?;
+            let should_render = match pipe_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(
+                                &mut running_plugin.store,
+                                &[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)?;
+                let pipes_to_block_or_unblock =
+                    pipes_to_block_or_unblock(running_plugin, Some(&pipe_message.source));
+                let plugin_render_asset = PluginRenderAsset::new(
+                    plugin_id,
+                    client_id,
+                    rendered_bytes.as_bytes().to_vec(),
+                )
+                .with_pipes(pipes_to_block_or_unblock);
+                plugin_render_assets.push(plugin_render_asset);
+            } else {
+                let pipes_to_block_or_unblock =
+                    pipes_to_block_or_unblock(running_plugin, Some(&pipe_message.source));
+                let plugin_render_asset = PluginRenderAsset::new(plugin_id, client_id, vec![])
+                    .with_pipes(pipes_to_block_or_unblock);
+                let _ = senders
+                    .send_to_plugin(PluginInstruction::UnblockCliPipes(vec![
+                        plugin_render_asset,
+                    ]))
+                    .context("failed to unblock input pipe");
+            }
+        },
+        Err(_e) => {
+            // no-op, this is probably an old plugin that does not have this interface
+            // we don't log this error because if we do the logs will be super crowded
+            let pipes_to_block_or_unblock =
+                pipes_to_block_or_unblock(running_plugin, Some(&pipe_message.source));
+            let plugin_render_asset = PluginRenderAsset::new(
+                plugin_id,
+                client_id,
+                vec![], // nothing to render
+            )
+            .with_pipes(pipes_to_block_or_unblock);
+            let _ = senders
+                .send_to_plugin(PluginInstruction::UnblockCliPipes(vec![
+                    plugin_render_asset,
+                ]))
+                .context("failed to unblock input pipe");
+        },
+    }
+    Ok(())
+}
+
+pub fn pipes_to_block_or_unblock(
+    running_plugin: &mut RunningPlugin,
+    current_pipe: Option<&PipeSource>,
+) -> HashMap<String, PipeStateChange> {
+    let mut pipe_state_changes = HashMap::new();
+    let mut input_pipes_to_unblock: HashSet<String> = running_plugin
+        .plugin_env
+        .input_pipes_to_unblock
+        .lock()
+        .unwrap()
+        .drain()
+        .collect();
+    let mut input_pipes_to_block: HashSet<String> = running_plugin
+        .plugin_env
+        .input_pipes_to_block
+        .lock()
+        .unwrap()
+        .drain()
+        .collect();
+    if let Some(PipeSource::Cli(current_pipe)) = current_pipe {
+        pipe_state_changes.insert(current_pipe.to_owned(), PipeStateChange::NoChange);
+    }
+    for pipe in input_pipes_to_block.drain() {
+        pipe_state_changes.insert(pipe, PipeStateChange::Block);
+    }
+    for pipe in input_pipes_to_unblock.drain() {
+        // unblock has priority over block if they happened simultaneously
+        pipe_state_changes.insert(pipe, PipeStateChange::Unblock);
+    }
+    pipe_state_changes
+}
diff --git a/zellij-server/src/plugins/plugin_loader.rs b/zellij-server/src/plugins/plugin_loader.rs
index 1f001db1..c14a324b 100644
--- a/zellij-server/src/plugins/plugin_loader.rs
+++ b/zellij-server/src/plugins/plugin_loader.rs
@@ -56,7 +56,7 @@ pub struct PluginLoader<'a> {
     store: Arc<Mutex<Store>>,
     plugin: PluginConfig,
     plugin_dir: &'a PathBuf,
-    tab_index: usize,
+    tab_index: Option<usize>,
     plugin_own_data_dir: PathBuf,
     size: Size,
     wasm_blob_on_hd: Option<(Vec<u8>, PathBuf)>,
@@ -133,7 +133,7 @@ impl<'a> PluginLoader<'a> {
         plugin_id: PluginId,
         client_id: ClientId,
         plugin: &PluginConfig,
-        tab_index: usize,
+        tab_index: Option<usize>,
         plugin_dir: PathBuf,
         plugin_cache: Arc<Mutex<HashMap<PathBuf, Module>>>,
         senders: ThreadSenders,
@@ -339,7 +339,7 @@ impl<'a> PluginLoader<'a> {
         store: Arc<Mutex<Store>>,
         plugin: PluginConfig,
         plugin_dir: &'a PathBuf,
-        tab_index: usize,
+        tab_index: Option<usize>,
         size: Size,
         path_to_default_shell: PathBuf,
         zellij_cwd: PathBuf,
@@ -814,7 +814,9 @@ impl<'a> PluginLoader<'a> {
             .import_object(store_mut, &module)
             .with_context(err_context)?;
         let mut mut_plugin = self.plugin.clone();
-        mut_plugin.set_tab_index(self.tab_index);
+        if let Some(tab_index) = self.tab_index {
+            mut_plugin.set_tab_index(tab_index);
+        }
         let plugin_env = PluginEnv {
             plugin_id: self.plugin_id,
             client_id: self.client_id,
@@ -830,6 +832,8 @@ impl<'a> PluginLoader<'a> {
             default_shell: self.default_shell.clone(),
             default_layout: self.default_layout.clone(),
             plugin_cwd: self.zellij_cwd.clone(),
+            input_pipes_to_unblock: Arc::new(Mutex::new(HashSet::new())),
+            input_pipes_to_block: Arc::new(Mutex::new(HashSet::new())),
         };
 
         let subscriptions = Arc::new(Mutex::new(HashSet::new()));
diff --git a/zellij-server/src/plugins/plugin_map.rs b/zellij-server/src/plugins/plugin_map.rs
index 067c178f..e03cf172 100644
--- a/zellij-server/src/plugins/plugin_map.rs
+++ b/zellij-server/src/plugins/plugin_map.rs
@@ -15,7 +15,7 @@ use zellij_utils::{
     data::EventType,
     data::PluginCapabilities,
     input::command::TerminalAction,
-    input::layout::{Layout, RunPlugin, RunPluginLocation},
+    input::layout::{Layout, PluginUserConfiguration, RunPlugin, RunPluginLocation},
     input::plugins::PluginConfig,
     ipc::ClientAttributes,
 };
@@ -157,13 +157,19 @@ impl PluginMap {
     pub fn all_plugin_ids_for_plugin_location(
         &self,
         plugin_location: &RunPluginLocation,
+        plugin_configuration: &PluginUserConfiguration,
     ) -> Result<Vec<PluginId>> {
         let err_context = || format!("Failed to get plugin ids for location {plugin_location}");
         let plugin_ids: Vec<PluginId> = self
             .plugin_assets
             .iter()
             .filter(|(_, (running_plugin, _subscriptions, _workers))| {
-                &running_plugin.lock().unwrap().plugin_env.plugin.location == plugin_location
+                let running_plugin = running_plugin.lock().unwrap();
+                let running_plugin_location = &running_plugin.plugin_env.plugin.location;
+                let running_plugin_configuration =
+                    &running_plugin.plugin_env.plugin.userspace_configuration;
+                running_plugin_location == plugin_location
+                    && running_plugin_configuration == plugin_configuration
             })
             .map(|((plugin_id, _client_id), _)| *plugin_id)
             .collect();
@@ -172,6 +178,49 @@ impl PluginMap {
         }
         Ok(plugin_ids)
     }
+    pub fn clone_plugin_assets(
+        &self,
+    ) -> HashMap<RunPluginLocation, HashMap<PluginUserConfiguration, Vec<(PluginId, ClientId)>>>
+    {
+        let mut cloned_plugin_assets: HashMap<
+            RunPluginLocation,
+            HashMap<PluginUserConfiguration, Vec<(PluginId, ClientId)>>,
+        > = HashMap::new();
+        for ((plugin_id, client_id), (running_plugin, _, _)) in self.plugin_assets.iter() {
+            let running_plugin = running_plugin.lock().unwrap();
+            let running_plugin_location = &running_plugin.plugin_env.plugin.location;
+            let running_plugin_configuration =
+                &running_plugin.plugin_env.plugin.userspace_configuration;
+            match cloned_plugin_assets.get_mut(running_plugin_location) {
+                Some(location_map) => match location_map.get_mut(running_plugin_configuration) {
+                    Some(plugin_instances_info) => {
+                        plugin_instances_info.push((*plugin_id, *client_id));
+                    },
+                    None => {
+                        location_map.insert(
+                            running_plugin_configuration.clone(),
+                            vec![(*plugin_id, *client_id)],
+                        );
+                    },
+                },
+                None => {
+                    let mut location_map = HashMap::new();
+                    location_map.insert(
+                        running_plugin_configuration.clone(),
+                        vec![(*plugin_id, *client_id)],
+                    );
+                    cloned_plugin_assets.insert(running_plugin_location.clone(), location_map);
+                },
+            }
+        }
+        cloned_plugin_assets
+    }
+    pub fn all_plugin_ids(&self) -> Vec<(PluginId, ClientId)> {
+        self.plugin_assets
+            .iter()
+            .map(|((plugin_id, client_id), _)| (*plugin_id, *client_id))
+            .collect()
+    }
     pub fn insert(
         &mut self,
         plugin_id: PluginId,
@@ -218,7 +267,7 @@ pub struct PluginEnv {
     pub permissions: Arc<Mutex<Option<HashSet<PermissionType>>>>,
     pub senders: ThreadSenders,
     pub wasi_env: WasiEnv,
-    pub tab_index: usize,
+    pub tab_index: Option<usize>,
     pub client_id: ClientId,
     #[allow(dead_code)]
     pub plugin_own_data_dir: PathBuf,
@@ -228,6 +277,8 @@ pub struct PluginEnv {
     pub default_shell: Option<TerminalAction>,
     pub default_layout: Box<Layout>,
     pub plugin_cwd: PathBuf,
+    pub input_pipes_to_unblock: Arc<Mutex<HashSet<String>>>,
+    pub input_pipes_to_block: Arc<Mutex<HashSet<String>>>,
 }
 
 impl PluginEnv {
diff --git a/zellij-server/src/plugins/unit/plugin_tests.rs b/zellij-server/src/plugins/unit/plugin_tests.rs
index ca843358..6791213c 100644
--- a/zellij-server/src/plugins/unit/plugin_tests.rs
+++ b/zellij-server/src/plugins/unit/plugin_tests.rs
@@ -204,35 +204,6 @@ macro_rules! grant_permissions_and_log_actions_in_thread_naked_variant {
     };
 }
 
-macro_rules! log_actions_in_thread_naked_variant {
-    ( $arc_mutex_log:expr, $exit_event:path, $receiver:expr, $exit_after_count:expr ) => {
-        std::thread::Builder::new()
-            .name("logger thread".to_string())
-            .spawn({
-                let log = $arc_mutex_log.clone();
-                let mut exit_event_count = 0;
-                move || loop {
-                    let (event, _err_ctx) = $receiver
-                        .recv()
-                        .expect("failed to receive event on channel");
-                    match event {
-                        $exit_event => {
-                            exit_event_count += 1;
-                            log.lock().unwrap().push(event);
-                            if exit_event_count == $exit_after_count {
-                                break;
-                            }
-                        },
-                        _ => {
-                            log.lock().unwrap().push(event);
-                        },
-                    }
-                }
-            })
-            .unwrap()
-    };
-}
-
 fn create_plugin_thread(
     zellij_cwd: Option<PathBuf>,
 ) -> (
@@ -372,7 +343,7 @@ fn create_plugin_thread_with_server_receiver(
                 client_attributes,
                 default_shell_action,
             )
-            .expect("TEST")
+            .expect("TEST");
         })
         .unwrap();
     let teardown = {
@@ -599,7 +570,7 @@ pub fn load_new_plugin_from_hd() {
         received_screen_instructions,
         ScreenInstruction::PluginBytes,
         screen_receiver,
-        2,
+        1,
         &PermissionType::ChangeApplicationState,
         cache_path,
         plugin_thread_sender,
@@ -632,11 +603,14 @@ pub fn load_new_plugin_from_hd() {
         .unwrap()
         .iter()
         .find_map(|i| {
-            if let ScreenInstruction::PluginBytes(plugin_bytes) = i {
-                for (plugin_id, client_id, plugin_bytes) in plugin_bytes {
-                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes).to_string();
+            if let ScreenInstruction::PluginBytes(plugin_render_assets) = i {
+                for plugin_render_asset in plugin_render_assets {
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let plugin_bytes = plugin_render_asset.bytes.clone();
+                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes.as_slice()).to_string();
                     if plugin_bytes.contains("InputReceived") {
-                        return Some((*plugin_id, *client_id, plugin_bytes));
+                        return Some((plugin_id, client_id, plugin_bytes));
                     }
                 }
             }
@@ -671,7 +645,7 @@ pub fn plugin_workers() {
         received_screen_instructions,
         ScreenInstruction::PluginBytes,
         screen_receiver,
-        3,
+        2,
         &PermissionType::ChangeApplicationState,
         cache_path,
         plugin_thread_sender,
@@ -708,11 +682,14 @@ pub fn plugin_workers() {
         .unwrap()
         .iter()
         .find_map(|i| {
-            if let ScreenInstruction::PluginBytes(plugin_bytes) = i {
-                for (plugin_id, client_id, plugin_bytes) in plugin_bytes {
-                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes).to_string();
+            if let ScreenInstruction::PluginBytes(plugin_render_assets) = i {
+                for plugin_render_asset in plugin_render_assets {
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let plugin_bytes = plugin_render_asset.bytes.clone();
+                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes.as_slice()).to_string();
                     if plugin_bytes.contains("Payload from worker") {
-                        return Some((*plugin_id, *client_id, plugin_bytes));
+                        return Some((plugin_id, client_id, plugin_bytes));
                     }
                 }
             }
@@ -747,7 +724,7 @@ pub fn plugin_workers_persist_state() {
         received_screen_instructions,
         ScreenInstruction::PluginBytes,
         screen_receiver,
-        5,
+        4,
         &PermissionType::ChangeApplicationState,
         cache_path,
         plugin_thread_sender,
@@ -774,12 +751,13 @@ pub fn plugin_workers_persist_state() {
     // we do this a second time so that the worker will log the first message on its own state and
     // then send us the "received 2 messages" indication we check for below, letting us know it
     // managed to persist its own state and act upon it
-    std::thread::sleep(std::time::Duration::from_millis(500));
+    //std::thread::sleep(std::time::Duration::from_millis(500));
     let _ = plugin_thread_sender.send(PluginInstruction::Update(vec![(
         None,
         Some(client_id),
         Event::SystemClipboardFailure,
     )]));
+    std::thread::sleep(std::time::Duration::from_millis(500));
     let _ = plugin_thread_sender.send(PluginInstruction::Update(vec![(
         None,
         Some(client_id),
@@ -792,11 +770,14 @@ pub fn plugin_workers_persist_state() {
         .unwrap()
         .iter()
         .find_map(|i| {
-            if let ScreenInstruction::PluginBytes(plugin_bytes) = i {
-                for (plugin_id, client_id, plugin_bytes) in plugin_bytes {
-                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes).to_string();
+            if let ScreenInstruction::PluginBytes(plugin_render_assets) = i {
+                for plugin_render_asset in plugin_render_assets {
+                    let plugin_bytes = plugin_render_asset.bytes.clone();
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes.as_slice()).to_string();
                     if plugin_bytes.contains("received 2 messages") {
-                        return Some((*plugin_id, *client_id, plugin_bytes));
+                        return Some((plugin_id, client_id, plugin_bytes));
                     }
                 }
             }
@@ -811,6 +792,7 @@ pub fn can_subscribe_to_hd_events() {
     let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
                                           // destructor removes the directory
     let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
     let (plugin_thread_sender, screen_receiver, teardown) =
         create_plugin_thread(Some(plugin_host_folder));
     let plugin_should_float = Some(false);
@@ -827,11 +809,15 @@ pub fn can_subscribe_to_hd_events() {
         rows: 20,
     };
     let received_screen_instructions = Arc::new(Mutex::new(vec![]));
-    let screen_thread = log_actions_in_thread!(
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
         received_screen_instructions,
         ScreenInstruction::PluginBytes,
         screen_receiver,
-        2
+        2,
+        &PermissionType::ChangeApplicationState,
+        cache_path,
+        plugin_thread_sender,
+        client_id
     );
 
     let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
@@ -861,11 +847,14 @@ pub fn can_subscribe_to_hd_events() {
         .unwrap()
         .iter()
         .find_map(|i| {
-            if let ScreenInstruction::PluginBytes(plugin_bytes) = i {
-                for (plugin_id, client_id, plugin_bytes) in plugin_bytes {
-                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes).to_string();
+            if let ScreenInstruction::PluginBytes(plugin_render_assets) = i {
+                for plugin_render_asset in plugin_render_assets {
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let plugin_bytes = plugin_render_asset.bytes.clone();
+                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes.as_slice()).to_string();
                     if plugin_bytes.contains("FileSystemCreate") {
-                        return Some((*plugin_id, *client_id, plugin_bytes));
+                        return Some((plugin_id, client_id, plugin_bytes));
                     }
                 }
             }
@@ -4471,6 +4460,7 @@ pub fn hide_self_plugin_command() {
     let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
                                           // destructor removes the directory
     let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
     let (plugin_thread_sender, screen_receiver, teardown) =
         create_plugin_thread(Some(plugin_host_folder));
     let plugin_should_float = Some(false);
@@ -4487,11 +4477,15 @@ pub fn hide_self_plugin_command() {
         rows: 20,
     };
     let received_screen_instructions = Arc::new(Mutex::new(vec![]));
-    let screen_thread = log_actions_in_thread!(
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
         received_screen_instructions,
         ScreenInstruction::SuppressPane,
         screen_receiver,
-        1
+        1,
+        &PermissionType::ChangeApplicationState,
+        cache_path,
+        plugin_thread_sender,
+        client_id
     );
 
     let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
@@ -4536,6 +4530,7 @@ pub fn show_self_plugin_command() {
     let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
                                           // destructor removes the directory
     let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
     let (plugin_thread_sender, screen_receiver, teardown) =
         create_plugin_thread(Some(plugin_host_folder));
     let plugin_should_float = Some(false);
@@ -4552,13 +4547,16 @@ pub fn show_self_plugin_command() {
         rows: 20,
     };
     let received_screen_instructions = Arc::new(Mutex::new(vec![]));
-    let screen_thread = log_actions_in_thread!(
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
         received_screen_instructions,
         ScreenInstruction::FocusPaneWithId,
         screen_receiver,
-        1
+        1,
+        &PermissionType::ChangeApplicationState,
+        cache_path,
+        plugin_thread_sender,
+        client_id
     );
-
     let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
     let _ = plugin_thread_sender.send(PluginInstruction::Load(
         plugin_should_float,
@@ -5634,3 +5632,341 @@ pub fn web_request_plugin_command() {
         .clone();
     assert_snapshot!(format!("{:#?}", new_tab_event));
 }
+
+#[test]
+#[ignore]
+pub fn unblock_input_plugin_command() {
+    let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
+                                          // destructor removes the directory
+    let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
+    let (plugin_thread_sender, screen_receiver, teardown) =
+        create_plugin_thread(Some(plugin_host_folder));
+    let plugin_should_float = Some(false);
+    let plugin_title = Some("test_plugin".to_owned());
+    let run_plugin = RunPlugin {
+        _allow_exec_host_cmd: false,
+        location: RunPluginLocation::File(PathBuf::from(&*PLUGIN_FIXTURE)),
+        configuration: Default::default(),
+    };
+    let tab_index = 1;
+    let client_id = 1;
+    let size = Size {
+        cols: 121,
+        rows: 20,
+    };
+    let received_screen_instructions = Arc::new(Mutex::new(vec![]));
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
+        received_screen_instructions,
+        ScreenInstruction::PluginBytes,
+        screen_receiver,
+        1,
+        &PermissionType::ReadCliPipes,
+        cache_path,
+        plugin_thread_sender,
+        client_id
+    );
+
+    let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
+    let _ = plugin_thread_sender.send(PluginInstruction::Load(
+        plugin_should_float,
+        false,
+        plugin_title,
+        run_plugin,
+        tab_index,
+        None,
+        client_id,
+        size,
+        None,
+        false,
+    ));
+    std::thread::sleep(std::time::Duration::from_millis(500));
+
+    let _ = plugin_thread_sender.send(PluginInstruction::CliPipe {
+        pipe_id: "input_pipe_id".to_owned(),
+        name: "message_name".to_owned(),
+        payload: Some("message_payload".to_owned()),
+        plugin: None, // broadcast
+        args: None,
+        configuration: None,
+        floating: None,
+        pane_id_to_replace: None,
+        pane_title: None,
+        cwd: None,
+        skip_cache: false,
+        cli_client_id: client_id,
+    });
+    screen_thread.join().unwrap(); // this might take a while if the cache is cold
+    teardown();
+    let plugin_bytes_events = received_screen_instructions
+        .lock()
+        .unwrap()
+        .iter()
+        .rev()
+        .find_map(|i| {
+            if let ScreenInstruction::PluginBytes(..) = i {
+                Some(i.clone())
+            } else {
+                None
+            }
+        })
+        .clone();
+    assert_snapshot!(format!("{:#?}", plugin_bytes_events));
+}
+
+#[test]
+#[ignore]
+pub fn block_input_plugin_command() {
+    let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
+                                          // destructor removes the directory
+    let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
+    let (plugin_thread_sender, screen_receiver, teardown) =
+        create_plugin_thread(Some(plugin_host_folder));
+    let plugin_should_float = Some(false);
+    let plugin_title = Some("test_plugin".to_owned());
+    let run_plugin = RunPlugin {
+        _allow_exec_host_cmd: false,
+        location: RunPluginLocation::File(PathBuf::from(&*PLUGIN_FIXTURE)),
+        configuration: Default::default(),
+    };
+    let tab_index = 1;
+    let client_id = 1;
+    let size = Size {
+        cols: 121,
+        rows: 20,
+    };
+    let received_screen_instructions = Arc::new(Mutex::new(vec![]));
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
+        received_screen_instructions,
+        ScreenInstruction::PluginBytes,
+        screen_receiver,
+        1,
+        &PermissionType::ReadCliPipes,
+        cache_path,
+        plugin_thread_sender,
+        client_id
+    );
+
+    let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
+    let _ = plugin_thread_sender.send(PluginInstruction::Load(
+        plugin_should_float,
+        false,
+        plugin_title,
+        run_plugin,
+        tab_index,
+        None,
+        client_id,
+        size,
+        None,
+        false,
+    ));
+    // extra long time because we only start the fs watcher on plugin load
+    std::thread::sleep(std::time::Duration::from_millis(5000));
+
+    let _ = plugin_thread_sender.send(PluginInstruction::CliPipe {
+        pipe_id: "input_pipe_id".to_owned(),
+        name: "message_name_block".to_owned(),
+        payload: Some("message_payload".to_owned()),
+        plugin: None, // broadcast
+        args: None,
+        configuration: None,
+        floating: None,
+        pane_id_to_replace: None,
+        pane_title: None,
+        cwd: None,
+        skip_cache: false,
+        cli_client_id: client_id,
+    });
+    screen_thread.join().unwrap(); // this might take a while if the cache is cold
+    teardown();
+    let plugin_bytes_events = received_screen_instructions
+        .lock()
+        .unwrap()
+        .iter()
+        .rev()
+        .find_map(|i| {
+            if let ScreenInstruction::PluginBytes(..) = i {
+                Some(i.clone())
+            } else {
+                None
+            }
+        })
+        .clone();
+    assert_snapshot!(format!("{:#?}", plugin_bytes_events));
+}
+
+#[test]
+#[ignore]
+pub fn pipe_output_plugin_command() {
+    let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
+                                          // destructor removes the directory
+    let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
+    let (plugin_thread_sender, server_receiver, screen_receiver, teardown) =
+        create_plugin_thread_with_server_receiver(Some(plugin_host_folder));
+    let plugin_should_float = Some(false);
+    let plugin_title = Some("test_plugin".to_owned());
+    let run_plugin = RunPlugin {
+        _allow_exec_host_cmd: false,
+        location: RunPluginLocation::File(PathBuf::from(&*PLUGIN_FIXTURE)),
+        configuration: Default::default(),
+    };
+    let tab_index = 1;
+    let client_id = 1;
+    let size = Size {
+        cols: 121,
+        rows: 20,
+    };
+    let received_screen_instructions = Arc::new(Mutex::new(vec![]));
+    let _screen_thread = grant_permissions_and_log_actions_in_thread_naked_variant!(
+        received_screen_instructions,
+        ScreenInstruction::Exit,
+        screen_receiver,
+        1,
+        &PermissionType::ChangeApplicationState,
+        cache_path,
+        plugin_thread_sender,
+        client_id
+    );
+    let received_server_instruction = Arc::new(Mutex::new(vec![]));
+    let server_thread = log_actions_in_thread!(
+        received_server_instruction,
+        ServerInstruction::CliPipeOutput,
+        server_receiver,
+        1
+    );
+
+    let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
+    let _ = plugin_thread_sender.send(PluginInstruction::Load(
+        plugin_should_float,
+        false,
+        plugin_title,
+        run_plugin,
+        tab_index,
+        None,
+        client_id,
+        size,
+        None,
+        false,
+    ));
+    std::thread::sleep(std::time::Duration::from_millis(500));
+
+    let _ = plugin_thread_sender.send(PluginInstruction::CliPipe {
+        pipe_id: "input_pipe_id".to_owned(),
+        name: "pipe_output".to_owned(),
+        payload: Some("message_payload".to_owned()),
+        plugin: None, // broadcast
+        args: None,
+        configuration: None,
+        floating: None,
+        pane_id_to_replace: None,
+        pane_title: None,
+        cwd: None,
+        skip_cache: false,
+        cli_client_id: client_id,
+    });
+    std::thread::sleep(std::time::Duration::from_millis(500));
+    teardown();
+    server_thread.join().unwrap(); // this might take a while if the cache is cold
+    let plugin_bytes_events = received_server_instruction
+        .lock()
+        .unwrap()
+        .iter()
+        .rev()
+        .find_map(|i| {
+            if let ServerInstruction::CliPipeOutput(..) = i {
+                Some(i.clone())
+            } else {
+                None
+            }
+        })
+        .clone();
+    assert_snapshot!(format!("{:#?}", plugin_bytes_events));
+}
+
+#[test]
+#[ignore]
+pub fn pipe_message_to_plugin_plugin_command() {
+    let temp_folder = tempdir().unwrap(); // placed explicitly in the test scope because its
+                                          // destructor removes the directory
+    let plugin_host_folder = PathBuf::from(temp_folder.path());
+    let cache_path = plugin_host_folder.join("permissions_test.kdl");
+    let (plugin_thread_sender, screen_receiver, teardown) =
+        create_plugin_thread(Some(plugin_host_folder));
+    let plugin_should_float = Some(false);
+    let plugin_title = Some("test_plugin".to_owned());
+    let run_plugin = RunPlugin {
+        _allow_exec_host_cmd: false,
+        location: RunPluginLocation::File(PathBuf::from(&*PLUGIN_FIXTURE)),
+        configuration: Default::default(),
+    };
+    let tab_index = 1;
+    let client_id = 1;
+    let size = Size {
+        cols: 121,
+        rows: 20,
+    };
+    let received_screen_instructions = Arc::new(Mutex::new(vec![]));
+    let screen_thread = grant_permissions_and_log_actions_in_thread!(
+        received_screen_instructions,
+        ScreenInstruction::PluginBytes,
+        screen_receiver,
+        2,
+        &PermissionType::ReadCliPipes,
+        cache_path,
+        plugin_thread_sender,
+        client_id
+    );
+
+    let _ = plugin_thread_sender.send(PluginInstruction::AddClient(client_id));
+    let _ = plugin_thread_sender.send(PluginInstruction::Load(
+        plugin_should_float,
+        false,
+        plugin_title,
+        run_plugin,
+        tab_index,
+        None,
+        client_id,
+        size,
+        None,
+        false,
+    ));
+    std::thread::sleep(std::time::Duration::from_millis(500));
+    let _ = plugin_thread_sender.send(PluginInstruction::CliPipe {
+        pipe_id: "input_pipe_id".to_owned(),
+        name: "pipe_message_to_plugin".to_owned(),
+        payload: Some("payload_sent_to_self".to_owned()),
+        plugin: None, // broadcast
+        args: None,
+        configuration: None,
+        floating: None,
+        pane_id_to_replace: None,
+        pane_title: None,
+        cwd: None,
+        skip_cache: false,
+        cli_client_id: client_id,
+    });
+    std::thread::sleep(std::time::Duration::from_millis(500));
+    teardown();
+    screen_thread.join().unwrap(); // this might take a while if the cache is cold
+    let plugin_bytes_event = received_screen_instructions
+        .lock()
+        .unwrap()
+        .iter()
+        .find_map(|i| {
+            if let ScreenInstruction::PluginBytes(plugin_render_assets) = i {
+                for plugin_render_asset in plugin_render_assets {
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let plugin_bytes = plugin_render_asset.bytes.clone();
+                    let plugin_bytes = String::from_utf8_lossy(plugin_bytes.as_slice()).to_string();
+                    if plugin_bytes.contains("Payload from self:") {
+                        return Some((plugin_id, client_id, plugin_bytes));
+                    }
+                }
+            }
+            None
+        });
+    assert_snapshot!(format!("{:#?}", plugin_bytes_event));
+}
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__block_input_plugin_command.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__block_input_plugin_command.snap
new file mode 100644
index 00000000..2034c6a0
--- /dev/null
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__block_input_plugin_command.snap
@@ -0,0 +1,62 @@
+---
+source: zellij-server/src/plugins/./unit/plugin_tests.rs
+assertion_line: 5812
+expression: "format!(\"{:#?}\", plugin_bytes_events)"
+---
+Some(
+    PluginBytes(
+        [
+            PluginRenderAsset {
+                client_id: 1,
+                plugin_id: 0,
+                bytes: [
+                    82,
+                    111,
+                    119,
+                    115,
+                    58,
+                    32,
+                    50,
+                    48,
+                    44,
+                    32,
+                    67,
+                    111,
+                    108,
+                    115,
+                    58,
+                    32,
+                    49,
+                    50,
+                    49,
+                    44,
+                    32,
+                    82,
+                    101,
+                    99,
+                    101,
+                    105,
+                    118,
+                    101,
+                    100,
+                    32,
+                    101,
+                    118,
+                    101,
+                    110,
+                    116,
+                    115,
+                    58,
+                    32,
+                    91,
+                    93,
+                    10,
+                    13,
+                ],
+                cli_pipes: {
+                    "input_pipe_id": Block,
+                },
+            },
+        ],
+    ),
+)
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__granted_permission_request_result.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__granted_permission_request_result.snap
index cc51bef2..9c3ecd06 100644
--- a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__granted_permission_request_result.snap
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__granted_permission_request_result.snap
@@ -1,6 +1,6 @@
 ---
 source: zellij-server/src/plugins/./unit/plugin_tests.rs
-assertion_line: 5189
+assertion_line: 5307
 expression: "format!(\"{:#?}\", permissions)"
 ---
 Some(
@@ -12,5 +12,7 @@ Some(
         OpenTerminalsOrPlugins,
         WriteToStdin,
         WebAccess,
+        ReadCliPipes,
+        MessageAndLaunchOtherPlugins,
     ],
 )
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_message_to_plugin_plugin_command.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_message_to_plugin_plugin_command.snap
new file mode 100644
index 00000000..b363559b
--- /dev/null
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_message_to_plugin_plugin_command.snap
@@ -0,0 +1,12 @@
+---
+source: zellij-server/src/plugins/./unit/plugin_tests.rs
+assertion_line: 5961
+expression: "format!(\"{:#?}\", plugin_bytes_event)"
+---
+Some(
+    (
+        0,
+        1,
+        "Payload from self: \"my_cool_payload\"\n\r",
+    ),
+)
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_output_plugin_command.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_output_plugin_command.snap
new file mode 100644
index 00000000..a7c24c9f
--- /dev/null
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__pipe_output_plugin_command.snap
@@ -0,0 +1,11 @@
+---
+source: zellij-server/src/plugins/./unit/plugin_tests.rs
+assertion_line: 5771
+expression: "format!(\"{:#?}\", plugin_bytes_events)"
+---
+Some(
+    CliPipeOutput(
+        "pipe_output",
+        "this_is_my_output",
+    ),
+)
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__request_plugin_permissions.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__request_plugin_permissions.snap
index d9a92db5..3d9f7cfe 100644
--- a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__request_plugin_permissions.snap
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__request_plugin_permissions.snap
@@ -1,6 +1,6 @@
 ---
 source: zellij-server/src/plugins/./unit/plugin_tests.rs
-assertion_line: 5101
+assertion_line: 5217
 expression: "format!(\"{:#?}\", new_tab_event)"
 ---
 Some(
@@ -14,5 +14,7 @@ Some(
         OpenTerminalsOrPlugins,
         WriteToStdin,
         WebAccess,
+        ReadCliPipes,
+        MessageAndLaunchOtherPlugins,
     ],
 )
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__send_message_to_plugin_plugin_command.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__send_message_to_plugin_plugin_command.snap
new file mode 100644
index 00000000..46c9617a
--- /dev/null
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__send_message_to_plugin_plugin_command.snap
@@ -0,0 +1,12 @@
+---
+source: zellij-server/src/plugins/./unit/plugin_tests.rs
+assertion_line: 5856
+expression: "format!(\"{:#?}\", plugin_bytes_event)"
+---
+Some(
+    (
+        0,
+        1,
+        "Payload from self: \"my_cool_payload\"\n\r",
+    ),
+)
diff --git a/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__unblock_input_plugin_command.snap b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__unblock_input_plugin_command.snap
new file mode 100644
index 00000000..ce6f0344
--- /dev/null
+++ b/zellij-server/src/plugins/unit/snapshots/zellij_server__plugins__plugin_tests__unblock_input_plugin_command.snap
@@ -0,0 +1,62 @@
+---
+source: zellij-server/src/plugins/./unit/plugin_tests.rs
+assertion_line: 5730
+expression: "format!(\"{:#?}\", plugin_bytes_events)"
+---
+Some(
+    PluginBytes(
+        [
+            PluginRenderAsset {
+                client_id: 1,
+                plugin_id: 0,
+                bytes: [
+                    82,
+                    111,
+                    119,
+                    115,
+                    58,
+                    32,
+                    50,
+                    48,
+                    44,
+                    32,
+                    67,
+                    111,
+                    108,
+                    115,
+                    58,
+                    32,
+                    49,
+                    50,
+                    49,
+                    44,
+                    32,
+                    82,
+                    101,
+                    99,
+                    101,
+                    105,
+                    118,
+                    101,
+                    100,
+                    32,
+                    101,
+                    118,
+                    101,
+                    110,
+                    116,
+                    115,
+                    58,
+                    32,
+                    91,
+                    93,
+                    10,
+                    13,
+                ],
+                cli_pipes: {
+                    "input_pipe_id": Unblock,
+                },
+            },
+        ],
+    ),
+)
diff --git a/zellij-server/src/plugins/wasm_bridge.rs b/zellij-server/src/plugins/wasm_bridge.rs
index 7851e870..f4679b5e 100644
--- a/zellij-server/src/plugins/wasm_bridge.rs
+++ b/zellij-server/src/plugins/wasm_bridge.rs
@@ -1,4 +1,7 @@
 use super::{PluginId, PluginInstruction};
+use crate::plugins::pipes::{
+    apply_pipe_message_to_plugin, pipes_to_block_or_unblock, PendingPipes, PipeStateChange,
+};
 use crate::plugins::plugin_loader::PluginLoader;
 use crate::plugins::plugin_map::{AtomicEvent, PluginEnv, PluginMap, RunningPlugin, Subscriptions};
 use crate::plugins::plugin_worker::MessageToWorker;
@@ -16,7 +19,7 @@ use wasmer::{Module, Store, Value};
 use zellij_utils::async_channel::Sender;
 use zellij_utils::async_std::task::{self, JoinHandle};
 use zellij_utils::consts::ZELLIJ_CACHE_DIR;
-use zellij_utils::data::{PermissionStatus, PermissionType};
+use zellij_utils::data::{PermissionStatus, PermissionType, PipeMessage, PipeSource};
 use zellij_utils::downloader::Downloader;
 use zellij_utils::input::permission::PermissionCache;
 use zellij_utils::notify_debouncer_full::{notify::RecommendedWatcher, Debouncer, FileIdMap};
@@ -24,22 +27,53 @@ use zellij_utils::plugin_api::event::ProtobufEvent;
 
 use zellij_utils::prost::Message;
 
+use crate::panes::PaneId;
 use crate::{
     background_jobs::BackgroundJob, screen::ScreenInstruction, thread_bus::ThreadSenders,
-    ui::loading_indication::LoadingIndication, ClientId,
+    ui::loading_indication::LoadingIndication, ClientId, ServerInstruction,
 };
 use zellij_utils::{
     data::{Event, EventType, PluginCapabilities},
     errors::prelude::*,
     input::{
         command::TerminalAction,
-        layout::{Layout, RunPlugin, RunPluginLocation},
+        layout::{Layout, PluginUserConfiguration, RunPlugin, RunPluginLocation},
         plugins::PluginsConfig,
     },
     ipc::ClientAttributes,
     pane_size::Size,
 };
 
+#[derive(Debug, Clone)]
+pub enum EventOrPipeMessage {
+    Event(Event),
+    PipeMessage(PipeMessage),
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct PluginRenderAsset {
+    // TODO: naming
+    pub client_id: ClientId,
+    pub plugin_id: PluginId,
+    pub bytes: Vec<u8>,
+    pub cli_pipes: HashMap<String, PipeStateChange>,
+}
+
+impl PluginRenderAsset {
+    pub fn new(plugin_id: PluginId, client_id: ClientId, bytes: Vec<u8>) -> Self {
+        PluginRenderAsset {
+            client_id,
+            plugin_id,
+            bytes,
+            ..Default::default()
+        }
+    }
+    pub fn with_pipes(mut self, cli_pipes: HashMap<String, PipeStateChange>) -> Self {
+        self.cli_pipes = cli_pipes;
+        self
+    }
+}
+
 pub struct WasmBridge {
     connected_clients: Arc<Mutex<Vec<ClientId>>>,
     plugins: PluginsConfig,
@@ -49,7 +83,8 @@ pub struct WasmBridge {
     plugin_cache: Arc<Mutex<HashMap<PathBuf, Module>>>,
     plugin_map: Arc<Mutex<PluginMap>>,
     next_plugin_id: PluginId,
-    cached_events_for_pending_plugins: HashMap<PluginId, Vec<Event>>,
+    plugin_ids_waiting_for_permission_request: HashSet<PluginId>,
+    cached_events_for_pending_plugins: HashMap<PluginId, Vec<EventOrPipeMessage>>,
     cached_resizes_for_pending_plugins: HashMap<PluginId, (usize, usize)>, // (rows, columns)
     cached_worker_messages: HashMap<PluginId, Vec<(ClientId, String, String, String)>>, // Vec<clientid,
     // worker_name,
@@ -64,6 +99,9 @@ pub struct WasmBridge {
     client_attributes: ClientAttributes,
     default_shell: Option<TerminalAction>,
     default_layout: Box<Layout>,
+    cached_plugin_map:
+        HashMap<RunPluginLocation, HashMap<PluginUserConfiguration, Vec<(PluginId, ClientId)>>>,
+    pending_pipes: PendingPipes,
 }
 
 impl WasmBridge {
@@ -96,6 +134,7 @@ impl WasmBridge {
             watcher,
             next_plugin_id: 0,
             cached_events_for_pending_plugins: HashMap::new(),
+            plugin_ids_waiting_for_permission_request: HashSet::new(),
             cached_resizes_for_pending_plugins: HashMap::new(),
             cached_worker_messages: HashMap::new(),
             loading_plugins: HashMap::new(),
@@ -105,17 +144,20 @@ impl WasmBridge {
             client_attributes,
             default_shell,
             default_layout,
+            cached_plugin_map: HashMap::new(),
+            pending_pipes: Default::default(),
         }
     }
     pub fn load_plugin(
         &mut self,
         run: &RunPlugin,
-        tab_index: usize,
+        tab_index: Option<usize>,
         size: Size,
         cwd: Option<PathBuf>,
         skip_cache: bool,
         client_id: Option<ClientId>,
-    ) -> Result<PluginId> {
+        cli_client_id: Option<ClientId>,
+    ) -> Result<(PluginId, ClientId)> {
         // returns the plugin id
         let err_context = move || format!("failed to load plugin");
 
@@ -179,6 +221,7 @@ impl WasmBridge {
                             plugin_id,
                             &mut loading_indication,
                             e,
+                            cli_client_id,
                         ),
                     }
                 }
@@ -210,16 +253,19 @@ impl WasmBridge {
                         plugin_id,
                         &mut loading_indication,
                         e,
+                        cli_client_id,
                     ),
                 }
-                let _ =
-                    senders.send_to_plugin(PluginInstruction::ApplyCachedEvents(vec![plugin_id]));
+                let _ = senders.send_to_plugin(PluginInstruction::ApplyCachedEvents {
+                    plugin_ids: vec![plugin_id],
+                    done_receiving_permissions: false,
+                });
             }
         });
         self.loading_plugins
             .insert((plugin_id, run.clone()), load_plugin_task);
         self.next_plugin_id += 1;
-        Ok(plugin_id)
+        Ok((plugin_id, client_id))
     }
     pub fn unload_plugin(&mut self, pid: PluginId) -> Result<()> {
         info!("Bye from plugin {}", &pid);
@@ -234,6 +280,14 @@ impl WasmBridge {
                 log::error!("Failed to remove cache dir for plugin: {:?}", e);
             }
         }
+        self.cached_plugin_map.clear();
+        let mut pipes_to_unblock = self.pending_pipes.unload_plugin(&pid);
+        for pipe_name in pipes_to_unblock.drain(..) {
+            let _ = self
+                .senders
+                .send_to_server(ServerInstruction::UnblockCliPipeInput(pipe_name))
+                .context("failed to unblock input pipe");
+        }
         Ok(())
     }
     pub fn reload_plugin(&mut self, run_plugin: &RunPlugin) -> Result<()> {
@@ -242,7 +296,8 @@ impl WasmBridge {
             return Ok(());
         }
 
-        let plugin_ids = self.all_plugin_ids_for_plugin_location(&run_plugin.location)?;
+        let plugin_ids = self
+            .all_plugin_ids_for_plugin_location(&run_plugin.location, &run_plugin.configuration)?;
         for plugin_id in &plugin_ids {
             let (rows, columns) = self.size_of_plugin_id(*plugin_id).unwrap_or((0, 0));
             self.cached_events_for_pending_plugins
@@ -315,6 +370,7 @@ impl WasmBridge {
                                     *plugin_id,
                                     &mut loading_indication,
                                     e,
+                                    None,
                                 ),
                             }
                         }
@@ -326,11 +382,15 @@ impl WasmBridge {
                                 *plugin_id,
                                 &mut loading_indication,
                                 &e,
+                                None,
                             );
                         }
                     },
                 }
-                let _ = senders.send_to_plugin(PluginInstruction::ApplyCachedEvents(plugin_ids));
+                let _ = senders.send_to_plugin(PluginInstruction::ApplyCachedEvents {
+                    plugin_ids,
+                    done_receiving_permissions: false,
+                });
             }
         });
         self.loading_plugins
@@ -402,42 +462,48 @@ impl WasmBridge {
                         let mut running_plugin = running_plugin.lock().unwrap();
                         let _s = _s; // guard to allow the task to complete before cleanup/shutdown
                         if running_plugin.apply_event_id(AtomicEvent::Resize, event_id) {
+                            let old_rows = running_plugin.rows;
+                            let old_columns = running_plugin.columns;
                             running_plugin.rows = new_rows;
                             running_plugin.columns = new_columns;
 
-                            let rendered_bytes = running_plugin
-                                .instance
-                                .clone()
-                                .exports
-                                .get_function("render")
-                                .map_err(anyError::new)
-                                .and_then(|render| {
-                                    render
-                                        .call(
-                                            &mut running_plugin.store,
-                                            &[
-                                                Value::I32(new_rows as i32),
-                                                Value::I32(new_columns as i32),
-                                            ],
-                                        )
-                                        .map_err(anyError::new)
-                                })
-                                .and_then(|_| wasi_read_string(&running_plugin.plugin_env.wasi_env))
-                                .with_context(err_context);
-                            match rendered_bytes {
-                                Ok(rendered_bytes) => {
-                                    let plugin_bytes = vec![(
-                                        plugin_id,
-                                        client_id,
-                                        rendered_bytes.as_bytes().to_vec(),
-                                    )];
-                                    senders
-                                        .send_to_screen(ScreenInstruction::PluginBytes(
-                                            plugin_bytes,
-                                        ))
-                                        .unwrap();
-                                },
-                                Err(e) => log::error!("{}", e),
+                            if old_rows != new_rows || old_columns != new_columns {
+                                let rendered_bytes = running_plugin
+                                    .instance
+                                    .clone()
+                                    .exports
+                                    .get_function("render")
+                                    .map_err(anyError::new)
+                                    .and_then(|render| {
+                                        render
+                                            .call(
+                                                &mut running_plugin.store,
+                                                &[
+                                                    Value::I32(new_rows as i32),
+                                                    Value::I32(new_columns as i32),
+                                                ],
+                                            )
+                                            .map_err(anyError::new)
+                                    })
+                                    .and_then(|_| {
+                                        wasi_read_string(&running_plugin.plugin_env.wasi_env)
+                                    })
+                                    .with_context(err_context);
+                                match rendered_bytes {
+                                    Ok(rendered_bytes) => {
+                                        let plugin_render_asset = PluginRenderAsset::new(
+                                            plugin_id,
+                                            client_id,
+                                            rendered_bytes.as_bytes().to_vec(),
+                                        );
+                                        senders
+                                            .send_to_screen(ScreenInstruction::PluginBytes(vec![
+                                                plugin_render_asset,
+                                            ]))
+                                            .unwrap();
+                                    },
+                                    Err(e) => log::error!("{}", e),
+                                }
                             }
                         }
                     }
@@ -484,10 +550,7 @@ impl WasmBridge {
                 let event_type =
                     EventType::from_str(&event.to_string()).with_context(err_context)?;
                 if (subs.contains(&event_type) || event_type == EventType::PermissionRequestResult)
-                    && ((pid.is_none() && cid.is_none())
-                        || (pid.is_none() && cid == Some(*client_id))
-                        || (cid.is_none() && pid == Some(*plugin_id))
-                        || (cid == Some(*client_id) && pid == Some(*plugin_id)))
+                    && Self::message_is_directed_at_plugin(pid, cid, plugin_id, client_id)
                 {
                     task::spawn({
                         let senders = self.senders.clone();
@@ -498,18 +561,18 @@ impl WasmBridge {
                         let _s = shutdown_sender.clone();
                         async move {
                             let mut running_plugin = running_plugin.lock().unwrap();
-                            let mut plugin_bytes = vec![];
+                            let mut plugin_render_assets = vec![];
                             let _s = _s; // guard to allow the task to complete before cleanup/shutdown
                             match apply_event_to_plugin(
                                 plugin_id,
                                 client_id,
                                 &mut running_plugin,
                                 &event,
-                                &mut plugin_bytes,
+                                &mut plugin_render_assets,
                             ) {
                                 Ok(()) => {
                                     let _ = senders.send_to_screen(ScreenInstruction::PluginBytes(
-                                        plugin_bytes,
+                                        plugin_render_assets,
                                     ));
                                 },
                                 Err(e) => {
@@ -532,7 +595,112 @@ impl WasmBridge {
             }
             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());
+                    cached_events.push(EventOrPipeMessage::Event(event.clone()));
+                }
+            }
+        }
+        Ok(())
+    }
+    pub fn pipe_messages(
+        &mut self,
+        mut messages: Vec<(Option<PluginId>, Option<ClientId>, PipeMessage)>,
+        shutdown_sender: Sender<()>,
+    ) -> Result<()> {
+        let plugins_to_update: Vec<(
+            PluginId,
+            ClientId,
+            Arc<Mutex<RunningPlugin>>,
+            Arc<Mutex<Subscriptions>>,
+        )> = self
+            .plugin_map
+            .lock()
+            .unwrap()
+            .running_plugins_and_subscriptions()
+            .iter()
+            .cloned()
+            .filter(|(plugin_id, _client_id, _running_plugin, _subscriptions)| {
+                !&self
+                    .cached_events_for_pending_plugins
+                    .contains_key(&plugin_id)
+            })
+            .collect();
+        for (message_pid, message_cid, pipe_message) in messages.drain(..) {
+            for (plugin_id, client_id, running_plugin, _subscriptions) in &plugins_to_update {
+                if Self::message_is_directed_at_plugin(
+                    message_pid,
+                    message_cid,
+                    plugin_id,
+                    client_id,
+                ) {
+                    if let PipeSource::Cli(pipe_id) = &pipe_message.source {
+                        self.pending_pipes
+                            .mark_being_processed(pipe_id, plugin_id, client_id);
+                    }
+                    task::spawn({
+                        let senders = self.senders.clone();
+                        let running_plugin = running_plugin.clone();
+                        let pipe_message = pipe_message.clone();
+                        let plugin_id = *plugin_id;
+                        let client_id = *client_id;
+                        let _s = shutdown_sender.clone();
+                        async move {
+                            let mut running_plugin = running_plugin.lock().unwrap();
+                            let mut plugin_render_assets = vec![];
+                            let _s = _s; // guard to allow the task to complete before cleanup/shutdown
+                            match apply_pipe_message_to_plugin(
+                                plugin_id,
+                                client_id,
+                                &mut running_plugin,
+                                &pipe_message,
+                                &mut plugin_render_assets,
+                                &senders,
+                            ) {
+                                Ok(()) => {
+                                    let _ = senders.send_to_screen(ScreenInstruction::PluginBytes(
+                                        plugin_render_assets,
+                                    ));
+                                },
+                                Err(e) => {
+                                    log::error!("{:?}", e);
+
+                                    // https://stackoverflow.com/questions/66450942/in-rust-is-there-a-way-to-make-literal-newlines-in-r-using-windows-c
+                                    let stringified_error =
+                                        format!("{:?}", e).replace("\n", "\n\r");
+
+                                    handle_plugin_crash(
+                                        plugin_id,
+                                        stringified_error,
+                                        senders.clone(),
+                                    );
+                                },
+                            }
+                        }
+                    });
+                }
+            }
+            let all_connected_clients: Vec<ClientId> = self
+                .connected_clients
+                .lock()
+                .unwrap()
+                .iter()
+                .copied()
+                .collect();
+            for (plugin_id, cached_events) in self.cached_events_for_pending_plugins.iter_mut() {
+                if message_pid.is_none() || message_pid.as_ref() == Some(plugin_id) {
+                    cached_events.push(EventOrPipeMessage::PipeMessage(pipe_message.clone()));
+                    if let PipeSource::Cli(pipe_id) = &pipe_message.source {
+                        for client_id in &all_connected_clients {
+                            if Self::message_is_directed_at_plugin(
+                                message_pid,
+                                message_cid,
+                                plugin_id,
+                                client_id,
+                            ) {
+                                self.pending_pipes
+                                    .mark_being_processed(pipe_id, plugin_id, client_id);
+                            }
+                        }
+                    }
                 }
             }
         }
@@ -541,16 +709,27 @@ impl WasmBridge {
     pub fn apply_cached_events(
         &mut self,
         plugin_ids: Vec<PluginId>,
+        done_receiving_permissions: bool,
         shutdown_sender: Sender<()>,
     ) -> Result<()> {
         let mut applied_plugin_paths = HashSet::new();
         for plugin_id in plugin_ids {
+            if !done_receiving_permissions
+                && self
+                    .plugin_ids_waiting_for_permission_request
+                    .contains(&plugin_id)
+            {
+                continue;
+            }
+            self.plugin_ids_waiting_for_permission_request
+                .remove(&plugin_id);
             self.apply_cached_events_and_resizes_for_plugin(plugin_id, shutdown_sender.clone())?;
             if let Some(run_plugin) = self.run_plugin_of_loading_plugin_id(plugin_id) {
                 applied_plugin_paths.insert(run_plugin.clone());
             }
             self.loading_plugins
                 .retain(|(p_id, _run_plugin), _| p_id != &plugin_id);
+            self.clear_plugin_map_cache();
         }
         for run_plugin in applied_plugin_paths.drain() {
             if self.pending_plugin_reloads.remove(&run_plugin) {
@@ -595,7 +774,9 @@ impl WasmBridge {
         shutdown_sender: Sender<()>,
     ) -> Result<()> {
         let err_context = || format!("Failed to apply cached events to plugin");
-        if let Some(events) = self.cached_events_for_pending_plugins.remove(&plugin_id) {
+        if let Some(events_or_pipe_messages) =
+            self.cached_events_for_pending_plugins.remove(&plugin_id)
+        {
             let all_connected_clients: Vec<ClientId> = self
                 .connected_clients
                 .lock()
@@ -610,41 +791,90 @@ impl WasmBridge {
                     .unwrap()
                     .get_running_plugin_and_subscriptions(plugin_id, *client_id)
                 {
-                    let subs = subscriptions.lock().unwrap().clone();
-                    for event in events.clone() {
-                        let event_type =
-                            EventType::from_str(&event.to_string()).with_context(err_context)?;
-                        if !subs.contains(&event_type) {
-                            continue;
-                        }
-                        task::spawn({
-                            let senders = self.senders.clone();
-                            let running_plugin = running_plugin.clone();
-                            let client_id = *client_id;
-                            let _s = shutdown_sender.clone();
-                            async move {
-                                let mut running_plugin = running_plugin.lock().unwrap();
-                                let mut plugin_bytes = vec![];
-                                let _s = _s; // guard to allow the task to complete before cleanup/shutdown
-                                match apply_event_to_plugin(
-                                    plugin_id,
-                                    client_id,
-                                    &mut running_plugin,
-                                    &event,
-                                    &mut plugin_bytes,
-                                ) {
-                                    Ok(()) => {
-                                        let _ = senders.send_to_screen(
-                                            ScreenInstruction::PluginBytes(plugin_bytes),
-                                        );
+                    task::spawn({
+                        let senders = self.senders.clone();
+                        let running_plugin = running_plugin.clone();
+                        let client_id = *client_id;
+                        let _s = shutdown_sender.clone();
+                        let events_or_pipe_messages = events_or_pipe_messages.clone();
+                        async move {
+                            let subs = subscriptions.lock().unwrap().clone();
+                            let _s = _s; // guard to allow the task to complete before cleanup/shutdown
+                            for event_or_pipe_message in events_or_pipe_messages {
+                                match event_or_pipe_message {
+                                    EventOrPipeMessage::Event(event) => {
+                                        match EventType::from_str(&event.to_string())
+                                            .with_context(err_context)
+                                        {
+                                            Ok(event_type) => {
+                                                if !subs.contains(&event_type) {
+                                                    continue;
+                                                }
+                                                let mut running_plugin =
+                                                    running_plugin.lock().unwrap();
+                                                let mut plugin_render_assets = vec![];
+                                                match apply_event_to_plugin(
+                                                    plugin_id,
+                                                    client_id,
+                                                    &mut running_plugin,
+                                                    &event,
+                                                    &mut plugin_render_assets,
+                                                ) {
+                                                    Ok(()) => {
+                                                        let _ = senders.send_to_screen(
+                                                            ScreenInstruction::PluginBytes(
+                                                                plugin_render_assets,
+                                                            ),
+                                                        );
+                                                    },
+                                                    Err(e) => {
+                                                        log::error!("{}", e);
+                                                    },
+                                                }
+                                            },
+                                            Err(e) => {
+                                                log::error!("Failed to apply event: {:?}", e);
+                                            },
+                                        }
                                     },
-                                    Err(e) => {
-                                        log::error!("{}", e);
+                                    EventOrPipeMessage::PipeMessage(pipe_message) => {
+                                        let mut running_plugin = running_plugin.lock().unwrap();
+                                        let mut plugin_render_assets = vec![];
+
+                                        match apply_pipe_message_to_plugin(
+                                            plugin_id,
+                                            client_id,
+                                            &mut running_plugin,
+                                            &pipe_message,
+                                            &mut plugin_render_assets,
+                                            &senders,
+                                        ) {
+                                            Ok(()) => {
+                                                let _ = senders.send_to_screen(
+                                                    ScreenInstruction::PluginBytes(
+                                                        plugin_render_assets,
+                                                    ),
+                                                );
+                                            },
+                                            Err(e) => {
+                                                log::error!("{:?}", e);
+
+                                                // https://stackoverflow.com/questions/66450942/in-rust-is-there-a-way-to-make-literal-newlines-in-r-using-windows-c
+                                                let stringified_error =
+                                                    format!("{:?}", e).replace("\n", "\n\r");
+
+                                                handle_plugin_crash(
+                                                    plugin_id,
+                                                    stringified_error,
+                                                    senders.clone(),
+                                                );
+                                            },
+                                        }
                                     },
                                 }
                             }
-                        });
-                    }
+                        }
+                    });
                 }
             }
         }
@@ -676,14 +906,55 @@ impl WasmBridge {
             .find(|((_plugin_id, run_plugin), _)| &run_plugin.location == plugin_location)
             .is_some()
     }
+    fn plugin_id_of_loading_plugin(
+        &self,
+        plugin_location: &RunPluginLocation,
+        plugin_configuration: &PluginUserConfiguration,
+    ) -> Option<PluginId> {
+        self.loading_plugins
+            .iter()
+            .find_map(|((plugin_id, run_plugin), _)| {
+                if &run_plugin.location == plugin_location
+                    && &run_plugin.configuration == plugin_configuration
+                {
+                    Some(*plugin_id)
+                } else {
+                    None
+                }
+            })
+    }
     fn all_plugin_ids_for_plugin_location(
         &self,
         plugin_location: &RunPluginLocation,
+        plugin_configuration: &PluginUserConfiguration,
     ) -> Result<Vec<PluginId>> {
         self.plugin_map
             .lock()
             .unwrap()
-            .all_plugin_ids_for_plugin_location(plugin_location)
+            .all_plugin_ids_for_plugin_location(plugin_location, plugin_configuration)
+    }
+    pub fn all_plugin_and_client_ids_for_plugin_location(
+        &mut self,
+        plugin_location: &RunPluginLocation,
+        plugin_configuration: &PluginUserConfiguration,
+    ) -> Vec<(PluginId, Option<ClientId>)> {
+        if self.cached_plugin_map.is_empty() {
+            self.cached_plugin_map = self.plugin_map.lock().unwrap().clone_plugin_assets();
+        }
+        match self
+            .cached_plugin_map
+            .get(plugin_location)
+            .and_then(|m| m.get(plugin_configuration))
+        {
+            Some(plugin_and_client_ids) => plugin_and_client_ids
+                .iter()
+                .map(|(plugin_id, client_id)| (*plugin_id, Some(*client_id)))
+                .collect(),
+            None => vec![],
+        }
+    }
+    pub fn all_plugin_ids(&self) -> Vec<(PluginId, ClientId)> {
+        self.plugin_map.lock().unwrap().all_plugin_ids()
     }
     fn size_of_plugin_id(&self, plugin_id: PluginId) -> Option<(usize, usize)> {
         // (rows/colums)
@@ -793,6 +1064,117 @@ impl WasmBridge {
 
         permission_cache.write_to_file().with_context(err_context)
     }
+    pub fn cache_plugin_events(&mut self, plugin_id: PluginId) {
+        self.plugin_ids_waiting_for_permission_request
+            .insert(plugin_id);
+        self.cached_events_for_pending_plugins
+            .entry(plugin_id)
+            .or_insert_with(Default::default);
+    }
+
+    // gets all running plugins details matching this run_plugin, if none are running, loads one and
+    // returns its details
+    pub fn get_or_load_plugins(
+        &mut self,
+        run_plugin: RunPlugin,
+        size: Size,
+        cwd: Option<PathBuf>,
+        skip_cache: bool,
+        should_float: bool,
+        should_be_open_in_place: bool,
+        pane_title: Option<String>,
+        pane_id_to_replace: Option<PaneId>,
+        cli_client_id: Option<ClientId>,
+    ) -> Vec<(PluginId, Option<ClientId>)> {
+        let all_plugin_ids = self.all_plugin_and_client_ids_for_plugin_location(
+            &run_plugin.location,
+            &run_plugin.configuration,
+        );
+        if all_plugin_ids.is_empty() {
+            if let Some(loading_plugin_id) =
+                self.plugin_id_of_loading_plugin(&run_plugin.location, &run_plugin.configuration)
+            {
+                return vec![(loading_plugin_id, None)];
+            }
+            match self.load_plugin(
+                &run_plugin,
+                None,
+                size,
+                cwd.clone(),
+                skip_cache,
+                None,
+                cli_client_id,
+            ) {
+                Ok((plugin_id, client_id)) => {
+                    drop(self.senders.send_to_screen(ScreenInstruction::AddPlugin(
+                        Some(should_float),
+                        should_be_open_in_place,
+                        run_plugin,
+                        pane_title,
+                        None,
+                        plugin_id,
+                        pane_id_to_replace,
+                        cwd,
+                        Some(client_id),
+                    )));
+                    vec![(plugin_id, Some(client_id))]
+                },
+                Err(e) => {
+                    log::error!("Failed to load plugin: {e}");
+                    if let Some(cli_client_id) = cli_client_id {
+                        let _ = self.senders.send_to_server(ServerInstruction::LogError(
+                            vec![format!("Failed to log plugin: {e}")],
+                            cli_client_id,
+                        ));
+                    }
+                    vec![]
+                },
+            }
+        } else {
+            all_plugin_ids
+        }
+    }
+    pub fn clear_plugin_map_cache(&mut self) {
+        self.cached_plugin_map.clear();
+    }
+    // returns the pipe names to unblock
+    pub fn update_cli_pipe_state(
+        &mut self,
+        pipe_state_changes: Vec<PluginRenderAsset>,
+    ) -> Vec<String> {
+        let mut pipe_names_to_unblock = vec![];
+        for pipe_state_change in pipe_state_changes {
+            let client_id = pipe_state_change.client_id;
+            let plugin_id = pipe_state_change.plugin_id;
+            for (cli_pipe_name, pipe_state_change) in pipe_state_change.cli_pipes {
+                pipe_names_to_unblock.append(&mut self.pending_pipes.update_pipe_state_change(
+                    &cli_pipe_name,
+                    pipe_state_change,
+                    &plugin_id,
+                    &client_id,
+                ));
+            }
+        }
+        let pipe_names_to_unblock =
+            pipe_names_to_unblock
+                .into_iter()
+                .fold(HashSet::new(), |mut acc, p| {
+                    acc.insert(p);
+                    acc
+                });
+        pipe_names_to_unblock.into_iter().collect()
+    }
+    fn message_is_directed_at_plugin(
+        message_pid: Option<PluginId>,
+        message_cid: Option<ClientId>,
+        plugin_id: &PluginId,
+        client_id: &ClientId,
+    ) -> bool {
+        message_pid.is_none() && message_cid.is_none()
+            || (message_pid.is_none() && message_cid == Some(*client_id))
+            || (message_cid.is_none() && message_pid == Some(*plugin_id))
+            || (message_cid == Some(*client_id) && message_pid == Some(*plugin_id))
+    }
 }
 
 fn handle_plugin_successful_loading(senders: &ThreadSenders, plugin_id: PluginId) {
@@ -805,6 +1187,7 @@ fn handle_plugin_loading_failure(
     plugin_id: PluginId,
     loading_indication: &mut LoadingIndication,
     error: impl std::fmt::Debug,
+    cli_client_id: Option<ClientId>,
 ) {
     log::error!("{:?}", error);
     let _ = senders.send_to_background_jobs(BackgroundJob::StopPluginLoadingAnimation(plugin_id));
@@ -813,6 +1196,12 @@ fn handle_plugin_loading_failure(
         plugin_id,
         loading_indication.clone(),
     ));
+    if let Some(cli_client_id) = cli_client_id {
+        let _ = senders.send_to_server(ServerInstruction::LogError(
+            vec![format!("{:?}", error)],
+            cli_client_id,
+        ));
+    }
 }
 
 // TODO: move to permissions?
@@ -850,7 +1239,7 @@ pub fn apply_event_to_plugin(
     client_id: ClientId,
     running_plugin: &mut RunningPlugin,
     event: &Event,
-    plugin_bytes: &mut Vec<(PluginId, ClientId, Vec<u8>)>,
+    plugin_render_assets: &mut Vec<PluginRenderAsset>,
 ) -> Result<()> {
     let instance = &running_plugin.instance;
     let plugin_env = &running_plugin.plugin_env;
@@ -897,7 +1286,14 @@ pub fn apply_event_to_plugin(
                     })
                     .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()));
+                let pipes_to_block_or_unblock = pipes_to_block_or_unblock(running_plugin, None);
+                let plugin_render_asset = PluginRenderAsset::new(
+                    plugin_id,
+                    client_id,
+                    rendered_bytes.as_bytes().to_vec(),
+                )
+                .with_pipes(pipes_to_block_or_unblock);
+                plugin_render_assets.push(plugin_render_asset);
             }
         },
         (PermissionStatus::Denied, permission) => {
diff --git a/zellij-server/src/plugins/zellij_exports.rs b/zellij-server/src/plugins/zellij_exports.rs
index 62dbb2b5..8791e128 100644
--- a/zellij-server/src/plugins/zellij_exports.rs
+++ b/zellij-server/src/plugins/zellij_exports.rs
@@ -18,7 +18,8 @@ use std::{
 use wasmer::{imports, AsStoreMut, Function, FunctionEnv, FunctionEnvMut, Imports};
 use wasmer_wasi::WasiEnv;
 use zellij_utils::data::{
-    CommandType, ConnectToSession, HttpVerb, PermissionStatus, PermissionType, PluginPermission,
+    CommandType, ConnectToSession, HttpVerb, MessageToPlugin, PermissionStatus, PermissionType,
+    PluginPermission,
 };
 use zellij_utils::input::permission::PermissionCache;
 
@@ -58,6 +59,7 @@ macro_rules! apply_action {
             $env.plugin_env.client_attributes.clone(),
             $env.plugin_env.default_shell.clone(),
             $env.plugin_env.default_layout.clone(),
+            None,
         ) {
             log::error!("{}: {:?}", $error_message(), e);
         }
@@ -240,6 +242,16 @@ fn host_run_plugin_command(env: FunctionEnvMut<ForeignFunctionEnv>) {
                     PluginCommand::RenameSession(new_session_name) => {
                         rename_session(env, new_session_name)
                     },
+                    PluginCommand::UnblockCliPipeInput(pipe_name) => {
+                        unblock_cli_pipe_input(env, pipe_name)
+                    },
+                    PluginCommand::BlockCliPipeInput(pipe_name) => {
+                        block_cli_pipe_input(env, pipe_name)
+                    },
+                    PluginCommand::CliPipeOutput(pipe_name, output) => {
+                        cli_pipe_output(env, pipe_name, output)?
+                    },
+                    PluginCommand::MessageToPlugin(message) => message_to_plugin(env, message)?,
                 },
                 (PermissionStatus::Denied, permission) => {
                     log::error!(
@@ -272,6 +284,39 @@ fn subscribe(env: &ForeignFunctionEnv, event_list: HashSet<EventType>) -> Result
         ))
 }
 
+fn unblock_cli_pipe_input(env: &ForeignFunctionEnv, pipe_name: String) {
+    env.plugin_env
+        .input_pipes_to_unblock
+        .lock()
+        .unwrap()
+        .insert(pipe_name);
+}
+
+fn block_cli_pipe_input(env: &ForeignFunctionEnv, pipe_name: String) {
+    env.plugin_env
+        .input_pipes_to_block
+        .lock()
+        .unwrap()
+        .insert(pipe_name);
+}
+
+fn cli_pipe_output(env: &ForeignFunctionEnv, pipe_name: String, output: String) -> Result<()> {
+    env.plugin_env
+        .senders
+        .send_to_server(ServerInstruction::CliPipeOutput(pipe_name, output))
+        .context("failed to send pipe output")
+}
+
+fn message_to_plugin(env: &ForeignFunctionEnv, message_to_plugin: MessageToPlugin) -> Result<()> {
+    env.plugin_env
+        .senders
+        .send_to_plugin(PluginInstruction::MessageFromPlugin {
+            source_plugin_id: env.plugin_env.plugin_id,
+            message: message_to_plugin,
+        })
+        .context("failed to send message to plugin")
+}
+
 fn unsubscribe(env: &ForeignFunctionEnv, event_list: HashSet<EventType>) -> Result<()> {
     env.subscriptions
         .lock()
@@ -325,6 +370,15 @@ fn request_permission(env: &ForeignFunctionEnv, permissions: Vec<PermissionType>
             ));
     }
 
+    // we do this so that messages that have arrived while the user is seeing the permission screen
+    // will be cached and reapplied once the permission is granted
+    let _ = env
+        .plugin_env
+        .senders
+        .send_to_plugin(PluginInstruction::CachePluginEvents {
+            plugin_id: env.plugin_env.plugin_id,
+        });
+
     env.plugin_env
         .senders
         .send_to_screen(ScreenInstruction::RequestPluginPermissions(
@@ -1353,6 +1407,10 @@ fn check_command_permission(
         | PluginCommand::DeleteAllDeadSessions
         | PluginCommand::RenameSession(..)
         | PluginCommand::RenameTab(..) => PermissionType::ChangeApplicationState,
+        PluginCommand::UnblockCliPipeInput(..)
+        | PluginCommand::BlockCliPipeInput(..)
+        | PluginCommand::CliPipeOutput(..) => PermissionType::ReadCliPipes,
+        PluginCommand::MessageToPlugin(..) => PermissionType::MessageAndLaunchOtherPlugins,
         _ => return (PermissionStatus::Granted, None),
     };
 
diff --git a/zellij-server/src/route.rs b/zellij-server/src/route.rs
index 17fcc3de..e3c85361 100644
--- a/zellij-server/src/route.rs
+++ b/zellij-server/src/route.rs
@@ -1,4 +1,4 @@
-use std::collections::VecDeque;
+use std::collections::{HashSet, VecDeque};
 use std::sync::{Arc, RwLock};
 
 use crate::thread_bus::ThreadSenders;
@@ -36,6 +36,7 @@ pub(crate) fn route_action(
     client_attributes: ClientAttributes,
     default_shell: Option<TerminalAction>,
     default_layout: Box<Layout>,
+    mut seen_cli_pipes: Option<&mut HashSet<String>>,
 ) -> Result<bool> {
     let mut should_break = false;
     let err_context = || format!("failed to route action for client {client_id}");
@@ -803,6 +804,57 @@ pub(crate) fn route_action(
                 .send_to_screen(ScreenInstruction::RenameSession(name, client_id))
                 .with_context(err_context)?;
         },
+        Action::CliPipe {
+            pipe_id,
+            mut name,
+            payload,
+            plugin,
+            args,
+            configuration,
+            floating,
+            in_place,
+            skip_cache,
+            cwd,
+            pane_title,
+            ..
+        } => {
+            if let Some(seen_cli_pipes) = seen_cli_pipes.as_mut() {
+                if !seen_cli_pipes.contains(&pipe_id) {
+                    seen_cli_pipes.insert(pipe_id.clone());
+                    senders
+                        .send_to_server(ServerInstruction::AssociatePipeWithClient {
+                            pipe_id: pipe_id.clone(),
+                            client_id,
+                        })
+                        .with_context(err_context)?;
+                }
+            }
+            if let Some(name) = name.take() {
+                let should_open_in_place = in_place.unwrap_or(false);
+                if should_open_in_place && pane_id.is_none() {
+                    log::error!("Was asked to open a new plugin in-place, but cannot identify the pane id... is the ZELLIJ_PANE_ID variable set?");
+                }
+                let pane_id_to_replace = if should_open_in_place { pane_id } else { None };
+                senders
+                    .send_to_plugin(PluginInstruction::CliPipe {
+                        pipe_id,
+                        name,
+                        payload,
+                        plugin,
+                        args,
+                        configuration,
+                        floating,
+                        pane_id_to_replace,
+                        cwd,
+                        pane_title,
+                        skip_cache,
+                        cli_client_id: client_id,
+                    })
+                    .with_context(err_context)?;
+            } else {
+                log::error!("Message must have a name");
+            }
+        },
     }
     Ok(should_break)
 }
@@ -833,13 +885,14 @@ pub(crate) fn route_thread_main(
 ) -> Result<()> {
     let mut retry_queue = VecDeque::new();
     let err_context = || format!("failed to handle instruction for client {client_id}");
+    let mut seen_cli_pipes = HashSet::new();
     'route_loop: loop {
         match receiver.recv() {
             Some((instruction, err_ctx)) => {
                 err_ctx.update_thread_ctx();
                 let rlocked_sessions = session_data.read().to_anyhow().with_context(err_context)?;
-                let handle_instruction = |instruction: ClientToServerMsg,
-                                          mut retry_queue: Option<
+                let mut handle_instruction = |instruction: ClientToServerMsg,
+                                              mut retry_queue: Option<
                     &mut VecDeque<ClientToServerMsg>,
                 >|
                  -> Result<bool> {
@@ -868,6 +921,7 @@ pub(crate) fn route_thread_main(
                                     rlocked_sessions.client_attributes.clone(),
                                     rlocked_sessions.default_shell.clone(),
                                     rlocked_sessions.layout.clone(),
+                                    Some(&mut seen_cli_pipes),
                                 )? {
                                     should_break = true;
                                 }
diff --git a/zellij-server/src/screen.rs b/zellij-server/src/screen.rs
index b11d09e3..c312480d 100644
--- a/zellij-server/src/screen.rs
+++ b/zellij-server/src/screen.rs
@@ -35,7 +35,7 @@ use crate::{
     output::Output,
     panes::sixel::SixelImageStore,
     panes::PaneId,
-    plugins::PluginInstruction,
+    plugins::{PluginInstruction, PluginRenderAsset},
     pty::{ClientTabIndexOrPaneId, PtyInstruction, VteBytes},
     tab::Tab,
     thread_bus::Bus,
@@ -138,7 +138,7 @@ type HoldForCommand = Option<RunCommand>;
 #[derive(Debug, Clone)]
 pub enum ScreenInstruction {
     PtyBytes(u32, VteBytes),
-    PluginBytes(Vec<(u32, ClientId, VteBytes)>), // u32 is plugin_id
+    PluginBytes(Vec<PluginRenderAsset>),
     Render,
     NewPane(
         PaneId,
@@ -285,7 +285,7 @@ pub enum ScreenInstruction {
         bool,         // should be opened in place
         RunPlugin,
         Option<String>, // pane title
-        usize,          // tab index
+        Option<usize>,  // tab index
         u32,            // plugin id
         Option<PaneId>,
         Option<PathBuf>, // cwd
@@ -822,7 +822,7 @@ impl Screen {
 
                     self.log_and_report_session_state()
                         .with_context(err_context)?;
-                    return self.render().with_context(err_context);
+                    return self.render(None).with_context(err_context);
                 },
                 Err(err) => Err::<(), _>(err).with_context(err_context).non_fatal(),
             }
@@ -956,7 +956,7 @@ impl Screen {
             }
             self.log_and_report_session_state()
                 .with_context(err_context)?;
-            self.render().with_context(err_context)
+            self.render(None).with_context(err_context)
         }
     }
 
@@ -994,7 +994,7 @@ impl Screen {
         }
         self.log_and_report_session_state()
             .with_context(err_context)?;
-        self.render().with_context(err_context)
+        self.render(None).with_context(err_context)
     }
 
     pub fn update_pixel_dimensions(&mut self, pixel_dimensions: PixelDimensions) {
@@ -1038,7 +1038,7 @@ impl Screen {
     }
 
     /// Renders this [`Screen`], which amounts to rendering its active [`Tab`].
-    pub fn render(&mut self) -> Result<()> {
+    pub fn render(&mut self, plugin_render_assets: Option<Vec<PluginRenderAsset>>) -> Result<()> {
         let err_context = "failed to render screen";
 
         let mut output = Output::new(
@@ -1059,13 +1059,20 @@ impl Screen {
         }
         if output.is_dirty() {
             let serialized_output = output.serialize().context(err_context)?;
-            self.bus
+            let _ = self
+                .bus
                 .senders
                 .send_to_server(ServerInstruction::Render(Some(serialized_output)))
-                .context(err_context)
-        } else {
-            Ok(())
+                .context(err_context);
+        }
+        if let Some(plugin_render_assets) = plugin_render_assets {
+            let _ = self
+                .bus
+                .senders
+                .send_to_plugin(PluginInstruction::UnblockCliPipes(plugin_render_assets))
+                .context("failed to unblock input pipe");
         }
+        Ok(())
     }
 
     /// Returns a mutable reference to this [`Screen`]'s tabs.
@@ -1264,7 +1271,7 @@ impl Screen {
         }
 
         self.log_and_report_session_state()
-            .and_then(|_| self.render())
+            .and_then(|_| self.render(None))
             .with_context(err_context)
     }
 
@@ -1694,7 +1701,7 @@ impl Screen {
 
         self.log_and_report_session_state()
             .context("failed to toggle tabs")?;
-        self.render()
+        self.render(None)
     }
 
     pub fn focus_plugin_pane(
@@ -1902,7 +1909,7 @@ impl Screen {
                 .with_context(err_context)?;
         }
         self.unblock_input()?;
-        self.render()?;
+        self.render(None)?;
         Ok(())
     }
     pub fn replace_pane(
@@ -2155,21 +2162,25 @@ pub(crate) fn screen_thread_main(
                     }
                 }
             },
-            ScreenInstruction::PluginBytes(mut plugin_bytes) => {
-                for (pid, client_id, vte_bytes) in plugin_bytes.drain(..) {
+            ScreenInstruction::PluginBytes(mut plugin_render_assets) => {
+                for plugin_render_asset in plugin_render_assets.iter_mut() {
+                    let plugin_id = plugin_render_asset.plugin_id;
+                    let client_id = plugin_render_asset.client_id;
+                    let vte_bytes = plugin_render_asset.bytes.drain(..).collect();
+
                     let all_tabs = screen.get_tabs_mut();
                     for tab in all_tabs.values_mut() {
-                        if tab.has_plugin(pid) {
-                            tab.handle_plugin_bytes(pid, client_id, vte_bytes)
+                        if tab.has_plugin(plugin_id) {
+                            tab.handle_plugin_bytes(plugin_id, client_id, vte_bytes)
                                 .context("failed to process plugin bytes")?;
                             break;
                         }
                     }
                 }
-                screen.render()?;
+                screen.render(Some(plugin_render_assets))?;
             },
             ScreenInstruction::Render => {
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::NewPane(
                 pid,
@@ -2227,7 +2238,7 @@ pub(crate) fn screen_thread_main(
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
 
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::OpenInPlaceEditor(pid, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
@@ -2235,7 +2246,7 @@ pub(crate) fn screen_thread_main(
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
 
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::TogglePaneEmbedOrFloating(client_id) => {
                 active_tab_and_connected_client_id!(screen, client_id, |tab: &mut Tab, client_id: ClientId| tab
@@ -2243,7 +2254,7 @@ pub(crate) fn screen_thread_main(
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
 
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::ToggleFloatingPanes(client_id, default_shell) => {
                 active_tab_and_connected_client_id!(screen, client_id, |tab: &mut Tab, client_id: ClientId| tab
@@ -2251,7 +2262,7 @@ pub(crate) fn screen_thread_main(
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
 
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::HorizontalSplit(
                 pid,
@@ -2280,7 +2291,7 @@ pub(crate) fn screen_thread_main(
                 }
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::VerticalSplit(
                 pid,
@@ -2309,7 +2320,7 @@ pub(crate) fn screen_thread_main(
                 }
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::WriteCharacter(bytes, client_id) => {
                 let mut state_changed = false;
@@ -2340,7 +2351,7 @@ pub(crate) fn screen_thread_main(
                     ?
                 );
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::SwitchFocus(client_id) => {
@@ -2350,7 +2361,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.focus_next_pane(client_id)
                 );
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::FocusNextPane(client_id) => {
@@ -2359,7 +2370,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.focus_next_pane(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::FocusPreviousPane(client_id) => {
@@ -2368,7 +2379,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.focus_previous_pane(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2379,14 +2390,14 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.move_focus_left(client_id),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::MoveFocusLeftOrPreviousTab(client_id) => {
                 screen.move_focus_left_or_previous_tab(client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::MoveFocusDown(client_id) => {
@@ -2396,7 +2407,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.move_focus_down(client_id),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2407,14 +2418,14 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.move_focus_right(client_id),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::MoveFocusRightOrNextTab(client_id) => {
                 screen.move_focus_right_or_next_tab(client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::MoveFocusUp(client_id) => {
@@ -2424,7 +2435,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.move_focus_up(client_id),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2437,7 +2448,7 @@ pub(crate) fn screen_thread_main(
                     ),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::DumpScreen(file, client_id, full) => {
@@ -2451,7 +2462,7 @@ pub(crate) fn screen_thread_main(
                     ),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::DumpLayout(default_shell, client_id) => {
@@ -2473,7 +2484,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.edit_scrollback(client_id),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::ScrollUp(client_id) => {
@@ -2483,7 +2494,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.scroll_active_terminal_up(client_id)
                 );
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::MovePane(client_id) => {
                 active_tab_and_connected_client_id!(
@@ -2491,7 +2502,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2501,7 +2512,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane_backwards(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2511,7 +2522,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane_down(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2521,7 +2532,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane_up(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2531,7 +2542,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane_right(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2541,7 +2552,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.move_active_pane_left(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2552,7 +2563,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .handle_scrollwheel_up(&point, 3, client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ScrollDown(client_id) => {
@@ -2561,7 +2572,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.scroll_active_terminal_down(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ScrollDownAt(point, client_id) => {
@@ -2571,7 +2582,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .handle_scrollwheel_down(&point, 3, client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ScrollToBottom(client_id) => {
@@ -2581,7 +2592,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_to_bottom(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ScrollToTop(client_id) => {
@@ -2591,7 +2602,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_to_top(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::PageScrollUp(client_id) => {
@@ -2601,7 +2612,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_up_page(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::PageScrollDown(client_id) => {
@@ -2611,7 +2622,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_down_page(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::HalfPageScrollUp(client_id) => {
@@ -2621,7 +2632,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_up_half_page(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::HalfPageScrollDown(client_id) => {
@@ -2631,7 +2642,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .scroll_active_terminal_down_half_page(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ClearScroll(client_id) => {
@@ -2641,7 +2652,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .clear_active_terminal_scroll(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::CloseFocusedPane(client_id) => {
@@ -2650,7 +2661,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.close_focused_pane(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2666,7 +2677,7 @@ pub(crate) fn screen_thread_main(
                     |tab| tab.set_pane_selectable(id, selectable),
                 );
 
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::ClosePane(id, client_id) => {
@@ -2726,7 +2737,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.update_active_pane_name(c, client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2736,7 +2747,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.undo_active_rename_pane(client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ToggleActiveTerminalFullscreen(client_id) => {
@@ -2746,7 +2757,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .toggle_active_pane_fullscreen(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
@@ -2755,24 +2766,24 @@ pub(crate) fn screen_thread_main(
                 for tab in screen.tabs.values_mut() {
                     tab.set_pane_frames(screen.draw_pane_frames);
                 }
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
             },
             ScreenInstruction::SwitchTabNext(client_id) => {
                 screen.switch_tab_next(None, true, client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::SwitchTabPrev(client_id) => {
                 screen.switch_tab_prev(None, true, client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::CloseTab(client_id) => {
                 screen.close_tab(client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::NewTab(
                 cwd,
@@ -2829,13 +2840,13 @@ pub(crate) fn screen_thread_main(
                             plugin_loading_message_cache.remove(plugin_id)
                         {
                             screen.update_plugin_loading_stage(*plugin_id, loading_indication);
-                            screen.render()?;
+                            screen.render(None)?;
                         }
                     }
                 }
 
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::GoToTab(tab_index, client_id) => {
                 let client_id_to_switch = if client_id.is_none() {
@@ -2857,7 +2868,7 @@ pub(crate) fn screen_thread_main(
                     Some(client_id) if pending_tab_ids.is_empty() => {
                         screen.go_to_tab(tab_index as usize, client_id)?;
                         screen.unblock_input()?;
-                        screen.render()?;
+                        screen.render(None)?;
                     },
                     _ => {
                         if let Some(client_id) = client_id {
@@ -2886,7 +2897,7 @@ pub(crate) fn screen_thread_main(
                 if let Some(client_id) = client_id {
                     if let Ok(tab_exists) = screen.go_to_tab_name(tab_name.clone(), client_id) {
                         screen.unblock_input()?;
-                        screen.render()?;
+                        screen.render(None)?;
                         if create && !tab_exists {
                             let tab_index = screen.get_new_tab_index();
                             screen.new_tab(tab_index, swap_layouts, Some(tab_name), client_id)?;
@@ -2908,17 +2919,17 @@ pub(crate) fn screen_thread_main(
             ScreenInstruction::UpdateTabName(c, client_id) => {
                 screen.update_active_tab_name(c, client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::UndoRenameTab(client_id) => {
                 screen.undo_active_rename_tab(client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::TerminalResize(new_size) => {
                 screen.resize_to_screen(new_size)?;
                 screen.log_and_report_session_state()?; // update tabs so that the ui indication will be send to the plugins
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::TerminalPixelDimensions(pixel_dimensions) => {
                 screen.update_pixel_dimensions(pixel_dimensions);
@@ -2934,12 +2945,12 @@ pub(crate) fn screen_thread_main(
             },
             ScreenInstruction::ChangeMode(mode_info, client_id) => {
                 screen.change_mode(mode_info, client_id)?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ChangeModeForAllClients(mode_info) => {
                 screen.change_mode_for_all_clients(mode_info)?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ToggleActiveSyncTab(client_id) => {
@@ -2949,65 +2960,65 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, _client_id: ClientId| tab.toggle_sync_panes_is_active()
                 );
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::LeftClick(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_left_click(&point, client_id), ?);
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::RightClick(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_right_click(&point, client_id), ?);
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::MiddleClick(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_middle_click(&point, client_id), ?);
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::LeftMouseRelease(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_left_mouse_release(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::RightMouseRelease(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_right_mouse_release(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::MiddleMouseRelease(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_middle_mouse_release(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::MouseHoldLeft(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_mouse_hold_left(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::MouseHoldRight(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_mouse_hold_right(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::MouseHoldMiddle(point, client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .handle_mouse_hold_middle(&point, client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::Copy(client_id) => {
                 active_tab!(screen, client_id, |tab: &mut Tab| tab
                     .copy_selection(client_id), ?);
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::Exit => {
                 break;
@@ -3015,7 +3026,7 @@ pub(crate) fn screen_thread_main(
             ScreenInstruction::ToggleTab(client_id) => {
                 screen.toggle_tab(client_id)?;
                 screen.unblock_input()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::AddClient(client_id, tab_position_to_focus, pane_id_to_focus) => {
                 screen.add_client(client_id)?;
@@ -3032,12 +3043,12 @@ pub(crate) fn screen_thread_main(
                     screen.go_to_tab(tab_position_to_focus, client_id)?;
                 }
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::RemoveClient(client_id) => {
                 screen.remove_client(client_id)?;
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::AddOverlay(overlay, _client_id) => {
                 screen.get_active_overlays_mut().pop();
@@ -3046,7 +3057,7 @@ pub(crate) fn screen_thread_main(
             },
             ScreenInstruction::RemoveOverlay(_client_id) => {
                 screen.get_active_overlays_mut().pop();
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::ConfirmPrompt(_client_id) => {
@@ -3063,7 +3074,7 @@ pub(crate) fn screen_thread_main(
             },
             ScreenInstruction::DenyPrompt(_client_id) => {
                 screen.get_active_overlays_mut().pop();
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::UpdateSearch(c, client_id) => {
@@ -3072,7 +3083,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.update_search_term(c, client_id), ?
                 );
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::SearchDown(client_id) => {
                 active_tab_and_connected_client_id!(
@@ -3080,7 +3091,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.search_down(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::SearchUp(client_id) => {
                 active_tab_and_connected_client_id!(
@@ -3088,7 +3099,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.search_up(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::SearchToggleCaseSensitivity(client_id) => {
@@ -3098,7 +3109,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab
                         .toggle_search_case_sensitivity(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::SearchToggleWrap(client_id) => {
@@ -3107,7 +3118,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.toggle_search_wrap(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::SearchToggleWholeWord(client_id) => {
@@ -3116,7 +3127,7 @@ pub(crate) fn screen_thread_main(
                     client_id,
                     |tab: &mut Tab, client_id: ClientId| tab.toggle_search_whole_words(client_id)
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.unblock_input()?;
             },
             ScreenInstruction::AddRedPaneFrameColorOverride(pane_ids, error_text) => {
@@ -3129,7 +3140,7 @@ pub(crate) fn screen_thread_main(
                         }
                     }
                 }
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::ClearPaneFrameColorOverride(pane_ids) => {
                 let all_tabs = screen.get_tabs_mut();
@@ -3141,7 +3152,7 @@ pub(crate) fn screen_thread_main(
                         }
                     }
                 }
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::PreviousSwapLayout(client_id) => {
                 active_tab_and_connected_client_id!(
@@ -3150,7 +3161,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.previous_swap_layout(Some(client_id)),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
                 screen.unblock_input()?;
             },
@@ -3161,7 +3172,7 @@ pub(crate) fn screen_thread_main(
                     |tab: &mut Tab, client_id: ClientId| tab.next_swap_layout(Some(client_id), true),
                     ?
                 );
-                screen.render()?;
+                screen.render(None)?;
                 screen.log_and_report_session_state()?;
                 screen.unblock_input()?;
             },
@@ -3325,7 +3336,19 @@ pub(crate) fn screen_thread_main(
                     } else {
                         log::error!("Must have pane id to replace or connected client_id if replacing a pane");
                     }
-                } else if let Some(active_tab) = screen.tabs.get_mut(&tab_index) {
+                } else if let Some(client_id) = client_id {
+                    active_tab!(screen, client_id, |active_tab: &mut Tab| {
+                        active_tab.new_pane(
+                            PaneId::Plugin(plugin_id),
+                            Some(pane_title),
+                            should_float,
+                            Some(run_plugin),
+                            None,
+                        )
+                    }, ?);
+                } else if let Some(active_tab) =
+                    tab_index.and_then(|tab_index| screen.tabs.get_mut(&tab_index))
+                {
                     active_tab.new_pane(
                         PaneId::Plugin(plugin_id),
                         Some(pane_title),
@@ -3338,7 +3361,7 @@ pub(crate) fn screen_thread_main(
                 }
                 if let Some(loading_indication) = plugin_loading_message_cache.remove(&plugin_id) {
                     screen.update_plugin_loading_stage(plugin_id, loading_indication);
-                    screen.render()?;
+                    screen.render(None)?;
                 }
                 screen.log_and_report_session_state()?;
                 screen.unblock_input()?;
@@ -3349,7 +3372,7 @@ pub(crate) fn screen_thread_main(
                 if !found_plugin {
                     plugin_loading_message_cache.insert(pid, loading_indication);
                 }
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::StartPluginLoadingIndication(pid, loading_indication) => {
                 let all_tabs = screen.get_tabs_mut();
@@ -3359,7 +3382,7 @@ pub(crate) fn screen_thread_main(
                         break;
                     }
                 }
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::ProgressPluginLoadingOffset(pid) => {
                 let all_tabs = screen.get_tabs_mut();
@@ -3369,7 +3392,7 @@ pub(crate) fn screen_thread_main(
                         break;
                     }
                 }
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::RequestStateUpdateForPlugins => {
                 let all_tabs = screen.get_tabs_mut();
@@ -3377,7 +3400,7 @@ pub(crate) fn screen_thread_main(
                     tab.update_input_modes()?;
                 }
                 screen.log_and_report_session_state()?;
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::LaunchOrFocusPlugin(
                 run_plugin,
@@ -3432,7 +3455,7 @@ pub(crate) fn screen_thread_main(
                                 move_to_focused_tab,
                                 client_id,
                             )? {
-                                screen.render()?;
+                                screen.render(None)?;
                                 screen.log_and_report_session_state()?;
                             } else {
                                 screen
@@ -3529,7 +3552,7 @@ pub(crate) fn screen_thread_main(
                 for tab in all_tabs.values_mut() {
                     if tab.has_non_suppressed_pane_with_pid(&pane_id) {
                         tab.suppress_pane(pane_id, client_id);
-                        drop(screen.render());
+                        drop(screen.render(None));
                         break;
                     }
                 }
@@ -3544,7 +3567,7 @@ pub(crate) fn screen_thread_main(
                 for tab in all_tabs.values_mut() {
                     if tab.has_pane_with_pid(&pane_id) {
                         match tab.rename_pane(new_name, pane_id) {
-                            Ok(()) => drop(screen.render()),
+                            Ok(()) => drop(screen.render(None)),
                             Err(e) => log::error!("Failed to rename pane: {:?}", e),
                         }
                         break;
@@ -3611,7 +3634,7 @@ pub(crate) fn screen_thread_main(
                 screen.unblock_input()?;
                 screen.log_and_report_session_state()?;
 
-                screen.render()?;
+                screen.render(None)?;
             },
             ScreenInstruction::DumpLayoutToHd => {
                 if screen.session_serialization {
diff --git a/zellij-server/src/unit/screen_tests.rs b/zellij-server/src/unit/screen_tests.rs
index 6296d60d..34076d70 100644
--- a/zellij-server/src/unit/screen_tests.rs
+++ b/zellij-server/src/unit/screen_tests.rs
@@ -125,6 +125,7 @@ fn send_cli_action_to_server(
             client_attributes.clone(),
             default_shell.clone(),
             default_layout.clone(),
+            None,
         )
         .unwrap();
     }
diff --git a/zellij-tile/src/lib.rs b/zellij-tile/src/lib.rs
index 39058c29..767f73c4 100644
--- a/zellij-tile/src/lib.rs
+++ b/zellij-tile/src/lib.rs
@@ -21,7 +21,7 @@ pub mod ui_components;
 
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
-use zellij_utils::data::Event;
+use zellij_utils::data::{Event, PipeMessage};
 
 // use zellij_tile::shim::plugin_api::event::ProtobufEvent;
 
@@ -37,6 +37,12 @@ pub trait ZellijPlugin: Default {
     fn update(&mut self, event: Event) -> bool {
         false
     } // return true if it should render
+    /// Will be called when data is being piped to the plugin, a PipeMessage.payload of None signifies the pipe
+    /// has ended
+    /// If the plugin returns `true` from this function, Zellij will know it should be rendered and call its `render` function.
+    fn pipe(&mut self, pipe_message: PipeMessage) -> bool {
+        false
+    } // return true if it should render
     /// Will be called either after an `update` that requested it, or when the plugin otherwise needs to be re-rendered (eg. on startup, or when the plugin is resized).
     /// The `rows` and `cols` values represent the "content size" of the plugin (this will not include its surrounding frame if the user has pane frames enabled).
     fn render(&mut self, rows: usize, cols: usize) {}
@@ -137,6 +143,21 @@ macro_rules! register_plugin {
         }
 
         #[no_mangle]
+        pub fn pipe() -> bool {
+            let err_context = "Failed to deserialize pipe message";
+            use std::convert::TryInto;
+            use zellij_tile::shim::plugin_api::pipe_message::ProtobufPipeMessage;
+            use zellij_tile::shim::prost::Message;
+            STATE.with(|state| {
+                let protobuf_bytes: Vec<u8> = $crate::shim::object_from_stdin().unwrap();
+                let protobuf_pipe_message: ProtobufPipeMessage =
+                    ProtobufPipeMessage::decode(protobuf_bytes.as_slice()).unwrap();
+                let pipe_message = protobuf_pipe_message.try_into().unwrap();
+                state.borrow_mut().pipe(pipe_message)
+            })
+        }
+
+        #[no_mangle]
         pub fn render(rows: i32, cols: i32) {
             STATE.with(|state| {
                 state.borrow_mut().render(rows as usize, cols as usize);
diff --git a/zellij-tile/src/shim.rs b/zellij-tile/src/shim.rs
index 6252a8f9..975e3499 100644
--- a/zellij-tile/src/shim.rs
+++ b/zellij-tile/src/shim.rs
@@ -694,6 +694,38 @@ pub fn rename_session(name: &str) {
     unsafe { host_run_plugin_command() };
 }
 
+/// Unblock the input side of a pipe, requesting the next message be sent if there is one
+pub fn unblock_cli_pipe_input(pipe_name: &str) {
+    let plugin_command = PluginCommand::UnblockCliPipeInput(pipe_name.to_owned());
+    let protobuf_plugin_command: ProtobufPluginCommand = plugin_command.try_into().unwrap();
+    object_to_stdout(&protobuf_plugin_command.encode_to_vec());
+    unsafe { host_run_plugin_command() };
+}
+
+/// Block the input side of a pipe, will only be released once this or another plugin unblocks it
+pub fn block_cli_pipe_input(pipe_name: &str) {
+    let plugin_command = PluginCommand::BlockCliPipeInput(pipe_name.to_owned());
+    let protobuf_plugin_command: ProtobufPluginCommand = plugin_command.try_into().unwrap();
+    object_to_stdout(&protobuf_plugin_command.encode_to_vec());
+    unsafe { host_run_plugin_command() };
+}
+
+/// Send output to the output side of a pipe, ths does not affect the input side of same pipe
+pub fn cli_pipe_output(pipe_name: &str, output: &str) {
+    let plugin_command = PluginCommand::CliPipeOutput(pipe_name.to_owned(), output.to_owned());
+    let protobuf_plugin_command: ProtobufPluginCommand = plugin_command.try_into().unwrap();
+    object_to_stdout(&protobuf_plugin_command.encode_to_vec());
+    unsafe { host_run_plugin_command() };
+}
+
+/// Send a message to a plugin, it will be launched if it is not already running
+pub fn pipe_message_to_plugin(message_to_plugin: MessageToPlugin) {
+    let plugin_command = PluginCommand::MessageToPlugin(message_to_plugin);
+    let protobuf_plugin_command: ProtobufPluginCommand = plugin_command.try_into().unwrap();
+    object_to_stdout(&protobuf_plugin_command.encode_to_vec());
+    unsafe { host_run_plugin_command() };
+}
+
 // Utility Functions
 
 #[allow(unused)]
diff --git a/zellij-utils/assets/prost/api.action.rs b/zellij-utils/assets/prost/api.action.rs
index 908fffa7..4096b074 100644
--- a/zellij-utils/assets/prost/api.action.rs
+++ b/zellij-utils/assets/prost/api.action.rs
@@ -5,7 +5,7 @@ pub struct Action {
     pub name: i32,
     #[prost(
         oneof = "action::OptionalPayload",
-        tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46"
+        tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47"
     )]
     pub optional_payload: ::core::option::Option<action::OptionalPayload>,
 }
@@ -104,10 +104,24 @@ pub mod action {
         RenameSessionPayload(::prost::alloc::string::String),
         #[prost(message, tag = "46")]
         LaunchPluginPayload(super::LaunchOrFocusPluginPayload),
+        #[prost(message, tag = "47")]
+        MessagePayload(super::CliPipePayload),
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CliPipePayload {
+    #[prost(string, optional, tag = "1")]
+    pub name: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(string, tag = "2")]
+    pub payload: ::prost::alloc::string::String,
+    #[prost(message, repeated, tag = "3")]
+    pub args: ::prost::alloc::vec::Vec<NameAndValue>,
+    #[prost(string, optional, tag = "4")]
+    pub plugin: ::core::option::Option<::prost::alloc::string::String>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct IdAndName {
     #[prost(bytes = "vec", tag = "1")]
     pub name: ::prost::alloc::vec::Vec<u8>,
@@ -410,6 +424,7 @@ pub enum ActionName {
     BreakPaneLeft = 79,
     RenameSession = 80,
     LaunchPlugin = 81,
+    CliPipe = 82,
 }
 impl ActionName {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -500,6 +515,7 @@ impl ActionName {
             ActionName::BreakPaneLeft => "BreakPaneLeft",
             ActionName::RenameSession => "RenameSession",
             ActionName::LaunchPlugin => "LaunchPlugin",
+            ActionName::CliPipe => "CliPipe",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -587,6 +603,7 @@ impl ActionName {
             "BreakPaneLeft" => Some(Self::BreakPaneLeft),
             "RenameSession" => Some(Self::RenameSession),
             "LaunchPlugin" => Some(Self::LaunchPlugin),
+            "CliPipe" => Some(Self::CliPipe),
             _ => None,
         }
     }
diff --git a/zellij-utils/assets/prost/api.pipe_message.rs b/zellij-utils/assets/prost/api.pipe_message.rs
new file mode 100644
index 00000000..96c566ff
--- /dev/null
+++ b/zellij-utils/assets/prost/api.pipe_message.rs
@@ -0,0 +1,52 @@
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PipeMessage {
+    #[prost(enumeration = "PipeSource", tag = "1")]
+    pub source: i32,
+    #[prost(string, optional, tag = "2")]
+    pub cli_source_id: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(uint32, optional, tag = "3")]
+    pub plugin_source_id: ::core::option::Option<u32>,
+    #[prost(string, tag = "4")]
+    pub name: ::prost::alloc::string::String,
+    #[prost(string, optional, tag = "5")]
+    pub payload: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(message, repeated, tag = "6")]
+    pub args: ::prost::alloc::vec::Vec<Arg>,
+    #[prost(bool, tag = "7")]
+    pub is_private: bool,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Arg {
+    #[prost(string, tag = "1")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub value: ::prost::alloc::string::String,
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum PipeSource {
+    Cli = 0,
+    Plugin = 1,
+}
+impl PipeSource {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            PipeSource::Cli => "Cli",
+            PipeSource::Plugin => "Plugin",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "Cli" => Some(Self::Cli),
+            "Plugin" => Some(Self::Plugin),
+            _ => None,
+        }
+    }
+}
diff --git a/zellij-utils/assets/prost/api.plugin_command.rs b/zellij-utils/assets/prost/api.plugin_command.rs
index 0dd5f681..d9b528ae 100644
--- a/zellij-utils/assets/prost/api.plugin_command.rs
+++ b/zellij-utils/assets/prost/api.plugin_command.rs
@@ -5,7 +5,7 @@ pub struct PluginCommand {
     pub name: i32,
     #[prost(
         oneof = "plugin_command::Payload",
-        tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46"
+        tags = "2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50"
     )]
     pub payload: ::core::option::Option<plugin_command::Payload>,
 }
@@ -104,10 +104,64 @@ pub mod plugin_command {
         DeleteDeadSessionPayload(::prost::alloc::string::String),
         #[prost(string, tag = "46")]
         RenameSessionPayload(::prost::alloc::string::String),
+        #[prost(string, tag = "47")]
+        UnblockCliPipeInputPayload(::prost::alloc::string::String),
+        #[prost(string, tag = "48")]
+        BlockCliPipeInputPayload(::prost::alloc::string::String),
+        #[prost(message, tag = "49")]
+        CliPipeOutputPayload(super::CliPipeOutputPayload),
+        #[prost(message, tag = "50")]
+        MessageToPluginPayload(super::MessageToPluginPayload),
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CliPipeOutputPayload {
+    #[prost(string, tag = "1")]
+    pub pipe_name: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub output: ::prost::alloc::string::String,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MessageToPluginPayload {
+    #[prost(string, optional, tag = "1")]
+    pub plugin_url: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(message, repeated, tag = "2")]
+    pub plugin_config: ::prost::alloc::vec::Vec<ContextItem>,
+    #[prost(string, tag = "3")]
+    pub message_name: ::prost::alloc::string::String,
+    #[prost(string, optional, tag = "4")]
+    pub message_payload: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(message, repeated, tag = "5")]
+    pub message_args: ::prost::alloc::vec::Vec<ContextItem>,
+    #[prost(message, optional, tag = "6")]
+    pub new_plugin_args: ::core::option::Option<NewPluginArgs>,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct NewPluginArgs {
+    #[prost(bool, optional, tag = "1")]
+    pub should_float: ::core::option::Option<bool>,
+    #[prost(message, optional, tag = "2")]
+    pub pane_id_to_replace: ::core::option::Option<PaneId>,
+    #[prost(string, optional, tag = "3")]
+    pub pane_title: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(string, optional, tag = "4")]
+    pub cwd: ::core::option::Option<::prost::alloc::string::String>,
+    #[prost(bool, tag = "5")]
+    pub skip_cache: bool,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct PaneId {
+    #[prost(enumeration = "PaneType", tag = "1")]
+    pub pane_type: i32,
+    #[prost(uint32, tag = "2")]
+    pub id: u32,
+}
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SwitchSessionPayload {
     #[prost(string, optional, tag = "1")]
     pub name: ::core::option::Option<::prost::alloc::string::String>,
@@ -318,6 +372,10 @@ pub enum CommandName {
     DeleteDeadSession = 73,
     DeleteAllDeadSessions = 74,
     RenameSession = 75,
+    UnblockCliPipeInput = 76,
+    BlockCliPipeInput = 77,
+    CliPipeOutput = 78,
+    MessageToPlugin = 79,
 }
 impl CommandName {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -402,6 +460,10 @@ impl CommandName {
             CommandName::DeleteDeadSession => "DeleteDeadSession",
             CommandName::DeleteAllDeadSessions => "DeleteAllDeadSessions",
             CommandName::RenameSession => "RenameSession",
+            CommandName::UnblockCliPipeInput => "UnblockCliPipeInput",
+            CommandName::BlockCliPipeInput => "BlockCliPipeInput",
+            CommandName::CliPipeOutput => "CliPipeOutput",
+            CommandName::MessageToPlugin => "MessageToPlugin",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -483,6 +545,36 @@ impl CommandName {
             "DeleteDeadSession" => Some(Self::DeleteDeadSession),
             "DeleteAllDeadSessions" => Some(Self::DeleteAllDeadSessions),
             "RenameSession" => Some(Self::RenameSession),
+            "UnblockCliPipeInput" => Some(Self::UnblockCliPipeInput),
+            "BlockCliPipeInput" => Some(Self::BlockCliPipeInput),
+            "CliPipeOutput" => Some(Self::CliPipeOutput),
+            "MessageToPlugin" => Some(Self::MessageToPlugin),
+            _ => None,
+        }
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum PaneType {
+    Terminal = 0,
+    Plugin = 1,
+}
+impl PaneType {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            PaneType::Terminal => "Terminal",
+            PaneType::Plugin => "Plugin",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "Terminal" => Some(Self::Terminal),
+            "Plugin" => Some(Self::Plugin),
             _ => None,
         }
     }
diff --git a/zellij-utils/assets/prost/api.plugin_permission.rs b/zellij-utils/assets/prost/api.plugin_permission.rs
index d33fa950..f928625d 100644
--- a/zellij-utils/assets/prost/api.plugin_permission.rs
+++ b/zellij-utils/assets/prost/api.plugin_permission.rs
@@ -8,6 +8,8 @@ pub enum PermissionType {
     OpenTerminalsOrPlugins = 4,
     WriteToStdin = 5,
     WebAccess = 6,
+    ReadCliPipes = 7,
+    MessageAndLaunchOtherPlugins = 8,
 }
 impl PermissionType {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -23,6 +25,10 @@ impl PermissionType {
             PermissionType::OpenTerminalsOrPlugins => "OpenTerminalsOrPlugins",
             PermissionType::WriteToStdin => "WriteToStdin",
             PermissionType::WebAccess => "WebAccess",
+            PermissionType::ReadCliPipes => "ReadCliPipes",
+            PermissionType::MessageAndLaunchOtherPlugins => {
+                "MessageAndLaunchOtherPlugins"
+            }
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -35,6 +41,8 @@ impl PermissionType {
             "OpenTerminalsOrPlugins" => Some(Self::OpenTerminalsOrPlugins),
             "WriteToStdin" => Some(Self::WriteToStdin),
             "WebAccess" => Some(Self::WebAccess),
+            "ReadCliPipes" => Some(Self::ReadCliPipes),
+            "MessageAndLaunchOtherPlugins" => Some(Self::MessageAndLaunchOtherPlugins),
             _ => None,
         }
     }
diff --git a/zellij-utils/assets/prost/generated_plugin_api.rs b/zellij-utils/assets/prost/generated_plugin_api.rs
index a7e73f65..ee70c470 100644
--- a/zellij-utils/assets/prost/generated_plugin_api.rs
+++ b/zellij-utils/assets/prost/generated_plugin_api.rs
@@ -20,6 +20,9 @@ pub mod api {
     pub mod message {
         include!("api.message.rs");
     }
+    pub mod pipe_message {
+        include!("api.pipe_message.rs");
+    }
     pub mod plugin_command {
         include!("api.plugin_command.rs");
     }
diff --git a/zellij-utils/src/cli.rs b/zellij-utils/src/cli.rs
index 97a0cd77..bf8f8880 100644
--- a/zellij-utils/src/cli.rs
+++ b/zellij-utils/src/cli.rs
@@ -289,6 +289,43 @@ pub enum Sessions {
     ConvertTheme {
         old_theme_file: PathBuf,
     },
+    /// Send data to one or more plugins, launch them if they are not running.
+    #[clap(override_usage(
+r#"
+zellij pipe [OPTIONS] [--] <PAYLOAD>
+
+* Send data to a specific plugin:
+
+zellij pipe --plugin file:/path/to/my/plugin.wasm --name my_pipe_name -- my_arbitrary_data
+
+* To all running plugins (that are listening):
+
+zellij pipe --name my_pipe_name -- my_arbitrary_data
+
+* Pipe data into this command's STDIN and get output from the plugin on this command's STDOUT
+
+tail -f /tmp/my-live-logfile | zellij pipe --name logs --plugin https://example.com/my-plugin.wasm | wc -l
+"#))]
+    Pipe {
+        /// The name of the pipe
+        #[clap(short, long, value_parser, display_order(1))]
+        name: Option<String>,
+        /// The data to send down this pipe (if blank, will listen to STDIN)
+        payload: Option<String>,
+
+        #[clap(short, long, value_parser, display_order(2))]
+        /// The args of the pipe
+        args: Option<PluginUserConfiguration>, // TODO: we might want to not re-use
+        // PluginUserConfiguration
+        /// The plugin url (eg. file:/tmp/my-plugin.wasm) to direct this pipe to, if not specified,
+        /// will be sent to all plugins, if specified and is not running, the plugin will be launched
+        #[clap(short, long, value_parser, display_order(3))]
+        plugin: Option<String>,
+        /// The plugin configuration (note: the same plugin with different configuration is
+        /// considered a different plugin for the purposes of determining the pipe destination)
+        #[clap(short('c'), long, value_parser, display_order(4))]
+        plugin_configuration: Option<PluginUserConfiguration>,
+    },
 }
 
 #[derive(Debug, Subcommand, Clone, Serialize, Deserialize)]
@@ -549,4 +586,79 @@ pub enum CliAction {
     RenameSession {
         name: String,
     },
+    /// Send data to one or more plugins, launch them if they are not running.
+    #[clap(override_usage(
+r#"
+zellij action pipe [OPTIONS] [--] <PAYLOAD>
+
+* Send data to a specific plugin:
+
+zellij action pipe --plugin file:/path/to/my/plugin.wasm --name my_pipe_name -- my_arbitrary_data
+
+* To all running plugins (that are listening):
+
+zellij action pipe --name my_pipe_name -- my_arbitrary_data
+
+* Pipe data into this command's STDIN and get output from the plugin on this command's STDOUT
+
+tail -f /tmp/my-live-logfile | zellij action pipe --name logs --plugin https://example.com/my-plugin.wasm | wc -l
+"#))]
+    Pipe {
+        /// The name of the pipe
+        #[clap(short, long, value_parser, display_order(1))]
+        name: Option<String>,
+        /// The data to send down this pipe (if blank, will listen to STDIN)
+        payload: Option<String>,
+
+        #[clap(short, long, value_parser, display_order(2))]
+        /// The args of the pipe
+        args: Option<PluginUserConfiguration>, // TODO: we might want to not re-use
+        // PluginUserConfiguration
+        /// The plugin url (eg. file:/tmp/my-plugin.wasm) to direct this pipe to, if not specified,
+        /// will be sent to all plugins, if specified and is not running, the plugin will be launched
+        #[clap(short, long, value_parser, display_order(3))]
+        plugin: Option<String>,
+        /// The plugin configuration (note: the same plugin with different configuration is
+        /// considered a different plugin for the purposes of determining the pipe destination)
+        #[clap(short('c'), long, value_parser, display_order(4))]
+        plugin_configuration: Option<PluginUserConfiguration>,
+        /// Launch a new plugin even if one is already running
+        #[clap(
+            short('l'),
+            long,
+            value_parser,
+            takes_value(false),
+            default_value("false"),
+            display_order(5)
+        )]
+        force_launch_plugin: bool,
+        /// If launching a new plugin, skip cache and force-compile the plugin
+        #[clap(
+            short('s'),
+            long,
+            value_parser,
+            takes_value(false),
+            default_value("false"),
+            display_order(6)
+        )]
+        skip_plugin_cache: bool,
+        /// If launching a plugin, should it be floating or not, defaults to floating
+        #[clap(short('f'), long, value_parser, display_order(7))]
+        floating_plugin: Option<bool>,
+        /// If launching a plugin, launch it in-place (on top of the current pane)
+        #[clap(
+            short('i'),
+            long,
+            value_parser,
+            conflicts_with("floating-plugin"),
+            display_order(8)
+        )]
+        in_place_plugin: Option<bool>,
+        /// If launching a plugin, specify its working directory
+        #[clap(short('w'), long, value_parser, display_order(9))]
+        plugin_cwd: Option<PathBuf>,
+        /// If launching a plugin, specify its pane title
+        #[clap(short('t'), long, value_parser, display_order(10))]
+        plugin_title: Option<String>,
+    },
 }
diff --git a/zellij-utils/src/data.rs b/zellij-utils/src/data.rs
index bed6f474..f1e01115 100644
--- a/zellij-utils/src/data.rs
+++ b/zellij-utils/src/data.rs
@@ -538,6 +538,8 @@ pub enum Permission {
     OpenTerminalsOrPlugins,
     WriteToStdin,
     WebAccess,
+    ReadCliPipes,
+    MessageAndLaunchOtherPlugins,
 }
 
 impl PermissionType {
@@ -554,6 +556,10 @@ impl PermissionType {
             PermissionType::OpenTerminalsOrPlugins => "Start new terminals and plugins".to_owned(),
             PermissionType::WriteToStdin => "Write to standard input (STDIN)".to_owned(),
             PermissionType::WebAccess => "Make web requests".to_owned(),
+            PermissionType::ReadCliPipes => "Control command line pipes and output".to_owned(),
+            PermissionType::MessageAndLaunchOtherPlugins => {
+                "Send messages to and launch other plugins".to_owned()
+            },
         }
     }
 }
@@ -975,6 +981,86 @@ impl CommandToRun {
     }
 }
 
+#[derive(Debug, Default, Clone)]
+pub struct MessageToPlugin {
+    pub plugin_url: Option<String>,
+    pub plugin_config: BTreeMap<String, String>,
+    pub message_name: String,
+    pub message_payload: Option<String>,
+    pub message_args: BTreeMap<String, String>,
+    /// these will only be used in case we need to launch a new plugin to send this message to,
+    /// since none are running
+    pub new_plugin_args: Option<NewPluginArgs>,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct NewPluginArgs {
+    pub should_float: Option<bool>,
+    pub pane_id_to_replace: Option<PaneId>,
+    pub pane_title: Option<String>,
+    pub cwd: Option<PathBuf>,
+    pub skip_cache: bool,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum PaneId {
+    Terminal(u32),
+    Plugin(u32),
+}
+
+impl MessageToPlugin {
+    pub fn new(message_name: impl Into<String>) -> Self {
+        MessageToPlugin {
+            message_name: message_name.into(),
+            ..Default::default()
+        }
+    }
+    pub fn with_plugin_url(mut self, url: impl Into<String>) -> Self {
+        self.plugin_url = Some(url.into());
+        self
+    }
+    pub fn with_plugin_config(mut self, plugin_config: BTreeMap<String, String>) -> Self {
+        self.plugin_config = plugin_config;
+        self
+    }
+    pub fn with_payload(mut self, payload: impl Into<String>) -> Self {
+        self.message_payload = Some(payload.into());
+        self
+    }
+    pub fn with_args(mut self, args: BTreeMap<String, String>) -> Self {
+        self.message_args = args;
+        self
+    }
+    pub fn new_plugin_instance_should_float(mut self, should_float: bool) -> Self {
+        let mut new_plugin_args = self.new_plugin_args.get_or_insert_with(Default::default);
+        new_plugin_args.should_float = Some(should_float);
+        self
+    }
+    pub fn new_plugin_instance_should_replace_pane(mut self, pane_id: PaneId) -> Self {
+        let mut new_plugin_args = self.new_plugin_args.get_or_insert_with(Default::default);
+        new_plugin_args.pane_id_to_replace = Some(pane_id);
+        self
+    }
+    pub fn new_plugin_instance_should_have_pane_title(
+        mut self,
+        pane_title: impl Into<String>,
+    ) -> Self {
+        let mut new_plugin_args = self.new_plugin_args.get_or_insert_with(Default::default);
+        new_plugin_args.pane_title = Some(pane_title.into());
+        self
+    }
+    pub fn new_plugin_instance_should_have_cwd(mut self, cwd: PathBuf) -> Self {
+        let mut new_plugin_args = self.new_plugin_args.get_or_insert_with(Default::default);
+        new_plugin_args.cwd = Some(cwd);
+        self
+    }
+    pub fn new_plugin_instance_should_skip_cache(mut self) -> Self {
+        let mut new_plugin_args = self.new_plugin_args.get_or_insert_with(Default::default);
+        new_plugin_args.skip_cache = true;
+        self
+    }
+}
+
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct ConnectToSession {
     pub name: Option<String>,
@@ -1014,6 +1100,39 @@ pub enum HttpVerb {
     Delete,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum PipeSource {
+    Cli(String), // String is the pipe_id of the CLI pipe (used for blocking/unblocking)
+    Plugin(u32), // u32 is the lugin id
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct PipeMessage {
+    pub source: PipeSource,
+    pub name: String,
+    pub payload: Option<String>,
+    pub args: BTreeMap<String, String>,
+    pub is_private: bool,
+}
+
+impl PipeMessage {
+    pub fn new(
+        source: PipeSource,
+        name: impl Into<String>,
+        payload: &Option<String>,
+        args: &Option<BTreeMap<String, String>>,
+        is_private: bool,
+    ) -> Self {
+        PipeMessage {
+            source,
+            name: name.into(),
+            payload: payload.clone(),
+            args: args.clone().unwrap_or_else(|| Default::default()),
+            is_private,
+        }
+    }
+}
+
 #[derive(Debug, Clone, EnumDiscriminants, ToString)]
 #[strum_discriminants(derive(EnumString, Hash, Serialize, Deserialize))]
 #[strum_discriminants(name(CommandType))]
@@ -1104,5 +1223,9 @@ pub enum PluginCommand {
         Vec<u8>,                  // body
         BTreeMap<String, String>, // context
     ),
-    RenameSession(String), // String -> new session name
+    RenameSession(String),         // String -> new session name
+    UnblockCliPipeInput(String),   // String => pipe name
+    BlockCliPipeInput(String),     // String => pipe name
+    CliPipeOutput(String, String), // String => pipe name, String => output
+    MessageToPlugin(MessageToPlugin),
 }
diff --git a/zellij-utils/src/errors.rs b/zellij-utils/src/errors.rs
index aff97f2d..26ad521c 100644
--- a/zellij-utils/src/errors.rs
+++ b/zellij-utils/src/errors.rs
@@ -393,6 +393,11 @@ pub enum PluginContext {
     PermissionRequestResult,
     DumpLayout,
     LogLayoutToHd,
+    CliPipe,
+    Message,
+    CachePluginEvents,
+    MessageFromPlugin,
+    UnblockCliPipes,
 }
 
 /// Stack call representations corresponding to the different types of [`ClientInstruction`]s.
@@ -413,6 +418,8 @@ pub enum ClientContext {
     DoneParsingStdinQuery,
     SwitchSession,
     SetSynchronisedOutput,
+    UnblockCliPipeInput,
+    CliPipeOutput,
 }
 
 /// Stack call representations corresponding to the different types of [`ServerInstruction`]s.
@@ -430,7 +437,11 @@ pub enum ServerContext {
     ConnStatus,
     ActiveClients,
     Log,
+    LogError,
     SwitchSession,
+    UnblockCliPipeInput,
+    CliPipeOutput,
+    AssociatePipeWithClient,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
diff --git a/zellij-utils/src/input/actions.rs b/zellij-utils/src/input/actions.rs
index 8af5d28e..74c6c1e2 100644
--- a/zellij-utils/src/input/actions.rs
+++ b/zellij-utils/src/input/actions.rs
@@ -13,6 +13,8 @@ use crate::input::config::{Config, ConfigError, KdlError};
 use crate::input::options::OnForceClose;
 use miette::{NamedSource, Report};
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+use uuid::Uuid;
 
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -256,6 +258,20 @@ pub enum Action {
     BreakPaneRight,
     BreakPaneLeft,
     RenameSession(String),
+    CliPipe {
+        pipe_id: String,
+        name: Option<String>,
+        payload: Option<String>,
+        args: Option<BTreeMap<String, String>>,
+        plugin: Option<String>,
+        configuration: Option<BTreeMap<String, String>>,
+        launch_new: bool,
+        skip_cache: bool,
+        floating: Option<bool>,
+        in_place: Option<bool>,
+        cwd: Option<PathBuf>,
+        pane_title: Option<String>,
+    },
 }
 
 impl Action {
@@ -582,6 +598,41 @@ impl Action {
                 )])
             },
             CliAction::RenameSession { name } => Ok(vec![Action::RenameSession(name)]),
+            CliAction::Pipe {
+                name,
+                payload,
+                args,
+                plugin,
+                plugin_configuration,
+                force_launch_plugin,
+                skip_plugin_cache,
+                floating_plugin,
+                in_place_plugin,
+                plugin_cwd,
+                plugin_title,
+            } => {
+                let current_dir = get_current_dir();
+                let cwd = plugin_cwd
+                    .map(|cwd| current_dir.join(cwd))
+                    .or_else(|| Some(current_dir));
+                let skip_cache = skip_plugin_cache;
+                let pipe_id = Uuid::new_v4().to_string();
+                Ok(vec![Action::CliPipe {
+                    pipe_id,
+                    name,
+                    payload,
+                    args: args.map(|a| a.inner().clone()), // TODO: no clone somehow
+                    plugin,
+                    configuration: plugin_configuration.map(|a| a.inner().clone()), // TODO: no clone
+                    // somehow
+                    launch_new: force_launch_plugin,
+                    floating: floating_plugin,
+                    in_place: in_place_plugin,
+                    cwd,
+                    pane_title: plugin_title,
+                    skip_cache,
+                }])
+            },
         }
     }
 }
diff --git a/zellij-utils/src/input/layout.rs b/zellij-utils/src/input/layout.rs
index 40ca7209..82dab15e 100644
--- a/zellij-utils/src/input/layout.rs
+++ b/zellij-utils/src/input/layout.rs
@@ -265,6 +265,9 @@ impl PluginUserConfiguration {
     pub fn inner(&self) -> &BTreeMap<String, String> {
         &self.0
     }
+    pub fn insert(&mut self, config_key: impl Into<String>, config_value: impl Into<String>) {
+        self.0.insert(config_key.into(), config_value.into());
+    }
 }
 
 impl FromStr for PluginUserConfiguration {
diff --git a/zellij-utils/src/ipc.rs b/zellij-utils/src/ipc.rs
index b6e83722..902d0235 100644
--- a/zellij-utils/src/ipc.rs
+++ b/zellij-utils/src/ipc.rs
@@ -103,6 +103,8 @@ pub enum ServerToClientMsg {
     Log(Vec<String>),
     LogError(Vec<String>),
     SwitchSession(ConnectToSession),
+    UnblockCliPipeInput(String),   // String -> pipe name
+    CliPipeOutput(String, String), // String -> pipe name, String -> Output
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
diff --git a/zellij-utils/src/lib.rs b/zellij-utils/src/lib.rs
index 05752e21..de5848ec 100644
--- a/zellij-utils/src/lib.rs
+++ b/zellij-utils/src/lib.rs
@@ -27,7 +27,7 @@ pub mod logging; // Requires log4rs
 pub use ::{
     anyhow, async_channel, async_std, clap, common_path, humantime, interprocess, lazy_static,
     libc, miette, nix, notify_debouncer_full, regex, serde, signal_hook, surf, tempfile, termwiz,
-    vte,
+    url, uuid, vte,
 };
 
 pub use ::prost;
diff --git a/zellij-utils/src/plugin_api/action.proto b/zellij-utils/src/plugin_api/action.proto
index da10d82a..0ed5b3b7 100644
--- a/zellij-utils/src/plugin_api/action.proto
+++ b/zellij-utils/src/plugin_api/action.proto
@@ -53,9 +53,17 @@ message Action {
     IdAndName rename_tab_payload = 44;
     string rename_session_payload = 45;
     LaunchOrFocusPluginPayload launch_plugin_payload = 46;
+    CliPipePayload message_payload = 47;
   }
 }
 
+message CliPipePayload {
+  optional string name = 1;
+  string payload = 2;
+  repeated NameAndValue args = 3;
+  optional string plugin = 4;
+}
+
 message IdAndName {
   bytes name = 1;
   uint32 id = 2;
@@ -227,6 +235,7 @@ enum ActionName {
     BreakPaneLeft = 79;
     RenameSession = 80;
     LaunchPlugin = 81;
+    CliPipe = 82;
 }
 
 message Position {
diff --git a/zellij-utils/src/plugin_api/action.rs b/zellij-utils/src/plugin_api/action.rs
index 71eaaa13..82de4c7e 100644
--- a/zellij-utils/src/plugin_api/action.rs
+++ b/zellij-utils/src/plugin_api/action.rs
@@ -1246,6 +1246,7 @@ impl TryFrom<Action> for ProtobufAction {
             | Action::Deny
             | Action::Copy
             | Action::DumpLayout
+            | Action::CliPipe { .. }
             | Action::SkipConfirm(..) => Err("Unsupported action"),
         }
     }
diff --git a/zellij-utils/src/plugin_api/mod.rs b/zellij-utils/src/plugin_api/mod.rs
index 40057fde..0e26485e 100644
--- a/zellij-utils/src/plugin_api/mod.rs
+++ b/zellij-utils/src/plugin_api/mod.rs
@@ -5,6 +5,7 @@ pub mod file;
 pub mod input_mode;
 pub mod key;
 pub mod message;
+pub mod pipe_message;
 pub mod plugin_command;
 pub mod plugin_ids;
 pub mod plugin_permission;
diff --git a/zellij-utils/src/plugin_api/pipe_message.proto b/zellij-utils/src/plugin_api/pipe_message.proto
new file mode 100644
index 00000000..5f488a75
--- /dev/null
+++ b/zellij-utils/src/plugin_api/pipe_message.proto
@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package api.pipe_message;
+
+message PipeMessage {
+    PipeSource source = 1;
+    optional string cli_source_id = 2;
+    optional uint32 plugin_source_id = 3;
+    string name = 4;
+    optional string payload = 5;
+    repeated Arg args = 6;
+    bool is_private = 7;
+}
+
+enum PipeSource {
+  Cli = 0;
+  Plugin = 1;
+}
+
+message Arg {
+  string key = 1;
+  string value = 2;
+}
diff --git a/zellij-utils/src/plugin_api/pipe_message.rs b/zellij-utils/src/plugin_api/pipe_message.rs
new file mode 100644
index 00000000..dbfe4930
--- /dev/null
+++ b/zellij-utils/src/plugin_api/pipe_message.rs
@@ -0,0 +1,71 @@
+pub use super::generated_api::api::pipe_message::{
+    Arg as ProtobufArg, PipeMessage as ProtobufPipeMessage, PipeSource as ProtobufPipeSource,
+};
+use crate::data::{PipeMessage, PipeSource};
+
+use std::convert::TryFrom;
+
+impl TryFrom<ProtobufPipeMessage> for PipeMessage {
+    type Error = &'static str;
+    fn try_from(protobuf_pipe_message: ProtobufPipeMessage) -> Result<Self, &'static str> {
+        let source = match (
+            ProtobufPipeSource::from_i32(protobuf_pipe_message.source),
+            protobuf_pipe_message.cli_source_id,
+            protobuf_pipe_message.plugin_source_id,
+        ) {
+            (Some(ProtobufPipeSource::Cli), Some(cli_source_id), _) => {
+                PipeSource::Cli(cli_source_id)
+            },
+            (Some(ProtobufPipeSource::Plugin), _, Some(plugin_source_id)) => {
+                PipeSource::Plugin(plugin_source_id)
+            },
+            _ => return Err("Invalid PipeSource or payload"),
+        };
+        let name = protobuf_pipe_message.name;
+        let payload = protobuf_pipe_message.payload;
+        let args = protobuf_pipe_message
+            .args
+            .into_iter()
+            .map(|arg| (arg.key, arg.value))
+            .collect();
+        let is_private = protobuf_pipe_message.is_private;
+        Ok(PipeMessage {
+            source,
+            name,
+            payload,
+            args,
+            is_private,
+        })
+    }
+}
+
+impl TryFrom<PipeMessage> for ProtobufPipeMessage {
+    type Error = &'static str;
+    fn try_from(pipe_message: PipeMessage) -> Result<Self, &'static str> {
+        let (source, cli_source_id, plugin_source_id) = match pipe_message.source {
+            PipeSource::Cli(input_pipe_id) => {
+                (ProtobufPipeSource::Cli as i32, Some(input_pipe_id), None)
+            },
+            PipeSource::Plugin(plugin_id) => {
+                (ProtobufPipeSource::Plugin as i32, None, Some(plugin_id))
+            },
+        };
+        let name = pipe_message.name;
+        let payload = pipe_message.payload;
+        let args: Vec<_> = pipe_message
+            .args
+            .into_iter()
+            .map(|(key, value)| ProtobufArg { key, value })
+            .collect();
+        let is_private = pipe_message.is_private;
+        Ok(ProtobufPipeMessage {
+            source,
+            cli_source_id,
+            plugin_source_id,
+            name,
+            payload,
+            args,
+            is_private,
+        })
+    }
+}
diff --git a/zellij-utils/src/plugin_api/plugin_command.proto b/zellij-utils/src/plugin_api/plugin_command.proto
index 53994a88..6ffb0345 100644
--- a/zellij-utils/src/plugin_api/plugin_command.proto
+++ b/zellij-utils/src/plugin_api/plugin_command.proto
@@ -87,6 +87,10 @@ enum CommandName {
   DeleteDeadSession = 73;
   DeleteAllDeadSessions = 74;
   RenameSession = 75;
+  UnblockCliPipeInput = 76;
+  BlockCliPipeInput = 77;
+  CliPipeOutput = 78;
+  MessageToPlugin = 79;
 }
 
 message PluginCommand {
@@ -137,9 +141,45 @@ message PluginCommand {
     WebRequestPayload web_request_payload = 44;
     string delete_dead_session_payload = 45;
     string rename_session_payload = 46;
+    string unblock_cli_pipe_input_payload = 47;
+    string block_cli_pipe_input_payload = 48;
+    CliPipeOutputPayload cli_pipe_output_payload = 49;
+    MessageToPluginPayload message_to_plugin_payload = 50;
   }
 }
 
+message CliPipeOutputPayload {
+  string pipe_name = 1;
+  string output = 2;
+}
+
+message MessageToPluginPayload {
+  optional string plugin_url = 1;
+  repeated ContextItem plugin_config = 2;
+  string message_name = 3;
+  optional string message_payload = 4;
+  repeated ContextItem message_args = 5;
+  optional NewPluginArgs new_plugin_args = 6;
+}
+
+message NewPluginArgs {
+    optional bool should_float = 1;
+    optional PaneId pane_id_to_replace = 2;
+    optional string pane_title = 3;
+    optional string cwd = 4;
+    bool skip_cache = 5;
+}
+
+message PaneId {
+  PaneType pane_type = 1;
+  uint32 id = 2;
+}
+
+enum PaneType {
+  Terminal = 0;
+  Plugin = 1;
+}
+
 message SwitchSessionPayload {
   optional string name = 1;
   optional uint32 tab_position = 2;
diff --git a/zellij-utils/src/plugin_api/plugin_command.rs b/zellij-utils/src/plugin_api/plugin_command.rs
index ed476687..fa570a26 100644
--- a/zellij-utils/src/plugin_api/plugin_command.rs
+++ b/zellij-utils/src/plugin_api/plugin_command.rs
@@ -3,9 +3,11 @@ pub use super::generated_api::api::{
     event::{EventNameList as ProtobufEventNameList, Header},
     input_mode::InputMode as ProtobufInputMode,
     plugin_command::{
-        plugin_command::Payload, CommandName, ContextItem, EnvVariable, ExecCmdPayload,
-        HttpVerb as ProtobufHttpVerb, IdAndNewName, MovePayload, OpenCommandPanePayload,
-        OpenFilePayload, PluginCommand as ProtobufPluginCommand, PluginMessagePayload,
+        plugin_command::Payload, CliPipeOutputPayload, CommandName, ContextItem, EnvVariable,
+        ExecCmdPayload, HttpVerb as ProtobufHttpVerb, IdAndNewName, MessageToPluginPayload,
+        MovePayload, NewPluginArgs as ProtobufNewPluginArgs, OpenCommandPanePayload,
+        OpenFilePayload, PaneId as ProtobufPaneId, PaneType as ProtobufPaneType,
+        PluginCommand as ProtobufPluginCommand, PluginMessagePayload,
         RequestPluginPermissionPayload, ResizePayload, RunCommandPayload, SetTimeoutPayload,
         SubscribePayload, SwitchSessionPayload, SwitchTabToPayload, UnsubscribePayload,
         WebRequestPayload,
@@ -14,7 +16,10 @@ pub use super::generated_api::api::{
     resize::ResizeAction as ProtobufResizeAction,
 };
 
-use crate::data::{ConnectToSession, HttpVerb, PermissionType, PluginCommand};
+use crate::data::{
+    ConnectToSession, HttpVerb, MessageToPlugin, NewPluginArgs, PaneId, PermissionType,
+    PluginCommand,
+};
 
 use std::collections::BTreeMap;
 use std::convert::TryFrom;
@@ -42,6 +47,33 @@ impl Into<ProtobufHttpVerb> for HttpVerb {
     }
 }
 
+impl TryFrom<ProtobufPaneId> for PaneId {
+    type Error = &'static str;
+    fn try_from(protobuf_pane_id: ProtobufPaneId) -> Result<Self, &'static str> {
+        match ProtobufPaneType::from_i32(protobuf_pane_id.pane_type) {
+            Some(ProtobufPaneType::Terminal) => Ok(PaneId::Terminal(protobuf_pane_id.id)),
+            Some(ProtobufPaneType::Plugin) => Ok(PaneId::Plugin(protobuf_pane_id.id)),
+            None => Err("Failed to convert PaneId"),
+        }
+    }
+}
+
+impl TryFrom<PaneId> for ProtobufPaneId {
+    type Error = &'static str;
+    fn try_from(pane_id: PaneId) -> Result<Self, &'static str> {
+        match pane_id {
+            PaneId::Terminal(id) => Ok(ProtobufPaneId {
+                pane_type: ProtobufPaneType::Terminal as i32,
+                id,
+            }),
+            PaneId::Plugin(id) => Ok(ProtobufPaneId {
+                pane_type: ProtobufPaneType::Plugin as i32,
+                id,
+            }),
+        }
+    }
+}
+
 impl TryFrom<ProtobufPluginCommand> for PluginCommand {
     type Error = &'static str;
     fn try_from(protobuf_plugin_command: ProtobufPluginCommand) -> Result<Self, &'static str> {
@@ -641,6 +673,62 @@ impl TryFrom<ProtobufPluginCommand> for PluginCommand {
                 },
                 _ => Err("Mismatched payload for RenameSession"),
             },
+            Some(CommandName::UnblockCliPipeInput) => match protobuf_plugin_command.payload {
+                Some(Payload::UnblockCliPipeInputPayload(pipe_name)) => {
+                    Ok(PluginCommand::UnblockCliPipeInput(pipe_name))
+                },
+                _ => Err("Mismatched payload for UnblockPipeInput"),
+            },
+            Some(CommandName::BlockCliPipeInput) => match protobuf_plugin_command.payload {
+                Some(Payload::BlockCliPipeInputPayload(pipe_name)) => {
+                    Ok(PluginCommand::BlockCliPipeInput(pipe_name))
+                },
+                _ => Err("Mismatched payload for BlockPipeInput"),
+            },
+            Some(CommandName::CliPipeOutput) => match protobuf_plugin_command.payload {
+                Some(Payload::CliPipeOutputPayload(CliPipeOutputPayload { pipe_name, output })) => {
+                    Ok(PluginCommand::CliPipeOutput(pipe_name, output))
+                },
+                _ => Err("Mismatched payload for PipeOutput"),
+            },
+            Some(CommandName::MessageToPlugin) => match protobuf_plugin_command.payload {
+                Some(Payload::MessageToPluginPayload(MessageToPluginPayload {
+                    plugin_url,
+                    plugin_config,
+                    message_name,
+                    message_payload,
+                    message_args,
+                    new_plugin_args,
+                })) => {
+                    let plugin_config: BTreeMap<String, String> = plugin_config
+                        .into_iter()
+                        .map(|e| (e.name, e.value))
+                        .collect();
+                    let message_args: BTreeMap<String, String> = message_args
+                        .into_iter()
+                        .map(|e| (e.name, e.value))
+                        .collect();
+                    Ok(PluginCommand::MessageToPlugin(MessageToPlugin {
+                        plugin_url,
+                        plugin_config,
+                        message_name,
+                        message_payload,
+                        message_args,
+                        new_plugin_args: new_plugin_args.and_then(|protobuf_new_plugin_args| {
+                            Some(NewPluginArgs {
+                                should_float: protobuf_new_plugin_args.should_float,
+                                pane_id_to_replace: protobuf_new_plugin_args
+                                    .pane_id_to_replace
+                                    .and_then(|p_id| PaneId::try_from(p_id).ok()),
+                                pane_title: protobuf_new_plugin_args.pane_title,
+                                cwd: protobuf_new_plugin_args.cwd.map(|cwd| PathBuf::from(cwd)),
+                                skip_cache: protobuf_new_plugin_args.skip_cache,
+                            })
+                        }),
+                    }))
+                },
+                _ => Err("Mismatched payload for PipeOutput"),
+            },
             None => Err("Unrecognized plugin command"),
         }
     }
@@ -1069,6 +1157,54 @@ impl TryFrom<PluginCommand> for ProtobufPluginCommand {
                 name: CommandName::RenameSession as i32,
                 payload: Some(Payload::RenameSessionPayload(new_session_name)),
             }),
+            PluginCommand::UnblockCliPipeInput(pipe_name) => Ok(ProtobufPluginCommand {
+                name: CommandName::UnblockCliPipeInput as i32,
+                payload: Some(Payload::UnblockCliPipeInputPayload(pipe_name)),
+            }),
+            PluginCommand::BlockCliPipeInput(pipe_name) => Ok(ProtobufPluginCommand {
+                name: CommandName::BlockCliPipeInput as i32,
+                payload: Some(Payload::BlockCliPipeInputPayload(pipe_name)),
+            }),
+            PluginCommand::CliPipeOutput(pipe_name, output) => Ok(ProtobufPluginCommand {
+                name: CommandName::CliPipeOutput as i32,
+                payload: Some(Payload::CliPipeOutputPayload(CliPipeOutputPayload {
+                    pipe_name,
+                    output,
+                })),
+            }),
+            PluginCommand::MessageToPlugin(message_to_plugin) => {
+                let plugin_config: Vec<_> = message_to_plugin
+                    .plugin_config
+                    .into_iter()
+                    .map(|(name, value)| ContextItem { name, value })
+                    .collect();
+                let message_args: Vec<_> = message_to_plugin
+                    .message_args
+                    .into_iter()
+                    .map(|(name, value)| ContextItem { name, value })
+                    .collect();
+                Ok(ProtobufPluginCommand {
+                    name: CommandName::MessageToPlugin as i32,
+                    payload: Some(Payload::MessageToPluginPayload(MessageToPluginPayload {
+                        plugin_url: message_to_plugin.plugin_url,
+                        plugin_config,
+                        message_name: message_to_plugin.message_name,
+                        message_payload: message_to_plugin.message_payload,
+                        message_args,
+                        new_plugin_args: message_to_plugin.new_plugin_args.map(|m_t_p| {
+                            ProtobufNewPluginArgs {
+                                should_float: m_t_p.should_float,
+                                pane_id_to_replace: m_t_p
+                                    .pane_id_to_replace
+                                    .and_then(|p_id| ProtobufPaneId::try_from(p_id).ok()),
+                                pane_title: m_t_p.pane_title,
+                                cwd: m_t_p.cwd.map(|cwd| cwd.display().to_string()),
+                                skip_cache: m_t_p.skip_cache,
+                            }
+                        }),
+                    })),
+                })
+            },
         }
     }
 }
diff --git a/zellij-utils/src/plugin_api/plugin_permission.proto b/zellij-utils/src/plugin_api/plugin_permission.proto
index a796c748..761384c1 100644
--- a/zellij-utils/src/plugin_api/plugin_permission.proto
+++ b/zellij-utils/src/plugin_api/plugin_permission.proto
@@ -10,4 +10,6 @@ enum PermissionType {
   OpenTerminalsOrPlugins = 4;
   WriteToStdin = 5;
   WebAccess = 6;
+  ReadCliPipes = 7;
+  MessageAndLaunchOtherPlugins = 8;
 }
diff --git a/zellij-utils/src/plugin_api/plugin_permission.rs b/zellij-utils/src/plugin_api/plugin_permission.rs
index c9f0d49f..4f258ac2 100644
--- a/zellij-utils/src/plugin_api/plugin_permission.rs
+++ b/zellij-utils/src/plugin_api/plugin_permission.rs
@@ -20,6 +20,10 @@ impl TryFrom<ProtobufPermissionType> for PermissionType {
             },
             ProtobufPermissionType::WriteToStdin => Ok(PermissionType::WriteToStdin),
             ProtobufPermissionType::WebAccess => Ok(PermissionType::WebAccess),
+            ProtobufPermissionType::ReadCliPipes => Ok(PermissionType::ReadCliPipes),
+            ProtobufPermissionType::MessageAndLaunchOtherPlugins => {
+                Ok(PermissionType::MessageAndLaunchOtherPlugins)
+            },
         }
     }
 }
@@ -41,6 +45,10 @@ impl TryFrom<PermissionType> for ProtobufPermissionType {
             },
             PermissionType::WriteToStdin => Ok(ProtobufPermissionType::WriteToStdin),
             PermissionType::WebAccess => Ok(ProtobufPermissionType::WebAccess),
+            PermissionType::ReadCliPipes => Ok(ProtobufPermissionType::ReadCliPipes),
+            PermissionType::MessageAndLaunchOtherPlugins => {
+                Ok(ProtobufPermissionType::MessageAndLaunchOtherPlugins)
+            },
         }
     }
 }