liminal-cli 0.3.32

//! The laptop deployment: one configuration and one production server lifecycle.

use liminal_server::server::runtime::{Deployment, run_with_deployment};
use signal_hook::{
    consts::signal::{SIGINT, SIGTERM},
    iterator::Signals,
};
use std::io::Write as _;
use std::net::{SocketAddr, TcpListener};
use std::os::unix::fs::{DirBuilderExt as _, OpenOptionsExt as _, PermissionsExt as _};
use std::path::{Path, PathBuf};
use std::sync::{
    Arc,
    atomic::{AtomicU8, Ordering},
};

pub(crate) mod assistant;
mod console;

#[cfg(test)]
#[path = "up_tests.rs"]
mod tests;

/// Starts the configured deployment, creating the laptop configuration only when absent.
///
/// # Errors
/// A configuration, listener, store or lifecycle act is refused by name.
pub fn run(path: Option<&Path>) -> anyhow::Result<()> {
    let explicit = path.is_some();
    let path = match path {
        Some(path) => path.to_path_buf(),
        None => default_path()?,
    };
    if !explicit && !path.exists() {
        let directory = path
            .parent()
            .ok_or_else(|| anyhow::anyhow!("configuration has no parent"))?;
        let text = configuration(directory)?;
        // All ports are checked together before creating even the private directory.
        let reserved = [
            reserve("127.0.0.1:7400".parse()?, "listen_address")?,
            reserve("127.0.0.1:7401".parse()?, "health_listen_address")?,
            reserve("127.0.0.1:7480".parse()?, "console.listen")?,
        ];
        write_configuration(&path, &text)?;
        drop(reserved);
    }
    let mut table: toml::Table = toml::from_str(&std::fs::read_to_string(&path)?)?;
    let Some(console) = console_configuration(&table)? else {
        liminal_server::run(&path)?;
        return Ok(());
    };
    let server = server_configuration(&path, &mut table)?;
    anyhow::ensure!(
        console.listen.ip().is_loopback() && console.listen.port() != 0,
        "console.listen must name a non-zero loopback port"
    );
    anyhow::ensure!(
        server.auth.is_none(),
        "console: Lys sign-in is not served by liminal up; keep the configured sign-in console server"
    );
    if console.assistant {
        anyhow::ensure!(
            server.agents_config().is_some() && server.services.profile == "full",
            "console.assistant requires [agents] and services.profile = 'full'"
        );
    }
    let mut reserved = Vec::new();
    reserved.push(reserve(server.listen_address, "listen_address")?);
    reserved.push(reserve(
        server.health_listen_address,
        "health_listen_address",
    )?);
    for address in &server.extra_listen_addresses {
        reserved.push(reserve(*address, "extra_listen_addresses")?);
    }
    if let Some(websocket) = &server.websocket {
        reserved.push(reserve(
            websocket.listen_address,
            "websocket.listen_address",
        )?);
    }
    if let Some(cluster) = &server.cluster {
        reserved.push(reserve(cluster.listen_address, "cluster.listen_address")?);
    }
    let listener = reserve(console.listen, "console.listen")?;
    drop(reserved);
    let shutdown = liminal_server::server::ShutdownHandle::new();
    let phase = Arc::new(AtomicU8::new(0));
    let signals = StopSignals::new(shutdown.clone(), Arc::clone(&phase))?;
    let mut deployment = ConsoleDeployment {
        listener: Some(listener),
        console,
        shutdown,
        phase: Arc::clone(&phase),
        path: path.clone(),
        server: server.listen_address,
    };
    run_with_deployment(&path, &server, &mut deployment)?;
    phase.store(4, Ordering::Release);
    println!("exchange stopped");
    drop(signals);
    Ok(())
}

/// Reads and validates an up configuration, including its optional console table.
///
/// # Errors
/// The file, a configuration field or a referenced schema is refused by name.
pub fn load_configuration(path: &Path) -> anyhow::Result<liminal_server::config::ServerConfig> {
    let mut table: toml::Table = toml::from_str(&std::fs::read_to_string(path)?)?;
    console_configuration(&table)?;
    server_configuration(path, &mut table)
}

fn server_configuration(
    path: &Path,
    table: &mut toml::Table,
) -> anyhow::Result<liminal_server::config::ServerConfig> {
    table.remove("console");
    let server: liminal_server::config::ServerConfig = std::mem::take(table).try_into()?;
    let mut server = liminal_server::config::apply_env_overrides(server)?;
    liminal_server::config::validate(&mut server, path.parent())?;
    Ok(server)
}

fn console_configuration(
    table: &toml::Table,
) -> anyhow::Result<Option<liminal_server::config::types::ConsoleConfig>> {
    table
        .get("console")
        .cloned()
        .map(toml::Value::try_into)
        .transpose()
        .map_err(Into::into)
}

fn default_path() -> anyhow::Result<PathBuf> {
    let home = std::env::var_os("HOME")
        .ok_or_else(|| anyhow::anyhow!("HOME is absent: use liminal up --config <file>"))?;
    anyhow::ensure!(
        !home.is_empty(),
        "HOME is empty: use liminal up --config <file>"
    );
    Ok(PathBuf::from(home).join(".liminal/liminal.toml"))
}

fn write_configuration(path: &Path, text: &str) -> anyhow::Result<()> {
    let directory = path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("configuration has no parent"))?;
    if !directory.exists() {
        std::fs::DirBuilder::new().mode(0o700).create(directory)?;
    }
    std::fs::set_permissions(directory, std::fs::Permissions::from_mode(0o700))?;
    let store = directory.join("store");
    if !store.exists() {
        std::fs::DirBuilder::new().mode(0o700).create(store)?;
    }
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .mode(0o600)
        .open(path)?;
    file.write_all(text.as_bytes())?;
    file.sync_all()?;
    Ok(())
}

fn reserve(address: SocketAddr, key: &str) -> anyhow::Result<TcpListener> {
    TcpListener::bind(address).map_err(|error| {
        let holder = std::process::Command::new("lsof")
            .args(["-nP", &format!("-iTCP:{}", address.port()), "-sTCP:LISTEN"])
            .output()
            .map_or_else(
                |_| String::new(),
                |output| String::from_utf8_lossy(&output.stdout).trim().to_owned(),
            );
        let holder = if holder.is_empty() {
            String::new()
        } else {
            format!("; held by {holder}")
        };
        anyhow::anyhow!(
            "port {} ({address}) refused: {error}{holder}; move it with {key} in the configuration",
            address.port()
        )
    })
}

fn configuration(directory: &Path) -> anyhow::Result<String> {
    let store = directory.join("store");
    let store = toml::Value::String(
        store
            .to_str()
            .ok_or_else(|| anyhow::anyhow!("HOME is not Unicode"))?
            .to_owned(),
    );
    let mut text = format!(
        r#"# Created by liminal up. Disabled optional sections are named explicitly:
# auth = absent (open exchange); cluster = absent; websocket = absent
# templates = absent; open.construction = absent; channel schemas = absent
listen_address = "127.0.0.1:7400"
extra_listen_addresses = []
health_listen_address = "127.0.0.1:7401"
drain_timeout_ms = 0
channels = []
routing_rules = []
persistence_path = {store}

[open]
reach = "loopback"

[services]
profile = "full"
[services.self_streams]
cost_rows = 10000
events_rows = 100000

[limits]
max_connections = 256
max_subscriptions_per_connection = 1024
max_conversations_per_connection = 32
max_pending_pushes_per_connection = 32
max_pending_conversation_replies_per_connection = 32
max_pending_replies_per_conversation = 8
max_connection_inbox_bytes = 4194304
max_connection_outbound_bytes = 4194304
max_subscription_inbox_depth = 4096
delivery_slice_budget = 32
max_channels = 512
reply_timeout_ms = 5000

[agents]
max_per_owner = 64
max_per_exchange = 64
max_field_bytes = 512
max_event_bytes = 16384
replay_page_rows = 64
max_watchers = 256
watcher_depth = 16
watcher_bytes = 262144
max_listened_subjects = 1024
max_subject_bytes = 4096

[participant]
wire_frame_limit = 65536
attach_receipt_ttl_ms = 60000
receipt_provenance_ttl_ms = 600000
live_receipt_server_report_threshold = 1024
max_live_attach_receipts_per_participant = 8
receipt_provenance_server_report_threshold = 4096
receipt_provenance_per_conversation_report_threshold = 256
max_receipt_provenance_per_participant = 64
max_retired_identity_slots_server = 1024
identity_slots = 4
observer_recovery_max_entries = 64
max_semantic_conversations_per_connection = 32
max_ordinary_record_entries = 1
max_ordinary_record_bytes = 131072
max_generated_marker_entries = 1
max_generated_marker_bytes = 4096
mandatory_transaction_bound_entries = 4
mandatory_transaction_bound_bytes = 16384
full_recovery_claim_entries = 4
full_recovery_claim_bytes = 16384
retained_capacity_entries = 2048
retained_capacity_bytes = 16777216
max_retained_record_rows = 1024
closure_episode_churn_limit = 1024
[participant.pruning]
keep_acknowledged_behind_checkpoint = 512

[console]
listen = "127.0.0.1:7480"
assistant = true
"#
    );
    text.push_str("\n[console.harness]\n");
    text.push_str(assistant::HARNESS);
    Ok(text)
}

struct ConsoleDeployment {
    listener: Option<TcpListener>,
    console: liminal_server::config::types::ConsoleConfig,
    shutdown: liminal_server::server::ShutdownHandle,
    phase: Arc<AtomicU8>,
    path: PathBuf,
    server: SocketAddr,
}

impl Deployment for ConsoleDeployment {
    fn shutdown(&self) -> liminal_server::server::ShutdownHandle {
        self.shutdown.clone()
    }

    fn serve(
        &mut self,
        supervisor: liminal_server::server::ConnectionSupervisor,
        services: Option<
            Arc<liminal_server::server::connection::services::LiminalConnectionServices>,
        >,
        shutdown: liminal_server::server::ShutdownHandle,
    ) -> Result<(), liminal_server::ServerError> {
        let listener =
            self.listener
                .take()
                .ok_or_else(|| liminal_server::ServerError::ListenerAccept {
                    message: "console listener already started".to_owned(),
                })?;
        let console = console::start(listener, supervisor)?;
        let mut assistant = if self.console.assistant {
            Some(
                assistant::Assistant::start(&self.console, self.server, &self.path, services)
                    .map_err(|error| lifecycle_error(&error))?,
            )
        } else {
            None
        };
        let url = format!("http://{}/", self.console.listen);
        self.phase.store(1, Ordering::Release);
        println!("{url}");
        open_browser(&url);
        self.shutdown.wait();
        self.phase.store(1, Ordering::Release);
        let assistant_stop = assistant
            .as_mut()
            .map_or(Ok(()), assistant::Assistant::stop);
        if assistant_stop.is_ok() {
            println!("assistant stopped");
        } else if let Err(error) = &assistant_stop {
            eprintln!("assistant stop: {error:#}");
        }
        self.phase.store(2, Ordering::Release);
        let stopped = console.shutdown();
        if stopped.is_ok() {
            println!("console stopped");
        }
        self.phase.store(3, Ordering::Release);
        shutdown.initiate();
        stopped.and_then(|()| assistant_stop.map_err(|error| lifecycle_error(&error)))
    }
}

struct StopSignals {
    handle: signal_hook::iterator::Handle,
    worker: Option<std::thread::JoinHandle<()>>,
}

impl StopSignals {
    fn new(
        shutdown: liminal_server::server::ShutdownHandle,
        phase: Arc<AtomicU8>,
    ) -> anyhow::Result<Self> {
        let mut signals = Signals::new([SIGINT, SIGTERM])?;
        let handle = signals.handle();
        let worker = std::thread::spawn(move || {
            for _signal in signals.forever() {
                if !shutdown.initiate() {
                    let waiting = match phase.load(Ordering::Acquire) {
                        0 => "startup",
                        1 => "assistant",
                        2 => "console listener",
                        3 => "exchange drain and store flush",
                        _ => "process exit",
                    };
                    eprintln!("still waiting for {waiting}");
                }
            }
        });
        Ok(Self {
            handle,
            worker: Some(worker),
        })
    }
}

impl Drop for StopSignals {
    fn drop(&mut self) {
        self.handle.close();
        if let Some(worker) = self.worker.take() {
            if worker.join().is_err() {
                eprintln!("up signal worker panicked");
            }
        }
    }
}

fn open_browser(url: &str) {
    let opener = if cfg!(target_os = "macos") {
        "open"
    } else {
        "xdg-open"
    };
    match std::process::Command::new(opener)
        .arg(url)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
    {
        Ok(mut child) => {
            let url = url.to_owned();
            std::thread::spawn(move || match child.wait() {
                Ok(status) if status.success() => {}
                Ok(status) => eprintln!("browser opener {opener} exited {status}; open {url}"),
                Err(error) => eprintln!("browser opener {opener}: {error}"),
            });
        }
        Err(error) => eprintln!("browser opener {opener} unavailable: {error}; open {url}"),
    }
}

fn lifecycle_error(error: &anyhow::Error) -> liminal_server::ServerError {
    liminal_server::ServerError::ListenerAccept {
        message: error.to_string(),
    }
}

/// Serves the assistant's in-memory ticket to the ACP harness's MCP child.
///
/// # Errors
/// The up configuration or its ticket tools failed.
pub fn run_tools(path: &Path) -> anyhow::Result<bool> {
    let mut table: toml::Table = toml::from_str(&std::fs::read_to_string(path)?)?;
    if console_configuration(&table)?.is_none() {
        return Ok(false);
    }
    let server = server_configuration(path, &mut table)?;
    let ticket = assistant::ticket(server.listen_address);
    let live = crate::Live::new(&ticket).map_err(anyhow::Error::msg)?;
    let mut tools = crate::Server::configured(live, &ticket).map_err(anyhow::Error::msg)?;
    tools.serve(&mut std::io::stdin().lock(), &mut std::io::stdout())?;
    Ok(true)
}