feat: new ya.exec() API and ya exec subcommand to execute an action and await its result (#3780)

CHANGELOG.md

@@ -20,6 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/):
 - Certificate authentication for SFTP VFS provider ([#3716])
 - New `hovered` condition specifying different icons for hovered files ([#3728])
 - Allow using `ps.sub()` in `init.lua` directly without a plugin ([#3638])
+- New `ya.exec()` API and `ya exec` subcommand to execute an action and await its result ([#3780])
 - New `sort_fallback` option to control fallback sorting behavior ([#3077])
 - New `fs.access()` API to access the filesystem ([#3668])
 - New `relay-notify-push` DDS event to customize the notification handler ([#3642])
@@ -1684,3 +1685,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/):
 [#3748]: https://github.com/sxyazi/yazi/pull/3748
 [#3757]: https://github.com/sxyazi/yazi/pull/3757
 [#3765]: https://github.com/sxyazi/yazi/pull/3765
+[#3780]: https://github.com/sxyazi/yazi/pull/3780

Cargo.lock

@@ -4626,9 +4626,9 @@ dependencies = [
 
 [[package]]
 name = "toml"
-version = "1.0.6+spec-1.1.0"
+version = "1.0.7+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "399b1124a3c9e16766831c6bba21e50192572cdd98706ea114f9502509686ffc"
+checksum = "dd28d57d8a6f6e458bc0b8784f8fdcc4b99a437936056fa122cb234f18656a96"
 dependencies = [
  "indexmap 2.13.0",
  "serde_core",
@@ -4641,27 +4641,27 @@ dependencies = [
 
 [[package]]
 name = "toml_datetime"
-version = "1.0.0+spec-1.1.0"
+version = "1.0.1+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32c2555c699578a4f59f0cc68e5116c8d7cabbd45e1409b989d4be085b53f13e"
+checksum = "9b320e741db58cac564e26c607d3cc1fdc4a88fd36c879568c07856ed83ff3e9"
 dependencies = [
  "serde_core",
 ]
 
 [[package]]
 name = "toml_parser"
-version = "1.0.9+spec-1.1.0"
+version = "1.0.10+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "702d4415e08923e7e1ef96cd5727c0dfed80b4d2fa25db9647fe5eb6f7c5a4c4"
+checksum = "7df25b4befd31c4816df190124375d5a20c6b6921e2cad937316de3fccd63420"
 dependencies = [
  "winnow",
 ]
 
 [[package]]
 name = "toml_writer"
-version = "1.0.6+spec-1.1.0"
+version = "1.0.7+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab16f14aed21ee8bfd8ec22513f7287cd4a91aa92e44edfe2c17ddd004e92607"
+checksum = "f17aaa1c6e3dc22b1da4b6bba97d066e354c7945cac2f7852d4e4e7ca7a6b56d"
 
 [[package]]
 name = "tracing"
@@ -5541,9 +5541,9 @@ checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
 
 [[package]]
 name = "winnow"
-version = "0.7.15"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945"
+checksum = "a90e88e4667264a994d34e6d1ab2d26d398dcdca8b7f52bec8668957517fc7d8"
 
 [[package]]
 name = "wit-bindgen"
@@ -5773,6 +5773,7 @@ dependencies = [
  "clap_complete_fig",
  "clap_complete_nushell",
  "crossterm 0.29.0",
+ "hashbrown 0.16.1",
  "serde",
  "serde_json",
  "tokio",
@@ -6015,7 +6016,6 @@ version = "26.2.2"
 dependencies = [
  "ansi-to-tui",
  "anyhow",
- "crossterm 0.29.0",
  "futures",
  "hashbrown 0.16.1",
  "libc",
@@ -6151,7 +6151,6 @@ version = "26.2.2"
 dependencies = [
  "anyhow",
  "crossterm 0.29.0",
- "libc",
  "ratatui",
  "tokio",
  "yazi-config",

Cargo.toml

@@ -72,7 +72,7 @@ thiserror           = "2.0.18"
 tokio               = { version = "1.50.0", features = [ "full" ] }
 tokio-stream        = "0.1.18"
 tokio-util          = "0.7.18"
-toml                = { version = "1.0.6" }
+toml                = { version = "1.0.7" }
 tracing             = { version = "0.1.44", features = [ "max_level_debug", "release_max_level_debug" ] }
 twox-hash           = { version = "2.1.2", default-features = false, features = [ "std", "random", "xxhash3_128" ] }
 typed-path          = "0.12.3"

yazi-cli/Cargo.toml

@@ -34,6 +34,7 @@ yazi-shim   = { path = "../yazi-shim", version = "26.2.2" }
 anyhow     = { workspace = true }
 clap       = { workspace = true }
 crossterm  = { workspace = true }
+hashbrown  = { workspace = true }
 serde      = { workspace = true }
 serde_json = { workspace = true }
 tokio      = { workspace = true }

yazi-cli/src/args.rs

@@ -17,10 +17,12 @@ pub(super) struct Args {
 
 #[derive(Subcommand)]
 pub(super) enum Command {
-	/// Emit a command to be executed by the current instance.
+	/// Emit an action to be executed by the current instance.
 	Emit(CommandEmit),
-	/// Emit a command to be executed by the specified instance.
+	/// Emit an action to be executed by the specified instance.
 	EmitTo(CommandEmitTo),
+	/// Execute an action on the current instance and print its result.
+	Exec(CommandExec),
 	/// Manage packages.
 	#[command(subcommand)]
 	Pkg(CommandPkg),
@@ -34,9 +36,9 @@ pub(super) enum Command {
 
 #[derive(clap::Args)]
 pub(super) struct CommandEmit {
-	/// Name of the command.
+	/// Name of the action.
 	pub(super) name: String,
-	/// Arguments of the command.
+	/// Arguments of the action.
 	#[arg(allow_hyphen_values = true, trailing_var_arg = true)]
 	pub(super) args: Vec<OsString>,
 }
@@ -45,13 +47,22 @@ pub(super) struct CommandEmit {
 pub(super) struct CommandEmitTo {
 	/// Receiver ID.
 	pub(super) receiver: Id,
-	/// Name of the command.
+	/// Name of the action.
 	pub(super) name:     String,
-	/// Arguments of the command.
+	/// Arguments of the action.
 	#[arg(allow_hyphen_values = true, trailing_var_arg = true)]
 	pub(super) args:     Vec<OsString>,
 }
 
+#[derive(clap::Args)]
+pub(super) struct CommandExec {
+	/// Name of the action.
+	pub(super) name: String,
+	/// Arguments of the action.
+	#[arg(allow_hyphen_values = true, trailing_var_arg = true)]
+	pub(super) args: Vec<OsString>,
+}
+
 #[derive(Subcommand)]
 pub(super) enum CommandPkg {
 	/// Add packages.
@@ -156,6 +167,29 @@ macro_rules! impl_emit_body {
 	};
 }
 
+macro_rules! impl_exec_body {
+	($name:ident) => {
+		impl $name {
+			#[allow(dead_code)]
+			pub(super) fn body(self, reply_to: Id) -> Result<String> {
+				#[derive(serde::Serialize)]
+				#[serde(untagged)]
+				enum Elem {
+					Id(Id),
+					Name(String),
+					Arg(Vec<u8>),
+				}
+
+				let action: Vec<_> = [Elem::Id(reply_to), Elem::Name(self.name)]
+					.into_iter()
+					.chain(self.args.into_iter().map(|s| Elem::Arg(s.into_encoded_bytes())))
+					.collect();
+				Ok(serde_json::to_string(&action)?)
+			}
+		}
+	};
+}
+
 macro_rules! impl_pub_body {
 	($name:ident) => {
 		impl $name {
@@ -178,5 +212,7 @@ macro_rules! impl_pub_body {
 impl_emit_body!(CommandEmit);
 impl_emit_body!(CommandEmitTo);
 
+impl_exec_body!(CommandExec);
+
 impl_pub_body!(CommandPub);
 impl_pub_body!(CommandPubTo);

yazi-cli/src/dds/draw.rs

@@ -0,0 +1,42 @@
+use anyhow::{Context, Result};
+use hashbrown::HashSet;
+use tokio::{io::AsyncWriteExt, time};
+use yazi_dds::{ClientReader, Payload, Stream, ember::EmberHi};
+use yazi_macro::try_format;
+
+use crate::dds::Dds;
+
+impl Dds {
+	/// Connect to an existing server and listen in on the messages that are being
+	/// sent by other yazi instances:
+	///   - If no server is running, fail right away;
+	///   - If a server is closed, attempt to reconnect forever.
+	pub(crate) async fn draw(kinds: HashSet<&str>) -> Result<()> {
+		async fn make(kinds: &HashSet<&str>) -> Result<ClientReader> {
+			let (lines, mut writer) = Stream::connect().await?;
+			let hi = Payload::new(EmberHi::borrowed(kinds.iter().copied()));
+			writer.write_all(try_format!("{hi}\n")?.as_bytes()).await?;
+			writer.flush().await?;
+			Ok(lines)
+		}
+
+		let mut lines = make(&kinds).await.context("No running Yazi instance found")?;
+		loop {
+			match lines.next_line().await? {
+				Some(s) => {
+					let kind = s.split(',').next();
+					if matches!(kind, Some(kind) if kinds.contains(kind)) {
+						println!("{s}");
+					}
+				}
+				None => loop {
+					time::sleep(time::Duration::from_secs(1)).await;
+					if let Ok(new) = make(&kinds).await {
+						lines = new;
+						break;
+					}
+				},
+			}
+		}
+	}
+}

yazi-cli/src/dds/exec.rs

@@ -0,0 +1,70 @@
+use std::str::FromStr;
+
+use anyhow::{Result, bail};
+use serde::Deserialize;
+use tokio::io::AsyncWriteExt;
+use yazi_dds::{ID, Payload, Stream, ember::{Ember, EmberHi}};
+use yazi_macro::try_format;
+use yazi_shared::{Id, data::Data};
+
+use crate::{CommandExec, CommandPub, dds::Dds};
+
+impl Dds {
+	pub(crate) async fn exec(cmd: CommandExec) -> anyhow::Result<Data> {
+		let receiver = CommandPub::receiver()?;
+		let req = cmd.body(*yazi_dds::ID)?;
+		let resp = Self::ask("dds-exec", receiver, &req, "dds-exec-result").await?;
+
+		#[derive(Deserialize)]
+		struct Body {
+			ok:    bool,
+			#[serde(default)]
+			value: Data,
+			#[serde(default)]
+			error: String,
+		}
+
+		let body = Body::deserialize(&resp)?;
+		if body.ok {
+			Ok(body.value)
+		} else if !body.error.is_empty() {
+			bail!("{}", body.error)
+		} else {
+			bail!("Unknown error")
+		}
+	}
+
+	/// Send one custom message and wait for a matching custom reply.
+	async fn ask(kind: &str, receiver: Id, body: &str, reply_kind: &str) -> Result<Data> {
+		Ember::validate(kind)?;
+		Ember::validate(reply_kind)?;
+
+		let payload = try_format!(
+			"{}\n{kind},{receiver},{ID},{body}\n",
+			Payload::new(EmberHi::borrowed([reply_kind])),
+		)?;
+
+		let (mut lines, mut writer) = Stream::connect().await?;
+		writer.write_all(payload.as_bytes()).await?;
+		writer.flush().await?;
+		drop(writer);
+
+		while let Ok(Some(line)) = lines.next_line().await {
+			match line.split(',').next() {
+				Some("hey") => {
+					if let Ok(Ember::Hey(hey)) = Payload::from_str(&line).map(|p| p.body) {
+						Self::ensure_version(Some(&hey.version))?;
+						Self::ensure_ability(&hey.peers, kind, receiver)?;
+					}
+				}
+				Some(kind) if kind == reply_kind => match Payload::from_str(&line)?.body {
+					Ember::Custom(body) => return Ok(body.data),
+					_ => bail!("Expected custom payload of kind `{reply_kind}`"),
+				},
+				_ => {}
+			}
+		}
+
+		bail!("Connection closed before receiving reply")
+	}
+}

yazi-cli/src/dds/mod.rs

@@ -0,0 +1,3 @@
+yazi_macro::mod_flat!(draw exec shot);
+
+pub(crate) struct Dds;

yazi-cli/src/dds/shot.rs

@@ -0,0 +1,76 @@
+use std::str::FromStr;
+
+use anyhow::{Result, bail};
+use hashbrown::HashMap;
+use tokio::io::AsyncWriteExt;
+use yazi_dds::{ID, Payload, Peer, Stream, ember::{Ember, EmberBye, EmberHi}};
+use yazi_macro::try_format;
+use yazi_shared::Id;
+
+use crate::dds::Dds;
+
+impl Dds {
+	/// Connect to an existing server to send a single message.
+	pub(crate) async fn shot(kind: &str, receiver: Id, body: &str) -> Result<()> {
+		Ember::validate(kind)?;
+
+		let payload = try_format!(
+			"{}\n{kind},{receiver},{ID},{body}\n{}\n",
+			Payload::new(EmberHi::borrowed([])),
+			Payload::new(EmberBye::borrowed())
+		)?;
+
+		let (mut lines, mut writer) = Stream::connect().await?;
+		writer.write_all(payload.as_bytes()).await?;
+		writer.flush().await?;
+		drop(writer);
+
+		let (mut peers, mut version) = Default::default();
+		while let Ok(Some(line)) = lines.next_line().await {
+			match line.split(',').next() {
+				Some("hey") => {
+					if let Ok(Ember::Hey(hey)) = Payload::from_str(&line).map(|p| p.body) {
+						(peers, version) = (hey.peers, Some(hey.version));
+					}
+				}
+				Some("bye") => break,
+				_ => {}
+			}
+		}
+
+		Self::ensure_version(version.as_deref())?;
+		Self::ensure_ability(&peers, kind, receiver)?;
+		Ok(())
+	}
+
+	pub(super) fn ensure_version(version: Option<&str>) -> Result<()> {
+		if version.as_deref() != Some(EmberHi::version()) {
+			bail!(
+				"Incompatible version (Ya {}, Yazi {}). Restart all `ya` and `yazi` processes if you upgrade either one.",
+				EmberHi::version(),
+				version.as_deref().unwrap_or("Unknown")
+			);
+		}
+		Ok(())
+	}
+
+	pub(super) fn ensure_ability(peers: &HashMap<Id, Peer>, kind: &str, receiver: Id) -> Result<()> {
+		match (receiver, peers.get(&receiver).map(|p| p.able(kind))) {
+			// Send to all receivers
+			(Id(0), _) if peers.is_empty() => {
+				bail!("No receiver found. Check if any receivers are running.")
+			}
+			(Id(0), _) if peers.values().all(|p| !p.able(kind)) => {
+				bail!("No receiver has the ability to receive `{kind}` messages.")
+			}
+			(Id(0), _) => Ok(()),
+
+			// Send to a specific receiver
+			(_, Some(true)) => Ok(()),
+			(_, Some(false)) => {
+				bail!("Receiver `{receiver}` does not have the ability to receive `{kind}` messages.")
+			}
+			(_, None) => bail!("Receiver `{receiver}` not found. Check if the receiver is running."),
+		}
+	}
+}

yazi-cli/src/main.rs

@@ -1,4 +1,4 @@
-yazi_macro::mod_pub!(package shared);
+yazi_macro::mod_pub!(dds package shared);
 
 yazi_macro::mod_flat!(args);
 
@@ -44,9 +44,7 @@ async fn run() -> anyhow::Result<()> {
 		Command::Emit(cmd) => {
 			yazi_boot::init_default();
 			yazi_dds::init();
-			if let Err(e) =
-				yazi_dds::Client::shot("dds-emit", CommandPub::receiver()?, &cmd.body()?).await
-			{
+			if let Err(e) = dds::Dds::shot("dds-emit", CommandPub::receiver()?, &cmd.body()?).await {
 				errln!("Cannot emit command: {e}")?;
 				std::process::exit(1);
 			}
@@ -55,12 +53,25 @@ async fn run() -> anyhow::Result<()> {
 		Command::EmitTo(cmd) => {
 			yazi_boot::init_default();
 			yazi_dds::init();
-			if let Err(e) = yazi_dds::Client::shot("dds-emit", cmd.receiver, &cmd.body()?).await {
+			if let Err(e) = dds::Dds::shot("dds-emit", cmd.receiver, &cmd.body()?).await {
 				errln!("Cannot emit command: {e}")?;
 				std::process::exit(1);
 			}
 		}
 
+		Command::Exec(cmd) => {
+			yazi_boot::init_default();
+			yazi_dds::init();
+
+			match dds::Dds::exec(cmd).await {
+				Ok(data) => outln!("{}", serde_json::to_string(&data)?)?,
+				Err(e) => {
+					errln!("Cannot execute command: {e}")?;
+					std::process::exit(1);
+				}
+			}
+		}
+
 		Command::Pkg(cmd) => {
 			package::init()?;
 
@@ -77,8 +88,7 @@ async fn run() -> anyhow::Result<()> {
 		Command::Pub(cmd) => {
 			yazi_boot::init_default();
 			yazi_dds::init();
-			if let Err(e) = yazi_dds::Client::shot(&cmd.kind, CommandPub::receiver()?, &cmd.body()?).await
-			{
+			if let Err(e) = dds::Dds::shot(&cmd.kind, CommandPub::receiver()?, &cmd.body()?).await {
 				errln!("Cannot send message: {e}")?;
 				std::process::exit(1);
 			}
@@ -87,7 +97,7 @@ async fn run() -> anyhow::Result<()> {
 		Command::PubTo(cmd) => {
 			yazi_boot::init_default();
 			yazi_dds::init();
-			if let Err(e) = yazi_dds::Client::shot(&cmd.kind, cmd.receiver, &cmd.body()?).await {
+			if let Err(e) = dds::Dds::shot(&cmd.kind, cmd.receiver, &cmd.body()?).await {
 				errln!("Cannot send message: {e}")?;
 				std::process::exit(1);
 			}
@@ -96,7 +106,7 @@ async fn run() -> anyhow::Result<()> {
 		Command::Sub(cmd) => {
 			yazi_boot::init_default();
 			yazi_dds::init();
-			yazi_dds::Client::draw(cmd.kinds.split(',').collect()).await?;
+			dds::Dds::draw(cmd.kinds.split(',').collect()).await?;
 
 			tokio::signal::ctrl_c().await?;
 		}

yazi-dds/src/client.rs

@@ -1,6 +1,6 @@
-use std::{iter, mem, str::FromStr};
+use std::{mem, str::FromStr};
 
-use anyhow::{Context, Result, bail};
+use anyhow::Result;
 use hashbrown::{HashMap, HashSet};
 use parking_lot::RwLock;
 use serde::{Deserialize, Serialize};
@@ -9,7 +9,7 @@ use tracing::error;
 use yazi_macro::try_format;
 use yazi_shared::{Id, RoCell};
 
-use crate::{ClientReader, ClientWriter, Payload, Pubsub, Server, Stream, ember::{Ember, EmberBye, EmberHey, EmberHi}};
+use crate::{ClientReader, ClientWriter, Payload, Pubsub, Server, Stream, ember::{Ember, EmberHey}};
 
 pub static ID: RoCell<Id> = RoCell::new();
 pub(super) static PEERS: RoCell<RwLock<HashMap<Id, Peer>>> = RoCell::new();
@@ -77,96 +77,6 @@ impl Client {
 		});
 	}
 
-	/// Connect to an existing server to send a single message.
-	pub async fn shot(kind: &str, receiver: Id, body: &str) -> Result<()> {
-		Ember::validate(kind)?;
-
-		let payload = try_format!(
-			"{}\n{kind},{receiver},{ID},{body}\n{}\n",
-			Payload::new(EmberHi::borrowed(iter::empty())),
-			Payload::new(EmberBye::borrowed())
-		)?;
-
-		let (mut lines, mut writer) = Stream::connect().await?;
-		writer.write_all(payload.as_bytes()).await?;
-		writer.flush().await?;
-		drop(writer);
-
-		let (mut peers, mut version) = Default::default();
-		while let Ok(Some(line)) = lines.next_line().await {
-			match line.split(',').next() {
-				Some("hey") => {
-					if let Ok(Ember::Hey(hey)) = Payload::from_str(&line).map(|p| p.body) {
-						(peers, version) = (hey.peers, Some(hey.version));
-					}
-				}
-				Some("bye") => break,
-				_ => {}
-			}
-		}
-
-		if version.as_deref() != Some(EmberHi::version()) {
-			bail!(
-				"Incompatible version (Ya {}, Yazi {}). Restart all `ya` and `yazi` processes if you upgrade either one.",
-				EmberHi::version(),
-				version.as_deref().unwrap_or("Unknown")
-			);
-		}
-
-		match (receiver, peers.get(&receiver).map(|p| p.able(kind))) {
-			// Send to all receivers
-			(Id(0), _) if peers.is_empty() => {
-				bail!("No receiver found. Check if any receivers are running.")
-			}
-			(Id(0), _) if peers.values().all(|p| !p.able(kind)) => {
-				bail!("No receiver has the ability to receive `{kind}` messages.")
-			}
-			(Id(0), _) => {}
-
-			// Send to a specific receiver
-			(_, Some(true)) => {}
-			(_, Some(false)) => {
-				bail!("Receiver `{receiver}` does not have the ability to receive `{kind}` messages.")
-			}
-			(_, None) => bail!("Receiver `{receiver}` not found. Check if the receiver is running."),
-		}
-
-		Ok(())
-	}
-
-	/// Connect to an existing server and listen in on the messages that are being
-	/// sent by other yazi instances:
-	///   - If no server is running, fail right away;
-	///   - If a server is closed, attempt to reconnect forever.
-	pub async fn draw(kinds: HashSet<&str>) -> Result<()> {
-		async fn make(kinds: &HashSet<&str>) -> Result<ClientReader> {
-			let (lines, mut writer) = Stream::connect().await?;
-			let hi = Payload::new(EmberHi::borrowed(kinds.iter().copied()));
-			writer.write_all(try_format!("{hi}\n")?.as_bytes()).await?;
-			writer.flush().await?;
-			Ok(lines)
-		}
-
-		let mut lines = make(&kinds).await.context("No running Yazi instance found")?;
-		loop {
-			match lines.next_line().await? {
-				Some(s) => {
-					let kind = s.split(',').next();
-					if matches!(kind, Some(kind) if kinds.contains(kind)) {
-						println!("{s}");
-					}
-				}
-				None => loop {
-					time::sleep(time::Duration::from_secs(1)).await;
-					if let Ok(new) = make(&kinds).await {
-						lines = new;
-						break;
-					}
-				},
-			}
-		}
-	}
-
 	pub(super) fn push<'a>(payload: impl Into<Payload<'a>>) -> Result<()> {
 		Ok(QUEUE_TX.send(try_format!("{}\n", payload.into())?)?)
 	}
@@ -221,5 +131,5 @@ impl Client {
 impl Peer {
 	pub(super) fn new(abilities: &HashSet<String>) -> Self { Self { abilities: abilities.clone() } }
 
-	pub(super) fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) }
+	pub fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) }
 }

yazi-dds/src/ember/hi.rs

@@ -18,9 +18,13 @@ pub struct EmberHi<'a> {
 impl<'a> EmberHi<'a> {
 	pub fn borrowed<I>(abilities: I) -> Ember<'a>
 	where
-		I: Iterator<Item = &'a str>,
+		I: IntoIterator<Item = &'a str>,
 	{
-		Self { abilities: abilities.map(Into::into).collect(), version: Self::version().into() }.into()
+		Self {
+			abilities: abilities.into_iter().map(Into::into).collect(),
+			version:   Self::version().into(),
+		}
+		.into()
 	}
 
 	pub fn version() -> &'static str {

yazi-dds/src/payload.rs

@@ -16,7 +16,7 @@ pub struct Payload<'a> {
 }
 
 impl<'a> Payload<'a> {
-	pub(super) fn new(body: Ember<'a>) -> Self { Self { receiver: Id(0), sender: *ID, body } }
+	pub fn new(body: Ember<'a>) -> Self { Self { receiver: Id(0), sender: *ID, body } }
 
 	pub(super) fn flush(&self) -> Result<()> {
 		writeln!(std::io::stdout(), "{self}")?;

yazi-dds/src/stream.rs

@@ -1,13 +1,13 @@
 use tokio::io::{BufReader, Lines, ReadHalf, WriteHalf};
 
-pub(super) struct Stream;
+pub struct Stream;
 
 use tokio::io::AsyncBufReadExt;
 
 #[cfg(unix)]
-pub(super) type ClientReader = Lines<BufReader<ReadHalf<tokio::net::UnixStream>>>;
+pub type ClientReader = Lines<BufReader<ReadHalf<tokio::net::UnixStream>>>;
 #[cfg(not(unix))]
-pub(super) type ClientReader = Lines<BufReader<ReadHalf<tokio::net::TcpStream>>>;
+pub type ClientReader = Lines<BufReader<ReadHalf<tokio::net::TcpStream>>>;
 
 #[cfg(unix)]
 pub(super) type ClientWriter = WriteHalf<tokio::net::UnixStream>;
@@ -21,14 +21,14 @@ pub(super) type ServerListener = tokio::net::TcpListener;
 
 impl Stream {
 	#[cfg(unix)]
-	pub(super) async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> {
+	pub async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> {
 		let stream = tokio::net::UnixStream::connect(Self::socket_file()).await?;
 		let (reader, writer) = tokio::io::split(stream);
 		Ok((BufReader::new(reader).lines(), writer))
 	}
 
 	#[cfg(not(unix))]
-	pub(super) async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> {
+	pub async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> {
 		let stream = tokio::net::TcpStream::connect("127.0.0.1:33581").await?;
 		let (reader, writer) = tokio::io::split(stream);
 		Ok((BufReader::new(reader).lines(), writer))

yazi-fm/src/app/app.rs

@@ -29,7 +29,7 @@ impl App {
 		macro_rules! drain_events {
 			() => {
 				for event in events.drain(..) {
-					Dispatcher::new(&mut app).dispatch(event)?;
+					Dispatcher::new(&mut app).dispatch(event);
 
 					need_render = NEED_RENDER.load(Ordering::Relaxed);
 					if need_render == 0 {

yazi-fm/src/dispatcher.rs

@@ -2,10 +2,11 @@ use std::sync::atomic::Ordering;
 
 use anyhow::Result;
 use crossterm::event::{KeyEvent, MouseEvent};
+use tokio::sync::mpsc;
 use tracing::warn;
 use yazi_actor::Ctx;
 use yazi_config::keymap::Key;
-use yazi_macro::{act, emit, succ};
+use yazi_macro::{act, emit};
 use yazi_shared::{data::Data, event::{ActionCow, Event, NEED_RENDER}};
 use yazi_widgets::input::InputMode;
 
@@ -20,7 +21,7 @@ impl<'a> Dispatcher<'a> {
 	pub(super) fn new(app: &'a mut App) -> Self { Self { app } }
 
 	#[inline]
-	pub(super) fn dispatch(&mut self, event: Event) -> Result<()> {
+	pub(super) fn dispatch(&mut self, event: Event) {
 		let result = match event {
 			Event::Call(action) => self.dispatch_call(action),
 			Event::Seq(actions) => self.dispatch_seq(actions),
@@ -32,64 +33,71 @@ impl<'a> Dispatcher<'a> {
 			Event::Paste(str) => self.dispatch_paste(str),
 		};
 
-		if let Err(err) = result {
-			warn!("Event dispatch error: {err:?}");
+		if let Err(e) = &result {
+			warn!("Event dispatch error: {e:?}");
+		}
+	}
+
+	fn dispatch_call(&mut self, action: ActionCow) -> Result<()> {
+		let tx = action.any::<mpsc::UnboundedSender<Result<Data>>>("__reply").cloned();
+		let result = Executor::new(self.app).execute(action);
+
+		if let Err(e) = &result {
+			warn!("Call dispatch error: {e:?}");
+		}
+		if let Some(tx) = tx {
+			tx.send(result).ok();
 		}
 		Ok(())
 	}
 
 	#[inline]
-	fn dispatch_call(&mut self, action: ActionCow) -> Result<Data> {
-		Executor::new(self.app).execute(action)
-	}
-
-	#[inline]
-	fn dispatch_seq(&mut self, mut actions: Vec<ActionCow>) -> Result<Data> {
+	fn dispatch_seq(&mut self, mut actions: Vec<ActionCow>) -> Result<()> {
 		if let Some(last) = actions.pop() {
 			self.dispatch_call(last)?;
 		}
 		if !actions.is_empty() {
 			emit!(Seq(actions));
 		}
-		succ!();
+		Ok(())
 	}
 
 	#[inline]
-	fn dispatch_render(&mut self, partial: bool) -> Result<Data> {
+	fn dispatch_render(&mut self, partial: bool) -> Result<()> {
 		if partial {
 			_ = NEED_RENDER.compare_exchange(0, 2, Ordering::Relaxed, Ordering::Relaxed);
 		} else {
 			NEED_RENDER.store(1, Ordering::Relaxed);
 		}
-		succ!()
+		Ok(())
 	}
 
 	#[inline]
-	fn dispatch_key(&mut self, key: KeyEvent) -> Result<Data> {
+	fn dispatch_key(&mut self, key: KeyEvent) -> Result<()> {
 		Router::new(self.app).route(Key::from(key))?;
-		succ!();
+		Ok(())
 	}
 
 	#[inline]
-	fn dispatch_mouse(&mut self, mouse: MouseEvent) -> Result<Data> {
+	fn dispatch_mouse(&mut self, mouse: MouseEvent) -> Result<()> {
 		let cx = &mut Ctx::active(&mut self.app.core, &mut self.app.term);
-		act!(app:mouse, cx, mouse)
+		act!(app:mouse, cx, mouse).map(|_| ())
 	}
 
 	#[inline]
-	fn dispatch_resize(&mut self) -> Result<Data> {
+	fn dispatch_resize(&mut self) -> Result<()> {
 		let cx = &mut Ctx::active(&mut self.app.core, &mut self.app.term);
-		act!(app:resize, cx, crate::Root::reflow as fn(_) -> _)
+		act!(app:resize, cx, crate::Root::reflow as fn(_) -> _).map(|_| ())
 	}
 
 	#[inline]
-	fn dispatch_focus(&mut self) -> Result<Data> {
+	fn dispatch_focus(&mut self) -> Result<()> {
 		let cx = &mut Ctx::active(&mut self.app.core, &mut self.app.term);
-		act!(app:focus, cx)
+		act!(app:focus, cx).map(|_| ())
 	}
 
 	#[inline]
-	fn dispatch_paste(&mut self, str: String) -> Result<Data> {
+	fn dispatch_paste(&mut self, str: String) -> Result<()> {
 		if self.app.core.input.visible {
 			let input = &mut self.app.core.input;
 			if input.mode() == InputMode::Insert {
@@ -98,6 +106,6 @@ impl<'a> Dispatcher<'a> {
 				input.replace_str(&str)?;
 			}
 		}
-		succ!();
+		Ok(())
 	}
 }

yazi-fm/src/executor.rs

@@ -14,7 +14,6 @@ impl<'a> Executor<'a> {
 	#[inline]
 	pub(super) fn new(app: &'a mut App) -> Self { Self { app } }
 
-	#[inline]
 	pub(super) fn execute(&mut self, action: ActionCow) -> Result<Data> {
 		match action.layer {
 			Layer::App => self.app(action),

yazi-plugin/Cargo.toml

@@ -58,6 +58,3 @@ uzers = { workspace = true }
 
 [target."cfg(windows)".dependencies]
 windows-sys = { version = "0.61.2", features = [ "Win32_System_JobObjects" ] }
-
-[target.'cfg(target_os = "macos")'.dependencies]
-crossterm = { workspace = true, features = [ "use-dev-tty", "libc" ] }

yazi-plugin/preset/plugins/dds.lua

@@ -1,20 +1,35 @@
 local M = {}
 
-function M:setup()
-	ps.sub_remote("dds-emit", function(action)
-		local i, args = 1, {}
-		for j = 2, #action do
-			local word = string.char(table.unpack(action[j]))
-			local key = word:match("^%-%-([^=]+)")
-			if not key then
-				i, args[i] = i + 1, word
-			elseif #key + 2 == #word then
-				args[key] = true
-			else
-				args[key] = word:sub(#key + 4)
-			end
+function M.parse_args(t, i)
+	local j, args = 1, {}
+	for i = i, #t do
+		local word = string.char(table.unpack(t[i]))
+		local key = word:match("^%-%-([^=]+)")
+		if not key then
+			j, args[j] = j + 1, word
+		elseif #key + 2 == #word then
+			args[key] = true
+		else
+			args[key] = word:sub(#key + 4)
 		end
-		ya.emit(action[1], args)
+	end
+	return args
+end
+
+function M:setup()
+	ps.sub_remote("dds-emit", function(t) ya.emit(t[1], M.parse_args(t, 2)) end)
+
+	ps.sub_remote("dds-exec", function(t)
+		ya.async(function()
+			local ok, value = pcall(ya.exec, t[2], M.parse_args(t, 3))
+			ps.pub_to(t[1], "dds-exec-result", ok and {
+				ok = true,
+				value = value,
+			} or {
+				ok = false,
+				error = tostring(value),
+			})
+		end)
 	end)
 end
 

yazi-plugin/src/utils/call.rs

@@ -1,7 +1,8 @@
-use mlua::{Function, Lua, Table};
+use mlua::{ExternalError, Function, Lua, Table};
+use tokio::sync::mpsc;
 use yazi_dds::Sendable;
 use yazi_macro::emit;
-use yazi_shared::{Layer, Source, event::Action};
+use yazi_shared::{Layer, Source, data::Data, event::Action};
 
 use super::Utils;
 
@@ -14,15 +15,18 @@ impl Utils {
 		})
 	}
 
-	pub(super) fn mgr_emit(lua: &Lua) -> mlua::Result<Function> {
-		lua.create_function(|lua, (name, args): (String, Table)| {
-			emit!(Call(Action {
-				name:   name.into(),
-				args:   Sendable::table_to_args(lua, args)?,
-				layer:  Layer::Mgr,
-				source: Source::Emit,
-			}));
-			Ok(())
+	pub(super) fn exec(lua: &Lua) -> mlua::Result<Function> {
+		lua.create_async_function(|lua, (name, args): (String, Table)| async move {
+			let mut action = Action::new(name, Source::Emit, Some(Layer::Mgr))?;
+			action.args = Sendable::table_to_args(&lua, args)?;
+
+			let (tx, mut rx) = mpsc::unbounded_channel::<anyhow::Result<Data>>();
+			emit!(Call(action.with_any("__reply", tx)));
+
+			Sendable::data_to_value(
+				&lua,
+				rx.recv().await.ok_or_else(|| "channel closed before action response".into_lua_err())??,
+			)
 		})
 	}
 }

yazi-plugin/src/utils/utils.rs

@@ -17,7 +17,7 @@ pub fn compose(
 
 			// Call
 			b"emit" => Utils::emit(lua)?,
-			b"mgr_emit" => Utils::mgr_emit(lua)?,
+			b"exec" => Utils::exec(lua)?,
 
 			// Image
 			b"image_info" => Utils::image_info(lua)?,

yazi-shared/src/data/data.rs

@@ -7,9 +7,10 @@ use serde::{Deserialize, Serialize};
 use crate::{Id, SStr, data::{DataAny, DataKey}, path::PathBufDyn, strand::{IntoStrand, StrandBuf}, url::{UrlBuf, UrlCow}};
 
 // --- Data
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
 #[serde(untagged)]
 pub enum Data {
+	#[default]
 	Nil,
 	Boolean(bool),
 	Integer(i64),
@@ -216,6 +217,13 @@ impl Data {
 		}
 	}
 
+	pub fn as_any<T: 'static>(&self) -> Option<&T> {
+		match self {
+			Self::Any(b) => (&**b).as_any().downcast_ref::<T>(),
+			_ => None,
+		}
+	}
+
 	pub fn into_string(self) -> Option<SStr> {
 		match self {
 			Self::String(s) => Some(s),

yazi-shared/src/event/action.rs

@@ -98,6 +98,10 @@ impl Action {
 
 	pub fn bool(&self, name: impl Into<DataKey>) -> bool { self.get(name).unwrap_or(false) }
 
+	pub fn any<T: 'static>(&self, name: impl Into<DataKey>) -> Option<&T> {
+		self.args.get(&name.into())?.as_any()
+	}
+
 	pub fn first<'a, T>(&'a self) -> Result<T>
 	where
 		T: TryFrom<&'a Data>,

yazi-term/Cargo.toml

@@ -26,8 +26,5 @@ crossterm = { workspace = true }
 ratatui   = { workspace = true }
 tokio     = { workspace = true }
 
-[target."cfg(unix)".dependencies]
-libc = { workspace = true }
-
 [target.'cfg(target_os = "macos")'.dependencies]
 crossterm = { workspace = true, features = [ "use-dev-tty", "libc" ] }