use super::channel_test_support::{Exchange, TestResult, finish_results, require};
use serde_json::{Value, json};
use std::io::{BufRead, BufReader};
#[cfg(unix)]
#[test]
fn agt001_unlisten_and_eof_leave_no_ticket_channel_receiver() -> TestResult {
use std::os::unix::net::UnixStream;
use std::sync::atomic::Ordering;
let exchange = Exchange::start()?;
let path = exchange.seat("alpha", Some(&["agents.alpha.inbox", "messages.s0"]))?;
let config = crate::config::Config::load(&path)?;
let rail = crate::network::Live::new(&config)?;
let mut server = super::Server::configured(rail, &config)?;
let counts = std::sync::Arc::clone(&server.listens);
let (mut host, transport) = UnixStream::pair()?;
let mut host_reader = BufReader::new(host.try_clone()?);
let mut input = BufReader::new(transport.try_clone()?);
let mut output = transport;
let observed = std::thread::scope(|scope| {
let runner = scope.spawn(move || server.serve(&mut input, &mut output));
let outcome = (|| {
request(
&mut host,
&mut host_reader,
1,
"initialize",
&json!({"protocolVersion":"2025-11-25"}),
)?;
require(
counts.count()? == 2 && counts.receivers.load(Ordering::SeqCst) == 4,
"two subjects must own four ticket channel threads",
)?;
for (id, subject, subjects, receivers) in
[(2, "agents.alpha.inbox", 1, 2), (3, "messages.s0", 0, 0)]
{
let answer = request(
&mut host,
&mut host_reader,
id,
"tools/call",
&json!({"name":"unlisten","arguments":{"subject":subject}}),
)?;
require(
answer["result"]["isError"] == false,
"unlisten must succeed",
)?;
require(
counts.count()? == subjects
&& counts.receivers.load(Ordering::SeqCst) == receivers,
"the unlisten answer must follow both stream closes and receiver joins",
)?;
}
request(
&mut host,
&mut host_reader,
4,
"tools/call",
&json!({"name":"listen","arguments":{"subject":"messages.s0"}}),
)?;
request(&mut host, &mut host_reader, 5, "ping", &json!({}))?;
require(
counts.count()? == 1 && counts.receivers.load(Ordering::SeqCst) == 2,
"listen must restore exactly its two receiver threads",
)
})();
let closed: TestResult = host.shutdown(std::net::Shutdown::Both).map_err(Into::into);
let joined: TestResult = runner
.join()
.map_err(|_| "the real ticket host panicked".into())
.and_then(|result| result.map_err(Into::into));
let empty = require(
counts.count()? == 0 && counts.receivers.load(Ordering::SeqCst) == 0,
"host EOF must leave no listened subject or ticket receiver",
);
finish_results([outcome, closed, joined, empty])
});
finish_results([observed, exchange.finish()])
}
#[cfg(unix)]
fn request(
output: &mut std::os::unix::net::UnixStream,
input: &mut impl BufRead,
id: u64,
method: &str,
params: &Value,
) -> Result<Value, Box<dyn std::error::Error>> {
super::framing::write_message(
output,
&json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}),
)?;
let mut bytes = Vec::new();
require(
input.read_until(b'\n', &mut bytes)? > 0,
"the ticket closed before answering",
)?;
let answer: Value =
serde_json::from_slice(&bytes).map_err(|_| "the ticket answer is not JSON")?;
require(
answer["id"] == id,
"the ticket answered a different request",
)?;
Ok(answer)
}