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

16 files changed, 599 insertions, 6 deletions

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)
+            },
         }
     }
 }