diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b1a3a6a..ec91cd8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,7 @@ jobs: ["internal/gitutil"]=80 ["internal/ui"]=75 ["internal/mcp"]=70 + ["internal/mcpclient"]=80 ) for PKG in "${!THRESHOLDS[@]}"; do diff --git a/.uf/dewey/learnings/mcp-transport-20260811T180634-jay-flowers.md b/.uf/dewey/learnings/mcp-transport-20260811T180634-jay-flowers.md new file mode 100644 index 0000000..11b51f3 --- /dev/null +++ b/.uf/dewey/learnings/mcp-transport-20260811T180634-jay-flowers.md @@ -0,0 +1,10 @@ +--- +tag: mcp-transport +author: jay-flowers +category: pattern +created_at: 2026-08-11T18:06:34Z +identity: mcp-transport-20260811T180634-jay-flowers +tier: draft +--- + +MCP Streamable HTTP Transport Pattern: When speaking MCP Streamable HTTP to a server like Dewey, the client must: (1) send an `initialize` handshake with `protocolVersion: "2025-03-26"`, `clientInfo.name/version`, and `capabilities: {}` before any `tools/call` invocations, (2) set `Accept: application/json, text/event-stream` and `Content-Type: application/json` headers on all requests, (3) capture the `Mcp-Session-Id` response header and attach it to subsequent requests, (4) wrap tool method names in a `tools/call` JSON-RPC envelope with `params.name` and `params.arguments`, (5) parse responses in dual format — check `Content-Type` for `application/json` (direct unmarshal) vs `text/event-stream` (scan for `data:` prefixed lines), (6) handle session recovery by resetting and re-initializing on HTTP 400/404. The `initialized` notification is skipped (Dewey doesn't enforce it). Bare `http.Post()` with plain JSON-RPC will get HTTP 400 from MCP endpoints. diff --git a/.uf/dewey/learnings/mcpclient-architecture-20260811T180643-jay-flowers.md b/.uf/dewey/learnings/mcpclient-architecture-20260811T180643-jay-flowers.md new file mode 100644 index 0000000..732c385 --- /dev/null +++ b/.uf/dewey/learnings/mcpclient-architecture-20260811T180643-jay-flowers.md @@ -0,0 +1,10 @@ +--- +tag: mcpclient-architecture +author: jay-flowers +category: pattern +created_at: 2026-08-11T18:06:43Z +identity: mcpclient-architecture-20260811T180643-jay-flowers +tier: draft +--- + +Shared MCP Client Architecture (replicator): The `internal/mcpclient/` package provides a reusable MCP Streamable HTTP client shared by both `memory.Client` (proxy to Dewey tools) and `doctor.deweyHealthProbe()`. Key design decisions: (1) `mcpclient.Config` struct with `Name`, `Version`, `Timeout`, `Logger` for dependency injection, (2) lazy session initialization on first `Call()` rather than at construction time (avoids blocking `NewClient()` if Dewey is down), (3) `sync.Mutex` protects session state (`sessionID`, `inited`) for concurrent goroutine safety, (4) `sync/atomic.Int64` for monotonically increasing JSON-RPC request IDs, (5) `io.LimitReader` at 10MB bounds response body reads, (6) `memory.UnavailableError` preserved via `wrapError()` bridge for backward compatibility with existing callers that use `errors.As`. The doctor went from ~70 lines of inline MCP code to 7 lines. diff --git a/.uf/dewey/learnings/review-council-20260811T180652-jay-flowers.md b/.uf/dewey/learnings/review-council-20260811T180652-jay-flowers.md new file mode 100644 index 0000000..f8093c0 --- /dev/null +++ b/.uf/dewey/learnings/review-council-20260811T180652-jay-flowers.md @@ -0,0 +1,10 @@ +--- +tag: review-council +author: jay-flowers +category: pattern +created_at: 2026-08-11T18:06:52Z +identity: review-council-20260811T180652-jay-flowers +tier: draft +--- + +Review Council Spec Review Patterns: When running the review council on spec artifacts, the most consistently flagged issues across all reviewers were: (1) Thread safety for shared mutable state — 4/5 reviewers flagged missing concurrency safety for session state shared across goroutines (CRITICAL), (2) SSE parsing edge cases — 3/5 reviewers flagged missing scenarios for empty body, malformed JSON, no data line (HIGH), (3) Package naming collisions — architect flagged `mcphttp` vs existing `mcp` package confusion, renamed to `mcpclient` (CRITICAL), (4) Timeout budget for multi-round-trip operations — 3/5 reviewers flagged lazy init adding latency (HIGH), (5) Missing coverage strategy — testing reviewer flagged constitution violation (CRITICAL). Lesson: spec review council catches real implementation bugs before code is written. The thread safety finding alone would have caused CI `-race` failures. diff --git a/AGENTS.md b/AGENTS.md index 4a1ba8c..7ad4269 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -411,6 +411,7 @@ internal/ org/ Cell domain logic (CRUD, epics, sessions, sync) comms/ Agent messaging + file reservations forge/ Orchestration (decompose, spawn, worktree, review, insights) + mcpclient/ MCP Streamable HTTP client (shared transport) memory/ Dewey proxy + deprecated tool stubs gitutil/ Git worktree operations (os/exec) doctor/ Health check engine diff --git a/README.md b/README.md index 596d709..e7f3cf6 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,7 @@ internal/ org/ Cell CRUD, epics, sessions, sync comms/ Agent messaging, file reservations forge/ Decomposition, spawning, worktrees, review, insights + mcpclient/ MCP Streamable HTTP client (shared transport) memory/ Dewey proxy, deprecated tool stubs gitutil/ Git worktree operations (os/exec) doctor/ Health check engine diff --git a/internal/doctor/checks.go b/internal/doctor/checks.go index 1098418..0b20a28 100644 --- a/internal/doctor/checks.go +++ b/internal/doctor/checks.go @@ -6,11 +6,7 @@ package doctor import ( - "bytes" - "encoding/json" "fmt" - "io" - "net/http" "os" "os/exec" "strings" @@ -18,6 +14,7 @@ import ( "github.com/unbound-force/replicator/internal/config" "github.com/unbound-force/replicator/internal/db" + "github.com/unbound-force/replicator/internal/mcpclient" ) // CheckResult holds the outcome of a single health check. @@ -119,78 +116,16 @@ func checkDewey(deweyURL string) CheckResult { } } -// deweyHealthProbe sends an MCP initialize request to verify Dewey is alive. -// This is a lightweight probe that does not establish a full session. +// deweyHealthProbe uses the shared MCP client to verify Dewey is alive. +// It sends an initialize handshake followed by a dewey_health tools/call. func deweyHealthProbe(deweyURL string) error { - reqBody := map[string]any{ - "jsonrpc": "2.0", - "method": "initialize", - "id": 1, - "params": map[string]any{ - "protocolVersion": "2025-03-26", - "capabilities": map[string]any{}, - "clientInfo": map[string]any{ - "name": "replicator-doctor", - "version": "1.0.0", - }, - }, - } - - body, err := json.Marshal(reqBody) - if err != nil { - return fmt.Errorf("marshal request: %w", err) - } - - req, err := http.NewRequest(http.MethodPost, deweyURL, bytes.NewReader(body)) - if err != nil { - return fmt.Errorf("create request: %w", err) - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Accept", "application/json, text/event-stream") - - client := &http.Client{Timeout: 5 * time.Second} - resp, err := client.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - respBody, _ := io.ReadAll(resp.Body) - return fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) - } - - // Read the SSE response — look for a JSON-RPC result in the event stream. - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return fmt.Errorf("read response: %w", err) - } - - // The response is SSE format: "event: message\ndata: {json}\n\n" - // Extract the JSON data line. - for _, line := range strings.Split(string(respBody), "\n") { - line = strings.TrimSpace(line) - if !strings.HasPrefix(line, "data: ") { - continue - } - data := strings.TrimPrefix(line, "data: ") - var rpcResp struct { - Result any `json:"result"` - Error *struct { - Message string `json:"message"` - } `json:"error"` - } - if err := json.Unmarshal([]byte(data), &rpcResp); err != nil { - return fmt.Errorf("parse response: %w", err) - } - if rpcResp.Error != nil { - return fmt.Errorf("dewey error: %s", rpcResp.Error.Message) - } - // Got a successful initialize response — Dewey is alive. - return nil - } - - return fmt.Errorf("no valid response from Dewey") + client := mcpclient.New(deweyURL, mcpclient.Config{ + Name: "replicator-doctor", + Version: "1.0.0", + Timeout: 5 * time.Second, + }) + _, err := client.Call("dewey_health", map[string]any{}) + return err } // checkConfigDir verifies the config directory exists. diff --git a/internal/doctor/checks_test.go b/internal/doctor/checks_test.go index f5a241d..d94467e 100644 --- a/internal/doctor/checks_test.go +++ b/internal/doctor/checks_test.go @@ -15,7 +15,8 @@ import ( // mcpHandler returns an http.HandlerFunc that mimics the MCP Streamable HTTP // transport. It validates POST method, Content-Type, Accept header, and -// JSON-RPC protocol fields. Responds with SSE-formatted JSON-RPC success. +// JSON-RPC protocol fields. Handles both "initialize" and "tools/call" methods +// with proper MCP response formats. func mcpHandler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { @@ -50,21 +51,47 @@ func mcpHandler() http.HandlerFunc { http.Error(w, "invalid jsonrpc version", http.StatusBadRequest) return } - if req["method"] == nil { + + method, _ := req["method"].(string) + if method == "" { http.Error(w, "missing method", http.StatusBadRequest) return } id, _ := req["id"].(float64) - resp := map[string]any{ - "jsonrpc": "2.0", - "result": map[string]any{ - "capabilities": map[string]any{}, - "protocolVersion": "2025-03-26", - "serverInfo": map[string]any{"name": "dewey", "version": "test"}, - }, - "id": int(id), + + var resp map[string]any + + switch method { + case "initialize": + resp = map[string]any{ + "jsonrpc": "2.0", + "result": map[string]any{ + "capabilities": map[string]any{}, + "protocolVersion": "2025-03-26", + "serverInfo": map[string]any{"name": "dewey", "version": "test"}, + }, + "id": int(id), + } + // Set session ID header for MCP session management. + w.Header().Set("Mcp-Session-Id", "test-session-id") + case "tools/call": + // Return a successful MCP tools/call response with content wrapper. + toolResult := `{"status":"ok"}` + resp = map[string]any{ + "jsonrpc": "2.0", + "result": map[string]any{ + "content": []map[string]any{ + {"type": "text", "text": toolResult}, + }, + }, + "id": int(id), + } + default: + http.Error(w, "unknown method", http.StatusBadRequest) + return } + respJSON, err := json.Marshal(resp) if err != nil { http.Error(w, "encode error", http.StatusInternalServerError) @@ -206,9 +233,28 @@ func TestCheckDewey_HTTPError(t *testing.T) { } func TestCheckDewey_RPCError(t *testing.T) { + // Return success for initialize but a JSON-RPC error for tools/call. + callCount := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req map[string]any + json.Unmarshal(body, &req) + + method, _ := req["method"].(string) + id, _ := req["id"].(float64) + callCount++ + w.Header().Set("Content-Type", "text/event-stream") - fmt.Fprint(w, "event: message\ndata: {\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32601,\"message\":\"method not found\"},\"id\":1}\n\n") + + if method == "initialize" { + resp := fmt.Sprintf(`{"jsonrpc":"2.0","result":{"capabilities":{},"protocolVersion":"2025-03-26","serverInfo":{"name":"dewey","version":"test"}},"id":%d}`, int(id)) + w.Header().Set("Mcp-Session-Id", "test-session") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", resp) + return + } + + // Return error for tools/call. + fmt.Fprintf(w, "event: message\ndata: {\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32601,\"message\":\"method not found\"},\"id\":%d}\n\n", int(id)) })) defer srv.Close() diff --git a/internal/mcpclient/client.go b/internal/mcpclient/client.go new file mode 100644 index 0000000..1ade4b5 --- /dev/null +++ b/internal/mcpclient/client.go @@ -0,0 +1,346 @@ +// Package mcpclient provides an MCP Streamable HTTP transport client. +// +// The Client type handles the MCP session lifecycle: initialize handshake, +// Mcp-Session-Id management, tools/call envelope wrapping, dual-format +// response parsing (SSE + plain JSON), session recovery on HTTP 400/404, +// and concurrency safety for shared usage across goroutines. +package mcpclient + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" +) + +// maxResponseBytes limits the response body read to prevent unbounded memory +// consumption on malformed responses. +const maxResponseBytes = 10 * 1024 * 1024 // 10MB + +// Config configures the MCP client. +type Config struct { + // Name is the client identity sent in clientInfo.name during initialize. + Name string + + // Version is the client version sent in clientInfo.version during initialize. + Version string + + // Timeout is the per-request HTTP timeout. Default: 10s. + Timeout time.Duration + + // Logger is an optional logger for session lifecycle events. + // If nil, the client operates silently. + Logger Logger +} + +// Logger is an optional interface for structured logging of session lifecycle events. +type Logger interface { + Info(msg string, keyvals ...any) + Warn(msg string, keyvals ...any) +} + +// Client is an MCP Streamable HTTP transport client that speaks to a remote +// MCP server (e.g., Dewey). It handles session lifecycle, envelope wrapping, +// and response parsing. It is safe for concurrent use. +type Client struct { + url string + config Config + http *http.Client + + mu sync.Mutex + sessionID string + inited bool + nextID atomic.Int64 +} + +// New creates an MCP client for the given endpoint URL. +func New(url string, cfg Config) *Client { + timeout := cfg.Timeout + if timeout == 0 { + timeout = 10 * time.Second + } + c := &Client{ + url: url, + config: cfg, + http: &http.Client{ + Timeout: timeout, + }, + } + c.nextID.Store(1) + return c +} + +// Call sends an MCP tools/call request, handling session initialization +// on first use and session recovery on HTTP 400/404. +// The method parameter is the MCP tool name (e.g., "dewey_health"). +// Returns the tool result as raw JSON. +func (c *Client) Call(method string, params any) (json.RawMessage, error) { + c.mu.Lock() + if !c.inited { + if err := c.initSession(); err != nil { + c.mu.Unlock() + return nil, err + } + } + sessionID := c.sessionID + c.mu.Unlock() + + result, statusCode, err := c.doToolsCall(method, params, sessionID) + if err != nil && (statusCode == http.StatusBadRequest || statusCode == http.StatusNotFound) { + // Session recovery: reset and retry once. + // The lock is held across initSession + doToolsCall to prevent + // concurrent goroutines from triggering redundant re-initialization + // (TOCTOU fix). doToolsCall is I/O-bound but recovery is rare, + // so serializing the retry path is acceptable. + c.mu.Lock() + c.inited = false + c.sessionID = "" + if err := c.initSession(); err != nil { + c.mu.Unlock() + return nil, err + } + sessionID = c.sessionID + + if c.config.Logger != nil { + c.config.Logger.Warn("session recovery triggered", "status", statusCode) + } + + result, _, err = c.doToolsCall(method, params, sessionID) + c.mu.Unlock() + if err != nil { + if c.config.Logger != nil { + c.config.Logger.Warn("session recovery failed", "error", err) + } + return nil, err + } + } else if err != nil { + return nil, err + } + + return result, nil +} + +// initSession sends the MCP initialize handshake. Must be called with c.mu held. +func (c *Client) initSession() error { + reqBody := map[string]any{ + "jsonrpc": "2.0", + "method": "initialize", + "id": c.nextID.Add(1) - 1, + "params": map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + "clientInfo": map[string]any{ + "name": c.config.Name, + "version": c.config.Version, + }, + }, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return &UnavailableError{Cause: fmt.Errorf("marshal initialize: %w", err)} + } + + req, err := http.NewRequest(http.MethodPost, c.url, bytes.NewReader(body)) + if err != nil { + return &UnavailableError{Cause: fmt.Errorf("create initialize request: %w", err)} + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + + resp, err := c.http.Do(req) + if err != nil { + return &UnavailableError{Cause: err} + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + return &UnavailableError{ + Cause: fmt.Errorf("initialize failed: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))), + } + } + + // Parse the response to verify it's a valid initialize response. + _, err = c.parseResponse(resp) + if err != nil { + return &UnavailableError{Cause: fmt.Errorf("initialize response: %w", err)} + } + + // Capture session ID if present. + if sid := resp.Header.Get("Mcp-Session-Id"); sid != "" { + c.sessionID = sid + } + + c.inited = true + + if c.config.Logger != nil { + c.config.Logger.Info("session initialized", "url", c.url) + } + + return nil +} + +// doToolsCall sends a tools/call request and returns the result. +func (c *Client) doToolsCall(method string, params any, sessionID string) (json.RawMessage, int, error) { + reqBody := map[string]any{ + "jsonrpc": "2.0", + "method": "tools/call", + "id": c.nextID.Add(1) - 1, + "params": map[string]any{ + "name": method, + "arguments": params, + }, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return nil, 0, &UnavailableError{Cause: fmt.Errorf("marshal tools/call: %w", err)} + } + + req, err := http.NewRequest(http.MethodPost, c.url, bytes.NewReader(body)) + if err != nil { + return nil, 0, &UnavailableError{Cause: fmt.Errorf("create tools/call request: %w", err)} + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + if sessionID != "" { + req.Header.Set("Mcp-Session-Id", sessionID) + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, 0, &UnavailableError{Cause: err} + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + return nil, resp.StatusCode, &UnavailableError{ + Cause: fmt.Errorf("tools/call failed: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))), + } + } + + result, err := c.parseToolsCallResponse(resp) + if err != nil { + return nil, resp.StatusCode, &UnavailableError{Cause: err} + } + + return result, resp.StatusCode, nil +} + +// jsonRPCResponse is a JSON-RPC 2.0 response envelope. +type jsonRPCResponse struct { + JSONRPC string `json:"jsonrpc"` + Result json.RawMessage `json:"result,omitempty"` + Error *jsonRPCError `json:"error,omitempty"` + ID any `json:"id"` +} + +// jsonRPCError is a JSON-RPC 2.0 error object. +type jsonRPCError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// mcpContent represents a single content block in an MCP tools/call result. +type mcpContent struct { + Type string `json:"type"` + Text string `json:"text"` +} + +// mcpToolResult represents the result field of an MCP tools/call response. +type mcpToolResult struct { + Content []mcpContent `json:"content"` +} + +// parseResponse reads and parses an HTTP response as a JSON-RPC response, +// handling both SSE (text/event-stream) and plain JSON (application/json) formats. +func (c *Client) parseResponse(resp *http.Response) (*jsonRPCResponse, error) { + respBody, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("read response: %w", err) + } + + ct := resp.Header.Get("Content-Type") + + // Plain JSON response. + if strings.HasPrefix(ct, "application/json") { + var rpcResp jsonRPCResponse + if err := json.Unmarshal(respBody, &rpcResp); err != nil { + return nil, fmt.Errorf("unmarshal JSON response: %w", err) + } + if rpcResp.Error != nil { + return nil, fmt.Errorf("JSON-RPC error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) + } + return &rpcResp, nil + } + + // SSE response: scan for data: lines. + // Empty Content-Type falls through to SSE scanning intentionally — + // some MCP servers omit the header on streamed responses. + if !strings.HasPrefix(ct, "text/event-stream") && ct != "" { + return nil, fmt.Errorf("unexpected content type %q", ct) + } + for _, line := range strings.Split(string(respBody), "\n") { + line = strings.TrimSpace(line) + var data string + if strings.HasPrefix(line, "data: ") { + data = strings.TrimPrefix(line, "data: ") + } else if strings.HasPrefix(line, "data:") { + data = strings.TrimPrefix(line, "data:") + } else { + continue + } + + var rpcResp jsonRPCResponse + if err := json.Unmarshal([]byte(data), &rpcResp); err != nil { + return nil, fmt.Errorf("unmarshal SSE data: %w", err) + } + if rpcResp.Error != nil { + return nil, fmt.Errorf("JSON-RPC error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) + } + return &rpcResp, nil + } + + return nil, fmt.Errorf("no valid response found in SSE stream") +} + +// parseToolsCallResponse parses an HTTP response as an MCP tools/call result, +// extracting content[0].text as the tool result. +func (c *Client) parseToolsCallResponse(resp *http.Response) (json.RawMessage, error) { + rpcResp, err := c.parseResponse(resp) + if err != nil { + return nil, err + } + + var toolResult mcpToolResult + if err := json.Unmarshal(rpcResp.Result, &toolResult); err != nil { + return nil, fmt.Errorf("unmarshal tool result: %w", err) + } + + if len(toolResult.Content) == 0 { + return nil, fmt.Errorf("empty content array in tools/call response") + } + + return json.RawMessage(toolResult.Content[0].Text), nil +} + +// UnavailableError indicates the MCP endpoint is not reachable or returned +// a non-recoverable error. +type UnavailableError struct { + Cause error +} + +func (e *UnavailableError) Error() string { + return fmt.Sprintf("mcp unavailable: %v", e.Cause) +} + +func (e *UnavailableError) Unwrap() error { + return e.Cause +} diff --git a/internal/mcpclient/client_test.go b/internal/mcpclient/client_test.go new file mode 100644 index 0000000..b4b081b --- /dev/null +++ b/internal/mcpclient/client_test.go @@ -0,0 +1,872 @@ +package mcpclient + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// mcpHandler is a stateful mock MCP server handler that tracks initialization +// state, validates MCP protocol compliance, and returns SSE-formatted responses. +type mcpHandler struct { + t *testing.T + + mu sync.Mutex + initialized bool + sessionID string + initCount atomic.Int64 + toolCallCount atomic.Int64 + + // overrides allow tests to customize behavior. + initStatus int // HTTP status for initialize (0 = 200) + noSessionID bool // omit Mcp-Session-Id header + toolHandler func(name string, args json.RawMessage) any // custom tool result + toolStatus int // HTTP status for tools/call (0 = 200) + plainJSON bool // respond with application/json instead of SSE + emptyBody bool // respond with empty body + malformedSSE bool // respond with malformed SSE + noDataLine bool // respond with SSE but no data: line + emptyContent bool // respond with empty content array + rejectCount int // reject this many tools/call with 400 before succeeding + rejectedSoFar int +} + +func (h *mcpHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // Validate common headers. + if r.Header.Get("Content-Type") != "application/json" { + h.t.Errorf("missing Content-Type: application/json") + } + accept := r.Header.Get("Accept") + if !strings.Contains(accept, "application/json") || !strings.Contains(accept, "text/event-stream") { + h.t.Errorf("Accept header missing required types: %q", accept) + } + + var req map[string]any + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "bad request body", http.StatusBadRequest) + return + } + + method, _ := req["method"].(string) + + switch method { + case "initialize": + h.handleInitialize(w, r, req) + case "tools/call": + h.handleToolsCall(w, r, req) + default: + http.Error(w, fmt.Sprintf("unknown method: %s", method), http.StatusBadRequest) + } +} + +func (h *mcpHandler) handleInitialize(w http.ResponseWriter, r *http.Request, req map[string]any) { + h.initCount.Add(1) + + if h.initStatus != 0 { + http.Error(w, "init error", h.initStatus) + return + } + + h.mu.Lock() + h.initialized = true + h.sessionID = "test-session-123" + h.mu.Unlock() + + if !h.noSessionID { + w.Header().Set("Mcp-Session-Id", "test-session-123") + } + + result := map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + "serverInfo": map[string]any{ + "name": "test-server", + "version": "1.0.0", + }, + } + + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": result, + } + + if h.plainJSON { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(rpcResp) + return + } + + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) +} + +func (h *mcpHandler) handleToolsCall(w http.ResponseWriter, r *http.Request, req map[string]any) { + h.toolCallCount.Add(1) + + // Validate session ID header matches issued session (only when the + // handler actually sent a session ID in the initialize response). + h.mu.Lock() + expectedSID := h.sessionID + issuedSID := !h.noSessionID && expectedSID != "" + h.mu.Unlock() + if issuedSID { + gotSID := r.Header.Get("Mcp-Session-Id") + if gotSID != expectedSID { + h.t.Errorf("Mcp-Session-Id = %q, want %q", gotSID, expectedSID) + } + } + + // Session rejection simulation. + h.mu.Lock() + if h.rejectCount > 0 && h.rejectedSoFar < h.rejectCount { + h.rejectedSoFar++ + h.initialized = false + h.mu.Unlock() + http.Error(w, "session expired", http.StatusBadRequest) + return + } + h.mu.Unlock() + + if h.toolStatus != 0 { + http.Error(w, "tool error", h.toolStatus) + return + } + + if h.emptyBody { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + return + } + + if h.malformedSSE { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "data: {not-valid-json}\n\n") + return + } + + if h.noDataLine { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\n\n") + return + } + + // Extract tool name and arguments from params. + params, _ := req["params"].(map[string]any) + toolName, _ := params["name"].(string) + argsRaw, _ := json.Marshal(params["arguments"]) + + var toolResult any + if h.toolHandler != nil { + toolResult = h.toolHandler(toolName, argsRaw) + } else { + toolResult = map[string]string{"status": "ok"} + } + + if h.emptyContent { + // Return valid JSON-RPC but empty content array. + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []any{}, + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + return + } + + resultJSON, _ := json.Marshal(toolResult) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []map[string]any{ + { + "type": "text", + "text": string(resultJSON), + }, + }, + }, + } + + data, _ := json.Marshal(rpcResp) + + if h.plainJSON { + w.Header().Set("Content-Type", "application/json") + w.Write(data) + return + } + + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) +} + +// newMCPServer creates a test server with the stateful MCP handler. +func newMCPServer(t *testing.T) (*httptest.Server, *mcpHandler) { + t.Helper() + h := &mcpHandler{t: t} + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + return srv, h +} + +// --- Happy Path Tests --- + +func TestCall_SuccessfulInitializeAndToolsCall(t *testing.T) { + srv, h := newMCPServer(t) + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + result, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call: %v", err) + } + + var parsed map[string]string + if err := json.Unmarshal(result, &parsed); err != nil { + t.Fatalf("unmarshal result: %v", err) + } + if parsed["status"] != "ok" { + t.Errorf("status = %q, want %q", parsed["status"], "ok") + } + + if h.initCount.Load() != 1 { + t.Errorf("init count = %d, want 1", h.initCount.Load()) + } + if h.toolCallCount.Load() != 1 { + t.Errorf("tool call count = %d, want 1", h.toolCallCount.Load()) + } +} + +func TestCall_SessionReuse(t *testing.T) { + srv, h := newMCPServer(t) + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + + // First call triggers init. + if _, err := client.Call("dewey_health", map[string]any{}); err != nil { + t.Fatalf("first Call: %v", err) + } + + // Second call should reuse session. + if _, err := client.Call("store_learning", map[string]any{"info": "test"}); err != nil { + t.Fatalf("second Call: %v", err) + } + + if h.initCount.Load() != 1 { + t.Errorf("init count = %d, want 1 (should not re-init)", h.initCount.Load()) + } + if h.toolCallCount.Load() != 2 { + t.Errorf("tool call count = %d, want 2", h.toolCallCount.Load()) + } +} + +// --- Error Path Tests --- + +func TestCall_InitializeHTTPError(t *testing.T) { + srv, h := newMCPServer(t) + h.initStatus = http.StatusInternalServerError + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("dewey_health", map[string]any{}) + if err == nil { + t.Fatal("expected error for init failure") + } + + var unavail *UnavailableError + if !errors.As(err, &unavail) { + t.Errorf("expected UnavailableError, got %T: %v", err, err) + } + + if !strings.Contains(err.Error(), "500") { + t.Errorf("error should mention HTTP status: %v", err) + } +} + +func TestCall_SSEResponseWithJSONRPCError(t *testing.T) { + // Custom server that returns a JSON-RPC error for tools/call. + errorSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + json.NewDecoder(r.Body).Decode(&req) + method, _ := req["method"].(string) + + if method == "initialize" { + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + }, + } + w.Header().Set("Mcp-Session-Id", "err-session") + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + return + } + + // Return JSON-RPC error for tools/call. + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "error": map[string]any{ + "code": -32600, + "message": "invalid request", + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + })) + defer errorSrv.Close() + + client := New(errorSrv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for JSON-RPC error response") + } + if !strings.Contains(err.Error(), "invalid request") { + t.Errorf("error should contain message: %v", err) + } +} + +func TestCall_ConnectionRefused(t *testing.T) { + client := New("http://127.0.0.1:1", Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for connection refused") + } + + var unavail *UnavailableError + if !errors.As(err, &unavail) { + t.Errorf("expected UnavailableError, got %T: %v", err, err) + } +} + +func TestCall_MalformedJSONInSSE(t *testing.T) { + srv, h := newMCPServer(t) + h.malformedSSE = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for malformed JSON") + } + if !strings.Contains(err.Error(), "unmarshal") { + t.Errorf("error should mention unmarshal: %v", err) + } +} + +func TestCall_TimeoutOnRequest(t *testing.T) { + // Slow server that takes too long. + slowSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(500 * time.Millisecond) + })) + defer slowSrv.Close() + + client := New(slowSrv.URL, Config{ + Name: "test-client", + Version: "1.0.0", + Timeout: 100 * time.Millisecond, + }) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected timeout error") + } + + var unavail *UnavailableError + if !errors.As(err, &unavail) { + t.Errorf("expected UnavailableError, got %T: %v", err, err) + } +} + +// --- Edge Case Tests --- + +func TestCall_InitializeMissingSessionHeader(t *testing.T) { + srv, h := newMCPServer(t) + h.noSessionID = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + result, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call should succeed without session header: %v", err) + } + + var parsed map[string]string + if err := json.Unmarshal(result, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if parsed["status"] != "ok" { + t.Errorf("status = %q, want %q", parsed["status"], "ok") + } +} + +func TestCall_PlainJSONResponse(t *testing.T) { + srv, h := newMCPServer(t) + h.plainJSON = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + result, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call with plain JSON response: %v", err) + } + + var parsed map[string]string + if err := json.Unmarshal(result, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if parsed["status"] != "ok" { + t.Errorf("status = %q, want %q", parsed["status"], "ok") + } +} + +func TestCall_EmptyResponseBody(t *testing.T) { + srv, h := newMCPServer(t) + h.emptyBody = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for empty body") + } + if !strings.Contains(err.Error(), "no valid response") { + t.Errorf("error should mention no valid response: %v", err) + } +} + +func TestCall_SSENoDataLine(t *testing.T) { + srv, h := newMCPServer(t) + h.noDataLine = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for no data line") + } + if !strings.Contains(err.Error(), "no valid response") { + t.Errorf("error should mention no valid response: %v", err) + } +} + +func TestCall_EmptyContentArray(t *testing.T) { + srv, h := newMCPServer(t) + h.emptyContent = true + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error for empty content") + } + if !strings.Contains(err.Error(), "empty content") { + t.Errorf("error should mention empty content: %v", err) + } +} + +// --- Contract Tests --- + +func TestCall_ToolsCallEnvelopeCorrectness(t *testing.T) { + var receivedReq map[string]any + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + json.NewDecoder(r.Body).Decode(&req) + method, _ := req["method"].(string) + + if method == "initialize" { + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + }, + } + w.Header().Set("Mcp-Session-Id", "envelope-session") + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + return + } + + receivedReq = req + + toolResult := map[string]string{"status": "ok"} + resultJSON, _ := json.Marshal(toolResult) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []map[string]any{ + {"type": "text", "text": string(resultJSON)}, + }, + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + })) + defer srv.Close() + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("dewey_health", map[string]any{"key": "value"}) + if err != nil { + t.Fatalf("Call: %v", err) + } + + // Verify the tools/call envelope. + if receivedReq["method"] != "tools/call" { + t.Errorf("method = %v, want %q", receivedReq["method"], "tools/call") + } + + params, ok := receivedReq["params"].(map[string]any) + if !ok { + t.Fatalf("params is not a map: %T", receivedReq["params"]) + } + if params["name"] != "dewey_health" { + t.Errorf("params.name = %v, want %q", params["name"], "dewey_health") + } + + args, ok := params["arguments"].(map[string]any) + if !ok { + t.Fatalf("params.arguments is not a map: %T", params["arguments"]) + } + if args["key"] != "value" { + t.Errorf("params.arguments.key = %v, want %q", args["key"], "value") + } +} + +func TestCall_CorrectHeadersOnInitialize(t *testing.T) { + var initHeaders http.Header + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + json.NewDecoder(r.Body).Decode(&req) + method, _ := req["method"].(string) + + if method == "initialize" { + initHeaders = r.Header.Clone() + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "protocolVersion": "2025-03-26", + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Mcp-Session-Id", "header-session") + fmt.Fprintf(w, "data: %s\n\n", data) + return + } + + // tools/call + toolResult := map[string]string{"ok": "true"} + resultJSON, _ := json.Marshal(toolResult) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []map[string]any{ + {"type": "text", "text": string(resultJSON)}, + }, + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "data: %s\n\n", data) + })) + defer srv.Close() + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err != nil { + t.Fatalf("Call: %v", err) + } + + if initHeaders.Get("Accept") != "application/json, text/event-stream" { + t.Errorf("Accept = %q, want %q", initHeaders.Get("Accept"), "application/json, text/event-stream") + } + if initHeaders.Get("Content-Type") != "application/json" { + t.Errorf("Content-Type = %q, want %q", initHeaders.Get("Content-Type"), "application/json") + } +} + +// --- Recovery Tests --- + +func TestCall_SessionRecoveryOn400(t *testing.T) { + srv, h := newMCPServer(t) + h.rejectCount = 1 // Reject first tools/call, accept after re-init. + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + result, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call should succeed after recovery: %v", err) + } + + var parsed map[string]string + if err := json.Unmarshal(result, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if parsed["status"] != "ok" { + t.Errorf("status = %q, want %q", parsed["status"], "ok") + } + + // Should have initialized twice (once originally, once for recovery). + if h.initCount.Load() != 2 { + t.Errorf("init count = %d, want 2", h.initCount.Load()) + } +} + +func TestCall_SessionRecoveryOn404(t *testing.T) { + // Use a custom server that returns 404 for the first tools/call. + var initCount atomic.Int64 + var toolCallCount atomic.Int64 + rejected := false + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + json.NewDecoder(r.Body).Decode(&req) + method, _ := req["method"].(string) + + if method == "initialize" { + initCount.Add(1) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + }, + } + w.Header().Set("Mcp-Session-Id", "recovery-session") + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + return + } + + toolCallCount.Add(1) + if !rejected { + rejected = true + http.Error(w, "session not found", http.StatusNotFound) + return + } + + toolResult := map[string]string{"status": "ok"} + resultJSON, _ := json.Marshal(toolResult) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []map[string]any{ + {"type": "text", "text": string(resultJSON)}, + }, + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + })) + defer srv.Close() + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + result, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call should succeed after 404 recovery: %v", err) + } + + var parsed map[string]string + if err := json.Unmarshal(result, &parsed); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if parsed["status"] != "ok" { + t.Errorf("status = %q, want %q", parsed["status"], "ok") + } + + if initCount.Load() != 2 { + t.Errorf("init count = %d, want 2", initCount.Load()) + } +} + +func TestCall_RetryFailure(t *testing.T) { + // Server that always returns 400 for tools/call. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req map[string]any + json.NewDecoder(r.Body).Decode(&req) + method, _ := req["method"].(string) + + if method == "initialize" { + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "protocolVersion": "2025-03-26", + }, + } + w.Header().Set("Mcp-Session-Id", "retry-session") + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) + return + } + + http.Error(w, "always failing", http.StatusBadRequest) + })) + defer srv.Close() + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("test", map[string]any{}) + if err == nil { + t.Fatal("expected error when retry also fails") + } + + var unavail *UnavailableError + if !errors.As(err, &unavail) { + t.Errorf("expected UnavailableError, got %T: %v", err, err) + } +} + +// --- Concurrency Tests --- + +func TestCall_ConcurrentInitialization(t *testing.T) { + srv, h := newMCPServer(t) + + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + + const goroutines = 10 + var wg sync.WaitGroup + errs := make([]error, goroutines) + + wg.Add(goroutines) + for i := range goroutines { + go func(idx int) { + defer wg.Done() + _, errs[idx] = client.Call("dewey_health", map[string]any{}) + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Errorf("goroutine %d: %v", i, err) + } + } + + // With the mutex, only one goroutine should initialize. + if h.initCount.Load() != 1 { + t.Errorf("init count = %d, want 1 (concurrent init should be serialized)", h.initCount.Load()) + } +} + +// --- Logger Tests --- + +func TestCall_LoggerCalledOnSessionEvents(t *testing.T) { + srv, h := newMCPServer(t) + h.rejectCount = 1 + + logger := &testLogger{} + client := New(srv.URL, Config{ + Name: "test-client", + Version: "1.0.0", + Logger: logger, + }) + + _, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call: %v", err) + } + + if logger.infoCount < 1 { + t.Errorf("expected at least 1 Info log, got %d", logger.infoCount) + } + if logger.warnCount < 1 { + t.Errorf("expected at least 1 Warn log (recovery), got %d", logger.warnCount) + } +} + +func TestCall_NoLoggerOperatesSilently(t *testing.T) { + srv, _ := newMCPServer(t) + + // No Logger configured — should not panic. + client := New(srv.URL, Config{Name: "test-client", Version: "1.0.0"}) + _, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("Call: %v", err) + } +} + +// --- Default Config Tests --- + +func TestNew_DefaultTimeout(t *testing.T) { + client := New("http://example.com", Config{Name: "test", Version: "1.0.0"}) + if client.http.Timeout != 10*time.Second { + t.Errorf("timeout = %v, want 10s", client.http.Timeout) + } +} + +func TestNew_CustomTimeout(t *testing.T) { + client := New("http://example.com", Config{ + Name: "test", + Version: "1.0.0", + Timeout: 5 * time.Second, + }) + if client.http.Timeout != 5*time.Second { + t.Errorf("timeout = %v, want 5s", client.http.Timeout) + } +} + +func TestUnavailableError_Unwrap(t *testing.T) { + cause := fmt.Errorf("connection refused") + unavail := &UnavailableError{Cause: cause} + + // Error() includes the cause message. + if got := unavail.Error(); got != "mcp unavailable: connection refused" { + t.Errorf("Error() = %q, want %q", got, "mcp unavailable: connection refused") + } + + // Unwrap() returns the original cause. + if got := unavail.Unwrap(); got != cause { + t.Errorf("Unwrap() = %v, want %v", got, cause) + } + + // errors.Is works through Unwrap chain. + wrapped := fmt.Errorf("outer: %w", unavail) + var target *UnavailableError + if !errors.As(wrapped, &target) { + t.Errorf("errors.As failed to find UnavailableError through wrapping") + } + if target.Unwrap() != cause { + t.Errorf("unwrapped cause = %v, want %v", target.Unwrap(), cause) + } +} + +// --- Helpers --- + +type testLogger struct { + mu sync.Mutex + infoCount int + warnCount int +} + +func (l *testLogger) Info(msg string, keyvals ...any) { + l.mu.Lock() + l.infoCount++ + l.mu.Unlock() +} + +func (l *testLogger) Warn(msg string, keyvals ...any) { + l.mu.Lock() + l.warnCount++ + l.mu.Unlock() +} diff --git a/internal/memory/proxy.go b/internal/memory/proxy.go index 126debe..66b1859 100644 --- a/internal/memory/proxy.go +++ b/internal/memory/proxy.go @@ -1,102 +1,49 @@ // Package memory provides a Dewey HTTP proxy client for semantic memory operations. // // The hivemind_store and hivemind_find tools proxy to Dewey's semantic search -// endpoints via JSON-RPC 2.0 over HTTP. Six secondary tools return deprecation -// messages pointing users to native Dewey tools. +// endpoints via MCP Streamable HTTP transport. Six secondary tools return +// deprecation messages pointing users to native Dewey tools. // // On connection failure, errors include a structured "DEWEY_UNAVAILABLE" code // so agents can degrade gracefully. package memory import ( - "bytes" "encoding/json" + "errors" "fmt" - "io" - "net/http" "time" + + "github.com/unbound-force/replicator/internal/mcpclient" ) -// Client is a Dewey HTTP proxy that forwards JSON-RPC calls. +// Client is a Dewey HTTP proxy that forwards MCP tools/call requests. type Client struct { - url string - http *http.Client + mcp *mcpclient.Client } // NewClient creates a Dewey proxy client with a 10-second timeout. func NewClient(deweyURL string) *Client { return &Client{ - url: deweyURL, - http: &http.Client{ + mcp: mcpclient.New(deweyURL, mcpclient.Config{ + Name: "replicator-memory", + // Version identifies this client in the MCP initialize handshake, + // not the replicator binary version. + Version: "1.0.0", Timeout: 10 * time.Second, - }, + }), } } -// jsonRPCRequest is a JSON-RPC 2.0 request envelope. -type jsonRPCRequest struct { - JSONRPC string `json:"jsonrpc"` - Method string `json:"method"` - Params any `json:"params"` - ID int `json:"id"` -} - -// jsonRPCResponse is a JSON-RPC 2.0 response envelope. -type jsonRPCResponse struct { - JSONRPC string `json:"jsonrpc"` - Result json.RawMessage `json:"result,omitempty"` - Error *jsonRPCError `json:"error,omitempty"` - ID int `json:"id"` -} - -// jsonRPCError is a JSON-RPC 2.0 error object. -type jsonRPCError struct { - Code int `json:"code"` - Message string `json:"message"` -} - -// Call sends a JSON-RPC 2.0 POST to the Dewey endpoint. -// Returns the result field on success, or a structured error on failure. +// Call sends an MCP tools/call request to the Dewey endpoint. +// The method parameter is the Dewey tool name (e.g., "dewey_health"). +// Returns the tool result as raw JSON, or a structured error on failure. func (c *Client) Call(method string, params any) (json.RawMessage, error) { - reqBody := jsonRPCRequest{ - JSONRPC: "2.0", - Method: method, - Params: params, - ID: 1, - } - - body, err := json.Marshal(reqBody) - if err != nil { - return nil, fmt.Errorf("marshal request: %w", err) - } - - resp, err := c.http.Post(c.url, "application/json", bytes.NewReader(body)) - if err != nil { - return nil, &UnavailableError{Cause: err} - } - defer resp.Body.Close() - - respBody, err := io.ReadAll(resp.Body) + result, err := c.mcp.Call(method, params) if err != nil { - return nil, fmt.Errorf("read response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - return nil, &UnavailableError{ - Cause: fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(respBody)), - } + return nil, wrapError(err) } - - var rpcResp jsonRPCResponse - if err := json.Unmarshal(respBody, &rpcResp); err != nil { - return nil, fmt.Errorf("unmarshal response: %w", err) - } - - if rpcResp.Error != nil { - return nil, fmt.Errorf("dewey error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) - } - - return rpcResp.Result, nil + return result, nil } // Health pings the Dewey endpoint to verify connectivity. @@ -183,3 +130,13 @@ func UnavailableResponse(err error) string { out, _ := json.MarshalIndent(resp, "", " ") return string(out) } + +// wrapError converts mcpclient.UnavailableError into memory.UnavailableError +// for backward compatibility with callers that check for *memory.UnavailableError. +func wrapError(err error) error { + var mcpUnavail *mcpclient.UnavailableError + if errors.As(err, &mcpUnavail) { + return &UnavailableError{Cause: mcpUnavail.Cause} + } + return err +} diff --git a/internal/memory/proxy_test.go b/internal/memory/proxy_test.go index 9732c81..c22e758 100644 --- a/internal/memory/proxy_test.go +++ b/internal/memory/proxy_test.go @@ -3,46 +3,129 @@ package memory import ( "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" + "strings" + "sync" "testing" - "time" ) -// newTestServer creates an httptest server that responds to JSON-RPC calls. -// The handler function receives the method and params and returns a result. -func newTestServer(t *testing.T, handler func(method string, params json.RawMessage) (any, *jsonRPCError)) *httptest.Server { - t.Helper() - return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - var req jsonRPCRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - http.Error(w, "bad request", http.StatusBadRequest) +// mcpMockHandler is a stateful mock MCP server for testing the memory proxy. +// It simulates MCP Streamable HTTP transport: initialize handshake, session ID, +// tools/call envelope validation, and SSE response format. +type mcpMockHandler struct { + t *testing.T + + mu sync.Mutex + initialized bool + + // toolHandler receives the tool name and raw arguments, returns a result. + toolHandler func(name string, args json.RawMessage) (any, error) +} + +func (h *mcpMockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var req map[string]any + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "bad request body", http.StatusBadRequest) + return + } + + method, _ := req["method"].(string) + + switch method { + case "initialize": + h.handleInitialize(w, req) + case "tools/call": + h.handleToolsCall(w, req) + default: + // Reject bare JSON-RPC methods (non-MCP) with 400. + http.Error(w, fmt.Sprintf("unknown method %q: MCP requires initialize + tools/call", method), http.StatusBadRequest) + } +} + +func (h *mcpMockHandler) handleInitialize(w http.ResponseWriter, req map[string]any) { + h.mu.Lock() + h.initialized = true + h.mu.Unlock() + + result := map[string]any{ + "protocolVersion": "2025-03-26", + "capabilities": map[string]any{}, + "serverInfo": map[string]any{ + "name": "mock-dewey", + "version": "1.0.0", + }, + } + + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": result, + } + + w.Header().Set("Mcp-Session-Id", "mock-session-id") + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) +} + +func (h *mcpMockHandler) handleToolsCall(w http.ResponseWriter, req map[string]any) { + params, _ := req["params"].(map[string]any) + toolName, _ := params["name"].(string) + argsRaw, _ := json.Marshal(params["arguments"]) + + var toolResult any + if h.toolHandler != nil { + var err error + toolResult, err = h.toolHandler(toolName, argsRaw) + if err != nil { + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "error": map[string]any{ + "code": -32600, + "message": err.Error(), + }, + } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) return } + } else { + toolResult = map[string]string{"status": "ok"} + } - paramsBytes, _ := json.Marshal(req.Params) - result, rpcErr := handler(req.Method, paramsBytes) + resultJSON, _ := json.Marshal(toolResult) + rpcResp := map[string]any{ + "jsonrpc": "2.0", + "id": req["id"], + "result": map[string]any{ + "content": []map[string]any{ + {"type": "text", "text": string(resultJSON)}, + }, + }, + } - resp := jsonRPCResponse{ - JSONRPC: "2.0", - ID: req.ID, - } - if rpcErr != nil { - resp.Error = rpcErr - } else { - resp.Result, _ = json.Marshal(result) - } + data, _ := json.Marshal(rpcResp) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprintf(w, "event: message\ndata: %s\n\n", data) +} - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(resp) - })) +// newMCPTestServer creates a test server with the MCP mock handler. +func newMCPTestServer(t *testing.T, handler func(name string, args json.RawMessage) (any, error)) *httptest.Server { + t.Helper() + h := &mcpMockHandler{t: t, toolHandler: handler} + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + return srv } func TestCall_Success(t *testing.T) { - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { return map[string]string{"status": "ok"}, nil }) - defer srv.Close() client := NewClient(srv.URL) result, err := client.Call("test_method", map[string]string{"key": "value"}) @@ -60,18 +143,17 @@ func TestCall_Success(t *testing.T) { } func TestCall_RPCError(t *testing.T) { - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - return nil, &jsonRPCError{Code: -32600, Message: "invalid request"} + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + return nil, fmt.Errorf("invalid request") }) - defer srv.Close() client := NewClient(srv.URL) _, err := client.Call("test_method", nil) if err == nil { t.Fatal("expected error for RPC error response") } - if got := err.Error(); got != "dewey error -32600: invalid request" { - t.Errorf("error = %q, want dewey error message", got) + if !strings.Contains(err.Error(), "invalid request") { + t.Errorf("error = %q, should mention invalid request", err.Error()) } } @@ -108,13 +190,12 @@ func TestCall_HTTPError(t *testing.T) { } func TestHealth_Success(t *testing.T) { - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - if method != "dewey_health" { - t.Errorf("method = %q, want %q", method, "dewey_health") + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + if name != "dewey_health" { + t.Errorf("tool name = %q, want %q", name, "dewey_health") } return map[string]string{"status": "healthy"}, nil }) - defer srv.Close() client := NewClient(srv.URL) if err := client.Health(); err != nil { @@ -131,13 +212,13 @@ func TestHealth_Failure(t *testing.T) { } func TestStore_Success(t *testing.T) { - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - if method != "store_learning" { - t.Errorf("method = %q, want %q", method, "store_learning") + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + if name != "store_learning" { + t.Errorf("tool name = %q, want %q", name, "store_learning") } var p map[string]string - json.Unmarshal(params, &p) + json.Unmarshal(args, &p) if p["information"] != "test learning" { t.Errorf("information = %q, want %q", p["information"], "test learning") } @@ -147,7 +228,6 @@ func TestStore_Success(t *testing.T) { return map[string]any{"id": "mem-123", "stored": true}, nil }) - defer srv.Close() client := NewClient(srv.URL) result, err := client.Store("test learning", "go,testing") @@ -165,13 +245,12 @@ func TestStore_Success(t *testing.T) { } func TestStore_NoTags(t *testing.T) { - var receivedParams map[string]any + var receivedArgs json.RawMessage - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - json.Unmarshal(params, &receivedParams) + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + receivedArgs = args return map[string]any{"stored": true}, nil }) - defer srv.Close() client := NewClient(srv.URL) _, err := client.Store("info only", "") @@ -179,7 +258,9 @@ func TestStore_NoTags(t *testing.T) { t.Fatalf("Store: %v", err) } - if _, hasTags := receivedParams["tags"]; hasTags { + var params map[string]any + json.Unmarshal(receivedArgs, ¶ms) + if _, hasTags := params["tags"]; hasTags { t.Error("tags should not be sent when empty") } } @@ -198,13 +279,13 @@ func TestStore_DeweyUnavailable(t *testing.T) { } func TestFind_Success(t *testing.T) { - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - if method != "semantic_search" { - t.Errorf("method = %q, want %q", method, "semantic_search") + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + if name != "semantic_search" { + t.Errorf("tool name = %q, want %q", name, "semantic_search") } var p map[string]any - json.Unmarshal(params, &p) + json.Unmarshal(args, &p) if p["query"] != "test query" { t.Errorf("query = %v, want %q", p["query"], "test query") } @@ -215,7 +296,6 @@ func TestFind_Success(t *testing.T) { }, }, nil }) - defer srv.Close() client := NewClient(srv.URL) result, err := client.Find("test query", "", 5) @@ -229,13 +309,12 @@ func TestFind_Success(t *testing.T) { } func TestFind_WithCollection(t *testing.T) { - var receivedParams map[string]any + var receivedArgs json.RawMessage - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - json.Unmarshal(params, &receivedParams) + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + receivedArgs = args return map[string]any{"results": []any{}}, nil }) - defer srv.Close() client := NewClient(srv.URL) _, err := client.Find("query", "learnings", 10) @@ -243,19 +322,20 @@ func TestFind_WithCollection(t *testing.T) { t.Fatalf("Find: %v", err) } - if receivedParams["source_type"] != "learnings" { - t.Errorf("source_type = %v, want %q", receivedParams["source_type"], "learnings") + var params map[string]any + json.Unmarshal(receivedArgs, ¶ms) + if params["source_type"] != "learnings" { + t.Errorf("source_type = %v, want %q", params["source_type"], "learnings") } } func TestFind_WithLimit(t *testing.T) { - var receivedParams map[string]any + var receivedArgs json.RawMessage - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - json.Unmarshal(params, &receivedParams) + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + receivedArgs = args return map[string]any{"results": []any{}}, nil }) - defer srv.Close() client := NewClient(srv.URL) _, err := client.Find("query", "", 7) @@ -263,20 +343,21 @@ func TestFind_WithLimit(t *testing.T) { t.Fatalf("Find: %v", err) } + var params map[string]any + json.Unmarshal(receivedArgs, ¶ms) // JSON numbers unmarshal as float64. - if receivedParams["limit"] != float64(7) { - t.Errorf("limit = %v, want 7", receivedParams["limit"]) + if params["limit"] != float64(7) { + t.Errorf("limit = %v, want 7", params["limit"]) } } func TestFind_ZeroLimit(t *testing.T) { - var receivedParams map[string]any + var receivedArgs json.RawMessage - srv := newTestServer(t, func(method string, params json.RawMessage) (any, *jsonRPCError) { - json.Unmarshal(params, &receivedParams) + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + receivedArgs = args return map[string]any{"results": []any{}}, nil }) - defer srv.Close() client := NewClient(srv.URL) _, err := client.Find("query", "", 0) @@ -284,7 +365,9 @@ func TestFind_ZeroLimit(t *testing.T) { t.Fatalf("Find: %v", err) } - if _, hasLimit := receivedParams["limit"]; hasLimit { + var params map[string]any + json.Unmarshal(receivedArgs, ¶ms) + if _, hasLimit := params["limit"]; hasLimit { t.Error("limit should not be sent when zero") } } @@ -311,9 +394,30 @@ func TestUnavailableResponse(t *testing.T) { } } -func TestNewClient_Timeout(t *testing.T) { - client := NewClient("http://example.com") - if client.http.Timeout != 10*time.Second { - t.Errorf("timeout = %v, want 10s", client.http.Timeout) +// TestCall_RejectsBareMethods verifies that the mock MCP server rejects bare +// JSON-RPC methods (the old behavior) and the new MCP transport succeeds. +// This is the regression test for issue #19. +func TestCall_RejectsBareMethods(t *testing.T) { + srv := newMCPTestServer(t, func(name string, args json.RawMessage) (any, error) { + return map[string]string{"status": "ok"}, nil + }) + + // The new client wraps in tools/call, so it should succeed. + client := NewClient(srv.URL) + _, err := client.Call("dewey_health", map[string]any{}) + if err != nil { + t.Fatalf("MCP client should succeed: %v", err) + } + + // Verify the mock rejects bare methods by sending a raw non-MCP request. + bareReq := `{"jsonrpc":"2.0","method":"dewey_health","params":{},"id":1}` + resp, err := http.Post(srv.URL, "application/json", strings.NewReader(bareReq)) + if err != nil { + t.Fatalf("bare request: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("bare method should be rejected with 400, got %d", resp.StatusCode) } } diff --git a/openspec/changes/fix-memory-client-mcp-transport/design.md b/openspec/changes/fix-memory-client-mcp-transport/design.md new file mode 100644 index 0000000..828c23d --- /dev/null +++ b/openspec/changes/fix-memory-client-mcp-transport/design.md @@ -0,0 +1,107 @@ +## Context + +`memory.Client.Call()` sends bare `http.Post()` to Dewey's MCP endpoint, which expects the MCP Streamable HTTP transport. The doctor health check (`internal/doctor/checks.go:deweyHealthProbe()`) was previously fixed and correctly speaks MCP, proving the pattern. The proxy client was not updated, causing all memory proxy tools to fail with HTTP 400. + +The MCP Streamable HTTP transport requires: +- `Accept: application/json, text/event-stream` header on all requests +- An `initialize` handshake before any `tools/call` invocations +- `Mcp-Session-Id` header management from initialize response +- Tool invocations wrapped in `tools/call` JSON-RPC method with `name` and `arguments` params +- SSE response parsing (`event: message\ndata: {json}\n\n`) + +## Goals / Non-Goals + +### Goals +- Make `memory.Client.Call()` speak MCP Streamable HTTP transport +- Manage MCP session lifecycle (initialize on first use, cache session ID) +- Parse both SSE and plain JSON responses correctly +- Wrap tool names in `tools/call` method envelope +- Preserve graceful degradation (`DEWEY_UNAVAILABLE` semantics) +- Extract shared MCP transport logic usable by both doctor and memory client +- Update tests to use MCP-compatible mock handlers +- Ensure thread safety for concurrent access to shared session state + +### Non-Goals +- Full MCP client library (only implement what's needed for `Call()`) +- Bidirectional SSE streaming (only request-response pattern) +- MCP `initialized` notification (Dewey does not enforce it; see spec note) +- MCP notifications or progress tokens +- Changing the `NewClient()` API or tool handler signatures +- `context.Context` propagation (existing codebase has zero context usage; would require API change) + +## Decisions + +### D1: Extract shared MCP client package as `internal/mcpclient/` + +**Decision**: Extract the MCP Streamable HTTP transport logic into a shared package called `internal/mcpclient/` (not `internal/mcphttp/`). + +**Rationale**: The project already has `internal/mcp/` which is the MCP JSON-RPC **server**. Naming the new package `mcphttp` would be ambiguous — it doesn't convey whether it's a client or server, and could be confused with an HTTP transport layer for the existing server. `mcpclient` clearly signals "MCP client" vs the existing `internal/mcp/` (server). The package contains a `Client` type that speaks MCP Streamable HTTP to external services like Dewey. + +**Constitution alignment**: Composability First — the shared helper is independently testable and usable by any component that needs to speak MCP Streamable HTTP. + +### D2: Lazy session initialization with concurrency safety + +**Decision**: Initialize the MCP session on the first `Call()` invocation, not at `NewClient()` construction time. Use `sync.Mutex` to serialize initialization and protect session state. + +**Rationale**: `NewClient()` is called at server startup (`cmd/replicator/serve.go:45`). Making it lazy means the server starts fast even when Dewey is down. Failed initialization returns `UnavailableError` (existing behavior preserved). Re-initialization on subsequent calls if session was never established. + +The `mcpclient.Client` is a shared instance used by all tool handlers. While the current MCP server processes requests sequentially (`server.go:103`), the shared package must be safe for concurrent use by design — future server changes or other consumers may introduce concurrency. Session initialization uses a mutex to ensure exactly one goroutine performs the `initialize` handshake. + +**Constitution alignment**: Composability First — the binary works standalone even when Dewey is unreachable. Testability — tests run with `-race` flag, so all shared state must be synchronized. + +**Latency note**: First-call latency includes the `initialize` round-trip overhead (~5ms for localhost Dewey, up to 100-500ms for remote endpoints). Session recovery also incurs this overhead. + +### D3: `tools/call` envelope wrapping + +**Decision**: `Call()` continues to accept a tool name (e.g., `"dewey_health"`) as its `method` parameter, but internally wraps it in the MCP `tools/call` JSON-RPC method with `{name: toolName, arguments: params}` as the params object. + +**Rationale**: This is transparent to callers — `Health()`, `Store()`, and `Find()` don't need to change. The MCP envelope is an internal transport detail. + +### D4: Dual-format response parsing (SSE + plain JSON) + +**Decision**: Handle both `text/event-stream` (SSE) and `application/json` responses. For SSE, use the same line-scanning approach as `deweyHealthProbe()` — read the full response body, split on newlines, find `data:` prefixed lines (with or without trailing space), unmarshal the JSON. For plain JSON, parse the body directly as JSON-RPC. + +**Rationale**: The MCP Streamable HTTP spec allows servers to respond with either content type. The `Accept` header includes both. The `initialize` response may come as plain JSON. Handling both formats is necessary for correct protocol implementation. + +The response body read is bounded via `io.LimitReader` (10MB limit) to prevent unbounded memory consumption. + +### D5: `Mcp-Session-Id` management + +**Decision**: Capture the `Mcp-Session-Id` header from the `initialize` response and attach it to all subsequent requests. If the header is absent, proceed without a session ID. + +**Rationale**: Required by the MCP Streamable HTTP spec. Without it, the server may reject follow-up requests or create new sessions per request. However, not all MCP servers set this header, so its absence is not an error. + +### D6: Configurable client identity and timeout + +**Decision**: The `mcpclient.Client` constructor accepts a `Config` struct with `Name` (client identity for `clientInfo.name`), `Version` (for `clientInfo.version`), and `Timeout` (per-request HTTP timeout, default 10s). + +**Rationale**: The shared package serves multiple consumers with different identities: `"replicator-memory"` for the proxy client, `"replicator-doctor"` for the health probe. The timeout must be configurable because the doctor uses 5s while the memory client uses 10s. + +### D7: Optional structured logging + +**Decision**: The `mcpclient.Client` optionally accepts a logger (compatible with `charmbracelet/log`) for session lifecycle events. If no logger is provided, the client operates silently. + +**Rationale**: Observability for session lifecycle (initialization, recovery, failures) is important for operators diagnosing Dewey connectivity issues. The logger is optional to preserve the existing silent behavior for consumers that don't need it. + +## Risks / Trade-offs + +### Session stale/expired + +If the MCP session expires (Dewey restart, timeout), subsequent calls will fail. **Mitigation**: On HTTP 400/404 from a `tools/call`, reset session state and retry once with a fresh `initialize`. This is a simple retry, not a complex reconnect loop. Session recovery is serialized with concurrent access via mutex. + +### Timeout budget for multi-step operations + +A single `Call()` can involve up to 4 HTTP round-trips in the worst case: `initialize` + `tools/call` + retry-`initialize` + retry-`tools/call`. With a 10-second per-request timeout, the worst-case wall-clock time is 40 seconds. **Accepted**: This is a pathological case (Dewey responding at exactly the timeout boundary). In practice, failures are fast (connection refused, HTTP 400). The per-request timeout (not per-`Call()` timeout) is simpler to implement and reason about. A future enhancement could add an overall deadline via `context.Context`. + +### Single-event assumption + +The SSE parser uses the first `data:` line per response. If Dewey starts streaming multi-event responses, the parser will only use the first one. **Accepted**: The current `tools/call` pattern returns single results. If streaming is needed later, this would be a new feature, not a fix for this bug. + +### Doctor code duplication during transition + +The doctor's `deweyHealthProbe()` will initially remain as-is. Task 3 migrates it to use the shared `internal/mcpclient/` package, resolving the duplication. If Task 3 is deferred (it is parallel-eligible), a tracking issue MUST be created before the PR is merged to ensure the migration happens in the next change cycle. + +### No `initialized` notification + +The MCP protocol requires an `initialized` notification after the `initialize` response. This client skips it because: (a) Dewey does not currently enforce it, (b) the doctor's working implementation also skips it, (c) it falls under the "full MCP client library" non-goal. If a future Dewey update requires it, the client will fail gracefully with `UnavailableError`. + diff --git a/openspec/changes/fix-memory-client-mcp-transport/proposal.md b/openspec/changes/fix-memory-client-mcp-transport/proposal.md new file mode 100644 index 0000000..d7167ee --- /dev/null +++ b/openspec/changes/fix-memory-client-mcp-transport/proposal.md @@ -0,0 +1,70 @@ +## Why + +`memory.Client.Call()` in `internal/memory/proxy.go` sends bare `http.Post()` to the Dewey MCP endpoint. The MCP Streamable HTTP transport requires specific headers, session lifecycle management, SSE response parsing, and a `tools/call` method envelope. Without these, Dewey returns HTTP 400 and all memory proxy tools (`hivemind_store`, `hivemind_find`, and all deprecated stubs) fail with `DEWEY_UNAVAILABLE`. + +This was identified as issue [#19](https://github.com/unbound-force/replicator/issues/19). The doctor health check (`internal/doctor/checks.go`) was already fixed in a prior change (`fix-dewey-doctor-check`) and correctly speaks MCP Streamable HTTP, proving the pattern works. The proxy client was not updated at that time. + +## What Changes + +Rewrite `memory.Client` to speak MCP Streamable HTTP transport: + +1. **Session lifecycle** — Send an `initialize` handshake on first use, capture `Mcp-Session-Id` from response headers, attach it to subsequent requests. +2. **Request formatting** — Set `Accept: application/json, text/event-stream` header. Wrap tool calls in the `tools/call` JSON-RPC method with tool name and arguments in the params envelope. +3. **SSE response parsing** — Parse `text/event-stream` responses by scanning for `data: ` prefixed lines and extracting the JSON-RPC result. +4. **Graceful degradation** — Preserve existing `DEWEY_UNAVAILABLE` error semantics. Session initialization failures degrade gracefully (no panic, no crash). +5. **Test updates** — Update `proxy_test.go` mock handlers to simulate MCP Streamable HTTP responses (SSE format, session headers). + +## Capabilities + +### New Capabilities +- `mcp-session-management`: Client maintains MCP session state (`Mcp-Session-Id`) across calls within a process lifetime, with concurrency-safe access +- `sse-response-parsing`: Client correctly parses both SSE-formatted and plain JSON responses from MCP Streamable HTTP endpoints +- `mcpclient-package`: Shared `internal/mcpclient/` package reusable by any component needing MCP Streamable HTTP transport + +### Modified Capabilities +- `memory.Client.Call()`: Switches from bare HTTP POST to full MCP Streamable HTTP transport (initialize + tools/call + SSE parsing). Delegates to new `mcpclient.Client` internally. +- `memory.Client.Health()`: Now works correctly against MCP-speaking Dewey endpoints (benefits from `Call()` transport fix — method signature and logic unchanged) +- `memory.Client.Store()`: Now works correctly against MCP-speaking Dewey endpoints (benefits from `Call()` transport fix — method signature and logic unchanged) +- `memory.Client.Find()`: Now works correctly against MCP-speaking Dewey endpoints (benefits from `Call()` transport fix — method signature and logic unchanged) + +### Removed Capabilities +- None + +## Impact + +- **`internal/mcpclient/`**: New shared package providing MCP Streamable HTTP client (`Client` type) with session lifecycle, SSE/JSON parsing, concurrency safety, configurable timeout/identity. +- **`internal/memory/proxy.go`**: Rewrite of `Call()` method to delegate to `mcpclient.Client`. `NewClient()` constructs an `mcpclient.Client` internally. Method signatures unchanged. +- **`internal/memory/proxy_test.go`**: Test mock handlers must simulate MCP Streamable HTTP (SSE responses, session headers, initialize handshake). +- **`internal/doctor/checks.go`**: Migrate `deweyHealthProbe()` to use `mcpclient.Client` instead of inline MCP implementation (removes duplication). +- **`cmd/replicator/serve.go`**: No changes expected — `NewClient()` signature stays the same. +- **`AGENTS.md`**: Update project structure to include `internal/mcpclient/` package. +- **Downstream tools**: All memory tools (`hivemind_store`, `hivemind_find`, deprecated stubs) benefit automatically since they proxy through `memory.Client`. + +## Constitution Alignment + +Assessed against the Unbound Force org constitution. + +### I. Autonomous Collaboration + +**Assessment**: PASS + +The memory proxy tools remain independently callable via MCP. The `Client` produces self-describing JSON responses. Inter-agent communication patterns are unaffected — only the transport between replicator and Dewey changes. + +### II. Composability First + +**Assessment**: PASS + +The binary continues to work standalone. Dewey integration degrades gracefully — when Dewey is unavailable or the MCP handshake fails, tools return `DEWEY_UNAVAILABLE` errors (existing behavior preserved). No new mandatory dependencies are introduced. + +### III. Observable Quality + +**Assessment**: PASS + +All tool responses remain JSON. The MCP transport change is internal to `memory.Client` — response shapes visible to callers are unchanged and continue to match the TypeScript version's structure. + +### IV. Testability + +**Assessment**: PASS + +Tests use `httptest.NewServer` with mock MCP handlers. No external services are required. The mock handlers simulate MCP Streamable HTTP responses (SSE format, session headers) to verify the full transport pipeline in isolation. + diff --git a/openspec/changes/fix-memory-client-mcp-transport/specs/mcp-transport.md b/openspec/changes/fix-memory-client-mcp-transport/specs/mcp-transport.md new file mode 100644 index 0000000..6a2c5ca --- /dev/null +++ b/openspec/changes/fix-memory-client-mcp-transport/specs/mcp-transport.md @@ -0,0 +1,239 @@ +## ADDED Requirements + +### Requirement: MCP Session Initialization + +The `mcpclient.Client` MUST send an MCP `initialize` handshake on first use before any `tools/call` invocations. The initialize request MUST include: +- `"method": "initialize"` +- `"params.protocolVersion": "2025-03-26"` (current MCP Streamable HTTP spec version) +- `"params.clientInfo.name"`: caller-provided name (e.g., `"replicator-memory"`, `"replicator-doctor"`) +- `"params.clientInfo.version"`: caller-provided version (e.g., `"1.0.0"`) +- `"params.capabilities": {}` + +The client MUST capture the `Mcp-Session-Id` response header and attach it to all subsequent requests. + +> **Note**: The MCP protocol requires an `initialized` notification after the `initialize` response. This client does NOT send the `initialized` notification — Dewey does not currently enforce it. This is a known deviation, accepted per the non-goal of "full MCP client library." If a future Dewey update requires the notification, session initialization will fail and the client will return `UnavailableError` (graceful degradation). + +#### Scenario: First call triggers initialization + +- **GIVEN** a `memory.Client` that has not yet been initialized +- **WHEN** `Call("dewey_health", {})` is invoked +- **THEN** the client MUST first send an `initialize` request to the Dewey endpoint +- **AND** capture the `Mcp-Session-Id` from the response headers +- **AND** then send the `tools/call` request with the session ID attached + +#### Scenario: Subsequent calls reuse session + +- **GIVEN** a `memory.Client` that has completed initialization +- **WHEN** `Call("store_learning", params)` is invoked +- **THEN** the client MUST NOT send another `initialize` request +- **AND** MUST include the cached `Mcp-Session-Id` header + +#### Scenario: Initialize returns HTTP error + +- **GIVEN** a Dewey endpoint that returns HTTP 500 on `initialize` +- **WHEN** `Call()` is invoked +- **THEN** it MUST return an `UnavailableError` +- **AND** the error message MUST include context about the initialization failure + +#### Scenario: Initialize response missing session header + +- **GIVEN** a Dewey endpoint that returns a valid `initialize` response but no `Mcp-Session-Id` header +- **WHEN** `Call()` is invoked +- **THEN** the client MUST proceed without a session ID (Dewey may not require it) +- **AND** subsequent requests MUST omit the `Mcp-Session-Id` header + +### Requirement: Concurrency Safety + +The `mcpclient.Client` MUST be safe for concurrent use by multiple goroutines. Session initialization MUST be serialized — if multiple goroutines call `Call()` concurrently on an uninitialized client, exactly one MUST perform the `initialize` handshake and all others MUST wait for it to complete. + +The `Mcp-Session-Id` field MUST be protected against concurrent read/write access (e.g., via `sync.Mutex` or `sync.RWMutex`). + +Session recovery (reset + re-initialize on HTTP 400/404) MUST also be serialized. If multiple goroutines detect session failure concurrently, only one MUST perform the re-initialization. Other goroutines MUST wait for recovery to complete and then retry with the new session. + +#### Scenario: Concurrent first calls safely initialize once + +- **GIVEN** a `mcpclient.Client` that has not yet been initialized +- **WHEN** 10 goroutines call `Call()` concurrently +- **THEN** exactly one `initialize` request MUST be sent to Dewey +- **AND** all goroutines MUST receive valid responses +- **AND** the test MUST pass under `-race` + +### Requirement: MCP Request Headers + +All requests to the Dewey endpoint MUST include: +- `Content-Type: application/json` +- `Accept: application/json, text/event-stream` + +#### Scenario: Correct headers on initialize + +- **GIVEN** a `mcpclient.Client` sending an initialize request +- **WHEN** the HTTP request is constructed +- **THEN** the `Accept` header MUST be `application/json, text/event-stream` +- **AND** the `Content-Type` header MUST be `application/json` + +### Requirement: tools/call Envelope Wrapping + +The `Call()` method MUST wrap the tool name and arguments in an MCP `tools/call` JSON-RPC method. The request body MUST have: +- `"method": "tools/call"` +- `"params.name": ` +- `"params.arguments": ` + +JSON-RPC request IDs SHOULD be monotonically increasing within a session to aid debugging. The `mcpclient.Client` maintains an internal counter. + +#### Scenario: Tool name wrapped in envelope + +- **GIVEN** a call to `Call("dewey_health", {})` +- **WHEN** the JSON-RPC request body is marshalled +- **THEN** the `method` field MUST be `"tools/call"` +- **AND** `params.name` MUST be `"dewey_health"` +- **AND** `params.arguments` MUST be `{}` + +### Requirement: Response Parsing + +The client MUST handle both `application/json` and `text/event-stream` response content types from the MCP endpoint. + +**For `text/event-stream` (SSE) responses**, the parser MUST: +- Scan response lines for `data:` prefixes (with or without trailing space) +- Extract and unmarshal the JSON-RPC response from the first `data:` line +- For `tools/call` responses: return the `result.content[0].text` field as `json.RawMessage` +- For `initialize` responses: return the `result` object directly (different shape — has `protocolVersion`, `capabilities`, etc.) + +**For `application/json` responses**, the parser MUST: +- Parse the body as a direct JSON-RPC response without SSE unwrapping + +The response body read SHOULD be bounded (e.g., `io.LimitReader` with 10MB limit) to prevent unbounded memory consumption on malformed responses. + +#### Scenario: Successful SSE response + +- **GIVEN** a Dewey endpoint that returns `event: message\ndata: {"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"{\"status\":\"ok\"}"}]}}\n\n` +- **WHEN** the client parses the response +- **THEN** the result MUST be the JSON value `{"status":"ok"}` + +#### Scenario: SSE response with JSON-RPC error + +- **GIVEN** a Dewey endpoint that returns `event: message\ndata: {"jsonrpc":"2.0","id":2,"error":{"code":-32600,"message":"invalid request"}}\n\n` +- **WHEN** the client parses the response +- **THEN** `Call()` MUST return an error containing `"invalid request"` + +#### Scenario: Plain JSON response (non-SSE) + +- **GIVEN** a Dewey endpoint that returns `Content-Type: application/json` with body `{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-03-26"}}` +- **WHEN** the client parses the response +- **THEN** the result MUST be parsed as a direct JSON-RPC response + +#### Scenario: Empty SSE response body + +- **GIVEN** a Dewey endpoint that returns HTTP 200 with an empty body +- **WHEN** the client parses the response +- **THEN** `Call()` MUST return an error indicating no valid response was found + +#### Scenario: Malformed JSON in SSE data line + +- **GIVEN** a Dewey endpoint that returns `data: {not-valid-json}` +- **WHEN** the client parses the response +- **THEN** `Call()` MUST return an error with unmarshal context + +#### Scenario: SSE response with no data line + +- **GIVEN** a Dewey endpoint that returns `event: message\n\n` (no `data:` line) +- **WHEN** the client parses the response +- **THEN** `Call()` MUST return an error indicating no valid response was found + +#### Scenario: Empty content array in tools/call response + +- **GIVEN** a Dewey endpoint that returns a `tools/call` response with `"content": []` +- **WHEN** the client parses the response +- **THEN** `Call()` MUST return an error (not panic with index-out-of-bounds) + +### Requirement: Session Recovery on Failure + +If a `tools/call` request fails with HTTP 400 or HTTP 404, the client MUST reset session state and retry once with a fresh `initialize` handshake. + +Session recovery MUST be serialized with concurrent access (see Concurrency Safety requirement). + +#### Scenario: Session expired and recovered (HTTP 400) + +- **GIVEN** a `mcpclient.Client` with a cached session ID +- **WHEN** a `tools/call` request returns HTTP 400 +- **THEN** the client MUST clear the cached session ID +- **AND** re-initialize with a new `initialize` request +- **AND** retry the original `tools/call` once + +#### Scenario: Session expired and recovered (HTTP 404) + +- **GIVEN** a `mcpclient.Client` with a cached session ID +- **WHEN** a `tools/call` request returns HTTP 404 +- **THEN** the client MUST follow the same recovery procedure as HTTP 400 + +#### Scenario: Retry also fails + +- **GIVEN** a `mcpclient.Client` attempting session recovery +- **WHEN** the re-initialized `tools/call` also fails +- **THEN** the client MUST return an `UnavailableError` +- **AND** MUST NOT retry further + +### Requirement: Timeout Budget + +The `mcpclient.Client` MUST accept a configurable per-request HTTP timeout. The timeout applies to each individual HTTP request, not to the entire `Call()` operation. The combined `initialize` + `tools/call` sequence has a maximum wall-clock time of `2 * timeout` on first call (or `4 * timeout` during session recovery with retry). + +The default timeout SHOULD be 10 seconds (matching the existing `memory.Client` behavior). + +#### Scenario: Timeout on initialize + +- **GIVEN** a Dewey endpoint that does not respond within the timeout +- **WHEN** `Call()` is invoked on an uninitialized client +- **THEN** the client MUST return an `UnavailableError` after the timeout expires + +### Requirement: Observability + +The `mcpclient.Client` SHOULD accept an optional logger (compatible with `charmbracelet/log`) for structured logging at key lifecycle points: +- `INFO`: Session initialized successfully (Dewey URL) +- `WARN`: Session recovery triggered (HTTP status code) +- `WARN`: Session recovery failed (error detail) + +If no logger is provided, the client MUST operate silently (no stdout/stderr output). + +## MODIFIED Requirements + +### Requirement: Call() Method Transport + +`Call()` MUST use `http.NewRequest` with explicit `Accept` and `Content-Type` headers instead of `http.Post()`. + +Previously: `Call()` used `c.http.Post(c.url, "application/json", ...)` which did not set the `Accept` header required by MCP Streamable HTTP. + +### Requirement: Health Check via MCP + +`Health()` MUST work correctly against MCP-speaking Dewey endpoints. + +Previously: `Health()` called `Call("dewey_health", {})` which sent bare JSON-RPC without the MCP envelope, causing HTTP 400. + +## REMOVED Requirements + +None — no requirements are being removed. + +## Coverage Strategy + +The `internal/mcpclient/` package MUST achieve ≥80% line coverage. The following paths require dedicated test cases: + +| Path | Test Category | +|------|---------------| +| Successful initialize + tools/call | Happy path | +| Session reuse (no re-init) | Happy path | +| Initialize HTTP error (500) | Error path | +| Initialize missing session header | Edge case | +| tools/call envelope correctness | Contract | +| SSE response parsing (valid) | Happy path | +| SSE response with JSON-RPC error | Error path | +| Plain JSON response (non-SSE) | Edge case | +| Empty response body | Edge case | +| Malformed JSON in SSE | Error path | +| No data line in SSE | Edge case | +| Empty content array | Edge case | +| Session recovery on 400 | Recovery | +| Session recovery on 404 | Recovery | +| Retry failure → UnavailableError | Recovery | +| Concurrent initialization | Concurrency | +| Timeout on request | Error path | + +Each scenario in this spec MUST have at least one corresponding test function. + diff --git a/openspec/changes/fix-memory-client-mcp-transport/tasks.md b/openspec/changes/fix-memory-client-mcp-transport/tasks.md new file mode 100644 index 0000000..e36ba67 --- /dev/null +++ b/openspec/changes/fix-memory-client-mcp-transport/tasks.md @@ -0,0 +1,34 @@ + + +## 1. Create shared MCP client package + +- [x] 1.1 Create `internal/mcpclient/` package with `Client` type that handles MCP Streamable HTTP transport: `initialize` handshake, `Mcp-Session-Id` management, `tools/call` envelope wrapping, dual-format response parsing (SSE + plain JSON), session recovery on HTTP 400/404, concurrency safety via `sync.Mutex`, configurable timeout and client identity via `Config` struct, optional structured logging, and `io.LimitReader` bounds on response bodies. Following TDD: write failing tests from Task 1.2 first, then implement to make them pass. Files: `internal/mcpclient/client.go` +- [x] 1.2 Write tests for the MCP client using `httptest.NewServer` with a stateful mock MCP handler that tracks initialization state, returns SSE-formatted responses, simulates session headers, and supports error scenarios. Must cover all scenarios from the coverage strategy: happy paths, error paths, edge cases (empty body, malformed JSON, no data line, missing session header, plain JSON response, empty content array), session recovery (400 + 404), concurrency safety (10 goroutines, must pass under `-race`), and timeout behavior. Files: `internal/mcpclient/client_test.go` + +## 2. Integrate MCP client into memory proxy + +- [x] 2.1 Rewrite `memory.Client` to use `mcpclient.Client` for all calls. Update `Call()` to delegate to the MCP client. Update `NewClient()` to construct an `mcpclient.Client` internally with `Name: "replicator-memory"`, `Version: "1.0.0"`, `Timeout: 10s`. Keep `Health()`, `Store()`, and `Find()` signatures unchanged. Update GoDoc comment on `Call()` to reflect MCP transport semantics. Following TDD: write/update failing tests from Task 2.2 first. Files: `internal/memory/proxy.go` +- [x] 2.2 Update `proxy_test.go` mock handlers to simulate MCP Streamable HTTP responses (SSE format, session headers, `tools/call` envelope validation). Verify `Health()`, `Store()`, and `Find()` work end-to-end through the MCP transport. Include a regression scenario: mock that rejects bare JSON-RPC (non-MCP) requests, confirming the old behavior would fail and the new behavior succeeds. Files: `internal/memory/proxy_test.go` + +## 3. Migrate doctor to shared client + +- [x] 3.1 [P] Refactor `deweyHealthProbe()` in `internal/doctor/checks.go` to use `mcpclient.Client` with `Name: "replicator-doctor"`, `Version: "1.0.0"`, `Timeout: 5s` instead of its inline MCP implementation. Remove the duplicated SSE parsing, header management, and initialize logic. Verify existing doctor tests still pass. Files: `internal/doctor/checks.go` + +## 4. Verification + +- [x] 4.1 Run `make check` — all tests pass, go vet clean (includes parity tests) +- [x] 4.2 Run `make check-coverage` — coverage ratchets pass +- [x] 4.3 Verify constitution alignment: Autonomous Collaboration (tools remain independently callable), Composability First (graceful degradation preserved), Observable Quality (JSON response shapes unchanged), Testability (all tests use `httptest`, no external services, `-race` passes) +- [x] 4.4 Update `AGENTS.md` project structure to include `internal/mcpclient/` package description + +