use super::wire::{Frame, Pipe};
use serde_json::json;
use tokio::io::AsyncWriteExt as _;
#[tokio::test]
async fn r1_real_fake_drains_forty_requests_through_a_1024_byte_transport() -> anyhow::Result<()> {
use std::process::Stdio;
let directory = tempfile::tempdir()?;
let socket = directory.path().join("control.sock");
let listener = tokio::net::UnixListener::bind(&socket)?;
let executable = std::env::current_exe()?
.parent()
.and_then(std::path::Path::parent)
.ok_or_else(|| anyhow::anyhow!("test executable root absent"))?
.join("assistant-fake");
let mut child = tokio::process::Command::new(executable)
.args(["--burst", "40", "--pipe-burst", "--control"])
.arg(&socket)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.kill_on_drop(true)
.spawn()?;
let mut control = listener.accept().await?.0;
let control_drain =
tokio::spawn(async move { tokio::io::copy(&mut control, &mut tokio::io::sink()).await });
let mut child_input = child
.stdin
.take()
.ok_or_else(|| anyhow::anyhow!("child stdin absent"))?;
let mut child_output = child
.stdout
.take()
.ok_or_else(|| anyhow::anyhow!("child stdout absent"))?;
let (host, transport) = tokio::io::duplex(1024);
let (read, write) = tokio::io::split(host);
let mut pipe = Pipe::start(read, write, 65536, 128, 1);
let (mut requests, mut responses) = tokio::io::split(transport);
let to_child = tokio::spawn(async move {
tokio::io::copy(&mut requests, &mut child_input).await?;
child_input.shutdown().await
});
let from_child = tokio::spawn(async move {
tokio::io::copy(&mut child_output, &mut responses).await?;
responses.shutdown().await
});
pipe.output
.request(1, "initialize", json!({"protocolVersion":2}))?;
let mut received = 0;
loop {
let frame = pipe
.input
.recv()
.await
.ok_or_else(|| anyhow::anyhow!("reader closed"))??
.ok_or_else(|| anyhow::anyhow!("fake stdout closed"))?;
match frame {
Frame::Response { id, result } => {
anyhow::ensure!(result.is_ok(), "fake refused request");
if id == 1 {
pipe.output.request(
2,
"session/new",
json!({"cwd":directory.path(),"mcpServers":[]}),
)?;
} else if id == 2 {
pipe.output.request(3, "session/prompt", json!({"sessionId":"fake-session","prompt":[{"type":"text","text":"burst"}]}))?;
} else {
assert_eq!(id, 3);
break;
}
}
Frame::Request { id, method, .. } => {
assert_eq!(method, "session/request_permission");
received += 1;
pipe.output.response(
id,
json!({"outcome":{"outcome":"cancelled"},"padding":"x".repeat(8192)}),
)?;
}
Frame::Notification { .. } => {}
}
}
assert_eq!(
received, 40,
"--burst did not run against the child process"
);
drop(pipe.output);
pipe.writer.await??;
to_child.await??;
assert!(child.wait().await?.success());
from_child.await??;
pipe.reader.await?;
control_drain.await??;
Ok(())
}