liminal-cli 0.3.1

//! A real exchange and a child ticket, with blocking signal-driven reads.

use std::error::Error;
use std::io::{BufRead, BufReader};
use std::path::Path;
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
use std::sync::Arc;

use base64::Engine as _;
use ed25519_dalek::{Signer as _, SigningKey};
use liminal_server::server::connection::ConnectionSupervisor;
use liminal_server::server::listener::ServerListener;
use lys_pass::{Claims, Holder, Mode, Right};
use serde_json::{Value, json};

pub(super) type TestResult = Result<(), Box<dyn Error>>;

pub(super) fn require(ok: bool, words: &str) -> TestResult {
    if ok { Ok(()) } else { Err(words.into()) }
}

pub(super) fn finish_results(results: impl IntoIterator<Item = TestResult>) -> TestResult {
    let failures: Vec<_> = results.into_iter().filter_map(Result::err).collect();
    if failures.is_empty() {
        Ok(())
    } else {
        Err(failures
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join("; ")
            .into())
    }
}

pub(super) fn child_result(result: Result<(i32, String), Box<dyn Error>>) -> TestResult {
    let (code, stderr) = result?;
    require(code == 0, "the ticket exited unsuccessfully")?;
    require(
        stderr.is_empty(),
        "the ticket emitted stderr during the contract",
    )
}

/// One process under test; the environment names the separately built binary.
pub(super) struct Host {
    child: Option<Child>,
    input: Option<ChildStdin>,
    output: Option<BufReader<ChildStdout>>,
    next_id: u64,
}

/// The `liminal` binary cargo built for this package, beside the test
/// executable's `deps` directory. Running the package's tests builds it; a
/// lib-only run that did not is refused by name rather than guessed.
fn ticket_binary() -> Result<std::path::PathBuf, Box<dyn Error>> {
    let executable = std::env::current_exe()?;
    let profile = executable
        .parent()
        .and_then(Path::parent)
        .ok_or("the test executable has no profile directory")?;
    let binary = profile.join(format!("liminal{}", std::env::consts::EXE_SUFFIX));
    require(
        binary.is_file(),
        "the liminal binary is not built beside this test; run the package's tests",
    )?;
    Ok(binary)
}

impl Host {
    pub fn start(config: &Path, agent: Option<&str>) -> Result<Self, Box<dyn Error>> {
        let mut command = Command::new(ticket_binary()?);
        command.args(["mcp", "--config"]).arg(config);
        command.env_remove("LYS_AGENT").env_remove("LYS_AGENT_PASS");
        if let Some(agent) = agent {
            command
                .env("LYS_AGENT", agent)
                .env("LYS_AGENT_PASS", "fixture-run-proof");
        }
        let child = command
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()?;
        let mut host = Self {
            child: Some(child),
            input: None,
            output: None,
            next_id: 0,
        };
        let child = host.child.as_mut().ok_or("the child is absent")?;
        host.input = Some(child.stdin.take().ok_or("the child has no stdin")?);
        host.output = Some(BufReader::new(
            child.stdout.take().ok_or("the child has no stdout")?,
        ));
        Ok(host)
    }

    pub fn read(&mut self) -> Result<Value, Box<dyn Error>> {
        let mut bytes = Vec::new();
        let n = self
            .output
            .as_mut()
            .ok_or("the host stdout is closed")?
            .read_until(b'\n', &mut bytes)?;
        require(n > 0, "the ticket closed stdout before answering")?;
        require(
            bytes.last() == Some(&b'\n'),
            "the host received an incomplete line",
        )?;
        serde_json::from_slice(&bytes).map_err(|_| "the host received invalid JSON".into())
    }

    pub fn request(&mut self, method: &str, params: &Value) -> Result<Value, Box<dyn Error>> {
        self.next_id += 1;
        let line = json!({"jsonrpc":"2.0", "id":self.next_id, "method":method, "params":params});
        super::framing::write_message(
            self.input.as_mut().ok_or("the host stdin is closed")?,
            &line,
        )?;
        let answer = self.read()?;
        require(
            answer["id"] == self.next_id,
            "the answer does not carry the request id",
        )?;
        Ok(answer)
    }

    pub fn initialize(&mut self) -> Result<Value, Box<dyn Error>> {
        self.request(
            "initialize",
            &json!({"protocolVersion":"2025-11-25", "capabilities":{},
            "clientInfo":{"name":"agt-contract-host","version":"1"}}),
        )
    }

    pub fn tool(&mut self, name: &str, arguments: &Value) -> Result<Value, Box<dyn Error>> {
        let answer = self.request("tools/call", &json!({"name":name,"arguments":arguments}))?;
        answer
            .get("result")
            .cloned()
            .ok_or_else(|| "the ticket refused the tool method".into())
    }

    pub fn close(mut self) -> Result<(i32, String), Box<dyn Error>> {
        drop(self.input.take());
        drop(self.output.take());
        let output = self
            .child
            .take()
            .ok_or("the child is absent")?
            .wait_with_output()?;
        let code = output
            .status
            .code()
            .ok_or("the ticket died without an exit code")?;
        let stderr = String::from_utf8(output.stderr).map_err(|_| "ticket stderr is not UTF-8")?;
        Ok((code, stderr))
    }
}

impl Drop for Host {
    fn drop(&mut self) {
        drop(self.input.take());
        drop(self.output.take());
        if let Some(mut child) = self.child.take() {
            match child.try_wait() {
                Ok(Some(_)) => {}
                Ok(None) => {
                    if let Err(error) = child.kill() {
                        eprintln!("the owned test child could not be stopped: {error}");
                    }
                }
                Err(error) => eprintln!("the owned test child could not be inspected: {error}"),
            }
            if let Err(error) = child.wait() {
                eprintln!("the owned test child could not be joined: {error}");
            }
        }
    }
}

/// Keys and passes belong exclusively to this test's issuer.
pub(super) struct Exchange {
    listener: Option<ServerListener>,
    pub supervisor: ConnectionSupervisor,
    pub address: String,
    pub services: Arc<liminal_server::server::connection::LiminalConnectionServices>,
    console: Option<liminal_server::server::connection::ConsoleForTests>,
    directory: tempfile::TempDir,
    key: SigningKey,
}

impl Exchange {
    pub fn start() -> Result<Self, Box<dyn Error>> {
        Self::configured(true, None)
    }

    /// An exchange whose console taps hold `capacity` messages each.
    pub fn with_tap_capacity(capacity: usize) -> Result<Self, Box<dyn Error>> {
        Self::configured(true, Some(capacity))
    }

    pub fn ungated() -> Result<Self, Box<dyn Error>> {
        Self::configured(false, None)
    }

    fn configured(lys: bool, tap_capacity: Option<usize>) -> Result<Self, Box<dyn Error>> {
        let directory = tempfile::tempdir()?;
        let key = SigningKey::from_bytes(&[41; 32]);
        let keys = json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"agt-fixture",
            "alg":"EdDSA","use":"sig","x":encode(&key.verifying_key().to_bytes())}]});
        let keys_path = directory.path().join("keys.json");
        std::fs::write(&keys_path, serde_json::to_vec(&keys)?)?;
        let store = directory.path().join("store");
        std::fs::create_dir(&store)?;
        let mut channels = vec!["messages.opaque".to_owned()];
        channels.extend((0..17).map(|n| format!("messages.s{n}")));
        channels.push("messages.invalid".to_owned());
        let channels: Vec<_> = channels
            .iter()
            .map(|name| json!({"name":name,"durable":false}))
            .collect();
        let mut value = json!({"listen_address":"127.0.0.1:0", "health_listen_address":"127.0.0.1:0",
            "drain_timeout_ms":0, "channels":channels, "routing_rules":[], "persistence_path":store,
            "auth":{"token":"agt-fixture-worker", "lys":{"issuer":"https://agt.test","network_id":"agt-network","keys_path":keys_path,"agents":{}}},"limits":{"max_channels":64},
            "participant":participant()});
        if !lys {
            value
                .as_object_mut()
                .ok_or("the fixture configuration is not an object")?
                .remove("auth");
        }
        let mut config: liminal_server::config::ServerConfig = serde_json::from_value(value)?;
        let opaque = config
            .channels
            .iter_mut()
            .find(|channel| channel.name == "messages.opaque")
            .ok_or("the opaque fixture channel is absent")?;
        let document = json!({ liminal::channel::OPAQUE_KEYWORD: true });
        opaque.loaded_schema = Some(liminal_server::config::LoadedSchema {
            bytes: serde_json::to_vec(&document)?,
            document,
        });
        let invalid = config
            .channels
            .iter_mut()
            .find(|channel| channel.name == "messages.invalid")
            .ok_or("the invalid fixture channel is absent")?;
        invalid.loaded_schema = Some(liminal_server::config::LoadedSchema {
            bytes: br#"{"type":"object"}"#.to_vec(),
            document: json!({"type":"object"}),
        });
        let mut services =
            liminal_server::server::connection::LiminalConnectionServices::from_config(&config)?;
        if let Some(capacity) = tap_capacity {
            services = services.with_console_tap_capacity(capacity);
        }
        if lys {
            services = services.with_agent_fixture("agt-network", Arc::new(AgentPlacement))?;
            provision_agents(&services)?;
        }
        let services = Arc::new(services);
        let mut builder = ConnectionSupervisor::builder(services.clone()).limits(config.limits);
        if let Some(auth) = &config.auth {
            builder = builder.auth(auth)?;
        }
        let supervisor = builder.build()?;
        // The console taps are bound as in `run`; the fixture pumps them.
        let console = supervisor.bind_console_for_tests(
            &services,
            &config,
            &liminal_server::server::ShutdownHandle::new(),
        )?;
        let listener = ServerListener::bind(&config, supervisor.clone())?;
        Ok(Self {
            address: listener.local_addr().to_string(),
            listener: Some(listener),
            supervisor,
            services,
            console: Some(console),
            directory,
            key,
        })
    }

    pub fn seat(
        &self,
        name: &str,
        subjects: Option<&[&str]>,
    ) -> Result<std::path::PathBuf, Box<dyn Error>> {
        let path = self.directory.path().join(format!("{name}.toml"));
        let pass = self.directory.path().join(format!("{name}.pass"));
        std::fs::write(&pass, self.pass(name)?)?;
        let mut value = json!({"server":self.address,"channel":"agents.alpha.inbox","token_file":pass,
            "limits":{"frame_bytes":131_072,"record_bytes":4_194_304,"reply_timeout_ms":5000,
                "flow_steps":8,"request_bytes":65_536}});
        if let Some(subjects) = subjects {
            value["listen"] = json!({"subjects":subjects});
        }
        let table: toml::Table = serde_json::from_value(value)?;
        std::fs::write(&path, toml::to_string(&table)?)?;
        Ok(path)
    }

    pub fn mismatch_seat(&self) -> Result<std::path::PathBuf, Box<dyn Error>> {
        let path = self.seat("mismatch", None)?;
        let text = std::fs::read_to_string(&path)?;
        let mut table: toml::Table = toml::from_str(&text)?;
        table.remove("token_file");
        table.insert(
            "lys".to_owned(),
            toml::Value::try_from(json!({"issuer":"https://agt.test","agent":"beta"}))?,
        );
        std::fs::write(&path, toml::to_string(&table)?)?;
        Ok(path)
    }

    pub fn pass(&self, holder: &str) -> Result<String, Box<dyn Error>> {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_secs();
        // A subject right names one subject exactly; a listen also needs its tap.
        let mut subjects = vec![
            "agents.alpha.inbox".to_owned(),
            "agents.beta.inbox".to_owned(),
            "messages.opaque".to_owned(),
            "messages.invalid".to_owned(),
        ];
        subjects.extend((0..17).map(|n| format!("messages.s{n}")));
        let taps: Vec<_> = subjects
            .iter()
            .map(|id| format!("console.tap.{id}"))
            .collect();
        subjects.extend(taps);
        let mut rights: Vec<_> = subjects
            .into_iter()
            .map(|id| Right {
                resource: lys_pass::rights::Resource {
                    kind: "liminal.subject".to_owned(),
                    id,
                },
                actions: vec!["publish".to_owned(), "subscribe".to_owned()],
                mode: Mode::Outright,
                grant: "agt-fixture".to_owned(),
            })
            .collect();
        rights.extend(["alpha", "beta"].into_iter().map(|name| Right {
            resource: lys_pass::rights::Resource {
                kind: "liminal.agent".into(),
                id: name.into(),
            },
            actions: vec!["message".into(), "watch".into()],
            mode: Mode::Outright,
            grant: "agt-message".into(),
        }));
        let claims = Claims {
            iss: "https://agt.test".to_owned(),
            sub: holder.to_owned(),
            aud: "liminal".to_owned(),
            iat: now.saturating_sub(60),
            exp: now.saturating_add(600),
            nbf: None,
            holder: Holder {
                id: holder.to_owned(),
                kind: "agent".to_owned(),
                responsible: Some("fixture-person".to_owned()),
            },
            agent_key_thumbprints: Vec::new(),
            rights,
            rights_truncated: None,
        };
        let header = encode(
            serde_json::to_string(&json!({"alg":"EdDSA","typ":"JWT","kid":"agt-fixture"}))?
                .as_bytes(),
        );
        let payload = encode(&serde_json::to_vec(&claims)?);
        let signing = format!("{header}.{payload}");
        let signature = self.key.sign(signing.as_bytes());
        Ok(format!("{signing}.{}", encode(&signature.to_bytes())))
    }

    pub fn finish(mut self) -> TestResult {
        let result = self
            .listener
            .take()
            .ok_or("the test listener is absent")?
            .shutdown();
        self.supervisor.shutdown();
        Ok(result?)
    }

    pub fn pump(&self) -> TestResult {
        self.services.console().pump(self.services.as_ref())?;
        Ok(())
    }
}

impl Drop for Exchange {
    fn drop(&mut self) {
        if let Some(listener) = self.listener.take() {
            if let Err(error) = listener.shutdown() {
                eprintln!("the owned test listener could not be stopped: {error}");
            }
            drop(self.console.take());
            self.supervisor.shutdown();
        }
    }
}

/// The `[participant]` section the server needs before it runs a publisher.
fn participant() -> Value {
    json!({"wire_frame_limit":65_536, "attach_receipt_ttl_ms":60_000,
        "receipt_provenance_ttl_ms":600_000, "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":131_072, "max_generated_marker_entries":1,
        "max_generated_marker_bytes":4096, "mandatory_transaction_bound_entries":4,
        "mandatory_transaction_bound_bytes":16_384, "full_recovery_claim_entries":4,
        "full_recovery_claim_bytes":16_384, "retained_capacity_entries":2048,
        "retained_capacity_bytes":16_777_216, "max_retained_record_rows":1024,
        "closure_episode_churn_limit":1024})
}

fn encode(bytes: &[u8]) -> String {
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes)
}

pub(super) fn run(body: impl FnOnce(&Exchange, &mut Host) -> TestResult) -> TestResult {
    run_on(Exchange::start()?, body)
}

pub(super) fn run_on(
    exchange: Exchange,
    body: impl FnOnce(&Exchange, &mut Host) -> TestResult,
) -> TestResult {
    let config = exchange.seat("alpha", None)?;
    let mut host = Host::start(&config, None)?;
    let outcome = body(&exchange, &mut host);
    let closed = host.close();
    let finished = exchange.finish();
    finish_results([outcome, child_result(closed), finished])
}

#[derive(Debug)]
struct AgentPlacement;
impl liminal_server::services::lifecycle::Placer for AgentPlacement {
    fn place(
        &self,
        _operation: &str,
        child: (&str, &str),
        parent: (&str, &str),
    ) -> Result<(), String> {
        if child.1.is_empty()
            || parent.1.is_empty()
            || !matches!(child.0, "liminal.agent" | "liminal.subject")
        {
            return Err("fixture placement refused".into());
        }
        Ok(())
    }
}
fn provision_agents(
    services: &liminal_server::server::connection::LiminalConnectionServices,
) -> TestResult {
    use liminal_server::agents::records::{AgentAct, AgentEvent, AgentHarness, AgentSpec};
    use liminal_server::server::connection::ConnectionServices;
    let agents = services.agents().ok_or("fixture agents missing")?;
    for name in ["alpha", "beta"] {
        let words = serde_json::to_string(&AgentAct {
            name: name.into(),
            holder: "fixture-person".into(),
            owner: "fixture-person".into(),
            at_ms: 1,
            change: AgentEvent::Create(AgentSpec {
                harness: AgentHarness::Cdx,
                purpose: "Exchange contract fixture.".into(),
                service: None,
            }),
        })?;
        let draft = lys_pass::drafts::ApprovedDraft {
            id: format!("create-{name}"),
            app: "liminal".into(),
            grant: "fixture-approved".into(),
            target: lys_pass::Target::new("liminal.agent", name, "create_agent")?,
            request_digest: lys_pass::drafts::request_digest(&words),
            approved_by: vec!["fixture-owner".into()],
            mode: lys_pass::Mode::ByDraft,
            words,
        };
        agents.execute_approved(&draft)?;
        services.agent_created(name)?;
    }
    Ok(())
}