liminal-cli 0.3.5

//! All network acts travel on the ticket rail and are decided at the server.

use super::config::Seat;
use crate::network::{Live, Rail, TapCloser, TapSubscription};
use serde_json::{Value, json};
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;

pub(super) enum Delivery {
    Inbox(Value),
    Cancel,
    State(Value),
}

pub(super) struct Exchange {
    rail: Arc<Live>,
    pub name: String,
    content_bytes: usize,
    request_bytes: usize,
    stream: Mutex<Option<liminal_sdk::PushClient>>,
}

pub(super) struct Listeners {
    pub events: mpsc::Receiver<anyhow::Result<Delivery>>,
    closers: Vec<Arc<dyn TapCloser>>,
    readers: Vec<JoinHandle<()>>,
}

impl Exchange {
    pub fn new(seat: &Seat) -> anyhow::Result<Arc<Self>> {
        Ok(Arc::new(Self {
            rail: Arc::new(Live::new(&seat.ticket).map_err(anyhow::Error::msg)?),
            name: seat.name.clone(),
            content_bytes: seat.harness.content_bytes,
            request_bytes: seat.ticket.limits.request_bytes,
            stream: Mutex::new(None),
        }))
    }

    pub fn request(&self, subject: &str, input: &Value) -> anyhow::Result<Value> {
        anyhow::ensure!(
            serde_json::to_vec(input)?.len() <= self.request_bytes,
            "host request exceeds request_bytes {}",
            self.request_bytes
        );
        let answer = self
            .rail
            .request(subject, input, true)
            .map_err(|_| anyhow::anyhow!("host transport or admission failed on {subject}"))?
            .ok_or_else(|| anyhow::anyhow!("host network request has no reply"))?;
        anyhow::ensure!(
            serde_json::to_vec(&answer)?.len() <= self.request_bytes,
            "host reply exceeds request_bytes {}",
            self.request_bytes
        );
        if let Some(refusal) = answer.get("refusal").and_then(Value::as_str) {
            // A key-signed seat holds no run pass; its key never leaves this process.
            if let Ok(proof) = std::env::var("LYS_AGENT_PASS").map(zeroize::Zeroizing::new) {
                anyhow::ensure!(
                    !proof.is_empty() && !refusal.contains(proof.as_str()),
                    "server refusal contains credential bytes"
                );
            }
            // A refusal name is protocol metadata; an arbitrary body never becomes a diagnostic.
            anyhow::ensure!(
                refusal
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_'),
                "host received an invalid refusal name"
            );
            anyhow::bail!("host network act refused: {refusal}");
        }
        Ok(answer)
    }

    pub fn refuse_prompt(&self, reason: &str) -> anyhow::Result<()> {
        let answer = self.request(
            "agents.refuse_prompt",
            &json!({"agent":self.name,"reason":reason}),
        )?;
        anyhow::ensure!(
            answer["refused"] == true,
            "agents.refuse_prompt did not confirm the refusal message"
        );
        Ok(())
    }

    pub fn gone(&self, status: i32, resumable: bool) -> anyhow::Result<()> {
        let answer = self.request("agents.session_close", &json!({"agent":self.name,"present":false,"exit_status":status,"session_state":super::exit::session_end(resumable)}))?;
        anyhow::ensure!(
            answer["recorded"] == true,
            "agents.session_close did not confirm gone and the session ending"
        );
        Ok(())
    }

    pub fn publish(&self, item: &Value) -> anyhow::Result<()> {
        // A single connection preserves publication order and avoids a handshake per token.
        let payload = serde_json::to_vec(item)?;
        anyhow::ensure!(
            payload.len() <= self.content_bytes,
            "stream item exceeds content_bytes {}",
            self.content_bytes
        );
        let mut held = self
            .stream
            .lock()
            .map_err(|_| anyhow::anyhow!("stream connection lock poisoned"))?;
        if held.is_none() {
            let pass = zeroize::Zeroizing::new(
                self.rail
                    .pass_for("liminal")
                    .map_err(|_| anyhow::anyhow!("stream audience pass unavailable"))?,
            );
            let credential = zeroize::Zeroizing::new(liminal_sdk::lys_pass_credential(&pass));
            let client =
                liminal_sdk::PushClient::connect_with_auth(self.rail.server_address(), &credential)
                    .map_err(|_| anyhow::anyhow!("stream connection refused"))?;
            *held = Some(client);
        }
        let client = held
            .as_ref()
            .ok_or_else(|| anyhow::anyhow!("stream connection absent"))?;
        client
            .publish(&format!("agents.{}.stream", self.name), payload)
            .map_err(|_| anyhow::anyhow!("agent stream publication refused"))?;
        let verdict = client
            .flush()
            .map_err(|_| anyhow::anyhow!("agent stream flush failed"))?;
        drop(held);
        anyhow::ensure!(
            verdict.is_proven_accepted(),
            "agent stream verdict: {} refused, {} unresolved",
            verdict.failures().len(),
            verdict.unresolved()
        );
        Ok(())
    }

    pub fn close_stream(&self) -> anyhow::Result<()> {
        let client = self
            .stream
            .lock()
            .map_err(|_| anyhow::anyhow!("stream connection lock poisoned"))?
            .take();
        if let Some(client) = client {
            let verdict = client
                .close()
                .map_err(|_| anyhow::anyhow!("agent stream close failed"))?;
            anyhow::ensure!(
                verdict.is_proven_accepted()
                    && verdict.mode() == liminal_sdk::FlushMode::FlushedAndHalfClosed,
                "agent stream close did not accept every publication and half-close"
            );
        }
        Ok(())
    }

    pub async fn listen(self: &Arc<Self>, event_bound: usize) -> anyhow::Result<Listeners> {
        let exchange = Arc::clone(self);
        let streams = tokio::task::spawn_blocking(move || {
            let mut streams: Vec<TapSubscription> = Vec::new();
            for suffix in ["state", "cancel", "inbox"] {
                let Ok(stream) = exchange
                    .rail
                    .tap(&format!("agents.{}.{suffix}", exchange.name))
                else {
                    let mut failures = vec![format!(
                        "host subscribe failed on agents.{}.{suffix}",
                        exchange.name
                    )];
                    for stream in streams {
                        if stream.closer.close().is_err() {
                            failures.push("host subscription rollback failed".to_owned());
                        }
                    }
                    anyhow::bail!("host subscribe failed: {}", failures.join("; "));
                };
                streams.push(stream);
            }
            Ok::<_, anyhow::Error>(streams)
        })
        .await??;
        // Transit shares the unread-mail count bound and backpressures the SDK when full.
        let (sender, events) = mpsc::channel(event_bound);
        let mut listeners = Listeners {
            events,
            closers: Vec::new(),
            readers: Vec::new(),
        };
        for (stream, suffix) in streams.into_iter().zip(["state", "cancel", "inbox"]) {
            listeners.closers.push(Arc::clone(&stream.closer));
            let sender = sender.clone();
            listeners.readers.push(tokio::task::spawn_blocking(move || {
                loop {
                    let record = stream
                        .reader
                        .receive()
                        .map_err(|_| anyhow::anyhow!("host subscription ended on {suffix}"));
                    let ended = record.is_err();
                    let record = record.map(|value| match suffix {
                        "inbox" => Delivery::Inbox(value),
                        "cancel" => Delivery::Cancel,
                        _ => Delivery::State(value),
                    });
                    if sender.blocking_send(record).is_err() || ended {
                        return;
                    }
                }
            }));
        }
        Ok(listeners)
    }

    pub fn inbox(&self, value: &Value) -> anyhow::Result<Option<String>> {
        // The server's refusal is an outcome record, not another user prompt.
        if value["type"] == "agent.host_refusal" {
            return Ok(None);
        }
        let text = match value.get("text") {
            Some(Value::String(text)) => text.clone(),
            Some(_) => anyhow::bail!("agent inbox text is not a string"),
            None => serde_json::to_string(value)?,
        };
        anyhow::ensure!(
            text.len() <= self.content_bytes,
            "inbox prompt exceeds content_bytes {}",
            self.content_bytes
        );
        Ok(Some(text))
    }
}

impl Listeners {
    pub async fn close(mut self) -> anyhow::Result<()> {
        self.events.close();
        let mut errors = Vec::new();
        for closer in &self.closers {
            if closer.close().is_err() {
                errors.push("host subscription close failed".to_owned());
            }
        }
        for reader in self.readers.drain(..) {
            if reader.await.is_err() {
                errors.push("host subscription reader panicked".to_owned());
            }
        }
        anyhow::ensure!(
            errors.is_empty(),
            "host subscription cleanup failed: {}",
            errors.join("; ")
        );
        Ok(())
    }
}

impl Drop for Listeners {
    fn drop(&mut self) {
        self.events.close();
        for closer in &self.closers {
            if closer.close().is_err() {
                eprintln!("liminal agent subscription cleanup failed");
            }
        }
    }
}