fix: accept upstream SSE data without a space (#269)

## Summary

An upstream SSE field may omit the space after `data:`, but the live
HTTP reader currently discards that spelling before event normalization.
A valid unspaced stream consequently produces an empty completed
Responses output over HTTP and WebSocket, and the Messages loop misses
its tool call. An unspaced `data:[DONE]` also fails to stop a connection
that remains open.

Accept both spellings in the shared inference reader and retain the
existing `data: ` form for its consumers. Preserve the field value,
including additional whitespace, and recognize both exact completion
markers. This completes the live-transport boundary missing from the
decoder-only coverage in #236; the [SSE field grammar makes the space
optional](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream).

Production changes are confined to `response_lines`: no new parser,
lifecycle state, public type, database schema, dependency, or delivery
queue. Already accepted data lines remain unchanged. Add coverage
through real HTTP/WebSocket clients, SQLite close/reopen and a
subsequent upstream continuation request, plus recorded Messages tool
rounds. Correct the search test fixture to use GET and verify the
successful tool result reaches the next model request.

## Test Plan

Exact commit `74b28f187a57f8d0bd41c2f884ea919b21a378f4` verified from a
clean detached worktree with a separate build directory, Rust 1.98.0,
and macOS:

- Against unchanged production on main `60126bb`, the final tests yield
seven intended failures and eleven passing controls. The isolated
patched build passes all eighteen. HTTP and WebSocket failures show
`response.completed` with `output: []`; Messages stops after one round
instead of two; the reader drops unspaced fields or times out after the
unspaced marker.
- Cover mixed spacing, empty values, Unicode, LF/CRLF, case-sensitive
field names, malformed JSON passthrough, meaningful extra whitespace,
exact and near-miss completion markers, the 256 KiB line boundary, and
chunk timeout behavior. Existing split-UTF-8, error, cancellation,
slow-consumer, and lifecycle tests also pass.
- `cargo test --locked --workspace --all-features`: 1,086 passed; eight
PostgreSQL tests and one existing doctest ignored.
- Forty-five fresh-process repeat runs pass: 40 reader boundary tests,
70 Messages tests, and 75 HTTP/WebSocket restart tests, including three
concurrent test processes.
- Workspace formatting, all-target/all-feature check and Clippy with
warnings denied; binary builds, launcher contracts, and source-install
CLI E2E with a development-profile wheel pass.
- Python: 111 passed, three installation-only skips. Cassette
validation: 113/113, 258 turns. No recorded cassette files were
modified.
- All applicable changed-file hooks pass. Remaining repository-wide
hooks pass; the known Apple Git hang in the all-files large-file hook
was avoided by running that hook on the complete changed-file set.

Validation uses local HTTP/database fixtures and existing recorded
streams. No live-model/GPU or new PostgreSQL execution is claimed. This
is an optional-space fix, not a claim of complete SSE framing
conformance.

Signed-off-by: Chuyue Wang <stevenwang0805@outlook.com>
Co-authored-by: Francisco Javier Arceo <arceofrancisco@gmail.com>

Author: StevenWang-CY

crates/agentic-server-core/src/executor/inference.rs

@@ -190,7 +190,8 @@ pub fn call_inference(
     }
 }
 
-/// Convert a successful upstream response body into normalized SSE data lines.
+/// Convert a successful upstream response body into SSE data lines with a `data: ` prefix.
+/// The single space after the field colon is optional on the wire.
 pub(super) fn response_lines(
     resp: reqwest::Response,
     chunk_timeout: Duration,
@@ -215,12 +216,19 @@ pub(super) fn response_lines(
                     return;
                 }
             };
-            for line in lines {
-                match line.as_str() {
-                    "data: [DONE]" => return,
-                    l if l.starts_with("data: ") => yield Ok(line),
-                    _ => {}
+            for mut line in lines {
+                let Some(data) = line.strip_prefix("data:") else {
+                    continue;
+                };
+                if data.strip_prefix(' ').unwrap_or(data) == "[DONE]" {
+                    return;
+                }
+                // Keep one transport spelling for Responses and Messages consumers,
+                // without trimming whitespace that belongs to the field value.
+                if !data.starts_with(' ') {
+                    line.insert("data:".len(), ' ');
                 }
+                yield Ok(line);
             }
         }
     }
@@ -239,6 +247,123 @@ mod tests {
 
     use super::*;
 
+    async fn read_sse_body(body: String, keep_open: bool) -> Vec<ExecutorResult<String>> {
+        let app = axum::Router::new().route(
+            "/v1/responses",
+            post(move || {
+                let body = body.clone();
+                async move {
+                    let chunks = async_stream::stream! {
+                        yield Ok::<_, Infallible>(Bytes::from(body));
+                        if keep_open {
+                            std::future::pending::<()>().await;
+                        }
+                    };
+                    Response::builder()
+                        .header("content-type", "text/event-stream")
+                        .body(Body::from_stream(chunks))
+                        .unwrap()
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}/v1/responses", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        let result = call_inference(
+            "{}".to_owned(),
+            url,
+            Arc::new(reqwest::Client::new()),
+            None,
+            Duration::from_secs(1),
+        )
+        .collect()
+        .await;
+        server.abort();
+        let _ = server.await;
+        result
+    }
+
+    #[tokio::test]
+    async fn sse_data_spacing_preserves_field_values() {
+        let lines = read_sse_body(
+            concat!(
+                ":comment\r\nevent: ignored\r\nid: 1\r\nretry: 1000\r\n",
+                "Data: ignored\r\ndatabase: ignored\r\n data: ignored\r\n",
+                "data:{\"delta\":\"雪 ☃: data: text\"}\r\n\r\n",
+                "data: {\"spaced\":true}\n\n",
+                "data:  {\"leading_space\":true}\n\n",
+                "data:\t{\"tab\":true}\n\n",
+                "data:\n\ndata: \n\ndata:{malformed}\n\n",
+                "data:  [DONE]\n\ndata:[DONE]extra\n\n",
+            )
+            .to_owned(),
+            false,
+        )
+        .await
+        .into_iter()
+        .collect::<ExecutorResult<Vec<_>>>()
+        .unwrap();
+        assert_eq!(
+            lines,
+            [
+                "data: {\"delta\":\"雪 ☃: data: text\"}",
+                "data: {\"spaced\":true}",
+                "data:  {\"leading_space\":true}",
+                "data: \t{\"tab\":true}",
+                "data: ",
+                "data: ",
+                "data: {malformed}",
+                "data:  [DONE]",
+                "data: [DONE]extra",
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn sse_data_spacing_stops_at_both_done_markers() {
+        for separator in ["", " "] {
+            let body = format!("data: {{}}\n\ndata:{separator}[DONE]\r\n\r\ndata: ignored\n\n");
+            let lines = read_sse_body(body, true)
+                .await
+                .into_iter()
+                .collect::<ExecutorResult<Vec<_>>>()
+                .expect("[DONE] must stop without waiting for upstream EOF or a chunk timeout");
+            assert_eq!(lines, ["data: {}"]);
+        }
+    }
+
+    #[tokio::test]
+    async fn sse_data_spacing_preserves_line_limit() {
+        for separator in ["", " "] {
+            let prefix = format!("data:{separator}");
+            let line = format!("{prefix}{}", "x".repeat(MAX_SSE_LINE_BYTES - prefix.len()));
+            let accepted = read_sse_body(format!("{line}\n\n"), false).await;
+            assert_eq!(accepted.len(), 1);
+            assert_eq!(
+                accepted[0].as_ref().unwrap().len(),
+                MAX_SSE_LINE_BYTES + usize::from(separator.is_empty())
+            );
+
+            let rejected = read_sse_body(format!("{line}x\n\n"), false).await;
+            assert_eq!(rejected.len(), 1);
+            assert!(
+                rejected[0]
+                    .as_ref()
+                    .unwrap_err()
+                    .to_string()
+                    .contains("upstream SSE line exceeded")
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn sse_data_spacing_preserves_chunk_timeout_without_done() {
+        let lines = read_sse_body("data:{}\n\n".to_owned(), true).await;
+        assert_eq!(lines.len(), 2);
+        assert_eq!(lines[0].as_ref().unwrap(), "data: {}");
+        assert!(lines[1].as_ref().unwrap_err().to_string().contains("chunk timeout"));
+    }
+
     async fn oversized_body_server(status: StatusCode) -> (String, tokio::task::JoinHandle<()>) {
         let app = axum::Router::new().route(
             "/v1/responses",

crates/agentic-server-core/tests/messages_stream_test.rs

@@ -19,7 +19,7 @@ use agentic_core::tool::{ToolRegistry, WebSearchHandler};
 use agentic_core::types::messages::{GatewayToolMap, ToolParam, registry_tools};
 use axum::extract::State;
 use axum::response::{IntoResponse, Response};
-use axum::routing::post;
+use axum::routing::{get, post};
 use axum::{Json, Router};
 use futures::StreamExt;
 use http::StatusCode;
@@ -146,7 +146,7 @@ async fn spawn_mock_vllm_stream_then_error(
 async fn spawn_mock_search() -> (String, tokio::task::JoinHandle<()>) {
     let app = Router::new().route(
         "/v1/search",
-        post(|Json(_body): Json<Value>| async move {
+        get(|| async move {
             Json(serde_json::json!({
                 "results": {"web": [{"url": "https://www.rust-lang.org/", "title": "Rust",
                     "description": "d", "snippets": ["Rust 1.89.0 is the latest stable release."]}], "news": []},
@@ -188,8 +188,28 @@ async fn run_test_messages_stream(
 
 #[tokio::test]
 async fn messages_stream_presents_one_message_and_hides_gateway_tool() {
-    let (vllm_url, upstream, _v) = spawn_mock_vllm_stream(cassette_turn_streams()).await;
-    let (search_url, _s) = spawn_mock_search().await;
+    assert_messages_stream_presents_one_message(cassette_turn_streams()).await;
+}
+
+#[tokio::test]
+async fn messages_stream_accepts_unspaced_sse_data_through_gateway_tool_rounds() {
+    let streams = cassette_turn_streams()
+        .into_iter()
+        .map(|body| {
+            body.split_inclusive('\n')
+                .map(|line| {
+                    line.strip_prefix("data: ")
+                        .map_or_else(|| line.to_owned(), |data| format!("data:{data}"))
+                })
+                .collect()
+        })
+        .collect();
+    assert_messages_stream_presents_one_message(streams).await;
+}
+
+async fn assert_messages_stream_presents_one_message(streams: Vec<String>) {
+    let (vllm_url, upstream, vllm) = spawn_mock_vllm_stream(streams).await;
+    let (search_url, search) = spawn_mock_search().await;
     let exec_ctx = build_exec_ctx(&vllm_url, &search_url).await;
 
     let request = serde_json::json!({
@@ -210,13 +230,28 @@ async fn messages_stream_presents_one_message_and_hides_gateway_tool() {
     let stream = run_test_messages_stream(request, registry, Arc::clone(&exec_ctx)).await;
     let chunks: Vec<String> = stream.collect().await;
     let sse = chunks.join("");
+    vllm.abort();
+    search.abort();
+    let _ = tokio::join!(vllm, search);
 
     // Two upstream rounds ran (tool round + final).
     assert_eq!(
         upstream.calls.load(Ordering::SeqCst),
         2,
         "one tool round + one final round"
     );
+    let requests = upstream.requests.lock().await;
+    let tool_output = &requests[1]["messages"].as_array().unwrap().last().unwrap()["content"][0];
+    assert_eq!(tool_output["type"], "tool_result");
+    assert_ne!(
+        tool_output["is_error"], true,
+        "gateway search must succeed: {tool_output}"
+    );
+    assert!(
+        tool_output["content"]
+            .to_string()
+            .contains("https://www.rust-lang.org/")
+    );
 
     // Exactly one logical message lifecycle.
     assert_eq!(