liminal-cli 0.3.32

//! 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 embedded: Option<crate::up::assistant::Context>,
}

pub(super) async fn serve(seat: &Seat) -> anyhow::Result<Outcome> {
    serve_with(seat, None).await
}

async fn serve_with(
    seat: &Seat,
    embedded: Option<crate::up::assistant::Context>,
) -> anyhow::Result<Outcome> {
    let mut running = Running::new(seat, embedded).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;
    if let Some(embedded) = &running.embedded {
        if let Some(error) = errors.first() {
            embedded.notice.report(error)?;
        }
        embedded.ready();
    }
    Ok(Outcome {
        status: if status == 0 && !errors.is_empty() {
            1
        } else {
            status
        },
        errors,
    })
}

impl Running {
    async fn new(
        seat: &Seat,
        mut embedded: Option<crate::up::assistant::Context>,
    ) -> anyhow::Result<Self> {
        let term = signal(SignalKind::terminate())?;
        let interrupt = signal(SignalKind::interrupt())?;
        let lys = matches!(seat.ticket.credential, crate::Credential::Lys(_));
        let stdin = std::fs::metadata("/dev/stdin")
            .ok()
            .map(|meta| meta.file_type());
        let input: InputEnd =
            if let Some(stop) = embedded.as_mut().and_then(|embedded| embedded.stop.take()) {
                Box::pin(async move {
                    if stop.await.is_err() {
                        eprintln!("assistant lifecycle owner ended");
                    }
                    Ok(())
                })
            } else if watches_input(lys, stdin) {
                Box::pin(eof())
            } else {
                Box::pin(std::future::pending())
            };
        // Only a run pass has retained passes to revoke; a key-signed seat's
        // passes end at their own expiry, and an open seat holds none.
        let revoke = match &seat.ticket.credential {
            crate::Credential::Lys(table) => Self::revoke(seat, table)?,
            crate::Credential::Open { .. } => None,
            crate::Credential::Token(_) => {
                anyhow::bail!("agent run takes [lys] or [open], not a token_file")
            }
        };
        let exchange = Exchange::new(seat)?;
        let notice = embedded
            .as_ref()
            .map(|embedded| Arc::clone(&embedded.notice));
        let (child, pipe) = match launch(seat, notice) {
            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,
            embedded,
        })
    }

    /// The run pass's revocation, when Lys launched this seat with one.
    fn revoke(seat: &Seat, table: &crate::lys::LysTable) -> anyhow::Result<Option<Revoke>> {
        match std::env::var("LYS_AGENT_PASS") {
            Ok(pass) if table.key_file.as_os_str().is_empty() => Ok(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")
            }
            _ => Ok(None),
        }
    }

    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,
    notice: Option<Arc<crate::up::assistant::Notice>>,
) -> 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);
    // AGT-007 R4: the seat says the harness's environment.
    for name in &seat.harness.env_remove {
        command.env_remove(name);
    }
    command.envs(&seat.harness.env);
    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 stdout = crate::up::assistant::Observed::new(stdout, notice, seat.harness.acp_frame_bytes);
    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"))?
}

/// Whether the host ends when its stdin does (AGT-007 R1): only a Lys seat,
/// whose launcher holds a pipe or a socket there and closes it to end the
/// run. An open seat, a file, /dev/null or no stdin has no launcher to watch.
pub(super) fn watches_input(lys: bool, stdin: Option<std::fs::FileType>) -> bool {
    use std::os::unix::fs::FileTypeExt as _;
    lys && stdin.is_some_and(|kind| kind.is_fifo() || kind.is_socket())
}

async fn eof() -> anyhow::Result<()> {
    use std::os::unix::fs::OpenOptionsExt as _;
    let unstarted = |error: std::io::Error| {
        anyhow::anyhow!("the Lys host input watch on stdin could not start: {error}")
    };
    let flags = i32::try_from(rustix::fs::OFlags::NONBLOCK.bits())?;
    let file = std::fs::OpenOptions::new()
        .read(true)
        .custom_flags(flags)
        .open("/dev/stdin")
        .map_err(unstarted)?;
    let input = tokio::io::unix::AsyncFd::new(file).map_err(unstarted)?;
    loop {
        let mut ready = input
            .readable()
            .await
            .map_err(|error| anyhow::anyhow!("the Lys host input on stdin failed: {error}"))?;
        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!(
                    "the Lys host input on stdin failed: {error}"
                ));
            }
        }
    }
}

impl super::config::Harness {
    pub(crate) async fn serve_embedded(
        self,
        ticket: crate::Config,
        path: std::path::PathBuf,
        embedded: crate::up::assistant::Context,
    ) -> anyhow::Result<()> {
        let cwd = self.cwd.clone().map_or_else(std::env::current_dir, Ok)?;
        let seat = Seat {
            path,
            ticket,
            harness: self,
            name: "assistant".to_owned(),
            cwd,
        };
        let outcome = serve_with(&seat, Some(embedded)).await?;
        anyhow::ensure!(outcome.errors.is_empty(), "{}", outcome.errors.join("; "));
        anyhow::ensure!(
            matches!(outcome.status, 0 | 130 | 143),
            "Claude Code harness exited {}",
            outcome.status
        );
        Ok(())
    }
}