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

21 files changed, 773 insertions, 8 deletions

diff --git a/zellij-utils/assets/prost/api.action.rs b/zellij-utils/assets/prost/api.action.rs
index 908fffa77..4096b0740 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 000000000..96c566ff8
--- /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 0dd5f6814..d9b528ae6 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 d33fa950a..f928625da 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 a7e73f652..ee70c4709 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 97a0cd778..bf8f8880d 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 bed6f4747..f1e01115d 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<