sausage/cli/src/console.rs

//! Sequential console commands, line editing, and live notifications.
use crate::Host;
use anyhow::Result;
use rustyline::{DefaultEditor, ExternalPrinter, error::ReadlineError};
use sausage_vfs::{Layer, Open};
use serde_json::{Value, json};

const HELP: &str = "text                         chat with the agent
plugin.method {json}          call a plugin
ls [path], cat <path>, methods <plugin>
ps, log [plugin or pid] [-n N]
activate <plugin> <wasm>, revert <plugin>
stop <pid>, kill <pid>, help, quit";

/// Read a complete file through the host filesystem.
async fn read(host: &Host, path: &str) -> Result<String> {
    let file = host
        .filesystem()
        .await?
        .open(
            path,
            Open {
                follow: true,
                ..Open::default()
            },
        )
        .await?;
    let mut bytes = Vec::new();
    loop {
        let chunk = file.read(bytes.len() as u64, 65536).await?;
        if chunk.is_empty() {
            return Ok(String::from_utf8(bytes)?);
        }
        bytes.extend(chunk);
    }
}

/// Recognize the process IDs accepted by stop, kill, and log.
fn is_pid(text: &str) -> bool {
    text.rsplit_once('-').is_some()
        && text
            .get(text.len().saturating_sub(36)..)
            .is_some_and(|id| uuid::Uuid::try_parse(id).is_ok())
}

/// Render method signatures and their documentation.
fn methods(plugin: &str, doc: &Value) -> String {
    let mut lines = Vec::new();
    for method in doc["methods"].as_array().into_iter().flatten() {
        let params = method["params"].as_array().cloned().unwrap_or_default();
        let signature = params
            .iter()
            .map(|p| {
                format!(
                    "{}: {}",
                    p["name"].as_str().unwrap_or("?"),
                    p["schema"]["type"].as_str().unwrap_or("any")
                )
            })
            .collect::<Vec<_>>()
            .join(", ");
        lines.push(format!(
            "{plugin}.{}({signature}) -> {}",
            method["name"].as_str().unwrap_or("?"),
            method["result"]["schema"]["type"].as_str().unwrap_or("any")
        ));
        for field in ["summary", "description"] {
            if let Some(text) = method[field].as_str() {
                lines.push(format!("  {text}"));
            }
        }
        for p in params {
            if let Some(text) = p["description"].as_str() {
                lines.push(format!("  {}: {text}", p["name"].as_str().unwrap_or("?")));
            }
        }
    }
    lines.join("\n")
}

/// Parse journal filters, or treat invalid arguments as chat.
fn log_params(words: &[&str]) -> Option<Value> {
    let mut params = json!({});
    let mut words = words.iter();
    while let Some(word) = words.next() {
        if *word == "-n" {
            params["limit"] = json!(words.next()?.parse::<u64>().ok()?);
        } else if params.get("plugin").is_none() && params.get("pid").is_none() {
            params[if is_pid(word) { "pid" } else { "plugin" }] = json!(word);
        } else {
            return None;
        }
    }
    Some(params)
}

/// Execute one line and return its unlabelled result.
pub async fn execute_line(host: &Host, line: &str) -> Result<String> {
    let line = line.trim();
    if line.is_empty() {
        return Ok(String::new());
    }
    let (command, rest) = line.split_once(char::is_whitespace).unwrap_or((line, ""));
    let rest = rest.trim();
    let words: Vec<_> = rest.split_whitespace().collect();
    match (command, words.as_slice()) {
        ("help", []) => return Ok(HELP.into()),
        ("quit", []) => return Ok("Goodbye.".into()),
        ("cat", [path]) => return read(host, path).await,
        ("ls", [] | [_]) => {
            let mut entries = host
                .filesystem()
                .await?
                .list(words.first().copied().unwrap_or("/"))
                .await?;
            entries.sort_by(|a, b| a.name.cmp(&b.name));
            return Ok(entries
                .iter()
                .map(|e| format!("{}{}", e.name, if e.directory { "/" } else { "" }))
                .collect::<Vec<_>>()
                .join("\n"));
        }
        ("methods", [plugin]) => {
            let _activation = host.0.activations.lock().await;
            host.refresh_graph()
                .await
                .map_err(|e| anyhow::anyhow!("{e}"))?;
            return Ok(methods(
                plugin,
                &serde_json::from_str(
                    &read(host, &format!("/proc/core/plugins/{plugin}/discover.json")).await?,
                )?,
            ));
        }
        ("ps", []) => {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)?
                .as_millis();
            let mut processes: Vec<_> = host.processes().into_iter().collect();
            processes.sort_by(|a, b| a.0.cmp(&b.0));
            let mut lines = vec!["PID PLUGIN METHOD PARENT STATE AGE".into()];
            lines.extend(processes.iter().map(|(pid, p)| {
                format!(
                    "{pid} {} {} {} {} {}s",
                    p.plugin,
                    p.method,
                    p.parent.as_deref().unwrap_or("-"),
                    p.state,
                    now.saturating_sub(p.start_time) / 1000
                )
            }));
            return Ok(lines.join("\n"));
        }
        _ => {}
    }
    let explicit = (rest.is_empty() || rest.starts_with('{'))
        && command
            .split_once('.')
            .is_some_and(|(p, m)| !p.is_empty() && !m.is_empty());
    let (method, params) = match (command, words.as_slice()) {
        ("activate", [plugin, wasm]) => {
            ("core.activate".into(), json!({"plugin":plugin,"wasm":wasm}))
        }
        ("revert", [plugin]) => ("core.revert".into(), json!({"plugin":plugin})),
        ("stop" | "kill", [pid]) if is_pid(pid) => (format!("core.{command}"), json!({"pid":pid})),
        ("log", _) if log_params(&words).is_some() => {
            ("core.log".into(), log_params(&words).unwrap())
        }
        _ if explicit => (
            command.into(),
            if rest.is_empty() {
                json!({})
            } else {
                serde_json::from_str(rest)?
            },
        ),
        _ => ("agent.chat".into(), json!({"text":line})),
    };
    let value = match host.call(&method, params).await {
        Ok(value) => value,
        Err(error) => return Ok(format!("error: {}", serde_json::to_string_pretty(&error)?)),
    };
    if method == "agent.chat" && !explicit {
        return Ok(value["text"]
            .as_str()
            .or_else(|| value.as_str())
            .map(str::to_owned)
            .unwrap_or(serde_json::to_string_pretty(&value)?));
    }
    if method == "core.log" && command == "log" {
        return value
            .as_array()
            .unwrap()
            .iter()
            .map(|e| {
                let time = chrono::DateTime::parse_from_rfc3339(e["time"].as_str().unwrap())?
                    .with_timezone(&chrono::Local);
                let pid = e["pid"].as_str().unwrap();
                let prefix = format!("{}-", e["plugin"].as_str().unwrap());
                let short = pid
                    .strip_prefix(&prefix)
                    .map(|id| format!("{prefix}{}", &id[..8]))
                    .unwrap_or(pid.into());
                Ok(format!(
                    "{} {short} {} {}",
                    time.format("%H:%M:%S%.3f"),
                    e["stream"].as_str().unwrap(),
                    e["line"].as_str().unwrap()
                ))
            })
            .collect::<Result<Vec<_>>>()
            .map(|lines| lines.join("\n"));
    }
    Ok(serde_json::to_string_pretty(&value)?)
}

/// Dim text notifications and preserve other payloads.
pub fn notification_text(value: &Value) -> String {
    let params = &value["params"];
    if params["name"] != "text" {
        return value.to_string();
    }
    let payload = &params["params"];
    let text = payload["text"]
        .as_str()
        .or_else(|| payload.as_str())
        .map(str::to_owned)
        .unwrap_or(payload.to_string());
    format!("\x1b[2m{text}\x1b[0m")
}

/// Read and execute one line at a time. Interrupting a wait does not cancel its host process.
pub async fn run_input(host: Host) -> Result<()> {
    let terminal = std::io::IsTerminal::is_terminal(&std::io::stdin());
    let history = host.0.world.state_dir("core").join("history");
    let mut editor = if terminal {
        Some(DefaultEditor::new()?)
    } else {
        None
    };
    if let Some(editor) = &mut editor {
        let _ = editor.load_history(&history);
    }
    let mut printer = editor
        .as_mut()
        .map(|e| e.create_external_printer())
        .transpose()?;
    let (request, requests) =
        std::sync::mpsc::channel::<tokio::sync::oneshot::Sender<Result<Option<String>>>>();
    std::thread::spawn(move || {
        for response in requests {
            let line = (|| -> Result<_> {
                if let Some(editor) = &mut editor {
                    match editor.readline("> ") {
                        Ok(line) => {
                            editor.add_history_entry(&line)?;
                            editor.save_history(&history)?;
                            Ok(Some(line))
                        }
                        Err(ReadlineError::Interrupted | ReadlineError::Eof) => Ok(None),
                        Err(e) => Err(e.into()),
                    }
                } else {
                    std::io::stdin()
                        .lines()
                        .next()
                        .transpose()
                        .map_err(Into::into)
                }
            })();
            if response.send(line).is_err() {
                break;
            }
        }
    });
    let mut notifications = host.subscribe();
    let mut notification_task = tokio::task::JoinSet::new();
    notification_task.spawn(async move {
        loop {
            match notifications.recv().await {
                Ok(value) => {
                    let text = notification_text(&value);
                    if let Some(printer) = &mut printer {
                        let _ = printer.print(format!("{text}\n"));
                    } else {
                        println!("{text}");
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                Err(_) => break,
            }
        }
    });
    let mut interrupt = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
    loop {
        let (send, line) = tokio::sync::oneshot::channel();
        request.send(send)?;
        let line =
            tokio::select! { result = line => result??, _ = interrupt.recv() => return Ok(()) };
        let Some(line) = line else {
            return Ok(());
        };
        let text = tokio::select! {
            result = execute_line(&host, &line) => result.unwrap_or_else(|e| format!("error: {e}")),
            _ = interrupt.recv() => "stopped waiting; the call continues, see ps".into(),
        };
        if !text.is_empty() {
            println!("{text}");
        }
        if line.trim() == "quit" {
            return Ok(());
        }
    }
}