standard_plugin/daemon/

mod.rs

//! Daemon plugins: run inside `standardd` on the machines the plugin is
//! enabled on, never render, and may reach the machine through granted
//! system interfaces ([`process`], [`watch`], [`panes`]; with the `wasi`
//! feature, files, sockets and HTTP).
//!
//! A daemon plugin has an async [`DaemonPlugin::run`] loop driven by a
//! small single-threaded executor: the host calls the component's `drive`
//! export when the loop's next timer is due and after every event, and the
//! loop awaits [`Context::next_event`], [`Context::sleep`] and, with `wasi`,
//! WASI pollables (`daemon::io`). What happens on the machine arrives as
//! events too: [`Event::FileChanged`](crate::Event::FileChanged) for a
//! [`watch`], [`Event::ProcessOutput`](crate::Event::ProcessOutput) and
//! [`Event::ProcessExited`](crate::Event::ProcessExited) for a
//! [`process`], [`Event::PaneChanged`](crate::Event::PaneChanged) after
//! [`panes::subscribe`]. A *companion* is a daemon plugin that answers its
//! UI plugin's [`calls`](mod@crate::calls) in [`DaemonPlugin::call`];
//! [`Context::caller`] says who asked.
//!
//! Plugin methods take `&self`: the run loop and call handlers share the
//! plugin, so keep mutable state in a `RefCell` or `Cell`.
//!
//! ```rust,ignore
#![doc = include_str!("../../examples/together_companion.rs")]
//! ```

mod executor;
pub(crate) mod runtime;

/// HTTP: over `wasi:http` in a component (the `wasi` feature), and through
/// the test host natively.
#[cfg(any(feature = "wasi", not(target_arch = "wasm32")))]
pub mod http;
#[cfg(all(feature = "wasi", target_arch = "wasm32"))]
mod http_wasi;
#[cfg(all(feature = "wasi", target_arch = "wasm32"))]
pub mod io;

/// The WASI 0.2 bindings (`wasip2`), for what the SDK does not wrap.
#[cfg(all(feature = "wasi", target_arch = "wasm32"))]
pub use wasip2 as wasi;

use alloc::string::String;
use alloc::vec::Vec;
use core::future::Future;

use serde::de::DeserializeOwned;

use crate::api::{self, Json};
use crate::error::{Error, Result};
use crate::host;

pub use executor::{EVENT_QUEUE, NextEvent, NextEventUntil, Sleep};

/// The plugin event the host delivers the viewers' interest as
/// (`{ "surfaces": [...] }`); the runtime turns it into
/// [`Event::Interest`](crate::Event::Interest).
pub const INTEREST_EVENT: &str = "system.plugin.interest";

/// Which of the plugin's UI surfaces some viewer on the account shows now:
/// the union over every viewer, kept by the account and pushed to the
/// daemon half whenever it changes (a surface shown or hidden, a viewer
/// that opens or goes away). A daemon half that reads an outside source
/// only for its UI reads it while [`Interest::any`] holds and stops when it
/// goes away: nothing on screen, nothing read. Instances count as their
/// surface (`workers@<machine>` is `workers`).
///
/// Until the host has said anything (a host or account service without the
/// signal) the interest is *unknown*, and [`Interest::any`] and
/// [`Interest::shows`] answer `true`, so a plugin keeps working there.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Interest {
    surfaces: Option<alloc::collections::BTreeSet<String>>,
}

impl Interest {
    /// Interest in exactly `surfaces`.
    pub fn of<I, S>(surfaces: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        Self {
            surfaces: Some(surfaces.into_iter().map(Into::into).collect()),
        }
    }

    /// Whether the host has said anything yet.
    pub fn is_known(&self) -> bool {
        self.surfaces.is_some()
    }

    /// Whether some viewer shows any surface of the plugin (or the
    /// interest is unknown).
    pub fn any(&self) -> bool {
        self.surfaces.as_ref().is_none_or(|shown| !shown.is_empty())
    }

    /// Whether some viewer shows `surface` (or the interest is unknown).
    pub fn shows(&self, surface: &str) -> bool {
        self.surfaces
            .as_ref()
            .is_none_or(|shown| shown.contains(surface))
    }

    /// The surfaces shown, when known.
    pub fn surfaces(&self) -> impl Iterator<Item = &str> {
        self.surfaces.iter().flatten().map(String::as_str)
    }

    /// Reads the host's payload; `None` when it is not
    /// `{ "surfaces": [<string>, ...] }`.
    pub(crate) fn from_payload(payload: &str) -> Option<Self> {
        #[derive(serde::Deserialize)]
        struct Wire {
            surfaces: Vec<String>,
        }
        let wire: Wire = serde_json::from_str(payload).ok()?;
        Some(Self::of(wire.surfaces))
    }
}

/// A daemon (or companion) plugin.
pub trait DaemonPlugin: Sized + 'static {
    /// Called once after instantiation.
    fn activate(cx: &Context) -> Self;

    /// The plugin's long-running work, started after `activate` and dropped
    /// at `deactivate`. Await [`Context::next_event`] for events and
    /// [`Context::sleep`] between rounds of work.
    fn run(&self, cx: Context) -> impl Future<Output = ()> {
        let _ = cx;
        core::future::ready(())
    }

    /// Answers a call from the UI plugin with the same id. The default
    /// knows no method.
    fn call(&self, method: &str, request: Json, cx: &Context) -> Result<Json> {
        let _ = (request, cx);
        Err(Error::Invalid(alloc::format!("unknown method {method:?}")))
    }

    /// Called before the instance is dropped, after the run loop is.
    fn deactivate(&self) {}
}

/// How a pane changed ([`Event::PaneChanged`](crate::Event::PaneChanged)).
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum PaneChangeKind {
    Created,
    /// Title, size or agent state changed.
    Changed,
    Exited,
    Closed,
}

/// Who made a call, as the account routed it.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Caller {
    /// The machine the calling runtime runs on.
    pub machine_id: String,
    pub plugin_id: String,
    /// `ui` or `daemon`.
    pub kind: String,
    /// The caller's lease epoch, when a singleton daemon called.
    pub epoch: Option<u64>,
    /// The account controller lease when the call was made: the machine
    /// the user controls from and its fencing token.
    pub controller_machine_id: Option<String>,
    pub controller_fencing_token: Option<String>,
}

impl Caller {
    /// Whether the call came from the machine the user was controlling
    /// when it was made: act for the user only then.
    pub fn from_controller(&self) -> bool {
        self.controller_machine_id.as_deref() == Some(self.machine_id.as_str())
    }
}

/// A daemon plugin's handle to the host, its configuration and its
/// executor. Cheap to clone.
#[derive(Clone, Debug, Default)]
pub struct Context {
    config: String,
}

impl Context {
    pub(crate) fn new(config: String) -> Self {
        Self { config }
    }

    /// The configuration, deserialized (an empty document is `null`).
    pub fn config<T: DeserializeOwned>(&self) -> Result<T> {
        if self.config.trim().is_empty() {
            return Json(String::from("null")).parse();
        }
        Json(self.config.clone()).parse()
    }

    pub fn config_json(&self) -> Json {
        Json(self.config.clone())
    }

    pub fn values(&self) -> api::values::Values {
        api::values()
    }

    pub fn live(&self) -> api::live::Live {
        api::live()
    }

    pub fn events(&self) -> api::events::Events {
        api::events()
    }

    /// Exactly-once effects across the daemons that run a fleet plugin (a
    /// singleton's claims carry its lease epoch). No grant.
    pub fn claims(&self) -> api::claims::Claims {
        api::claims::Claims
    }

    pub fn calls(&self) -> api::calls::Calls {
        api::calls()
    }

    /// The daemon's monotonic clock as of the current `drive` call.
    pub fn now_ms(&self) -> u64 {
        executor::now_ms()
    }

    /// The next event the host delivers (values, live messages, plugin
    /// events, account changes). Events queue while nothing awaits them,
    /// up to [`EVENT_QUEUE`]; past that the oldest are dropped.
    pub fn next_event(&self) -> NextEvent {
        NextEvent::new()
    }

    /// Which of the plugin's UI surfaces some viewer shows now
    /// ([`Interest`]). Changes arrive as
    /// [`Event::Interest`](crate::Event::Interest).
    pub fn interest(&self) -> Interest {
        runtime::current_interest()
    }

    /// The next event, or `None` once the daemon's clock reaches `at_ms`
    /// (never, with `None`): the one wait a loop that reads on a schedule
    /// and reacts to events needs. The timer goes with it, so an event that
    /// arrives first leaves no wake behind.
    pub fn next_event_until(&self, at_ms: Option<u64>) -> NextEventUntil {
        NextEventUntil::new(at_ms)
    }

    /// Completes `ms` milliseconds from now.
    pub fn sleep(&self, ms: u64) -> Sleep {
        Sleep::until(executor::now_ms().saturating_add(ms))
    }

    /// Completes at `at_ms` on the daemon's clock.
    pub fn sleep_until(&self, at_ms: u64) -> Sleep {
        Sleep::until(at_ms)
    }

    /// Runs `task` beside the run loop until it completes or the plugin
    /// is deactivated.
    pub fn spawn(&self, task: impl Future<Output = ()> + 'static) {
        executor::spawn(task);
    }

    /// Who made the call being answered, inside [`DaemonPlugin::call`];
    /// `None` elsewhere.
    pub fn caller(&self) -> Option<Caller> {
        runtime::current_caller()
    }

    /// The machine this daemon runs on, by the account's machine id: the
    /// id a UI plugin targets with [`Target::Machine`](crate::Target::Machine)
    /// and finds in [`account::state`](crate::account::state).
    pub fn machine_id(&self) -> String {
        host::daemon_machine_id()
    }

    /// The account this daemon is enrolled in, once the daemon has read it.
    pub fn account_id(&self) -> Option<String> {
        host::daemon_account_id()
    }

    /// The lease epoch a singleton runs under (every write of its account
    /// session carries it); `None` for a fleet plugin. A plugin that records
    /// who did something (a claim, a value) can tag it with its epoch.
    pub fn lease_epoch(&self) -> Option<u64> {
        host::daemon_lease_epoch()
    }

    /// The Standard Code build this daemon runs: its channel
    /// (`branch:<name>`, `team`, ...), version and commit. `None` where it
    /// is not known.
    pub fn release(&self) -> Option<Release> {
        host::daemon_release()
    }
}

/// The Standard Code build a daemon runs ([`Context::release`]).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Release {
    /// `branch:<name>` for a branch build, else the published channel's
    /// name (`team`, `canary`, `production`).
    pub channel: String,
    pub version: String,
    pub git_sha: String,
}

impl Release {
    /// The Git ref the channel follows: `refs/heads/<name>` for a branch
    /// channel, else `refs/heads/main`.
    pub fn git_ref(&self) -> String {
        match self.channel.strip_prefix("branch:") {
            Some(branch) if !branch.is_empty() => alloc::format!("refs/heads/{branch}"),
            _ => "refs/heads/main".into(),
        }
    }
}

/// The plugin's environment: the daemon user's login environment reduced to
/// `HOME`, `USER`, `LOGNAME`, `PATH` (the login shell's), `LANG`, `LC_*`,
/// `TMPDIR` and `SHELL` (all of it under `machine.full`). Children spawned
/// through [`process`] inherit exactly these; the daemon's own environment
/// never reaches a plugin. No grant.
pub mod env {
    use super::*;

    /// Every variable, sorted by name.
    pub fn vars() -> Vec<(String, String)> {
        host::daemon_environment()
    }

    /// One variable.
    pub fn var(name: &str) -> Option<String> {
        vars()
            .into_iter()
            .find(|(key, _)| key == name)
            .map(|(_, value)| value)
    }

    /// The user's home directory (`HOME`): where children run unless a
    /// [`Command`](super::process::Command) names another directory.
    pub fn home_dir() -> Option<String> {
        var("HOME").filter(|home| home.starts_with('/'))
    }
}

/// Spawn programs on the daemon's machine. Grant: `process.exec:<program>`
/// naming the program exactly as spawned (`process.exec:*` for any), or
/// `machine.full`. Children are killed when the plugin stops.
pub mod process {
    use super::*;

    /// A spawned program.
    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub struct Child {
        pub pid: u32,
    }

    /// Where a child's standard stream goes.
    #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
    pub enum Stdio {
        /// Nowhere.
        Null,
        /// To the plugin: output arrives as
        /// [`Event::ProcessOutput`](crate::Event::ProcessOutput), input goes
        /// through [`Child::write`].
        Piped,
        /// The daemon's log, prefixed with the plugin id (output only).
        #[default]
        Log,
    }

    /// A program with its arguments, environment, directory and streams. It
    /// inherits the plugin's environment ([`env`](super::env)) and runs in
    /// the user's home unless told otherwise.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub struct Command {
        pub program: String,
        pub args: Vec<String>,
        /// Variables set on top of the plugin's environment ([`env`](super::env)).
        pub env: Vec<(String, String)>,
        /// Variables removed from the plugin's environment first.
        pub env_remove: Vec<String>,
        /// Start from an empty environment: only [`Command::env`].
        pub clear_env: bool,
        /// The working directory; the user's home when `None`.
        pub cwd: Option<String>,
        pub stdin: Stdio,
        pub stdout: Stdio,
        pub stderr: Stdio,
    }

    impl Command {
        /// `program` with no stdin and its output in the daemon's log.
        pub fn new(program: &str) -> Self {
            Self {
                program: String::from(program),
                args: Vec::new(),
                env: Vec::new(),
                env_remove: Vec::new(),
                clear_env: false,
                cwd: None,
                stdin: Stdio::Null,
                stdout: Stdio::Log,
                stderr: Stdio::Log,
            }
        }

        pub fn arg(mut self, arg: &str) -> Self {
            self.args.push(String::from(arg));
            self
        }

        pub fn args(mut self, args: &[&str]) -> Self {
            self.args.extend(args.iter().map(|arg| String::from(*arg)));
            self
        }

        pub fn env(mut self, key: &str, value: &str) -> Self {
            self.env.push((String::from(key), String::from(value)));
            self
        }

        /// Removes `key` from the inherited environment (`GIT_DIR`, say).
        pub fn env_remove(mut self, key: &str) -> Self {
            self.env_remove.push(String::from(key));
            self
        }

        pub fn clear_env(mut self) -> Self {
            self.clear_env = true;
            self
        }

        pub fn cwd(mut self, cwd: &str) -> Self {
            self.cwd = Some(String::from(cwd));
            self
        }

        pub fn stdin(mut self, stdin: Stdio) -> Self {
            self.stdin = stdin;
            self
        }

        pub fn stdout(mut self, stdout: Stdio) -> Self {
            self.stdout = stdout;
            self
        }

        pub fn stderr(mut self, stderr: Stdio) -> Self {
            self.stderr = stderr;
            self
        }

        pub fn spawn(&self) -> Result<Child> {
            host::process_run(self).map(|pid| Child { pid })
        }
    }

    /// Spawns `program` with `args`: no stdin, output to the daemon's log.
    pub fn spawn(program: &str, args: &[&str], cwd: Option<&str>) -> Result<Child> {
        let args: Vec<String> = args.iter().map(|arg| String::from(*arg)).collect();
        host::process_spawn(program, &args, cwd).map(|pid| Child { pid })
    }

    impl Child {
        /// Blocks the plugin until the program exits; its exit status (a
        /// signal is negated). Prefer awaiting
        /// [`Event::ProcessExited`](crate::Event::ProcessExited).
        pub fn wait(self) -> Result<i32> {
            host::process_wait(self.pid)
        }

        /// The exit status once the program exited.
        pub fn try_wait(self) -> Result<Option<i32>> {
            host::process_try_wait(self.pid)
        }

        /// Writes to a [`Stdio::Piped`] stdin.
        pub fn write(self, bytes: &[u8]) -> Result<()> {
            host::process_write(self.pid, bytes)
        }

        /// Closes a piped stdin: the program reads end of file.
        pub fn close_stdin(self) -> Result<()> {
            host::process_close_stdin(self.pid)
        }

        /// Kills the program's process group.
        pub fn kill(self) -> Result<()> {
            host::process_kill(self.pid)
        }
    }
}

/// File and directory change notifications, debounced by the host and
/// delivered as [`Event::FileChanged`](crate::Event::FileChanged). Grant:
/// `fs.read:<path>` covering the path. At most 256 watches per plugin.
pub mod watch {
    use super::*;

    /// A watch; dropping it keeps watching, [`Watch::unwatch`] stops.
    #[derive(Debug, PartialEq, Eq)]
    pub struct Watch {
        pub handle: u32,
    }

    /// What a watch covers.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub struct Options {
        /// A directory's whole tree (the default), or its own entries only.
        pub recursive: bool,
        /// Globs relative to the watched path whose changes never arrive:
        /// `*` within a component, `?` one character, `**` any number of
        /// components; a glob without `/` matches a component at any depth
        /// (`target`, `node_modules`, `*.log`), one with `/` is anchored at
        /// the watched path (`.git/objects`), and a leading `/` anchors a
        /// single name there, as in `.gitignore` (`/target` is the watched
        /// path's own `target` only). Excluding a directory excludes
        /// everything in it. At most 64.
        pub exclude: Vec<String>,
    }

    impl Default for Options {
        fn default() -> Self {
            Self {
                recursive: true,
                exclude: Vec::new(),
            }
        }
    }

    impl Options {
        /// A directory's own entries only.
        pub fn shallow() -> Self {
            Self {
                recursive: false,
                exclude: Vec::new(),
            }
        }

        /// Drops changes the glob matches.
        pub fn exclude(mut self, glob: &str) -> Self {
            self.exclude.push(String::from(glob));
            self
        }
    }

    /// Watches a file, or a directory and everything under it.
    pub fn watch(path: &str) -> Result<Watch> {
        watch_with(path, &Options::default())
    }

    /// Watches `path` as `options` say.
    pub fn watch_with(path: &str, options: &Options) -> Result<Watch> {
        host::watch(path, options).map(|handle| Watch { handle })
    }

    impl Watch {
        pub fn unwatch(self) -> Result<()> {
            host::unwatch(self.handle)
        }
    }
}

/// WebSockets the host holds for the plugin (`daemon-net`). Grant:
/// `socket.connect:<host>:<port>` for each server (`wss://` only). News
/// arrives as [`crate::Event::Plugin`] named [`net::WEBSOCKET_EVENT`];
/// [`net::SocketEvent::from_event`] reads it.
pub mod net {
    use super::*;

    /// The plugin event a socket's news arrives as.
    pub const WEBSOCKET_EVENT: &str = "system.net.websocket";

    /// An open (or opening) WebSocket.
    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub struct WebSocket {
        pub socket: u32,
    }

    /// What happened to a socket.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub enum SocketEvent {
        Open { socket: u32 },
        Message { socket: u32, text: String },
        Closed { socket: u32, reason: String },
    }

    impl SocketEvent {
        /// The socket news an event carries, if it is some.
        pub fn from_event(event: &crate::Event) -> Option<Self> {
            let crate::Event::Plugin { name, payload } = event else {
                return None;
            };
            if name != WEBSOCKET_EVENT {
                return None;
            }
            let value: serde_json::Value = serde_json::from_str(&payload.0).ok()?;
            let socket = u32::try_from(value.get("socket")?.as_u64()?).ok()?;
            let text = |field: &str| -> String {
                value
                    .get(field)
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or_default()
                    .into()
            };
            match value.get("kind")?.as_str()? {
                "open" => Some(Self::Open { socket }),
                "message" => Some(Self::Message {
                    socket,
                    text: text("text"),
                }),
                "closed" => Some(Self::Closed {
                    socket,
                    reason: text("reason"),
                }),
                _ => None,
            }
        }
    }

    impl WebSocket {
        /// Starts connecting to `url` with extra request `headers`; the
        /// socket's `Open` or `Closed` event follows.
        pub fn open(url: &str, headers: &[(String, String)]) -> Result<Self> {
            host::websocket_open(url, headers).map(|socket| Self { socket })
        }

        /// Sends one text frame (64 KiB at most).
        pub fn send(&self, text: &str) -> Result<()> {
            host::websocket_send(self.socket, text)
        }

        /// Closes it; its `Closed` event follows.
        pub fn close(self) {
            host::websocket_close(self.socket);
        }
    }
}

/// Panes on the daemon's machine. Grants: `panes.read`, `panes.write`.
pub mod panes {
    use super::*;
    pub use crate::api::account::Pane;

    pub fn list() -> Result<Vec<Pane>> {
        host::panes()
    }

    /// A pane [`create`] opened: its id and generation, together naming
    /// this run of it.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub struct Created {
        pub id: String,
        /// 1 for a new pane; a restart advances it.
        pub generation: u64,
    }

    impl Created {
        /// `<id>@<generation>`: the key state kept for this run of the pane
        /// goes under.
        pub fn key(&self) -> String {
            alloc::format!("{}@{}", self.id, self.generation)
        }
    }

    /// A pane to open: its directory (a project root of this machine or a
    /// directory inside one), its command (the user's shell when none) and
    /// variables on top of its login environment.
    ///
    /// ```rust,ignore
    /// let pane = NewPane::new("/work/app/web")
    ///     .command("pnpm dev")
    ///     .env("PORT", "4000")
    ///     .create()?;
    /// ```
    #[derive(Clone, Debug, Default, PartialEq, Eq)]
    pub struct NewPane {
        pub cwd: String,
        pub command: Option<String>,
        pub env: Vec<(String, String)>,
        /// The pane's title; the plugin's id when none.
        pub title: Option<String>,
    }

    impl NewPane {
        pub fn new(cwd: &str) -> Self {
            Self {
                cwd: cwd.into(),
                ..Self::default()
            }
        }

        pub fn command(mut self, command: &str) -> Self {
            self.command = Some(command.into());
            self
        }

        /// Sets one variable (at most 64; a name is not empty and has no
        /// `=`).
        pub fn env(mut self, name: &str, value: &str) -> Self {
            self.env.retain(|(key, _)| key != name);
            self.env.push((name.into(), value.into()));
            self
        }

        /// Titles the pane (1 to 128 characters, no control characters)
        /// instead of the plugin's id.
        pub fn title(mut self, title: &str) -> Self {
            self.title = Some(title.into());
            self
        }

        /// Opens it. `Invalid` for a directory outside every project root,
        /// variables a pane cannot be given or a title it cannot show.
        pub fn create(&self) -> Result<Created> {
            // Untitled panes use `create`, which every host has.
            match &self.title {
                None => host::pane_create(&self.cwd, self.command.as_deref(), &self.env),
                Some(title) => host::pane_create_with(
                    &self.cwd,
                    self.command.as_deref(),
                    &self.env,
                    Some(title),
                ),
            }
            .map(|(id, generation)| Created { id, generation })
        }
    }

    /// Opens a pane in `cwd` (a project root or a directory inside one),
    /// running `command` or the user's shell; [`NewPane`] adds variables.
    pub fn create(cwd: &str, command: Option<&str>) -> Result<Created> {
        NewPane {
            cwd: cwd.into(),
            command: command.map(Into::into),
            env: Vec::new(),
            title: None,
        }
        .create()
    }

    /// Types `bytes` into a pane.
    pub fn input(pane: &str, bytes: &[u8]) -> Result<()> {
        host::pane_input(pane, bytes)
    }

    pub fn close(pane: &str) -> Result<()> {
        host::pane_close(pane)
    }

    /// Asks for [`Event::PaneChanged`](crate::Event::PaneChanged) for every
    /// pane on this machine.
    pub fn subscribe() -> Result<()> {
        host::panes_subscribe(true)
    }

    pub fn unsubscribe() -> Result<()> {
        host::panes_subscribe(false)
    }

    /// Blocks the plugin until `pane` exits or closes, for at most
    /// `timeout_ms` (30 s at most); whether it did.
    pub fn wait(pane: &str, timeout_ms: u32) -> Result<bool> {
        host::pane_wait(pane, timeout_ms)
    }
}