use std::{
io::{Read, Write},
net::TcpListener,
process::Stdio,
thread,
};
use scv_protocol::{ClientMessage, PROTOCOL_VERSION, PeerInfo, ServerEvent};
use tokio::{
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
process::Command,
time::{Duration, timeout},
};
#[tokio::test]
async fn server_handshake_and_session_start() {
let workspace = tempfile::tempdir().unwrap();
let config_home = tempfile::tempdir().unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_scv-server"))
.arg("--stdio")
.env("OPENAI_API_KEY", "test-only")
.env("XDG_CONFIG_HOME", config_home.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut input = child.stdin.take().unwrap();
let output = child.stdout.take().unwrap();
input
.write_all(
format!(
concat!(
"{{\"type\":\"initialize\",\"request_id\":\"1\",\"protocol_version\":{},\"client\":{{\"name\":\"test\",\"version\":\"0\"}}}}\n",
"{{\"type\":\"session.start\",\"request_id\":\"2\",\"cwd\":{:?}}}\n"
),
PROTOCOL_VERSION,
workspace.path().display().to_string()
)
.as_bytes(),
)
.await
.unwrap();
input.shutdown().await.unwrap();
drop(input);
let mut lines = BufReader::new(output).lines();
let first = timeout(Duration::from_secs(3), lines.next_line())
.await
.unwrap()
.unwrap()
.unwrap();
let second = timeout(Duration::from_secs(3), lines.next_line())
.await
.unwrap()
.unwrap()
.unwrap();
assert!(matches!(
serde_json::from_str::<ServerEvent>(&first).unwrap(),
ServerEvent::Initialized { .. }
));
assert!(matches!(
serde_json::from_str::<ServerEvent>(&second).unwrap(),
ServerEvent::SessionStarted { .. }
));
assert!(
timeout(Duration::from_secs(3), child.wait())
.await
.unwrap()
.unwrap()
.success()
);
}
#[tokio::test]
async fn server_completes_a_streamed_turn_with_a_fake_provider() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let provider = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut request = [0u8; 64 * 1024];
let read = stream.read(&mut request).unwrap();
let request = String::from_utf8_lossy(&request[..read]);
assert!(request.starts_with("POST /v1/responses HTTP/1.1"));
let body = concat!(
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello \"}\n\n",
"event: response.output_text.delta\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"from fake\"}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":3}}}\n\n"
);
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
});
let workspace = tempfile::tempdir().unwrap();
let config_home = tempfile::tempdir().unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_scv-server"))
.args([
"--stdio",
"--model",
"fake-model",
"--base-url",
&format!("http://{address}/v1"),
])
.env("OPENAI_API_KEY", "test-only")
.env("XDG_CONFIG_HOME", config_home.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut input = child.stdin.take().unwrap();
let output = child.stdout.take().unwrap();
let mut lines = BufReader::new(output).lines();
send(
&mut input,
&ClientMessage::Initialize {
request_id: "init".into(),
protocol_version: PROTOCOL_VERSION,
client: PeerInfo {
name: "integration-test".into(),
version: "0".into(),
},
},
)
.await;
send(
&mut input,
&ClientMessage::SessionStart {
request_id: "session".into(),
cwd: workspace.path().display().to_string(),
provider: None,
model: None,
base_url: None,
no_tools: None,
},
)
.await;
let initialized = next_event(&mut lines).await;
assert!(matches!(initialized, ServerEvent::Initialized { .. }));
let session_id = match next_event(&mut lines).await {
ServerEvent::SessionStarted { session_id, .. } => session_id,
event => panic!("expected session.started, received {event:?}"),
};
send(
&mut input,
&ClientMessage::TurnStart {
request_id: "turn".into(),
session_id,
prompt: "say hello".into(),
},
)
.await;
let mut streamed = String::new();
let _usage = loop {
match next_event(&mut lines).await {
ServerEvent::AssistantDelta { content, .. } => streamed.push_str(&content),
ServerEvent::TurnCompleted { usage, .. } => break usage,
ServerEvent::TurnFailed { code, message, .. } => {
panic!("turn failed with {code}: {message}")
}
_ => {}
}
};
assert_eq!(streamed, "hello from fake");
input.shutdown().await.unwrap();
drop(input);
provider.join().unwrap();
assert!(
timeout(Duration::from_secs(3), child.wait())
.await
.unwrap()
.unwrap()
.success()
);
}
#[tokio::test]
async fn a_provider_stream_error_fails_the_turn_instead_of_completing_empty() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let provider = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
read_json_body(&mut stream);
let body = concat!(
"event: error\n",
"data: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"model is not available\"}}\n\n"
);
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
});
let workspace = tempfile::tempdir().unwrap();
let config_home = tempfile::tempdir().unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_scv-server"))
.args([
"--stdio",
"--model",
"fake-model",
"--base-url",
&format!("http://{address}/v1"),
])
.env("OPENAI_API_KEY", "test-only")
.env("XDG_CONFIG_HOME", config_home.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut input = child.stdin.take().unwrap();
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
send(
&mut input,
&ClientMessage::Initialize {
request_id: "init".into(),
protocol_version: PROTOCOL_VERSION,
client: PeerInfo {
name: "integration-test".into(),
version: "0".into(),
},
},
)
.await;
send(
&mut input,
&ClientMessage::SessionStart {
request_id: "session".into(),
cwd: workspace.path().display().to_string(),
provider: None,
model: None,
base_url: None,
no_tools: None,
},
)
.await;
assert!(matches!(
next_event(&mut lines).await,
ServerEvent::Initialized { .. }
));
let session_id = match next_event(&mut lines).await {
ServerEvent::SessionStarted { session_id, .. } => session_id,
event => panic!("expected session.started, received {event:?}"),
};
send(
&mut input,
&ClientMessage::TurnStart {
request_id: "turn".into(),
session_id,
prompt: "say hello".into(),
},
)
.await;
let (code, message) = loop {
match next_event(&mut lines).await {
ServerEvent::TurnFailed { code, message, .. } => break (code, message),
ServerEvent::TurnCompleted { .. } => panic!("a provider error completed the turn"),
ServerEvent::AssistantCompleted { content, .. } => {
panic!("a provider error produced an assistant message {content:?}")
}
_ => {}
}
};
assert_eq!(code, "provider_error");
assert!(message.contains("model is not available"), "{message}");
input.shutdown().await.unwrap();
drop(input);
provider.join().unwrap();
assert!(
timeout(Duration::from_secs(3), child.wait())
.await
.unwrap()
.unwrap()
.success()
);
}
#[tokio::test]
async fn tool_results_are_replayed_after_their_calls() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let provider = thread::spawn(move || {
let responses = [
concat!(
"data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\\\"README.md\\\"}\"}\n\n",
"data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"read\"}}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{}}\n\n"
),
concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"read it\"}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{}}\n\n"
),
];
let mut bodies = Vec::new();
for body in responses {
let (mut stream, _) = listener.accept().unwrap();
bodies.push(read_json_body(&mut stream));
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).unwrap();
}
bodies
});
let workspace = tempfile::tempdir().unwrap();
std::fs::write(workspace.path().join("README.md"), "fixture text").unwrap();
let config_home = tempfile::tempdir().unwrap();
let mut child = Command::new(env!("CARGO_BIN_EXE_scv-server"))
.args([
"--stdio",
"--model",
"fake-model",
"--base-url",
&format!("http://{address}/v1"),
])
.env("OPENAI_API_KEY", "test-only")
.env("XDG_CONFIG_HOME", config_home.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.unwrap();
let mut input = child.stdin.take().unwrap();
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
send(
&mut input,
&ClientMessage::Initialize {
request_id: "init".into(),
protocol_version: PROTOCOL_VERSION,
client: PeerInfo {
name: "integration-test".into(),
version: "0".into(),
},
},
)
.await;
send(
&mut input,
&ClientMessage::SessionStart {
request_id: "session".into(),
cwd: workspace.path().display().to_string(),
provider: None,
model: None,
base_url: None,
no_tools: None,
},
)
.await;
assert!(matches!(
next_event(&mut lines).await,
ServerEvent::Initialized { .. }
));
let session_id = match next_event(&mut lines).await {
ServerEvent::SessionStarted { session_id, .. } => session_id,
event => panic!("expected session.started, received {event:?}"),
};
send(
&mut input,
&ClientMessage::TurnStart {
request_id: "turn".into(),
session_id,
prompt: "read the README".into(),
},
)
.await;
loop {
match next_event(&mut lines).await {
ServerEvent::TurnCompleted { .. } => break,
ServerEvent::TurnFailed { code, message, .. } => {
panic!("turn failed with {code}: {message}")
}
_ => {}
}
}
input.shutdown().await.unwrap();
drop(input);
let bodies = provider.join().unwrap();
let replayed = bodies[1]["input"].as_array().unwrap();
assert_eq!(replayed.len(), 3, "{replayed:?}");
assert_eq!(
replayed[1],
serde_json::json!({
"type":"function_call",
"call_id":"call_1",
"name":"read",
"arguments":"{\"path\":\"README.md\"}"
})
);
assert_eq!(replayed[2]["type"], "function_call_output");
assert_eq!(replayed[2]["call_id"], "call_1");
assert!(
replayed[2]["output"]
.as_str()
.unwrap()
.contains("fixture text")
);
assert!(
timeout(Duration::from_secs(3), child.wait())
.await
.unwrap()
.unwrap()
.success()
);
}
fn read_json_body(stream: &mut std::net::TcpStream) -> serde_json::Value {
let mut request = Vec::new();
let mut byte = [0u8; 1];
while !request.ends_with(b"\r\n\r\n") {
stream.read_exact(&mut byte).unwrap();
request.push(byte[0]);
}
let head = String::from_utf8(request).unwrap();
let length: usize = head
.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
name.eq_ignore_ascii_case("content-length")
.then(|| value.trim().parse().unwrap())
})
.expect("request has a Content-Length");
let mut body = vec![0u8; length];
stream.read_exact(&mut body).unwrap();
serde_json::from_slice(&body).unwrap()
}
async fn send(input: &mut tokio::process::ChildStdin, message: &ClientMessage) {
input
.write_all(format!("{}\n", serde_json::to_string(message).unwrap()).as_bytes())
.await
.unwrap();
input.flush().await.unwrap();
}
async fn next_event(
lines: &mut tokio::io::Lines<BufReader<tokio::process::ChildStdout>>,
) -> ServerEvent {
let line = timeout(Duration::from_secs(3), lines.next_line())
.await
.unwrap()
.unwrap()
.expect("server closed stdout before the expected event");
serde_json::from_str(&line).unwrap()
}