a2aproject / a2aproject/a2a-rs
a2a-client: residual bytes at end of stream are discarded, hiding errors and truncated events
- Dominant language
- Rust
- Stars
- 75
- Forks
- 19
- Avg merge
- 11h 27m
- Merged PRs (30d)
- 21
Description
## Summary
`parse_sse_bytes` discards whatever is left in its buffer when the byte stream ends:
```rust
None => return None,
```
Anything the server sent that was not terminated by an SSE event boundary (`\n\n`, `\r\r`, `\r\n\r\n`) is dropped, and the caller sees an **empty, successfully-closed stream** — no events and no error.
## What is already handled
The plain-JSON-envelope case reported in #79 is **mostly** fixed on `main` already, by a content-type check in `send_streaming_message` (`a2a-client/src/jsonrpc.rs`):
```rust
let is_event_stream = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.map(|v| v.as_bytes().starts_with(b"text/event-stream"))
.unwrap_or(true);
```
A response that declares `application/json` is parsed as a `JsonRpcResponse` and its error surfaced. That covers `a2a-server`'s own pre-stream errors, which do set the header.
The REST binding is fine for a different reason: `post_streaming`/`get_streaming` check `resp.status().is_success()` first, and a REST error carries a non-2xx status.
## What is still lost
Three cases still reach `parse_sse_bytes` and end silently.
1. **No `Content-Type` at all.** `.unwrap_or(true)` deliberately treats a header-less response as a stream — the comment explains why (inspecting the body would hang a stream that has not sent anything yet). A JSON-RPC error envelope from a server that omits the header is therefore buffered, never boundary-terminated, and dropped.
2. **`Content-Type: text/event-stream` with a plain envelope body.** A server that sets the streaming header and then answers with a JSON-RPC error object — or a proxy that rewrites the body but keeps the header — produces the same silent empty stream.
3. **A truncated final SSE event.** If a server writes `data: {...}` and closes the connection without the trailing blank line, that last event is dropped. This one loses **data**, not just errors, and is independent of any error path.
All three are the same missing behavior: the residual buffer is never examined.
## Why this matters here
`a2acli` is the visible casualty:
- `a2acli task subscribe ` against such a server prints nothing and exits `0`, which under `A2ACLI_EXIT_002` reads as "the CLI conducted and reported the turn successfully".
- `send --stream`'s fallback to polling (`A2ACLI_TASK_POLL_004`) keys off an `Err` from `send_streaming_message`. An empty stream is not an `Err`, so the fallback never runs and the command reports nothing at all.
- #186 (Tier 2 stream resumption) has to distinguish "the stream ended" from "the stream was cut". It cannot, while a cut stream and a clean end are the same observable.
## Scope
- [ ] At end of stream, if the buffer holds anything other than whitespace, interpret it rather than dropping it.
- [ ] Parse a residual SSE-shaped buffer (`data:` lines) through the same `parse_event` callback the framed path uses, so a truncated final event is delivered — this must work for both the JSON-RPC and REST callbacks.
- [ ] Otherwise parse it as a `JsonRpcResponse` envelope and surface a contained error (or result) — covering cases 1 and 2.
- [ ] If it is neither, emit an error rather than ending cleanly: a non-empty unparseable tail means the stream was cut, and silence is the one answer that is always wrong.
- [ ] Trailing whitespace and newlines must still end the stream cleanly, since servers commonly send them.
- [ ] Tests for: a header-less JSON-RPC error body, an SSE-declared response carrying a plain envelope, a truncated final event on both bindings, a whitespace-only tail, and an unparseable tail.
## Relationship to #79
#79 proposed the residual-buffer fallback before the content-type check existed, so it addresses case 1/2 but not case 3, and it parses the tail only as a `JsonRpcResponse` — which cannot deliver a truncated REST event, since the REST callback expects a bare `StreamResponse`. It also ends the stream silently when the tail fails to parse. Credit to @arkavo-com for finding this via cross-SDK interop testing; this issue carries the remaining work.
Contributor guide
Assessment
This issue has not been assessed yet.