From fe8d6ae45e3207cc689ded4f3e7f6e3383635cb4 Mon Sep 17 00:00:00 2001 From: Emre K <110906681+kocaemre@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:13:58 +0200 Subject: [PATCH] Parse multi-line Modern SSE events Signed-off-by: Emre K <110906681+kocaemre@users.noreply.github.com> --- pkg/vmcp/client/modern.go | 48 +++++++++++++++----- pkg/vmcp/client/modern_test.go | 80 ++++++++++++++++++++++++---------- 2 files changed, 95 insertions(+), 33 deletions(-) diff --git a/pkg/vmcp/client/modern.go b/pkg/vmcp/client/modern.go index f5c703f7c6..3ae6aa6e91 100644 --- a/pkg/vmcp/client/modern.go +++ b/pkg/vmcp/client/modern.go @@ -378,36 +378,62 @@ func readModernEnvelope(resp *http.Response, wantID int64) (json.RawMessage, *mo // readModernSSE scans an SSE body for the response whose id matches wantID, // consuming (ignoring) any server->client requests/notifications interleaved on -// the stream. A stream that ends without a matching response yields errWrongEra. +// the stream. Consecutive data: lines in one SSE event are joined with "\n" per +// the SSE spec before JSON decoding. A stream that ends without a matching +// response yields errWrongEra. func readModernSSE(body io.Reader, wantID int64) (json.RawMessage, *modernRPCError, error) { sc := bufio.NewScanner(body) // Cap the token at maxResponseSize (the doc-promised bound) so a valid single // data: event up to that size decodes; the outer io.LimitReader already bounds // the total, so this cannot over-allocate. sc.Buffer(make([]byte, 0, 64*1024), maxResponseSize) - for sc.Scan() { - data, ok := strings.CutPrefix(sc.Text(), "data:") - if !ok { - continue + + var data [][]byte + flush := func() (json.RawMessage, *modernRPCError, bool, error) { + if len(data) == 0 { + return nil, nil, false, nil } + payload := bytes.Join(data, []byte("\n")) + data = nil + var env modernRPCEnvelope - if json.Unmarshal([]byte(strings.TrimSpace(data)), &env) != nil { - continue + if json.Unmarshal(payload, &env) != nil { + return nil, nil, false, nil } if env.Method != "" { - continue // server->client request/notification; not our response + return nil, nil, false, nil // server->client request/notification; not our response } if !modernIDMatches(env.ID, wantID) { - continue + return nil, nil, false, nil } if env.Error == nil && len(env.Result) == 0 { - return nil, nil, errWrongEra + return nil, nil, true, errWrongEra + } + return env.Result, env.Error, true, nil + } + + for sc.Scan() { + line := sc.Text() + if line == "" { + result, rpcErr, matched, err := flush() + if matched || err != nil { + return result, rpcErr, err + } + continue } - return env.Result, env.Error, nil + value, ok := strings.CutPrefix(line, "data:") + if !ok { + continue + } + data = append(data, []byte(strings.TrimPrefix(value, " "))) } if err := sc.Err(); err != nil { return nil, nil, fmt.Errorf("%w: reading SSE stream: %w", errModernTransient, err) } + result, rpcErr, matched, err := flush() + if matched || err != nil { + return result, rpcErr, err + } return nil, nil, errWrongEra } diff --git a/pkg/vmcp/client/modern_test.go b/pkg/vmcp/client/modern_test.go index b851ac9b55..6f69fc3dd4 100644 --- a/pkg/vmcp/client/modern_test.go +++ b/pkg/vmcp/client/modern_test.go @@ -165,34 +165,70 @@ func TestModernCall_Decode(t *testing.T) { assert.Equal(t, []string{"2026-07-28"}, out.SupportedVersions) } -// TestModernCall_SSEResponse verifies the dual-body reader handles a -// text/event-stream response, ignoring interleaved notifications and returning +// TestModernCall_SSEResponse verifies the dual-body reader handles +// text/event-stream responses, ignoring interleaved notifications and returning // the final matching JSON-RPC response. func TestModernCall_SSEResponse(t *testing.T) { t.Parallel() - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - // Echo the request's JSON-RPC id: modernCall's SSE reader matches the - // response frame by id (unlike the JSON path), and the id comes from a - // shared counter, so it cannot be hardcoded. - var req struct { - ID json.RawMessage `json:"id"` - } - body, _ := io.ReadAll(r.Body) - require.NoError(t, json.Unmarshal(body, &req)) + tests := []struct { + name string + writeEvent func(t *testing.T, w http.ResponseWriter, id json.RawMessage) + }{ + { + name: "single-line response after notification", + writeEvent: func(_ *testing.T, w http.ResponseWriter, id json.RawMessage) { + _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{}}\n\n")) + _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"id\":" + string(id) + + ",\"result\":{\"resultType\":\"complete\",\"ok\":true}}\n\n")) + }, + }, + { + name: "multi-line data response", + writeEvent: func(_ *testing.T, w http.ResponseWriter, id json.RawMessage) { + _, _ = w.Write([]byte("event: message\n")) + _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"id\":" + string(id) + ",\n")) + _, _ = w.Write([]byte("data: \"result\":{\"resultType\":\"complete\",\"ok\":true}}\n\n")) + }, + }, + { + name: "CRLF response without trailing blank line", + writeEvent: func(_ *testing.T, w http.ResponseWriter, id json.RawMessage) { + _, _ = w.Write([]byte("event: message\r\n")) + _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\"}\r\n\r\n")) + _, _ = w.Write([]byte("event: message\r\n")) + _, _ = w.Write([]byte("data:{\"jsonrpc\":\"2.0\",\"id\":" + string(id) + + ",\"result\":{\"resultType\":\"complete\",\"ok\":true}}")) + }, + }, + } - w.Header().Set("Content-Type", "text/event-stream") - // A progress notification (has "method", no id match) then the response. - _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/progress\",\"params\":{}}\n\n")) - _, _ = w.Write([]byte("data: {\"jsonrpc\":\"2.0\",\"id\":" + string(req.ID) + - ",\"result\":{\"resultType\":\"complete\",\"ok\":true}}\n\n")) - })) - t.Cleanup(srv.Close) + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() - var out map[string]any - require.NoError(t, modernCall(context.Background(), srv.Client(), srv.URL, "server/discover", nil, "", nil, &out)) - assert.Equal(t, "complete", out["resultType"]) - assert.Equal(t, true, out["ok"]) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Echo the request's JSON-RPC id: modernCall's SSE reader matches the + // response frame by id (unlike the JSON path), and the id comes from a + // shared counter, so it cannot be hardcoded. + var req struct { + ID json.RawMessage `json:"id"` + } + body, _ := io.ReadAll(r.Body) + require.NoError(t, json.Unmarshal(body, &req)) + + w.Header().Set("Content-Type", "text/event-stream") + tt.writeEvent(t, w, req.ID) + })) + t.Cleanup(srv.Close) + + var out map[string]any + require.NoError(t, modernCall(context.Background(), srv.Client(), srv.URL, "server/discover", nil, "", nil, &out)) + assert.Equal(t, "complete", out["resultType"]) + assert.Equal(t, true, out["ok"]) + }) + } } // TestModernCall_ErrorMapping verifies the era/error classification: a valid