diff --git a/internal/mcp/http.go b/internal/mcp/http.go index a717266..c4d96e4 100644 --- a/internal/mcp/http.go +++ b/internal/mcp/http.go @@ -35,13 +35,14 @@ type PostResponse struct { type httpTransport struct { poster Poster url string + extra map[string]string // static headers, e.g. an Authorization bearer mu sync.Mutex session string // Mcp-Session-Id, echoed back when the server issues one } -func newHTTPTransport(post Poster, endpoint string) *httpTransport { - return &httpTransport{poster: post, url: endpoint} +func newHTTPTransport(post Poster, endpoint string, extra map[string]string) *httpTransport { + return &httpTransport{poster: post, url: endpoint, extra: extra} } func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) { @@ -49,7 +50,7 @@ func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse if err != nil { return nil, err } - frame, err := decodeFrame(body) + frame, err := decodeFrame(body, req.ID) if err != nil { return nil, err } @@ -57,6 +58,18 @@ func (t *httpTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse if err := json.Unmarshal(frame, &resp); err != nil { return nil, fmt.Errorf("mcp: decode response: %w", err) } + // The id check the stdio transport already did. Without it a server that + // sends a request of its own (sampling/createMessage, roots/list) mid-stream + // has that request accepted as the answer: it unmarshals into an rpcResponse + // with neither result nor error, and the call reports success with nothing + // in it. An empty string and no error is the one answer that lies — the act + // is logged as done and the tool never ran. + if resp.ID == nil || *resp.ID != req.ID { + return nil, fmt.Errorf("mcp: response id mismatch (wanted %d)", req.ID) + } + if resp.Error == nil && len(resp.Result) == 0 { + return nil, errors.New("mcp: response carries neither result nor error") + } return &resp, nil } @@ -71,7 +84,14 @@ func (t *httpTransport) send(ctx context.Context, req *rpcRequest) ([]byte, erro if err != nil { return nil, err } - hdr := map[string]string{"Accept": "application/json, text/event-stream"} + hdr := map[string]string{} + // Configured headers first, so nothing here can be overwritten by them: + // a real remote server needs a bearer token, and the Vikunja one on + // loopback is only reachable without one because it is unauthenticated. + for k, v := range t.extra { + hdr[k] = v + } + hdr["Accept"] = "application/json, text/event-stream" t.mu.Lock() if t.session != "" { hdr["Mcp-Session-Id"] = t.session @@ -114,9 +134,10 @@ func headerGet(h map[string]string, key string) string { } // decodeFrame pulls the JSON object out of a body that is either raw JSON or -// SSE. For SSE we take the LAST data: payload that parses, which is the -// response — earlier frames on the stream are progress notifications. -func decodeFrame(body []byte) ([]byte, error) { +// SSE. For SSE we take the last data: payload that parses AND carries our own +// id with a result or an error in it. Matching on the presence of an "id" key +// alone is not enough: a JSON-RPC request from the server has one too. +func decodeFrame(body []byte, id int64) ([]byte, error) { trimmed := bytes.TrimSpace(body) if len(trimmed) == 0 { return nil, errors.New("mcp: empty response body") @@ -136,19 +157,28 @@ func decodeFrame(body []byte) ([]byte, error) { if payload == "" { continue } - var probe map[string]json.RawMessage + var probe struct { + ID *int64 `json:"id"` + Result json.RawMessage `json:"result"` + Error json.RawMessage `json:"error"` + Method string `json:"method"` + } if json.Unmarshal([]byte(payload), &probe) != nil { continue } - if _, isResp := probe["id"]; isResp { - last = []byte(payload) + if probe.Method != "" || probe.ID == nil || *probe.ID != id { + continue } + if len(probe.Result) == 0 && len(probe.Error) == 0 { + continue + } + last = []byte(payload) } if err := sc.Err(); err != nil { return nil, fmt.Errorf("mcp: read event stream: %w", err) } if last == nil { - return nil, errors.New("mcp: no JSON-RPC response in event stream") + return nil, fmt.Errorf("mcp: no JSON-RPC response for id %d in event stream", id) } return last, nil } diff --git a/internal/mcp/stdio.go b/internal/mcp/stdio.go index d597875..86e60e3 100644 --- a/internal/mcp/stdio.go +++ b/internal/mcp/stdio.go @@ -2,6 +2,7 @@ package mcp import ( "bufio" + "bytes" "context" "encoding/json" "errors" @@ -9,12 +10,17 @@ import ( "io" "os" "os/exec" - "strings" "sync" ) // maxLine bounds one JSON-RPC frame from a subprocess. A tool result bigger // than this is a misbehaving server, not something to buffer. +// +// The bound is enforced by bufio.Scanner's own buffer limit, not by measuring +// the line after it was assembled. Measuring afterwards is not a bound: a +// server that emits 500 MB with no newline would have all 500 MB in mavend's +// heap before the check could reject it, which on the deploy target is an OOM +// kill of the core daemon. const maxLine = 1 << 20 // 1 MiB // stdioTransport speaks newline-delimited JSON-RPC to a child process. This is @@ -23,12 +29,27 @@ const maxLine = 1 << 20 // 1 MiB // // Args are argv, never a shell string — the same discipline internal/tool // keeps, for the same reason. +// +// Reading happens on its own goroutine, feeding frames down a channel. That is +// what makes a call abandonable: bufio never observes a context, so a server +// that accepts a request and then writes nothing at all would otherwise block +// the reader forever with the transport lock held, and every other server in +// the manager behind it. type stdioTransport struct { - mu sync.Mutex - cmd *exec.Cmd - in io.WriteCloser - out *bufio.Reader - dead bool + cmd *exec.Cmd + in io.WriteCloser + lines chan []byte + stop chan struct{} // closed by Close, so the reader can give up + + // callMu serialises whole calls, so two callers cannot consume each + // other's frames off the shared channel. It is deliberately NOT the lock + // alive() takes: a hung call must not make the manager's health check + // block on it. + callMu sync.Mutex + + mu sync.Mutex + dead bool + readErr error } func newStdioTransport(ctx context.Context, argv []string, env []string, dir string) (*stdioTransport, error) { @@ -52,38 +73,92 @@ func newStdioTransport(ctx context.Context, argv []string, env []string, dir str if err := cmd.Start(); err != nil { return nil, fmt.Errorf("mcp: start %q: %w", argv[0], err) } - return &stdioTransport{cmd: cmd, in: in, out: bufio.NewReaderSize(out, 64<<10)}, nil + t := &stdioTransport{ + cmd: cmd, + in: in, + lines: make(chan []byte), + stop: make(chan struct{}), + } + go t.readLoop(out) + return t, nil +} + +// readLoop pushes one frame per line onto t.lines until the pipe ends. The +// scanner's own buffer limit is the frame bound: a line longer than maxLine +// ends the scan with bufio.ErrTooLong having buffered at most maxLine, rather +// than assembling the whole thing first and rejecting it afterwards. +func (t *stdioTransport) readLoop(out io.Reader) { + defer close(t.lines) + sc := bufio.NewScanner(out) + sc.Buffer(make([]byte, 0, 64<<10), maxLine) + for sc.Scan() { + line := bytes.TrimSpace(sc.Bytes()) + if len(line) == 0 { + continue + } + frame := append([]byte(nil), line...) + select { + case t.lines <- frame: + case <-t.stop: + return + } + } + err := sc.Err() + switch { + case errors.Is(err, bufio.ErrTooLong): + err = fmt.Errorf("mcp: frame exceeds %d bytes", maxLine) + case err == nil: + err = io.EOF + } + t.mu.Lock() + t.readErr = err + t.mu.Unlock() } func (t *stdioTransport) Call(ctx context.Context, req *rpcRequest) (*rpcResponse, error) { - t.mu.Lock() - defer t.mu.Unlock() - if t.dead { + t.callMu.Lock() + defer t.callMu.Unlock() + if !t.alive() { return nil, ErrClosed } - if err := t.write(req); err != nil { - t.dead = true + t.mu.Lock() + err := t.write(req) + t.mu.Unlock() + if err != nil { + _ = t.Close() return nil, err } // Read until the frame with our id turns up; anything else on the pipe is // a notification or a server-initiated request we do not answer. for { - if err := ctx.Err(); err != nil { - return nil, err + select { + case <-ctx.Done(): + // A server that took the request and answered nothing is not a + // server this connection can be reused with: the next call would + // read into a pipe whose state we no longer know. Drop it and let + // the manager re-dial. + _ = t.Close() + return nil, ctx.Err() + case line, ok := <-t.lines: + if !ok { + t.mu.Lock() + rerr := t.readErr + t.mu.Unlock() + _ = t.Close() + if rerr == nil { + rerr = ErrClosed + } + return nil, fmt.Errorf("mcp: read: %w", rerr) + } + var resp rpcResponse + if err := json.Unmarshal(line, &resp); err != nil { + continue // not a response frame; ignore rather than break the turn + } + if resp.ID == nil || *resp.ID != req.ID { + continue + } + return &resp, nil } - line, err := t.readLine() - if err != nil { - t.dead = true - return nil, err - } - var resp rpcResponse - if err := json.Unmarshal(line, &resp); err != nil { - continue // not a response frame; ignore rather than break the turn - } - if resp.ID == nil || *resp.ID != req.ID { - continue - } - return &resp, nil } } @@ -96,6 +171,7 @@ func (t *stdioTransport) Notify(ctx context.Context, method string, params any) return t.write(&rpcRequest{JSONRPC: "2.0", Method: method, Params: params}) } +// write must be called with t.mu held. func (t *stdioTransport) write(req *rpcRequest) error { req.JSONRPC = "2.0" raw, err := json.Marshal(req) @@ -108,31 +184,20 @@ func (t *stdioTransport) write(req *rpcRequest) error { return nil } -func (t *stdioTransport) readLine() ([]byte, error) { - for { - line, err := t.out.ReadString('\n') - if err != nil { - if len(strings.TrimSpace(line)) == 0 { - return nil, fmt.Errorf("mcp: read: %w", err) - } - return []byte(line), nil - } - if len(line) > maxLine { - return nil, fmt.Errorf("mcp: frame exceeds %d bytes", maxLine) - } - if s := strings.TrimSpace(line); s != "" { - return []byte(s), nil - } - } -} - +// Close is idempotent: a call that abandoned a silent pipe calls it, and so +// does the manager. func (t *stdioTransport) Close() error { t.mu.Lock() - defer t.mu.Unlock() + if t.dead { + t.mu.Unlock() + return nil + } t.dead = true + close(t.stop) if t.in != nil { _ = t.in.Close() } + t.mu.Unlock() if t.cmd.Process != nil { _ = t.cmd.Process.Kill() _ = t.cmd.Wait() @@ -145,5 +210,8 @@ func (t *stdioTransport) Close() error { func (t *stdioTransport) alive() bool { t.mu.Lock() defer t.mu.Unlock() - return !t.dead + if t.dead { + return false + } + return t.readErr == nil } diff --git a/internal/mcp/stdio_test.go b/internal/mcp/stdio_test.go index 8e43b59..b75b47d 100644 --- a/internal/mcp/stdio_test.go +++ b/internal/mcp/stdio_test.go @@ -9,6 +9,7 @@ import ( "os/exec" "strings" "testing" + "time" ) // The stdio transport is tested against a real subprocess — this test binary, @@ -47,6 +48,20 @@ func fakeStdioServer() { _ = out.Flush() continue } + // "mute" answers the handshake and then goes silent: a python server + // that hit an unhandled exception in its own read loop but did not + // exit is the ordinary way to get here. + if os.Getenv("MAVEN_MCP_FAKE") == "mute" && req.Method == "tools/call" { + select {} // never answer, never exit + } + // "flood" writes one enormous line with no newline in it. + if os.Getenv("MAVEN_MCP_FAKE") == "flood" && req.Method == "tools/call" { + for i := 0; i < 64; i++ { + _, _ = out.Write(make([]byte, 1<<20)) + } + _ = out.Flush() + continue + } result, rerr := h(req.Method, req.Params) resp := map[string]any{"jsonrpc": "2.0", "id": *req.ID} if rerr != nil { @@ -147,3 +162,55 @@ func TestStdioMissingCommand(t *testing.T) { t.Logf("err = %q", st[0].Err) } } + +// A stdio server that accepts a call and then answers nothing must not wedge +// the manager. Before the read moved onto its own goroutine, the read held the +// transport lock, Refresh took that lock through alive() while holding the +// manager lock, and from then on Tools, Status and Call blocked for EVERY +// server — including turns that touch no MCP tool at all. +func TestStdioSilentServerDoesNotWedgeTheManager(t *testing.T) { + m := stdioManager(t, "mute") + defer m.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := m.Call(ctx, "fake", "read_thing", nil) + done <- err + }() + select { + case err := <-done: + if err == nil { + t.Fatal("a call into a silent server must fail, not succeed") + } + case <-time.After(5 * time.Second): + t.Fatal("the call never returned: the context is not observed during the read") + } + + // The manager must still answer while (and after) that call was stuck. + ready := make(chan struct{}) + go func() { + m.Refresh(context.Background()) + m.Tools() + m.Status() + close(ready) + }() + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("Refresh/Tools/Status deadlocked behind the hung call") + } +} + +// One frame is bounded by the reader's buffer, not measured after the whole +// thing has already been assembled in mavend's heap. +func TestStdioOversizedFrameIsRefused(t *testing.T) { + m := stdioManager(t, "flood") + defer m.Close() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if _, err := m.Call(ctx, "fake", "read_thing", nil); err == nil { + t.Fatal("a 64 MiB frame must be refused") + } +}