use std::sync::{Arc, Mutex};
use std::time::Duration;
use super::{run_frontend_http, run_frontend_websocket_runtime, RuntimeHttpCredential};
use crate::acp_server::{self, AcpServer};
use crate::{
FrontendAttachSnapshot, FrontendAttachment, FrontendEvent, FrontendOperationInvocation,
FrontendOperationResult, FrontendResponse, FrontendRuntime, FrontendRuntimeDescriptor,
FrontendRuntimeError, RuntimeLeaseSnapshot, RuntimeSubmitError,
};
use async_trait::async_trait;
use futures::{SinkExt, StreamExt};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::broadcast;
fn fixture() -> Value {
serde_json::from_str(include_str!(
"../../../../sdk/frontend/test/fixtures/conformance.json"
))
.unwrap()
}
fn fixture_snapshot() -> FrontendAttachSnapshot {
let fixture = fixture();
FrontendAttachSnapshot {
descriptor: serde_json::from_value(fixture["descriptor"].clone()).unwrap(),
history: serde_json::from_value(fixture["history"]["messages"].clone()).unwrap(),
history_cursor: fixture["history"]["cursor"].as_u64().unwrap(),
replay: serde_json::from_value(fixture["replay"].clone()).unwrap(),
}
}
#[derive(Debug, PartialEq)]
struct BindingObservation {
descriptor: Value,
history: Value,
history_cursor: u64,
replay: Value,
request: Value,
live: Value,
ordered_sdk_calls: Value,
named_errors: Value,
}
fn expected_observation() -> BindingObservation {
let fixture = fixture();
BindingObservation {
descriptor: fixture["descriptor"].clone(),
history: fixture["history"]["messages"].clone(),
history_cursor: fixture["history"]["cursor"].as_u64().unwrap(),
replay: fixture["replay"].clone(),
request: fixture["request"].clone(),
live: fixture["live"].clone(),
ordered_sdk_calls: fixture["ordered_sdk_calls"].clone(),
named_errors: fixture["errors"]["observed"].clone(),
}
}
fn observation(
descriptor: Value,
history: Value,
history_cursor: u64,
replay: Value,
live: Value,
ordered_sdk_calls: Value,
named_errors: Value,
) -> BindingObservation {
let request = replay
.as_array()
.and_then(|events| events.first())
.and_then(|event| event.pointer("/payload/request"))
.cloned()
.expect("fixture replay carries the typed request");
BindingObservation {
descriptor,
history,
history_cursor,
replay,
request,
live,
ordered_sdk_calls,
named_errors,
}
}
struct FixtureRuntime {
snapshot: FrontendAttachSnapshot,
events: broadcast::Sender<FrontendEvent>,
calls: Mutex<Vec<Value>>,
}
impl FixtureRuntime {
fn new() -> Arc<Self> {
let (events, _) = broadcast::channel(64);
Arc::new(Self {
snapshot: fixture_snapshot(),
events,
calls: Mutex::new(Vec::new()),
})
}
fn event_sender(&self) -> broadcast::Sender<FrontendEvent> {
self.events.clone()
}
fn emit_live(&self) {
let fixture = fixture();
for event in serde_json::from_value::<Vec<FrontendEvent>>(fixture["live"].clone()).unwrap()
{
let _ = self.events.send(event);
}
}
fn record(&self, method: &str, params: Value) {
self.calls
.lock()
.unwrap()
.push(json!({"method": method, "params": params}));
}
fn observed_calls(&self) -> Value {
Value::Array(self.calls.lock().unwrap().clone())
}
}
fn response_request_id(response: &FrontendResponse) -> u64 {
match response {
FrontendResponse::Approval { request_id, .. }
| FrontendResponse::Elicitation { request_id, .. }
| FrontendResponse::Other { request_id, .. } => *request_id,
}
}
fn lease() -> RuntimeLeaseSnapshot {
serde_json::from_value(json!({
"controller": null,
"observers": [],
"lease_ttl_ms": 30_000
}))
.unwrap()
}
#[async_trait]
impl FrontendRuntime for FixtureRuntime {
async fn describe(&self) -> Result<FrontendRuntimeDescriptor, FrontendRuntimeError> {
Ok(self.snapshot.descriptor.clone())
}
async fn attach(
&self,
_history_limit: usize,
) -> Result<FrontendAttachment, FrontendRuntimeError> {
Ok(FrontendAttachment::from_snapshot(
self.snapshot.clone(),
self.events.subscribe(),
))
}
async fn send_input(self: Arc<Self>, prompt: String) -> Result<(), FrontendRuntimeError> {
let fixture = fixture();
if prompt == fixture["competing_prompt"] {
return Err(RuntimeSubmitError::Busy.into());
}
assert_eq!(prompt, fixture["prompt"]);
self.record("frontend.v2.send_input", json!({"prompt": prompt}));
self.emit_live();
Ok(())
}
async fn submit(&self, prompt: String) -> Result<String, FrontendRuntimeError> {
if prompt == fixture()["competing_prompt"] {
return Err(RuntimeSubmitError::Busy.into());
}
self.emit_live();
Ok(fixture()["reply"].as_str().unwrap().to_string())
}
async fn interrupt(&self) -> Result<bool, FrontendRuntimeError> {
Ok(false)
}
async fn steer(&self, _prompt: String) -> Result<(), FrontendRuntimeError> {
Ok(())
}
async fn respond(&self, response: FrontendResponse) -> Result<(), FrontendRuntimeError> {
if response_request_id(&response) == fixture()["request"]["id"].as_u64().unwrap() {
self.record(
"frontend.v2.respond",
json!({"response": serde_json::to_value(&response).unwrap()}),
);
Ok(())
} else {
Err(FrontendRuntimeError::UnsupportedAction("respond"))
}
}
async fn invoke(
&self,
_operation: FrontendOperationInvocation,
) -> Result<FrontendOperationResult, FrontendRuntimeError> {
Err(FrontendRuntimeError::UnsupportedAction("invoke"))
}
async fn lease_snapshot(&self) -> Result<RuntimeLeaseSnapshot, FrontendRuntimeError> {
Ok(lease())
}
async fn take_control(&self) -> Result<RuntimeLeaseSnapshot, FrontendRuntimeError> {
Ok(lease())
}
async fn heartbeat(&self) -> Result<RuntimeLeaseSnapshot, FrontendRuntimeError> {
Ok(lease())
}
async fn detach(&self) -> Result<RuntimeLeaseSnapshot, FrontendRuntimeError> {
Ok(lease())
}
}
async fn local_observation() -> BindingObservation {
let fixture = fixture();
let runtime = FixtureRuntime::new();
let descriptor = serde_json::to_value(runtime.describe().await.unwrap()).unwrap();
let mut attachment = runtime.attach(1000).await.unwrap();
let history = serde_json::to_value(&attachment.history).unwrap();
let history_cursor = attachment.history_cursor;
let mut replay = Vec::new();
while let Some(event) = attachment.next_replay_event() {
replay.push(event);
}
runtime
.clone()
.send_input(fixture["prompt"].as_str().unwrap().into())
.await
.unwrap();
let mut live = Vec::new();
while live.len() < fixture["live"].as_array().unwrap().len() {
live.push(attachment.next_event().await.unwrap());
}
let response: FrontendResponse =
serde_json::from_value(fixture["semantic_actions"][3]["params"]["response"].clone())
.unwrap();
runtime.respond(response).await.unwrap();
runtime.detach().await.unwrap();
let ordered_sdk_calls = runtime.observed_calls();
let busy = runtime
.clone()
.send_input(fixture["competing_prompt"].as_str().unwrap().into())
.await
.unwrap_err();
let unsupported: FrontendResponse =
serde_json::from_value(fixture["unsupported_response"].clone()).unwrap();
let unsupported = runtime.respond(unsupported).await.unwrap_err();
observation(
descriptor,
history,
history_cursor,
serde_json::to_value(replay).unwrap(),
serde_json::to_value(live).unwrap(),
ordered_sdk_calls,
json!([busy.code(), unsupported.code()]),
)
}
fn headers(
request: reqwest::RequestBuilder,
token: &str,
client_id: &str,
) -> reqwest::RequestBuilder {
request
.bearer_auth(token)
.header("x-supercode-client-id", client_id)
.header(
"x-supercode-permissions",
"observe,interact,approve,terminate",
)
}
async fn read_sse_events(response: reqwest::Response, count: usize) -> Vec<FrontendEvent> {
let mut stream = response.bytes_stream();
let mut pending = String::new();
let mut events = Vec::new();
while events.len() < count {
let chunk = tokio::time::timeout(Duration::from_secs(3), stream.next())
.await
.expect("fixture SSE timed out")
.expect("fixture SSE closed")
.unwrap();
pending.push_str(&String::from_utf8_lossy(&chunk));
while let Some(boundary) = pending.find('\n') {
let line = pending[..boundary].trim_end_matches('\r').to_string();
pending.drain(..=boundary);
if let Some(data) = line.strip_prefix("data: ") {
events.push(serde_json::from_str(data).unwrap());
}
}
}
events
}
async fn http_observation() -> BindingObservation {
let fixture = fixture();
let runtime = FixtureRuntime::new();
let token = "fixture-http-token";
let server = run_frontend_http(
runtime.clone(),
runtime.event_sender(),
"127.0.0.1:0",
Arc::from(token),
)
.await
.unwrap();
let base = format!("http://{}", server.address());
let client = reqwest::Client::new();
let stream = headers(
client.get(format!("{base}/frontend/events")),
token,
"fixture-http",
)
.send()
.await
.unwrap();
let mut descriptor = Value::Null;
let mut attachment = Value::Null;
for (id, action) in fixture["semantic_actions"]
.as_array()
.unwrap()
.iter()
.enumerate()
{
let envelope: Value = headers(client.post(format!("{base}/rpc")), token, "fixture-http")
.json(&json!({
"jsonrpc":"2.0", "id":id + 1,
"method":action["method"], "params":action["params"]
}))
.send()
.await
.unwrap()
.json()
.await
.unwrap();
assert!(envelope.get("error").is_none(), "{envelope}");
match action["method"].as_str().unwrap() {
"frontend.v2.describe" => descriptor = envelope["result"].clone(),
"frontend.v2.attach" => attachment = envelope["result"].clone(),
_ => {}
}
}
let live = read_sse_events(stream, fixture["live"].as_array().unwrap().len()).await;
let ordered_sdk_calls = runtime.observed_calls();
let errors = [
json!({"method":"frontend.v2.send_input","params":{"prompt":fixture["competing_prompt"]}}),
json!({"method":"frontend.v2.respond","params":{"response":fixture["unsupported_response"]}}),
];
let mut names = Vec::new();
for (index, error) in errors.iter().enumerate() {
let envelope: Value = headers(client.post(format!("{base}/rpc")), token, "fixture-http")
.json(&json!({"jsonrpc":"2.0","id":100 + index,"method":error["method"],"params":error["params"]}))
.send()
.await
.unwrap()
.json()
.await
.unwrap();
names.push(envelope["error"]["name"].clone());
}
observation(
descriptor,
attachment["history"].clone(),
attachment["history_cursor"].as_u64().unwrap(),
attachment["replay"].clone(),
serde_json::to_value(live).unwrap(),
ordered_sdk_calls,
Value::Array(names),
)
}
async fn next_ws_json<S>(socket: &mut S) -> Value
where
S: futures::Stream<
Item = Result<
tokio_tungstenite::tungstenite::Message,
tokio_tungstenite::tungstenite::Error,
>,
> + Unpin,
{
let message = tokio::time::timeout(Duration::from_secs(3), socket.next())
.await
.expect("fixture WebSocket timed out")
.expect("fixture WebSocket closed")
.unwrap();
serde_json::from_str(message.to_text().unwrap()).unwrap()
}
async fn ws_request<S>(
socket: &mut S,
id: u64,
method: &str,
params: Value,
) -> (Value, Vec<FrontendEvent>)
where
S: futures::Stream<
Item = Result<
tokio_tungstenite::tungstenite::Message,
tokio_tungstenite::tungstenite::Error,
>,
> + futures::Sink<
tokio_tungstenite::tungstenite::Message,
Error = tokio_tungstenite::tungstenite::Error,
> + Unpin,
{
socket
.send(tokio_tungstenite::tungstenite::Message::Text(
json!({"jsonrpc":"2.0","id":id,"method":method,"params":params})
.to_string()
.into(),
))
.await
.unwrap();
let mut events = Vec::new();
loop {
let message = next_ws_json(socket).await;
if message["method"] == "frontend.v2.event" {
events.push(serde_json::from_value(message["params"]["event"].clone()).unwrap());
} else if message["id"] == id {
return (message, events);
}
}
}
async fn websocket_observation() -> BindingObservation {
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
let fixture = fixture();
let runtime = FixtureRuntime::new();
let server = run_frontend_websocket_runtime(
runtime.clone(),
runtime.event_sender(),
"127.0.0.1:0",
vec![RuntimeHttpCredential::owner("fixture-ws-token")],
)
.await
.unwrap();
let mut request = format!("ws://{}/frontend/v2", server.address())
.into_client_request()
.unwrap();
request
.headers_mut()
.insert("authorization", "Bearer fixture-ws-token".parse().unwrap());
request
.headers_mut()
.insert("x-supercode-client-id", "fixture-ws".parse().unwrap());
request.headers_mut().insert(
"x-supercode-permissions",
"observe,interact,approve,terminate".parse().unwrap(),
);
let (mut socket, _) = tokio_tungstenite::connect_async(request).await.unwrap();
let mut descriptor = Value::Null;
let mut attachment = Value::Null;
let mut live = Vec::new();
for (index, action) in fixture["semantic_actions"]
.as_array()
.unwrap()
.iter()
.enumerate()
{
if action["method"] == "frontend.v2.detach" {
continue;
}
let (envelope, emitted) = ws_request(
&mut socket,
(index + 1) as u64,
action["method"].as_str().unwrap(),
action["params"].clone(),
)
.await;
live.extend(emitted);
assert!(envelope.get("error").is_none(), "{envelope}");
match action["method"].as_str().unwrap() {
"frontend.v2.describe" => descriptor = envelope["result"].clone(),
"frontend.v2.attach" => attachment = envelope["result"].clone(),
_ => {}
}
}
while live.len() < fixture["live"].as_array().unwrap().len() {
let message = next_ws_json(&mut socket).await;
if message["method"] == "frontend.v2.event" {
live.push(serde_json::from_value(message["params"]["event"].clone()).unwrap());
}
}
let errors = [
(
"frontend.v2.send_input",
json!({"prompt":fixture["competing_prompt"]}),
),
(
"frontend.v2.respond",
json!({"response":fixture["unsupported_response"]}),
),
];
let mut names = Vec::new();
for (index, (method, params)) in errors.into_iter().enumerate() {
let (envelope, _) = ws_request(&mut socket, 100 + index as u64, method, params).await;
names.push(envelope["error"]["name"].clone());
}
let detach = fixture["semantic_actions"]
.as_array()
.unwrap()
.iter()
.find(|action| action["method"] == "frontend.v2.detach")
.unwrap();
let (detached, _) = ws_request(
&mut socket,
200,
"frontend.v2.detach",
detach["params"].clone(),
)
.await;
assert!(detached.get("error").is_none(), "{detached}");
let ordered_sdk_calls = runtime.observed_calls();
observation(
descriptor,
attachment["history"].clone(),
attachment["history_cursor"].as_u64().unwrap(),
attachment["replay"].clone(),
serde_json::to_value(live).unwrap(),
ordered_sdk_calls,
Value::Array(names),
)
}
async fn write_acp(writer: &mut (impl AsyncWriteExt + Unpin), value: Value) {
writer
.write_all(format!("{value}\n").as_bytes())
.await
.unwrap();
writer.flush().await.unwrap();
}
async fn read_acp(reader: &mut (impl tokio::io::AsyncBufRead + Unpin)) -> Value {
let mut line = String::new();
reader.read_line(&mut line).await.unwrap();
serde_json::from_str(&line).unwrap()
}
async fn acp_request(
writer: &mut (impl AsyncWriteExt + Unpin),
reader: &mut (impl tokio::io::AsyncBufRead + Unpin),
id: u64,
method: &str,
mut params: Value,
) -> (Value, Vec<FrontendEvent>) {
params["sessionId"] = json!(fixture()["session_id"]);
write_acp(
writer,
json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}),
)
.await;
let mut events = Vec::new();
loop {
let message = read_acp(reader).await;
if message["method"] == "frontend.v2.event" {
events.push(serde_json::from_value(message["params"]["event"].clone()).unwrap());
} else if message["id"] == id {
return (message, events);
}
}
}
async fn acp_observation() -> BindingObservation {
let fixture = fixture();
let runtime = FixtureRuntime::new();
let acp = AcpServer::new(runtime.clone()).await.unwrap();
let (server_io, client_io) = tokio::io::duplex(64 * 1024);
let (server_read, server_write) = tokio::io::split(server_io);
let (client_read, mut client_write) = tokio::io::split(client_io);
let task = tokio::spawn(acp_server::run_stdio(
acp,
BufReader::new(server_read),
server_write,
));
let mut client_read = BufReader::new(client_read);
write_acp(
&mut client_write,
json!({"jsonrpc":"2.0","id":900,"method":"initialize","params":{"protocolVersion":1}}),
)
.await;
let _ = read_acp(&mut client_read).await;
write_acp(
&mut client_write,
json!({"jsonrpc":"2.0","id":901,"method":"session/load","params":{"sessionId":fixture["session_id"]}}),
)
.await;
let _ = read_acp(&mut client_read).await;
let mut descriptor = Value::Null;
let mut attachment = Value::Null;
let mut live = Vec::new();
for (index, action) in fixture["semantic_actions"]
.as_array()
.unwrap()
.iter()
.enumerate()
{
if action["method"] == "frontend.v2.detach" {
continue;
}
let (envelope, emitted) = acp_request(
&mut client_write,
&mut client_read,
(index + 1) as u64,
action["method"].as_str().unwrap(),
action["params"].clone(),
)
.await;
live.extend(emitted);
assert!(envelope.get("error").is_none(), "{envelope}");
match action["method"].as_str().unwrap() {
"frontend.v2.describe" => descriptor = envelope["result"].clone(),
"frontend.v2.attach" => attachment = envelope["result"].clone(),
_ => {}
}
}
while live.len() < fixture["live"].as_array().unwrap().len() {
let message = read_acp(&mut client_read).await;
if message["method"] == "frontend.v2.event" {
live.push(serde_json::from_value(message["params"]["event"].clone()).unwrap());
}
}
let errors = [
(
"frontend.v2.send_input",
json!({"prompt":fixture["competing_prompt"]}),
),
(
"frontend.v2.respond",
json!({"response":fixture["unsupported_response"]}),
),
];
let mut names = Vec::new();
for (index, (method, params)) in errors.into_iter().enumerate() {
let (envelope, _) = acp_request(
&mut client_write,
&mut client_read,
100 + index as u64,
method,
params,
)
.await;
names.push(envelope["error"]["name"].clone());
}
let detach = fixture["semantic_actions"]
.as_array()
.unwrap()
.iter()
.find(|action| action["method"] == "frontend.v2.detach")
.unwrap();
let (detached, _) = acp_request(
&mut client_write,
&mut client_read,
200,
"frontend.v2.detach",
detach["params"].clone(),
)
.await;
assert!(detached.get("error").is_none(), "{detached}");
let ordered_sdk_calls = runtime.observed_calls();
client_write.shutdown().await.unwrap();
task.await.unwrap().unwrap();
observation(
descriptor,
attachment["history"].clone(),
attachment["history_cursor"].as_u64().unwrap(),
attachment["replay"].clone(),
serde_json::to_value(live).unwrap(),
ordered_sdk_calls,
Value::Array(names),
)
}
#[tokio::test]
async fn every_rust_binding_reduces_the_complete_shared_fixture_identically() {
let expected = expected_observation();
assert_eq!(local_observation().await, expected, "local binding");
assert_eq!(http_observation().await, expected, "HTTP/SSE binding");
assert_eq!(websocket_observation().await, expected, "WebSocket binding");
assert_eq!(acp_observation().await, expected, "ACP binding");
}