diff --git a/docs/openapi.json b/docs/openapi.json index 7f969aee..6c1fde10 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -133,7 +133,7 @@ } }, "202": { - "description": "JSON-RPC notification accepted" + "description": "JSON-RPC notification accepted, or long-running request accepted with its response delivered over SSE" }, "400": { "description": "Invalid ACP envelope", diff --git a/research/acp/friction.md b/research/acp/friction.md index fc2e4bd9..08733f38 100644 --- a/research/acp/friction.md +++ b/research/acp/friction.md @@ -287,3 +287,13 @@ Update this file continuously during the migration. - Owner: Unassigned. - Status: resolved - Links: `server/packages/sandbox-agent/src/router.rs`, `server/packages/sandbox-agent/src/desktop_runtime.rs`, `sdks/typescript/src/client.ts`, `frontend/packages/inspector/src/components/debug/DesktopTab.tsx` + +- Date: 2026-07-20 +- Area: Long-running ACP requests over streamable HTTP +- Issue: `session/prompt` kept its POST open until the agent completed the turn, making successful execution depend on client and proxy response-header timeouts despite an existing SSE response channel. +- Impact: Long turns could lose their POST connection after several minutes, terminate request handling, or require downstream clients to configure unusually long HTTP timeouts. +- Proposed direction: Return `202 Accepted` after `session/prompt` is written to the agent, then deliver its correlated JSON-RPC result or error through the existing SSE stream. Keep short requests synchronous and preserve immediate HTTP errors before acceptance. +- Decision: Proposed and implemented for review. +- Owner: Unassigned. +- Status: in_progress +- Links: `server/packages/acp-http-adapter/src/process.rs`, `server/packages/sandbox-agent/tests/v1_api/acp_transport.rs`, https://github.com/rivet-dev/sandbox-agent/issues/305 diff --git a/sdks/acp-http-client/src/index.ts b/sdks/acp-http-client/src/index.ts index f8ac668e..98e4402f 100644 --- a/sdks/acp-http-client/src/index.ts +++ b/sdks/acp-http-client/src/index.ts @@ -454,9 +454,9 @@ class StreamableHttpAcpTransport { return; } - // SSE failure is non-fatal: the POST request/response flow still works. - // Exiting the loop allows ensureSseLoop() to restart it on the next POST. - return; + // Prompt responses can be delivered exclusively over SSE after a 202. + // Reconnect without waiting for another POST, replaying from Last-Event-ID. + await delay(150); } } } diff --git a/sdks/acp-http-client/tests/smoke.test.ts b/sdks/acp-http-client/tests/smoke.test.ts index 0e0403d6..765d71a9 100644 --- a/sdks/acp-http-client/tests/smoke.test.ts +++ b/sdks/acp-http-client/tests/smoke.test.ts @@ -181,4 +181,41 @@ describe("AcpHttpClient integration", () => { await client.disconnect(); }); + + it("reconnects SSE without another POST while a prompt is in flight", async () => { + const serverId = `acp-http-client-reconnect-${Date.now().toString(36)}`; + let remainingSseFailures = 3; + const reconnectingFetch: typeof fetch = async (input, init) => { + if (init?.method === "GET" && remainingSseFailures > 0) { + remainingSseFailures -= 1; + throw new TypeError("simulated SSE connection failure"); + } + return globalThis.fetch(input, init); + }; + + const client = new AcpHttpClient({ + baseUrl, + token, + fetch: reconnectingFetch, + transport: { + path: `/v1/acp/${encodeURIComponent(serverId)}`, + bootstrapQuery: { agent: "mock" }, + }, + }); + + await client.initialize(); + const session = await client.newSession({ + cwd: process.cwd(), + mcpServers: [], + }); + const prompt = await client.prompt({ + sessionId: session.sessionId, + prompt: [{ type: "text", text: "reconnect the event stream" }], + }); + + expect(remainingSseFailures).toBe(0); + expect(prompt.stopReason).toBe("end_turn"); + + await client.disconnect(); + }); }); diff --git a/sdks/typescript/src/generated/openapi.ts b/sdks/typescript/src/generated/openapi.ts index 537102fd..e9823711 100644 --- a/sdks/typescript/src/generated/openapi.ts +++ b/sdks/typescript/src/generated/openapi.ts @@ -1005,6 +1005,15 @@ export interface components { directory: string; skillName: string; }; + McpCommand: string | string[]; + /** @enum {string} */ + McpRemoteTransport: "http" | "sse"; + McpOAuthConfig: { + clientId?: string | null; + clientSecret?: string | null; + scope?: string | null; + }; + McpOAuthConfigOrDisabled: components["schemas"]["McpOAuthConfig"] | boolean; }; responses: never; parameters: never; @@ -1083,7 +1092,7 @@ export interface operations { "application/json": components["schemas"]["AcpEnvelope"]; }; }; - /** @description JSON-RPC notification accepted */ + /** @description JSON-RPC notification accepted, or long-running request accepted with its response delivered over SSE */ 202: { content: never; }; diff --git a/server/ARCHITECTURE.md b/server/ARCHITECTURE.md index 4ac7de8a..254f24c3 100644 --- a/server/ARCHITECTURE.md +++ b/server/ARCHITECTURE.md @@ -104,11 +104,10 @@ Each session tracks: POST /v1/acp/{serverId}?agent=... initialize ACP server, auto-install agent ↓ POST /v1/acp/{serverId} session/new -POST /v1/acp/{serverId} session/prompt - ↓ GET /v1/acp/{serverId} Subscribe to ACP SSE stream +POST /v1/acp/{serverId} session/prompt → 202 Accepted ↓ -JSON-RPC response envelopes Answer questions / reply to permissions +JSON-RPC response over SSE Answer questions / reply to permissions ↓ DELETE /v1/acp/{serverId} Close ACP server ``` diff --git a/server/packages/acp-http-adapter/src/process.rs b/server/packages/acp-http-adapter/src/process.rs index cec1d27f..4c12cb12 100644 --- a/server/packages/acp-http-adapter/src/process.rs +++ b/server/packages/acp-http-adapter/src/process.rs @@ -17,6 +17,9 @@ use crate::registry::LaunchSpec; const RING_BUFFER_SIZE: usize = 1024; const STDERR_TAIL_SIZE: usize = 16; +// These requests can outlive ordinary HTTP response-header timeouts. Their JSON-RPC +// response is correlated by id and delivered through the existing SSE stream. +const ASYNC_RESPONSE_METHODS: &[&str] = &["session/prompt"]; #[derive(Debug, Error)] pub enum AdapterError { @@ -184,6 +187,11 @@ impl AdapterRuntime { "post: stdin write complete, waiting for response" ); + if ASYNC_RESPONSE_METHODS.contains(&method.as_str()) { + self.spawn_async_response_waiter(method, key, id_value.clone(), rx, write_ms); + return Ok(PostOutcome::Accepted); + } + let wait_start = Instant::now(); match tokio::time::timeout(self.request_timeout, rx).await { Ok(Ok(response)) => { @@ -254,48 +262,116 @@ impl AdapterRuntime { } } + fn spawn_async_response_waiter( + &self, + method: String, + key: String, + id: Value, + rx: oneshot::Receiver, + write_ms: u64, + ) { + let pending = self.pending.clone(); + let sender = self.sender.clone(); + let ring = self.ring.clone(); + let sequence = self.sequence.clone(); + let request_timeout = self.request_timeout; + + tracing::info!( + method = %method, + id = %key, + write_ms = write_ms, + "post: request accepted; response will be delivered over SSE" + ); + + tokio::spawn(async move { + match tokio::time::timeout(request_timeout, rx).await { + Ok(Ok(_)) => { + tracing::info!( + method = %method, + id = %key, + "post: asynchronous response delivered over SSE" + ); + } + Ok(Err(_)) => { + tracing::error!( + method = %method, + id = %key, + "post: response channel dropped before asynchronous response" + ); + } + Err(_) => { + let removed = pending.lock().await.remove(&key).is_some(); + tracing::error!( + method = %method, + id = %key, + timeout_ms = request_timeout.as_millis() as u64, + "post: TIMEOUT waiting for asynchronous agent response" + ); + if removed { + broadcast_payload( + &sender, + &ring, + &sequence, + json_rpc_error(id, "timed out waiting for agent response"), + ) + .await; + } + } + } + }); + } + async fn subscribe( &self, last_event_id: Option, - ) -> (Vec<(u64, Value)>, broadcast::Receiver) { - let replay = { + ) -> (Vec<(u64, Value)>, u64, broadcast::Receiver) { + // Subscribe before taking the replay snapshot so a concurrently published + // message is guaranteed to appear in at least one source. The watermark + // below removes messages that appear in both. + let receiver = self.sender.subscribe(); + let (replay, replay_watermark) = { let ring = self.ring.lock().await; - ring.iter() + let replay_watermark = ring.back().map(|message| message.sequence).unwrap_or(0); + let replay = ring + .iter() .filter(|message| { - if let Some(last_event_id) = last_event_id { - message.sequence > last_event_id - } else { - true - } + last_event_id.is_none_or(|last_event_id| message.sequence > last_event_id) }) .map(|message| (message.sequence, message.payload.clone())) - .collect::>() + .collect::>(); + (replay, replay_watermark) }; - (replay, self.sender.subscribe()) + (replay, replay_watermark, receiver) } pub async fn sse_stream( self: Arc, last_event_id: Option, ) -> impl Stream> + Send + 'static { - let (replay, rx) = self.subscribe(last_event_id).await; - let replay_stream = stream::iter(replay.into_iter().map(|(sequence, payload)| { - let event = Event::default() - .event("message") - .id(sequence.to_string()) - .data(payload.to_string()); - Ok(event) - })); - - let live_stream = BroadcastStream::new(rx).filter_map(|item| async move { + self.payload_stream(last_event_id) + .await + .map(|(sequence, payload)| { + Ok(Event::default() + .event("message") + .id(sequence.to_string()) + .data(payload.to_string())) + }) + } + + /// Stream of sequenced raw JSON-RPC payloads. + pub async fn payload_stream( + self: Arc, + last_event_id: Option, + ) -> impl Stream + Send + 'static { + let (replay, replay_watermark, rx) = self.subscribe(last_event_id).await; + let replay_stream = stream::iter(replay); + + let live_stream = BroadcastStream::new(rx).filter_map(move |item| async move { match item { - Ok(message) => { - let event = Event::default() - .event("message") - .id(message.sequence.to_string()) - .data(message.payload.to_string()); - Some(Ok(event)) + Ok(message) if message.sequence > replay_watermark => { + Some((message.sequence, message.payload)) } + Ok(_) => None, Err(_) => None, } }); @@ -310,15 +386,9 @@ impl AdapterRuntime { self: Arc, last_event_id: Option, ) -> impl Stream + Send + 'static { - let (replay, rx) = self.subscribe(last_event_id).await; - let replay_stream = stream::iter(replay.into_iter().map(|(_sequence, payload)| payload)); - let live_stream = BroadcastStream::new(rx).filter_map(|item| async move { - match item { - Ok(message) => Some(message.payload), - Err(_) => None, - } - }); - replay_stream.chain(live_stream) + self.payload_stream(last_event_id) + .await + .map(|(_sequence, payload)| payload) } pub async fn shutdown(&self) { @@ -331,7 +401,22 @@ impl AdapterRuntime { "shutting down agent process" ); - self.pending.lock().await.clear(); + let pending_ids = { + let mut pending = self.pending.lock().await; + pending.drain().map(|(id, _)| id).collect::>() + }; + for id in pending_ids { + if let Ok(id) = serde_json::from_str(&id) { + broadcast_payload( + &self.sender, + &self.ring, + &self.sequence, + json_rpc_error(id, "agent process stopped before responding"), + ) + .await; + } + } + let mut child = self.child.lock().await; match child.try_wait() { Ok(Some(_)) => {} @@ -413,19 +498,7 @@ impl AdapterRuntime { // see it in order after preceding notifications. This lets the // SSE translation task detect turn completion after all // session/update events have been processed. - let seq = sequence.fetch_add(1, Ordering::SeqCst) + 1; - let message = StreamMessage { - sequence: seq, - payload, - }; - { - let mut guard = ring.lock().await; - guard.push_back(message.clone()); - while guard.len() > RING_BUFFER_SIZE { - guard.pop_front(); - } - } - let _ = sender.send(message); + broadcast_payload(&sender, &ring, &sequence, payload).await; continue; } else { tracing::warn!( @@ -433,6 +506,7 @@ impl AdapterRuntime { has_error = has_error, "agent stdout: response has no matching pending request (orphan)" ); + continue; } } @@ -446,21 +520,7 @@ impl AdapterRuntime { "agent stdout: notification/event → SSE broadcast" ); - let seq = sequence.fetch_add(1, Ordering::SeqCst) + 1; - let message = StreamMessage { - sequence: seq, - payload, - }; - - { - let mut guard = ring.lock().await; - guard.push_back(message.clone()); - while guard.len() > RING_BUFFER_SIZE { - guard.pop_front(); - } - } - - let _ = sender.send(message); + broadcast_payload(&sender, &ring, &sequence, payload).await; } tracing::info!( @@ -513,13 +573,36 @@ impl AdapterRuntime { let pending = self.pending.clone(); tokio::spawn(async move { - let status = { - let mut guard = child.lock().await; - guard.wait().await.ok() + let status = loop { + let status = { + let mut guard = child.lock().await; + guard.try_wait() + }; + match status { + Ok(Some(status)) => break Some(status), + Ok(None) => tokio::time::sleep(Duration::from_millis(50)).await, + Err(_) => break None, + } }; let age_ms = spawned_at.elapsed().as_millis() as u64; - let pending_count = pending.lock().await.len(); + let pending_ids = { + let mut pending = pending.lock().await; + pending.drain().map(|(id, _)| id).collect::>() + }; + let pending_count = pending_ids.len(); + + for id in pending_ids { + if let Ok(id) = serde_json::from_str(&id) { + broadcast_payload( + &sender, + &ring, + &sequence, + json_rpc_error(id, "agent process stopped before responding"), + ) + .await; + } + } if let Some(status) = status { tracing::warn!( @@ -539,21 +622,7 @@ impl AdapterRuntime { } }); - let seq = sequence.fetch_add(1, Ordering::SeqCst) + 1; - let message = StreamMessage { - sequence: seq, - payload, - }; - - { - let mut guard = ring.lock().await; - guard.push_back(message.clone()); - while guard.len() > RING_BUFFER_SIZE { - guard.pop_front(); - } - } - - let _ = sender.send(message); + broadcast_payload(&sender, &ring, &sequence, payload).await; } else { tracing::error!( age_ms = age_ms, @@ -620,6 +689,227 @@ impl AdapterRuntime { } } +async fn broadcast_payload( + sender: &broadcast::Sender, + ring: &Mutex>, + sequence: &AtomicU64, + payload: Value, +) { + // Keep sequence allocation, replay insertion, and live publication ordered. + // Otherwise concurrent timeout, exit, and stdout tasks can publish a newer + // sequence before an older one and break Last-Event-ID replay. + let mut guard = ring.lock().await; + let sequence = sequence.fetch_add(1, Ordering::SeqCst) + 1; + let message = StreamMessage { sequence, payload }; + guard.push_back(message.clone()); + while guard.len() > RING_BUFFER_SIZE { + guard.pop_front(); + } + let _ = sender.send(message); +} + +fn json_rpc_error(id: Value, message: &str) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "error": { + "code": -32603, + "message": message, + } + }) +} + fn id_key(value: &Value) -> String { serde_json::to_string(value).unwrap_or_else(|_| "null".to_string()) } + +#[cfg(all(test, unix))] +mod tests { + use std::collections::HashMap; + use std::path::PathBuf; + + use super::*; + + #[tokio::test] + async fn asynchronous_request_response_is_delivered_on_stream() { + let runtime = Arc::new( + AdapterRuntime::start( + LaunchSpec { + program: PathBuf::from("sh"), + args: vec![ + "-c".to_string(), + concat!( + "IFS= read -r line; ", + "printf '%s\\n' ", + "'{\"jsonrpc\":\"2.0\",\"id\":7,\"result\":{\"ok\":true}}'" + ) + .to_string(), + ], + env: HashMap::new(), + }, + Duration::from_secs(1), + ) + .await + .expect("start adapter"), + ); + let mut stream = Box::pin(runtime.clone().value_stream(Some(0)).await); + + let outcome = runtime + .post(json!({ + "jsonrpc": "2.0", + "id": 7, + "method": "session/prompt", + "params": {} + })) + .await + .expect("accept request"); + assert!(matches!(outcome, PostOutcome::Accepted)); + + let response = tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("stream response before timeout") + .expect("response event"); + assert_eq!(response["id"], 7); + assert_eq!(response["result"]["ok"], true); + } + + #[tokio::test] + async fn asynchronous_request_timeout_is_delivered_on_stream() { + let runtime = Arc::new( + AdapterRuntime::start( + LaunchSpec { + program: PathBuf::from("sh"), + args: vec![ + "-c".to_string(), + concat!( + "IFS= read -r line; sleep 0.1; ", + "printf '%s\\n' ", + "'{\"jsonrpc\":\"2.0\",\"id\":42,\"result\":{\"late\":true}}'" + ) + .to_string(), + ], + env: HashMap::new(), + }, + Duration::from_millis(25), + ) + .await + .expect("start adapter"), + ); + let mut stream = Box::pin(runtime.clone().value_stream(Some(0)).await); + + let outcome = runtime + .post(json!({ + "jsonrpc": "2.0", + "id": 42, + "method": "session/prompt", + "params": {} + })) + .await + .expect("accept request"); + assert!(matches!(outcome, PostOutcome::Accepted)); + + let response = tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("stream response before timeout") + .expect("response event"); + assert_eq!(response["id"], 42); + assert_eq!(response["error"]["code"], -32603); + assert_eq!( + response["error"]["message"], + "timed out waiting for agent response" + ); + + let mut saw_late_response = false; + for _ in 0..2 { + let Ok(Some(event)) = + tokio::time::timeout(Duration::from_millis(250), stream.next()).await + else { + break; + }; + if event["id"] == 42 && event.get("result").is_some() { + saw_late_response = true; + } + } + assert!(!saw_late_response); + } + + #[tokio::test] + async fn asynchronous_request_process_exit_is_delivered_on_stream() { + let runtime = Arc::new( + AdapterRuntime::start( + LaunchSpec { + program: PathBuf::from("sh"), + args: vec!["-c".to_string(), "IFS= read -r line; exit 1".to_string()], + env: HashMap::new(), + }, + Duration::from_secs(1), + ) + .await + .expect("start adapter"), + ); + let mut stream = Box::pin(runtime.clone().value_stream(Some(0)).await); + + let outcome = runtime + .post(json!({ + "jsonrpc": "2.0", + "id": 99, + "method": "session/prompt", + "params": {} + })) + .await + .expect("accept request"); + assert!(matches!(outcome, PostOutcome::Accepted)); + + let response = tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("stream response before timeout") + .expect("response event"); + assert_eq!(response["id"], 99); + assert_eq!(response["error"]["code"], -32603); + assert_eq!( + response["error"]["message"], + "agent process stopped before responding" + ); + } + + #[tokio::test] + async fn asynchronous_request_shutdown_is_delivered_on_stream() { + let runtime = Arc::new( + AdapterRuntime::start( + LaunchSpec { + program: PathBuf::from("sh"), + args: vec!["-c".to_string(), "IFS= read -r line; sleep 10".to_string()], + env: HashMap::new(), + }, + Duration::from_secs(30), + ) + .await + .expect("start adapter"), + ); + let mut stream = Box::pin(runtime.clone().value_stream(Some(0)).await); + + let outcome = runtime + .post(json!({ + "jsonrpc": "2.0", + "id": 100, + "method": "session/prompt", + "params": {} + })) + .await + .expect("accept request"); + assert!(matches!(outcome, PostOutcome::Accepted)); + + tokio::time::timeout(Duration::from_secs(1), runtime.shutdown()) + .await + .expect("shutdown completes"); + let response = tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("stream response before timeout") + .expect("response event"); + assert_eq!(response["id"], 100); + assert_eq!( + response["error"]["message"], + "agent process stopped before responding" + ); + } +} diff --git a/server/packages/sandbox-agent/src/acp_proxy_runtime.rs b/server/packages/sandbox-agent/src/acp_proxy_runtime.rs index e0a4f991..eb86ce36 100644 --- a/server/packages/sandbox-agent/src/acp_proxy_runtime.rs +++ b/server/packages/sandbox-agent/src/acp_proxy_runtime.rs @@ -7,7 +7,7 @@ use std::time::Duration; use acp_http_adapter::process::{AdapterError, AdapterRuntime, PostOutcome}; use acp_http_adapter::registry::LaunchSpec; use axum::response::sse::Event; -use futures::Stream; +use futures::{Stream, StreamExt}; use sandbox_agent_agent_management::agents::{AgentId, AgentManager, InstallOptions}; use sandbox_agent_error::SandboxError; use sandbox_agent_opencode_adapter::{AcpDispatch, AcpDispatchResult, AcpPayloadStream}; @@ -54,6 +54,23 @@ pub struct AcpServerInstanceInfo { pub type PinBoxSseStream = std::pin::Pin> + Send>>; +type PinBoxPayloadStream = std::pin::Pin + Send>>; + +impl ProxyInstance { + async fn annotated_payload_stream(&self, last_event_id: Option) -> PinBoxPayloadStream { + let stream = self.runtime.clone().payload_stream(last_event_id).await; + let agent = self.agent; + let runtime = self.runtime.clone(); + Box::pin(stream.then(move |(sequence, value)| { + let runtime = runtime.clone(); + async move { + let value = annotate_agent_error(agent, value); + let value = annotate_agent_stderr(value, &runtime).await; + (sequence, value) + } + })) + } +} impl AcpProxyRuntime { pub fn new(agent_manager: Arc) -> Self { @@ -179,7 +196,16 @@ impl AcpProxyRuntime { last_event_id: Option, ) -> Result { let instance = self.get_instance(server_id).await?; - let stream = instance.runtime.clone().sse_stream(last_event_id).await; + let stream = + instance + .annotated_payload_stream(last_event_id) + .await + .map(|(sequence, payload)| { + Ok(Event::default() + .event("message") + .id(sequence.to_string()) + .data(payload.to_string())) + }); Ok(Box::pin(stream)) } @@ -461,7 +487,10 @@ impl AcpDispatch for AcpProxyRuntime { .get_instance(&server_id) .await .map_err(|e| e.to_string())?; - let stream = instance.runtime.clone().value_stream(last_event_id).await; + let stream = instance + .annotated_payload_stream(last_event_id) + .await + .map(|(_sequence, payload)| payload); Ok(Box::pin(stream) as AcpPayloadStream) }) } diff --git a/server/packages/sandbox-agent/src/router.rs b/server/packages/sandbox-agent/src/router.rs index 195a5cdd..edd7e649 100644 --- a/server/packages/sandbox-agent/src/router.rs +++ b/server/packages/sandbox-agent/src/router.rs @@ -3141,7 +3141,7 @@ async fn get_v1_acp_servers( request_body = AcpEnvelope, responses( (status = 200, description = "JSON-RPC response envelope", body = AcpEnvelope), - (status = 202, description = "JSON-RPC notification accepted"), + (status = 202, description = "JSON-RPC notification accepted, or long-running request accepted with its response delivered over SSE"), (status = 406, description = "Client does not accept JSON responses", body = ProblemDetails), (status = 415, description = "Unsupported media type", body = ProblemDetails), (status = 400, description = "Invalid ACP envelope", body = ProblemDetails), diff --git a/server/packages/sandbox-agent/tests/v1_api/acp_transport.rs b/server/packages/sandbox-agent/tests/v1_api/acp_transport.rs index 147d8443..2795d332 100644 --- a/server/packages/sandbox-agent/tests/v1_api/acp_transport.rs +++ b/server/packages/sandbox-agent/tests/v1_api/acp_transport.rs @@ -154,21 +154,24 @@ async fn acp_round_trip_and_replay() { &[], ) .await; - assert_eq!(status, StatusCode::OK); - assert_eq!( - parse_json(&body)["result"]["echoedMethod"], - "session/prompt" - ); - - let first_chunk = read_first_sse_data_with_last_id(&test_app.app, "server-replay", 0).await; - let first_event_id = parse_sse_event_id(&first_chunk); - let first_event = parse_sse_data(&first_chunk); - assert_eq!(first_event["method"], "server/echo"); - - let second_chunk = - read_first_sse_data_with_last_id(&test_app.app, "server-replay", first_event_id).await; - let second_event_id = parse_sse_event_id(&second_chunk); - assert!(second_event_id > first_event_id); + assert_eq!(status, StatusCode::ACCEPTED); + assert!(body.is_empty()); + + let mut last_event_id = 0; + let mut events = Vec::new(); + for _ in 0..4 { + let chunk = + read_first_sse_data_with_last_id(&test_app.app, "server-replay", last_event_id).await; + let event_id = parse_sse_event_id(&chunk); + assert!(event_id > last_event_id); + last_event_id = event_id; + events.push(parse_sse_data(&chunk)); + } + + assert_eq!(events[0]["method"], "server/echo"); + assert!(events + .iter() + .any(|event| { event["id"] == 2 && event["result"]["echoedMethod"] == "session/prompt" })); } #[cfg(unix)]