liminal-cli 0.3.32

//! The embedded ACP host and the exchange's own assistant presence projection.

use std::collections::VecDeque;
use std::path::Path;
use std::pin::Pin;
use std::sync::{
    Arc, Mutex,
    atomic::{AtomicBool, Ordering},
};
use std::task::{Context as TaskContext, Poll};

use liminal_server::agents::runtime::Agents;
use liminal_server::config::types::ConsoleConfig;
use liminal_server::server::connection::services::{
    ConnectionServices as _, LiminalConnectionServices,
};
use serde_json::{Value, json};
use tokio::io::{AsyncRead, ReadBuf};
use tokio::sync::oneshot;

pub const GUIDANCE: &str = "the assistant needs Claude Code signed in on this computer: run claude once and sign in, then start liminal up again";
pub(super) const HARNESS: &str = r#"id = "claude-code"
program = "claude-agent-acp"
args = []
# cwd = absent (the directory where liminal up was started); session = absent
prompt_queue = 32
content_bytes = 65536
acp_frame_bytes = 262144
pending_permissions = 1024
write_queue = 2048
read_queue = 1
stream_queue = 128
issuer_response_bytes = 65536
listed_sessions = 50
title_chars = 120
env = {}
env_remove = ["CLAUDECODE"]
"#;

pub(super) struct Assistant {
    runtime: tokio::runtime::Runtime,
    task: Option<tokio::task::JoinHandle<anyhow::Result<()>>>,
    stop: Option<oneshot::Sender<()>>,
}

impl Assistant {
    pub fn start(
        console: &ConsoleConfig,
        server: std::net::SocketAddr,
        path: &Path,
        services: Option<Arc<LiminalConnectionServices>>,
    ) -> anyhow::Result<Self> {
        let agents = services
            .and_then(|services| services.agents().cloned())
            .ok_or_else(|| {
                anyhow::anyhow!("console.assistant requires the full services profile and [agents]")
            })?;
        let now = u64::try_from(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)?
                .as_millis(),
        )?;
        agents
            .registry()
            .place_own("assistant", "open:assistant", now)?;
        agents.created("assistant")?;
        let (ready, started) = oneshot::channel();
        let notice = Arc::new(Notice {
            agents,
            said: AtomicBool::new(false),
            words: Mutex::new(None),
            ready: Mutex::new(Some(ready)),
        });
        let harness: crate::agent::config::Harness = console.harness.clone().map_or_else(
            || toml::from_str(HARNESS),
            |table| toml::Value::Table(table).try_into(),
        )?;
        harness.validate()?;
        let ticket = ticket(server);
        let path = path.to_path_buf();
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()?;
        let (stop, stopped) = oneshot::channel();
        let context = Context {
            notice: Arc::clone(&notice),
            stop: Some(stopped),
            replies: VecDeque::new(),
            text: String::new(),
            server: server.to_string(),
            bound: harness.content_bytes,
            prompt_bound: harness.prompt_queue,
        };
        let task = runtime.spawn(async move {
            let result = harness.serve_embedded(ticket, path, context).await;
            let reported = result
                .as_ref()
                .err()
                .map_or(Ok(()), |error| notice.report(&error.to_string()));
            notice.ready();
            reported.and(result)
        });
        // The host subscribes or reports its failed start before releasing readiness.
        runtime
            .block_on(started)
            .map_err(|_| anyhow::anyhow!("assistant startup ended without its readiness signal"))?;
        Ok(Self {
            runtime,
            task: Some(task),
            stop: Some(stop),
        })
    }

    pub fn stop(&mut self) -> anyhow::Result<()> {
        if let Some(stop) = self.stop.take() {
            if stop.send(()).is_err() {
                eprintln!("assistant host had already ended");
            }
        }
        if let Some(task) = self.task.take() {
            // A reported harness refusal does not prevent a clean deployment stop.
            match self.runtime.block_on(task)? {
                Ok(()) => {}
                Err(error) => eprintln!("assistant host ended: {error}"),
            }
        }
        Ok(())
    }
}

impl Drop for Assistant {
    fn drop(&mut self) {
        if let Err(error) = self.stop() {
            eprintln!("assistant cleanup: {error}");
        }
    }
}

pub(super) fn ticket(server: std::net::SocketAddr) -> crate::Config {
    crate::Config {
        server: server.to_string(),
        channel: "console.topology".to_owned(),
        credential: crate::Credential::Open {
            name: Some("assistant".to_owned()),
        },
        agent: Some("assistant".to_owned()),
        listen: crate::config::ListenConfig::default(),
        limits: crate::Limits {
            frame_bytes: 65_536,
            record_bytes: 1_048_576,
            reply_timeout_ms: 10_000,
            flow_steps: 8,
            request_bytes: 65_536,
            connect_timeout_ms: 2000,
            answer_bytes: 16_777_216,
        },
    }
}

pub struct Notice {
    agents: Arc<Agents>,
    said: AtomicBool,
    words: Mutex<Option<String>>,
    ready: Mutex<Option<oneshot::Sender<()>>>,
}

impl Notice {
    fn ready(&self) {
        match self.ready.lock() {
            Ok(mut ready) => {
                if let Some(ready) = ready.take() {
                    if ready.send(()).is_err() {
                        eprintln!("assistant startup observer ended");
                    }
                }
            }
            Err(error) => eprintln!("assistant startup lock: {error}"),
        }
    }

    pub(crate) fn capture(&self, line: &[u8]) {
        if let Some(words) = harness_words(line) {
            match self.words.lock() {
                Ok(mut held) => {
                    *held = Some(words);
                }
                Err(error) => eprintln!("assistant diagnostic lock: {error}"),
            }
        }
    }

    pub(crate) fn report(&self, fallback: &str) -> anyhow::Result<()> {
        if self.said.swap(true, Ordering::AcqRel) {
            return Ok(());
        }
        let words = self
            .words
            .lock()
            .map_err(|_| anyhow::anyhow!("assistant diagnostic lock poisoned"))?
            .take()
            .unwrap_or_else(|| fallback.to_owned());
        let reason = format!("{words}; {GUIDANCE}");
        let mut presence = self
            .agents
            .retained()
            .value("agents.assistant.presence")?
            .unwrap_or_else(|| json!({}));
        presence["present"] = json!(false);
        presence["reason"] = json!(reason);
        presence["words"] = json!(words);
        presence["action"] = json!(GUIDANCE);
        self.agents
            .retained()
            .publish(&[("agents.assistant.presence".to_owned(), presence)])?;
        Ok(())
    }
}

pub struct Context {
    pub(crate) notice: Arc<Notice>,
    pub(crate) stop: Option<oneshot::Receiver<()>>,
    replies: VecDeque<Option<String>>,
    text: String,
    server: String,
    bound: usize,
    prompt_bound: usize,
}

impl Context {
    pub(crate) fn ready(&self) {
        self.notice.ready();
    }

    pub(crate) fn inbox(&mut self, value: &Value) -> anyhow::Result<()> {
        if !value["text"].is_string()
            || value.get("type").is_some()
            || value.get("config").is_some()
            || value.get("session").is_some()
        {
            return Ok(());
        }
        if self.replies.len() > self.prompt_bound {
            return Ok(());
        }
        let subject = value["reply_to"].as_str().map(str::to_owned).or_else(|| {
            value["from"]
                .as_str()
                .map(|name| format!("agents.{name}.inbox"))
        });
        if let Some(subject) = &subject {
            let name = subject
                .strip_prefix("agents.")
                .and_then(|subject| subject.strip_suffix(".inbox"));
            anyhow::ensure!(
                name.is_some_and(crate::config::is_open_name),
                "assistant reply_to must name a seat inbox"
            );
        }
        self.replies.push_back(subject);
        Ok(())
    }

    pub(crate) async fn update(
        &mut self,
        method: Option<&str>,
        value: &Value,
        failure: Option<i64>,
    ) -> anyhow::Result<()> {
        let failed = failure.is_some();
        if self.replies.front().is_some_and(Option::is_some)
            && method == Some("session/update")
            && value["update"]["sessionUpdate"] == "agent_message_chunk"
        {
            if let Some(text) = value["update"]["content"]["text"].as_str() {
                anyhow::ensure!(
                    text.len() <= self.bound.saturating_sub(self.text.len()),
                    "assistant reply exceeds content_bytes"
                );
                self.text.push_str(text);
            }
        }
        let complete = failed
            || value.get("stopReason").is_some()
            || (method == Some("session/update")
                && value["update"]["sessionUpdate"] == "state_update"
                && value["update"]["state"] == "idle");
        if let Some(code) = failure.filter(|_| !self.replies.is_empty()) {
            self.notice.report(&format!(
                "Claude Code harness refusal {code}: no message was supplied"
            ))?;
        }
        if complete {
            let text = std::mem::take(&mut self.text);
            let reply = self.replies.pop_front();
            if matches!(reply, Some(None)) {
                // An answer with nobody to go to is said, so its sender's absence is never silent.
                println!("assistant answer went nowhere: the message named no sender");
            }
            if let Some(Some(subject)) = reply {
                let server = self.server.clone();
                tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
                    let credential = crate::Credential::Open {
                        name: Some("assistant".to_owned()),
                    }
                    .open_record();
                    let client = liminal_sdk::PushClient::connect_with_auth(&server, &credential)?;
                    client.publish(
                        &subject,
                        serde_json::to_vec(&json!({"from":"assistant", "text":text}))?,
                    )?;
                    let verdict = client.close()?;
                    anyhow::ensure!(verdict.is_proven_accepted(), "assistant reply was refused");
                    Ok(())
                })
                .await??;
            }
        }
        Ok(())
    }
}

/// Observes only structured ACP error messages before the existing parser drops them.
pub struct Observed<R> {
    read: R,
    notice: Option<Arc<Notice>>,
    line: Vec<u8>,
    bound: usize,
}

impl<R> Observed<R> {
    pub(crate) const fn new(read: R, notice: Option<Arc<Notice>>, bound: usize) -> Self {
        Self {
            read,
            notice,
            line: Vec::new(),
            bound,
        }
    }
}

impl<R: AsyncRead + Unpin> AsyncRead for Observed<R> {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut TaskContext<'_>,
        buffer: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let this = self.get_mut();
        let before = buffer.filled().len();
        let result = Pin::new(&mut this.read).poll_read(cx, buffer);
        if let Some(notice) = &this.notice {
            for byte in &buffer.filled()[before..] {
                if *byte == b'\n' {
                    notice.capture(&this.line);
                    this.line.clear();
                } else if this.line.len() < this.bound {
                    this.line.push(*byte);
                }
            }
        }
        result
    }
}

pub(super) fn harness_words(line: &[u8]) -> Option<String> {
    serde_json::from_slice::<Value>(line)
        .ok()?
        .pointer("/error/message")?
        .as_str()
        .map(str::to_owned)
}