oikos/cmd/nomos/mcp.go

package main

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"
)

// ─── MCP Streamable HTTP client ────────────────────────────────────────

type mcpClient struct {
	baseURL   string
	token     string // OIKOS_MCP_BEARER_TOKEN — api's combinedAuth requires it on every request (no dev-open bypass)
	sessionID string
	http      *http.Client
	nextID    int
	mu        sync.Mutex // one client serializes its own MCP calls (the pool gives each session its own client, so this never blocks another session)
	// toolsCache holds the last tools/list result. The tool list is static
	// for the lifetime of one MCP connection — it only changes when the api
	// process (re)registers tools, i.e. on a restart, which this client
	// already detects and reacts to via reconnectLocked. Without this,
	// buildTools (called at the start of EVERY chat turn, including every
	// auto-continuation resume) paid a full tools/list round-trip every
	// single time for a list that's almost always identical to the last one.
	// Guarded separately from mu (not reused) so a cache check never
	// contends with an in-flight doRequest call for a different method.
	toolsMu    sync.Mutex
	toolsCache []toolDef
}

func newMCPClient(baseURL, token string) (*mcpClient, error) {
	c := &mcpClient{
		baseURL: baseURL,
		token:   token,
		http:    &http.Client{Timeout: 120 * time.Second},
	}

	resp, err := c.doRequest("initialize", map[string]any{
		"protocolVersion": "2024-11-05",
		"capabilities":    map[string]any{},
		"clientInfo":      map[string]any{"name": "nomos", "version": "2.0"},
	})
	if err != nil {
		return nil, fmt.Errorf("initialize: %w", err)
	}
	if resp.sessionID == "" {
		return nil, fmt.Errorf("no session ID in initialize response")
	}
	c.sessionID = resp.sessionID

	c.doRequest("notifications/initialized", map[string]any{})

	slog.Info("nomos: mcp connected", "session", c.sessionID[:16]+"...")
	return c, nil
}

type mcpJSONRPCResponse struct {
	sessionID string
	Result    json.RawMessage `json:"result"`
	Error     json.RawMessage `json:"error"`
}

// errStaleSession signals that the MCP server rejected our session id (e.g.
// after an api/MCP restart), so the client should re-initialize and retry.
var errStaleSession = fmt.Errorf("mcp session stale")

// doRequest serializes MCP calls and transparently re-initializes the session
// if the server has forgotten it (common after an api redeploy), retrying the
// original call once. Without this, an api restart permanently breaks nomos
// until it is itself restarted.
func (c *mcpClient) doRequest(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	resp, err := c.send(method, params)
	if err != nil && method != "initialize" && errors.Is(err, errStaleSession) {
		slog.Warn("nomos: mcp session stale, reconnecting")
		if rerr := c.reconnectLocked(); rerr != nil {
			return nil, fmt.Errorf("mcp reconnect: %w (original: %v)", rerr, err)
		}
		return c.send(method, params)
	}
	return resp, err
}

// reconnectLocked re-initializes the MCP session. The caller must hold c.mu.
func (c *mcpClient) reconnectLocked() error {
	c.sessionID = ""
	// A reconnect means the api process was restarted (or forgot us) — its
	// tool registration may have changed, so the cached list is no longer
	// trustworthy.
	c.toolsMu.Lock()
	c.toolsCache = nil
	c.toolsMu.Unlock()
	resp, err := c.send("initialize", map[string]any{
		"protocolVersion": "2024-11-05",
		"capabilities":    map[string]any{},
		"clientInfo":      map[string]any{"name": "nomos", "version": "2.0"},
	})
	if err != nil {
		return err
	}
	if resp.sessionID == "" {
		return fmt.Errorf("no session ID on re-initialize")
	}
	c.sessionID = resp.sessionID
	_, _ = c.send("notifications/initialized", map[string]any{})
	slog.Info("nomos: mcp reconnected", "session", c.sessionID[:16]+"...")
	return nil
}

// send performs one MCP round-trip. It does not lock; callers hold c.mu.
func (c *mcpClient) send(method string, params map[string]any) (*mcpJSONRPCResponse, error) {
	c.nextID++
	body, _ := json.Marshal(map[string]any{
		"jsonrpc": "2.0",
		"method":  method,
		"params":  params,
		"id":      c.nextID,
	})

	req, err := http.NewRequest(http.MethodPost, c.baseURL, bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/json, text/event-stream")
	if c.sessionID != "" {
		req.Header.Set("Mcp-Session-Id", c.sessionID)
	}
	if c.token != "" {
		req.Header.Set("Authorization", "Bearer "+c.token)
	}

	resp, err := c.http.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	// A rejected/unknown session comes back as 4xx (commonly 400/404).
	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusBadRequest {
		return nil, errStaleSession
	}

	result := &mcpJSONRPCResponse{}
	result.sessionID = resp.Header.Get("Mcp-Session-Id")

	scanner := bufio.NewScanner(resp.Body)
	scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
	gotData := false
	for scanner.Scan() {
		line := scanner.Text()
		if strings.HasPrefix(line, "data: ") {
			gotData = true
			data := line[6:]
			if err := json.Unmarshal([]byte(data), result); err != nil {
				return nil, fmt.Errorf("parse response: %w", err)
			}
		}
	}

	if result.Error != nil {
		return nil, fmt.Errorf("rpc error: %s", string(result.Error))
	}

	// Empty body with no result and a session set: the server likely dropped
	// our session. Notifications legitimately return no data, so exempt them.
	if !gotData && result.Result == nil && method != "notifications/initialized" {
		return nil, errStaleSession
	}

	if result.sessionID != "" {
		c.sessionID = result.sessionID
	}

	return result, nil
}

func (c *mcpClient) callTool(name string, args map[string]any) (any, error) {
	resp, err := c.doRequest("tools/call", map[string]any{
		"name":      name,
		"arguments": args,
	})
	if err != nil {
		return nil, err
	}

	var toolResult struct {
		Content []struct {
			Type string `json:"type"`
			Text string `json:"text"`
		} `json:"content"`
	}
	if err := json.Unmarshal(resp.Result, &toolResult); err != nil {
		return string(resp.Result), nil
	}

	var texts []string
	for _, c := range toolResult.Content {
		if c.Type == "text" {
			var parsed any
			if json.Unmarshal([]byte(c.Text), &parsed) == nil {
				return parsed, nil
			}
			texts = append(texts, c.Text)
		}
	}
	if len(texts) == 1 {
		return texts[0], nil
	}
	return texts, nil
}

func (c *mcpClient) listTools() ([]string, error) {
	resp, err := c.doRequest("tools/list", map[string]any{})
	if err != nil {
		return nil, err
	}
	var tr struct {
		Tools []struct {
			Name        string `json:"name"`
			Description string `json:"description"`
		} `json:"tools"`
	}
	if err := json.Unmarshal(resp.Result, &tr); err != nil {
		return nil, err
	}
	var names []string
	for _, t := range tr.Tools {
		names = append(names, t.Name)
	}
	return names, nil
}

func (c *mcpClient) close() {
}

// ─── Per-session MCP client pool ────────────────────────────────────────
//
// A single shared mcpClient serializes EVERY tool call across EVERY
// concurrently-running task through one mutex (see mcpClient.mu) — `run`
// executes its SSH command synchronously inside that lock and is capped at
// up to 10 minutes, so one task mid-`run` stalled every other task's tool
// calls, even trivial reads, behind it. The MCP *server* has no per-
// connection state to protect (newServer in internal/mcp/server.go returns
// one shared *mcp.Server instance whose tool handlers close only over the DB
// pool, which is already safe for concurrent use) — the mutex existed purely
// because the *client* reused one stateful transport session, not because
// the server needed it. Giving each task's own session its own client
// removes the cross-task serialization entirely: a task's own tool calls
// stay sequential (which they already are — the agent loop calls tools one
// at a time within a turn), but no longer block anyone else's.
type mcpClientPool struct {
	baseURL string
	token   string
	mu      sync.Mutex
	clients map[string]*pooledMCPClient
}

type pooledMCPClient struct {
	client   *mcpClient
	lastUsed time.Time
}

func newMCPClientPool(baseURL, token string) *mcpClientPool {
	return &mcpClientPool{baseURL: baseURL, token: token, clients: make(map[string]*pooledMCPClient)}
}

// get returns the client for sessionID, creating and initializing one (a
// real MCP handshake) on first use. Session ids that don't identify a real
// persisted conversation ("" / "ephemeral", the no-DB-store path; "query",
// the structured /query endpoint) still get exactly one dedicated,
// reused client each via the same map — just keyed on a fixed string instead
// of a real session id — so that traffic doesn't pay a fresh handshake per
// request while still never sharing a connection with an actual task.
func (p *mcpClientPool) get(sessionID string) (*mcpClient, error) {
	key := sessionID
	if key == "" {
		key = "ephemeral"
	}

	p.mu.Lock()
	if pc, ok := p.clients[key]; ok {
		pc.lastUsed = time.Now()
		p.mu.Unlock()
		return pc.client, nil
	}
	p.mu.Unlock()

	// Initialize outside the lock — it's a network round-trip, and holding
	// the pool mutex for it would serialize unrelated sessions' first calls
	// behind each other, undermining the whole point of this pool.
	c, err := newMCPClient(p.baseURL, p.token)
	if err != nil {
		return nil, err
	}

	p.mu.Lock()
	// Another goroutine may have created one for the same key while we were
	// initializing (two of this session's tool calls racing on a cold
	// start); keep whichever won, close out the loser's connection (a no-op
	// today, but future-proof if mcpClient.close ever does real teardown).
	if existing, ok := p.clients[key]; ok {
		p.mu.Unlock()
		c.close()
		return existing.client, nil
	}
	p.clients[key] = &pooledMCPClient{client: c, lastUsed: time.Now()}
	p.mu.Unlock()
	return c, nil
}

// mcpClientIdleTimeout is how long an idle session's MCP client is kept
// before eviction — long enough to outlive a single slow `run` (capped at 10
// minutes server-side) plus normal think-time between a task's tool calls,
// short enough not to accumulate one abandoned connection per finished task
// forever.
const mcpClientIdleTimeout = 20 * time.Minute

// sweep evicts clients idle past mcpClientIdleTimeout. Call on a ticker.
func (p *mcpClientPool) sweep() {
	cutoff := time.Now().Add(-mcpClientIdleTimeout)
	p.mu.Lock()
	defer p.mu.Unlock()
	for key, pc := range p.clients {
		if pc.lastUsed.Before(cutoff) {
			pc.client.close()
			delete(p.clients, key)
		}
	}
}

func (p *mcpClientPool) closeAll() {
	p.mu.Lock()
	defer p.mu.Unlock()
	for key, pc := range p.clients {
		pc.client.close()
		delete(p.clients, key)
	}
}