use liminal_sdk::{PushClient, SubscriptionStream, lys_pass_credential};
use serde::Deserialize;
use serde_json::{Value, json};
use std::io::{BufRead as _, BufReader, Write as _};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::sync::mpsc::{self, Receiver, SyncSender};
use std::thread::JoinHandle;
use zeroize::Zeroizing;
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Fixture {
server: String,
channel: String,
issuer: String,
agent: String,
run_pass_file: PathBuf,
watch_pass_file: PathBuf,
decision_url: String,
pull_url: String,
revoke_count_url: String,
operator_pass_file: PathBuf,
content_bytes: usize,
event_bound: usize,
#[serde(default = "frame_bytes")]
frame_bytes: usize,
}
const fn frame_bytes() -> usize {
262_144
}
impl Fixture {
fn validate(&self) -> anyhow::Result<()> {
let address: std::net::SocketAddr = self.server.parse()?;
anyhow::ensure!(
address.ip().is_loopback(),
"agent fixture exchange must be loopback"
);
for url in [
&self.issuer,
&self.decision_url,
&self.pull_url,
&self.revoke_count_url,
] {
let uri: ureq::http::Uri = url.parse()?;
let authority = uri
.authority()
.ok_or_else(|| anyhow::anyhow!("agent fixture URL has no authority"))?;
anyhow::ensure!(
uri.scheme_str() == Some("http")
&& ["localhost", "127.0.0.1", "[::1]"].contains(&authority.host())
&& !authority.as_str().contains('@')
&& uri.query().is_none(),
"agent fixture URLs must be loopback http without credentials or queries"
);
}
for path in [
&self.run_pass_file,
&self.watch_pass_file,
&self.operator_pass_file,
] {
anyhow::ensure!(
path.is_absolute(),
"agent fixture pass path must be absolute"
);
}
anyhow::ensure!(
self.content_bytes > 0
&& self.event_bound > 0
&& self.frame_bytes >= self.content_bytes,
"agent fixture bounds refused"
);
Ok(())
}
}
pub enum Event {
Connected(UnixStream),
Control(Value),
Network(String, Value),
Exit(i32),
Failed,
}
pub struct Lab {
fixture: Fixture,
directory: Option<tempfile::TempDir>,
child: Child,
input: Option<std::process::ChildStdin>,
receiver: Receiver<Event>,
sender: SyncSender<Event>,
control: Option<UnixStream>,
closes: Vec<Arc<dyn Fn() -> Result<(), liminal_sdk::SdkError> + Send + Sync>>,
readers: Vec<JoinHandle<()>>,
publisher: PushClient,
reaped: bool,
expected_exit: Option<i32>,
observed_exit: Option<i32>,
}
impl Lab {
pub fn new(scenario: &str, extra: &[&str]) -> anyhow::Result<Self> {
let root = PathBuf::from(std::env::var_os("AGT005_FIXTURE_ROOT").ok_or_else(|| {
anyhow::anyhow!(
"AGT005_FIXTURE_ROOT must name the local real-exchange fixture directory"
)
})?);
anyhow::ensure!(root.is_absolute(), "AGT005_FIXTURE_ROOT must be absolute");
let text = Zeroizing::new(std::fs::read_to_string(
root.join(format!("{scenario}.toml")),
)?);
let fixture: Fixture = toml::from_str(&text)
.map_err(|_| anyhow::anyhow!("AGT005 fixture descriptor refused"))?;
fixture.validate()?;
let directory = tempfile::tempdir()?;
let control_path = directory.path().join("signals.sock");
let control_listener = UnixListener::bind(&control_path)?;
let path = directory.path().join("seat.toml");
let mut args = vec![
"--control".to_owned(),
control_path
.to_str()
.ok_or_else(|| anyhow::anyhow!("fixture path is not Unicode"))?
.to_owned(),
];
args.extend(["--frame-bytes".to_owned(), fixture.frame_bytes.to_string()]);
args.extend(extra.iter().map(|arg| (*arg).to_owned()));
let seat = json!({"server":fixture.server,"channel":fixture.channel,"agent":fixture.agent,
"lys":{"issuer":fixture.issuer},"limits":{"frame_bytes":fixture.frame_bytes,"record_bytes":fixture.frame_bytes,"reply_timeout_ms":5000,"flow_steps":16,"request_bytes":fixture.frame_bytes},
"harness":{"id":"codex","program":env!("CARGO_BIN_EXE_liminal-agent-acp-fake"),"args":args,"content_bytes":fixture.content_bytes,"acp_frame_bytes":fixture.frame_bytes}});
std::fs::write(&path, toml::to_string(&seat)?)?;
let watch_pass = Zeroizing::new(std::fs::read_to_string(&fixture.watch_pass_file)?);
let credential = Zeroizing::new(lys_pass_credential(watch_pass.trim()));
let publisher = PushClient::connect_with_auth(&fixture.server, &credential)?;
let (sender, receiver) = mpsc::sync_channel(fixture.event_bound);
let run_pass = Zeroizing::new(std::fs::read_to_string(&fixture.run_pass_file)?);
let child = Command::new(env!("CARGO_BIN_EXE_liminal"))
.args(["agent", "run", "--config"])
.arg(&path)
.env("LYS_AGENT", &fixture.agent)
.env("LYS_AGENT_PASS", run_pass.trim())
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
let mut lab = Self {
fixture,
directory: Some(directory),
child,
input: None,
receiver,
sender,
control: None,
closes: Vec::new(),
readers: Vec::new(),
publisher,
reaped: false,
expected_exit: None,
observed_exit: None,
};
lab.input = lab.child.stdin.take();
lab.start_observers(control_listener)?;
let connection = lab.until(|event| matches!(event, Event::Connected(_)))?;
if let Event::Connected(stream) = connection {
lab.control = Some(stream);
}
lab.until(
|event| matches!(event, Event::Control(value) if value["event"] == "initialize"),
)?;
Ok(lab)
}
fn start_observers(&mut self, control_listener: UnixListener) -> anyhow::Result<()> {
let pid = rustix::process::Pid::from_raw(i32::try_from(self.child.id())?)
.ok_or_else(|| anyhow::anyhow!("fixture host pid absent"))?;
let exit_sender = self.sender.clone();
self.readers.push(std::thread::spawn(move || {
let result = rustix::process::waitid(
rustix::process::WaitId::Pid(pid),
rustix::process::WaitIdOptions::EXITED | rustix::process::WaitIdOptions::NOWAIT,
);
let event = match result {
Ok(Some(status)) => status
.exit_status()
.or_else(|| status.terminating_signal().map(|signal| 128 + signal))
.map_or(Event::Failed, Event::Exit),
_ => Event::Failed,
};
let sent = exit_sender.send(event);
drop(sent);
}));
let control_sender = self.sender.clone();
let bound = self.fixture.frame_bytes;
self.readers.push(std::thread::spawn(move || {
let result: anyhow::Result<()> = (|| {
let (stream, _) = control_listener.accept()?;
control_sender.send(Event::Connected(stream.try_clone()?))?;
let mut reader = BufReader::new(stream);
while let Some(value) = read_line(&mut reader, bound)? {
if control_sender.send(Event::Control(value)).is_err() {
return Ok(());
}
}
Ok(())
})();
if result.is_err() {
let sent = control_sender.send(Event::Failed);
drop(sent);
}
}));
Ok(())
}
pub fn watch(&mut self, leaf: &str) -> anyhow::Result<()> {
let pass = Zeroizing::new(std::fs::read_to_string(&self.fixture.watch_pass_file)?);
let credential = Zeroizing::new(lys_pass_credential(pass.trim()));
let subject = format!("agents.{}.{leaf}", self.fixture.agent);
let stream = SubscriptionStream::open_with_auth(
&self.fixture.server,
&subject,
Vec::new(),
&credential,
)?;
self.closes.push(Arc::new(stream.close_handle()?));
let sender = self.sender.clone();
let bound = self.fixture.frame_bytes;
self.readers.push(std::thread::spawn(move || {
let result: anyhow::Result<()> = (|| {
loop {
let record = stream.recv()?;
anyhow::ensure!(
record.payload().len() <= bound,
"agent fixture network frame bound exceeded"
);
let value: Value = serde_json::from_slice(record.payload())?;
if sender.send(Event::Network(subject.clone(), value)).is_err() {
return Ok(());
}
}
})();
if result.is_err() {
let sent = sender.send(Event::Failed);
drop(sent);
}
}));
Ok(())
}
pub fn until(&mut self, predicate: impl Fn(&Event) -> bool) -> anyhow::Result<Event> {
loop {
let event = self.receiver.recv()?;
if predicate(&event) {
return Ok(event);
}
match event {
Event::Exit(status) if self.expected_exit == Some(status) => {
self.observed_exit = Some(status);
}
Event::Exit(status) => {
anyhow::bail!("host exited {status} before the required signal")
}
Event::Failed => anyhow::bail!("fixture reader failed before the required signal"),
_ => {}
}
}
}
pub const fn expect_exit(&mut self, status: i32) {
self.expected_exit = Some(status);
}
pub fn send(&self, leaf: &str, text: &str) -> anyhow::Result<()> {
self.send_value(leaf, &json!({"text":text}))
}
pub fn send_value(&self, leaf: &str, value: &Value) -> anyhow::Result<()> {
self.publisher.publish(
&format!("agents.{}.{leaf}", self.fixture.agent),
serde_json::to_vec(value)?,
)?;
anyhow::ensure!(
self.publisher.flush()?.is_proven_accepted(),
"fixture message was refused or unresolved"
);
Ok(())
}
pub fn command(&mut self, command: &str) -> anyhow::Result<()> {
let stream = self
.control
.as_mut()
.ok_or_else(|| anyhow::anyhow!("fake control stream absent"))?;
stream.write_all(serde_json::to_string(&json!({"command":command}))?.as_bytes())?;
stream.write_all(b"\n")?;
stream.flush()?;
Ok(())
}
pub fn decide_and_pull(&self, decision: &str, count: usize) -> anyhow::Result<()> {
let mut decision_response = ureq::post(&self.fixture.decision_url)
.header("content-type", "application/json")
.send(serde_json::to_vec(
&json!({"agent":self.fixture.agent,"decision":decision,"count":count}),
)?)?;
let result: Value = serde_json::from_slice(
&decision_response
.body_mut()
.with_config()
.limit(u64::try_from(self.fixture.frame_bytes)?)
.read_to_vec()?,
)?;
anyhow::ensure!(
result["decided"] == json!(count),
"fixture issuer did not decide every requested draft"
);
let pass = Zeroizing::new(std::fs::read_to_string(&self.fixture.operator_pass_file)?);
let mut header =
ureq::http::HeaderValue::from_bytes(format!("Bearer {}", pass.trim()).as_bytes())?;
header.set_sensitive(true);
let response = ureq::post(&self.fixture.pull_url)
.header("authorization", header)
.header("content-type", "application/json")
.send(b"{}".as_slice())?;
anyhow::ensure!(
response.status().is_success(),
"real exchange did not accept the explicit draft pull"
);
Ok(())
}
pub fn terminate(&self) -> anyhow::Result<()> {
let pid = rustix::process::Pid::from_raw(i32::try_from(self.child.id())?)
.ok_or_else(|| anyhow::anyhow!("fixture host pid absent"))?;
rustix::process::kill_process(pid, rustix::process::Signal::TERM)?;
Ok(())
}
pub fn revoke_count(&self) -> anyhow::Result<usize> {
let mut response = ureq::get(&self.fixture.revoke_count_url).call()?;
let answer: Value = serde_json::from_slice(
&response
.body_mut()
.with_config()
.limit(u64::try_from(self.fixture.frame_bytes)?)
.read_to_vec()?,
)?;
Ok(usize::try_from(answer["revokes"].as_u64().ok_or_else(
|| anyhow::anyhow!("fixture issuer supplied no revoke count"),
)?)?)
}
pub fn finish(mut self, expected: i32) -> anyhow::Result<usize> {
if self.observed_exit != Some(expected) {
self.until(|event| matches!(event, Event::Exit(status) if *status == expected))?;
}
let status = self.child.wait()?;
self.reaped = true;
anyhow::ensure!(
status.code() == Some(expected),
"host exit did not match the expected status"
);
self.revoke_count()
}
}
fn read_line(reader: &mut BufReader<UnixStream>, bound: usize) -> anyhow::Result<Option<Value>> {
use std::io::Read as _;
let mut line = Vec::new();
if reader
.take(u64::try_from(bound)?.saturating_add(1))
.read_until(b'\n', &mut line)?
== 0
{
return Ok(None);
}
anyhow::ensure!(
line.len() <= bound && line.last() == Some(&b'\n'),
"agent fixture control frame bound or newline refused"
);
Ok(Some(serde_json::from_slice(&line)?))
}
impl Drop for Lab {
fn drop(&mut self) {
let (sender, replacement) = mpsc::channel();
drop(sender);
drop(std::mem::replace(&mut self.receiver, replacement));
self.input.take();
self.control.take();
for close in &self.closes {
if close().is_err() {
eprintln!("agent fixture subscription close failed");
}
}
if !self.reaped {
if let Ok(pid) = i32::try_from(self.child.id()) {
if let Some(pid) = rustix::process::Pid::from_raw(pid) {
if rustix::process::kill_process(pid, rustix::process::Signal::TERM).is_err() {
eprintln!("agent fixture host termination failed");
}
}
}
if self.child.wait().is_err() {
eprintln!("agent fixture host reap failed");
}
}
if let Some(directory) = &self.directory {
match UnixStream::connect(directory.path().join("signals.sock")) {
Ok(stream) => drop(stream),
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused
) => {}
Err(_) => eprintln!("agent fixture accept cancellation failed"),
}
}
for reader in self.readers.drain(..) {
if reader.join().is_err() {
eprintln!("agent fixture reader panicked");
}
}
if let Some(directory) = self.directory.take() {
if directory.close().is_err() {
eprintln!("agent fixture temporary directory cleanup failed");
}
}
}
}