liminal-cli 0.3.4

//! Signals end the host; every exit closes the child, subscriptions and run audience.

use super::config::Seat;
use super::containment::Contained;
use super::exchange::Exchange;
use super::exit::Revoke;
use super::wire::Pipe;
use std::process::Stdio;
use std::sync::Arc;
use tokio::process::Command;
use tokio::signal::unix::{SignalKind, signal};

pub(super) struct Outcome {
    pub status: i32,
    pub errors: Vec<String>,
}

type ExitTask = Option<tokio::task::JoinHandle<anyhow::Result<i32>>>;
type InputEnd = std::pin::Pin<Box<dyn std::future::Future<Output = anyhow::Result<()>> + Send>>;

pub(super) struct Running {
    child: Contained,
    pub pipe: Option<Pipe>,
    pub exited: ExitTask,
    pub exchange: Arc<Exchange>,
    revoke: Option<Revoke>,
    pub term: tokio::signal::unix::Signal,
    pub interrupt: tokio::signal::unix::Signal,
    pub input: InputEnd,
    pub listeners: Option<super::exchange::Listeners>,
    pub writer_consumed: bool,
    resumable: bool,
}

pub(super) async fn serve(seat: &Seat) -> anyhow::Result<Outcome> {
    let mut running = Running::new(seat).await?;
    let mut errors = Vec::new();
    let result = match running.open(seat).await {
        Ok(session) => {
            running.resumable = session.resumable;
            running.session(&seat.harness, session, &mut errors).await
        }
        Err((status, error)) => {
            if let Some(error) = error {
                errors.push(error);
            }
            Ok(status)
        }
    };
    let status = match result {
        Ok(status) => status,
        Err(error) => {
            errors.push(error.to_string());
            1
        }
    };
    running.close(status, &mut errors).await;
    Ok(Outcome {
        status: if status == 0 && !errors.is_empty() {
            1
        } else {
            status
        },
        errors,
    })
}

impl Running {
    async fn new(seat: &Seat) -> anyhow::Result<Self> {
        let term = signal(SignalKind::terminate())?;
        let interrupt = signal(SignalKind::interrupt())?;
        let input = Box::pin(eof());
        let crate::Credential::Lys(table) = &seat.ticket.credential else {
            anyhow::bail!("agent run requires [lys]");
        };
        // Only a run pass has retained passes to revoke; a key-signed seat's
        // passes end at their own expiry.
        let revoke = match std::env::var("LYS_AGENT_PASS") {
            Ok(pass) if table.key_file.as_os_str().is_empty() => Some(Revoke::new(
                &table.issuer,
                &seat.name,
                pass,
                seat.harness.issuer_response_bytes,
                seat.ticket.limits.reply_timeout_ms,
            )?),
            _ if table.key_file.as_os_str().is_empty() => {
                anyhow::bail!("LYS_AGENT_PASS is missing or not Unicode")
            }
            _ => None,
        };
        let exchange = Exchange::new(seat)?;
        let (child, pipe) = match launch(seat) {
            Ok(launched) => launched,
            Err(error) => {
                let Some(revoke) = revoke else {
                    return Err(error);
                };
                let revoked = tokio::task::spawn_blocking(move || revoke.send()).await?;
                return match revoked {
                    Ok(()) => Err(error),
                    Err(revoked) => Err(anyhow::anyhow!("{error}; {revoked}")),
                };
            }
        };
        let exited = Some(child.observe_exit());
        Ok(Self {
            child,
            pipe: Some(pipe),
            exited,
            exchange,
            revoke,
            term,
            interrupt,
            input,
            listeners: None,
            writer_consumed: false,
            resumable: false,
        })
    }

    async fn open(
        &mut self,
        seat: &Seat,
    ) -> Result<super::handshake::Session, (i32, Option<String>)> {
        let Some(pipe) = self.pipe.as_mut() else {
            return Err((1, Some("ACP pipe is absent".to_owned())));
        };
        tokio::select! {
            opened = super::handshake::open(pipe, &seat.cwd, &seat.path, seat.harness.session.as_deref(), &seat.harness) => opened.map_err(|error| (1, Some(error.to_string()))),
            result = observe(&mut self.exited) => match result { Ok(status) => Err((status, None)), Err(error) => Err((1, Some(error.to_string()))) },
            _ = self.term.recv() => Err((143, None)),
            _ = self.interrupt.recv() => Err((130, None)),
            result = &mut self.input => match result { Ok(()) => Err((0, None)), Err(error) => Err((1, Some(error.to_string()))) },
        }
    }

    async fn close(&mut self, status: i32, errors: &mut Vec<String>) {
        if let Some(pipe) = self.pipe.take() {
            pipe.reader.abort();
            pipe.writer.abort();
            if let Err(error) = pipe.reader.await {
                if !error.is_cancelled() {
                    errors.push("ACP reader panicked".to_owned());
                }
            }
            if !self.writer_consumed {
                match pipe.writer.await {
                    Ok(Ok(())) => {}
                    Ok(Err(error)) => errors.push(error.to_string()),
                    Err(error) if error.is_cancelled() => {}
                    Err(_) => errors.push("ACP writer panicked".to_owned()),
                }
            }
            drop(pipe.output);
        }
        if let Err(error) = self.child.kill() {
            errors.push(error.to_string());
        }
        if self.exited.is_some() {
            if let Err(error) = observe(&mut self.exited).await {
                errors.push(error.to_string());
            }
        }
        if let Err(error) = self.child.reap().await {
            errors.push(error.to_string());
        }
        if let Some(listeners) = self.listeners.take() {
            if let Err(error) = listeners.close().await {
                errors.push(error.to_string());
            }
        }
        if let Err(error) = self.exchange.close_stream() {
            errors.push(error.to_string());
        }
        let closed = Arc::clone(&self.exchange);
        let resumable = self.resumable;
        match tokio::task::spawn_blocking(move || closed.gone(status, resumable)).await {
            Ok(Ok(())) => {}
            Ok(Err(error)) => errors.push(error.to_string()),
            Err(_) => errors.push("host ending worker panicked".to_owned()),
        }
        if let Some(revoke) = self.revoke.take() {
            match tokio::task::spawn_blocking(move || revoke.send()).await {
                Ok(Ok(())) => {}
                Ok(Err(error)) => errors.push(error.to_string()),
                Err(_) => errors.push("run pass revoke worker panicked".to_owned()),
            }
        }
    }
}

fn launch(seat: &Seat) -> anyhow::Result<(Contained, Pipe)> {
    let executable = super::catalogue::executable(&seat.harness.program)?;
    let mut command = Command::new(executable);
    // The launch profile owns model routing; human harness diagnostics may contain payloads.
    command
        .args(&seat.harness.args)
        .current_dir(&seat.cwd)
        .process_group(0)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .kill_on_drop(true);
    let mut child = Contained::new(
        command
            .spawn()
            .map_err(|error| anyhow::anyhow!("ACP spawn failed: {}", error.kind()))?,
    )?;
    let stdout = child
        .child()
        .stdout
        .take()
        .ok_or_else(|| anyhow::anyhow!("ACP child stdout is not piped"))?;
    let stdin = child
        .child()
        .stdin
        .take()
        .ok_or_else(|| anyhow::anyhow!("ACP child stdin is not piped"))?;
    let pipe = Pipe::start(
        stdout,
        stdin,
        seat.harness.acp_frame_bytes,
        seat.harness.write_queue,
        seat.harness.read_queue,
    );
    Ok((child, pipe))
}

pub(super) async fn observe(
    task: &mut Option<tokio::task::JoinHandle<anyhow::Result<i32>>>,
) -> anyhow::Result<i32> {
    let result = task
        .as_mut()
        .ok_or_else(|| anyhow::anyhow!("ACP exit was already observed"))?
        .await;
    task.take();
    result.map_err(|_| anyhow::anyhow!("ACP exit observer panicked"))?
}

async fn eof() -> anyhow::Result<()> {
    use std::os::unix::fs::OpenOptionsExt as _;
    let flags = i32::try_from(rustix::fs::OFlags::NONBLOCK.bits())?;
    let file = std::fs::OpenOptions::new()
        .read(true)
        .custom_flags(flags)
        .open("/dev/stdin")?;
    let input = tokio::io::unix::AsyncFd::new(file)?;
    loop {
        let mut ready = input.readable().await?;
        match ready.try_io(|input| {
            rustix::io::read(input.get_ref(), &mut [0_u8; 1]).map_err(std::io::Error::from)
        }) {
            Ok(Ok(0)) => return Ok(()),
            Ok(Ok(_)) | Err(_) => {}
            Ok(Err(error)) => {
                return Err(anyhow::anyhow!("Lys host input failed: {}", error.kind()));
            }
        }
    }
}