Maven/cmd/mavend/mcp.go

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kami/maven/internal/config"
	"github.com/kami/maven/internal/ipc"
	"github.com/kami/maven/internal/mcp"
	"github.com/kami/maven/internal/store"
	"github.com/kami/maven/internal/webfetch"
)

// mcpRefreshInterval — how often the manager is asked to re-dial servers that
// are down. It is a tick, not a retry rate: mcp.Manager holds a per-server
// backoff that starts at DefaultReconnectEvery and doubles to
// MaxReconnectEvery, so a permanently misconfigured stdio server is not
// re-exec'd once a minute forever.
const mcpRefreshInterval = time.Minute

// mcpWiring — the MCP client, when the `mcp` block configures at least one
// enabled server. nil ⇒ nothing was configured, nothing is connected, and an
// allowlist row that happens to look like an MCP row refuses to run.
//
// It lives on the voice wiring because MCP tools ARE acts: they run through
// tool.Executor, the enabled allowlist and the confirm turn, which only exist
// on the voice/chat path. No voice surface ⇒ nothing that could call a tool.
type mcpWiring struct {
	mgr *mcp.Manager
	st  *store.Store
}

// wireMCP builds the manager. It does NOT dial: run does that, on its own
// goroutine, which is what makes "Maven starting is not contingent on someone
// else's process" true rather than merely intended.
//
// Dialing here used to be synchronous with a 30s budget, from wireVoice, from
// run. Connect dials serially and each HTTP dial is three requests against
// that server's timeout, so one black-holed endpoint cost 15s of boot and two
// cost the whole budget. On the passkey path wireVoice runs inside the unlock
// handler, so it delayed the answer to an unlock as well. Not failing and not
// blocking are different properties and only the first one held.
func wireMCP(cfg *config.Config, st *store.Store) *mcpWiring {
	servers := cfg.MCPServers()
	if len(servers) == 0 {
		return nil
	}
	limits := webfetch.Config{}
	if cfg.MCP != nil {
		limits.AllowHosts = cfg.MCP.AllowHosts
		limits.DenyHosts = cfg.MCP.DenyHosts
		limits.MaxBytes = cfg.MCP.MaxBytes
		limits.Timeout = time.Duration(cfg.MCP.Timeout)
		limits.HostInterval = time.Duration(cfg.MCP.HostInterval)
	}
	mgr, err := mcp.NewManager(mcp.WebfetchDoor(limits), servers)
	if err != nil {
		// Validation already ran in config.validate, so this is a programming
		// error rather than a config one. Still not fatal: MCP off is a working
		// Maven.
		log.Printf("mcp: not wired: %v", err)
		return nil
	}
	return &mcpWiring{mgr: mgr, st: st}
}

// connect dials every server and reconciles what came back. Called from run,
// under the daemon's context, so a shutdown during a slow dial is observed.
func (w *mcpWiring) connect(ctx context.Context) {
	if w == nil {
		return
	}
	w.mgr.Connect(ctx)
	w.propose(ctx)
}

// propose writes a 'proposed' allowlist row for every discovered tool, and
// reconciles the rows that already exist against what the server offers today.
// It does NOT enable anything: a configured server is a place Maven may look,
// not a capability she has. Kami enables what he wants on /tools, behind
// step-up, which is the same gate a shell tool goes through.
//
// Three things happen per discovered tool.
//
// A name not in the store becomes a proposal, carrying the tool's fingerprint.
//
// A name already in the store is reconciled against that fingerprint. A tool
// whose description, schema or readOnlyHint changed since it was approved drops
// back to 'proposed' and, if it stopped claiming read-only, to destructive=1.
// Insert-or-skip was not enough on its own: the cmd is a late-bound reference
// to a name the far end owns, so the server can redefine list_tasks into
// something that writes without the row changing at all.
//
// A row whose server is connected and no longer offers the tool is withdrawn.
func (w *mcpWiring) propose(ctx context.Context) {
	if w == nil {
		return
	}
	now := time.Now()
	fresh, changed := 0, 0
	seen := map[string]string{} // local name → "server/tool", for collisions
	for _, t := range w.mgr.Tools() {
		name := mcp.LocalName(t.Server, t.Name)
		remote := t.Server + "/" + t.Name
		// Two different tools can flatten to one local name: server "vik" with
		// tool "list_tasks" and server "vik_list" with tool "tasks" both give
		// "vik_list_tasks". The store keys rows by name, so the second would
		// land on the first one's row. Config-controlled and therefore rare,
		// but silently reusing a row is the wrong way to lose that race.
		if prev, dup := seen[name]; dup {
			log.Printf("mcp: %s and %s both map to the allowlist name %q — skipping the second, rename a server",
				prev, remote, name)
			continue
		}
		seen[name] = remote
		// No readOnlyHint ⇒ assume it mutates ⇒ the confirm turn. Being wrong
		// in this direction only costs a question.
		destructive := !t.ReadOnly
		provenance := fmt.Sprintf("mcp %s/%s", t.Server, t.Name)
		if t.Description != "" {
			provenance += ": " + t.Description
		}
		fp := mcp.Fingerprint(t)
		ok, err := w.st.ProposeMCPTool(ctx, name, mcp.Scope(t.Server),
			mcp.Cmd(t.Server, t.Name), destructive, provenance, fp, now)
		if err != nil {
			log.Printf("mcp: propose %s: %v", name, err)
			continue
		}
		if ok {
			fresh++
			continue
		}
		// The row already existed. Its provenance is whatever the server said
		// the first time; reconciling rewrites it, so what /tools shows is what
		// the server says now.
		ch, err := w.st.ReconcileMCPTool(ctx, name, fp, destructive, provenance, now)
		if err != nil {
			log.Printf("mcp: reconcile %s: %v", name, err)
			continue
		}
		if !ch.Changed {
			continue
		}
		changed++
		switch {
		case ch.Demoted && ch.Escalated:
			log.Printf("mcp: %s changed on the server and no longer claims read-only — disabled and marked destructive, re-approve it on /tools", name)
		case ch.Demoted:
			log.Printf("mcp: %s changed on the server since it was enabled — disabled, re-approve it on /tools", name)
		default:
			log.Printf("mcp: %s changed on the server; the proposal now shows the new description", name)
		}
	}
	w.withdrawGone(ctx, seen, now)
	if fresh > 0 {
		log.Printf("mcp: %d new tool proposal(s) waiting on /tools", fresh)
	}
	if changed > 0 {
		log.Printf("mcp: %d tool(s) changed since approval and need another look", changed)
	}
}

// withdrawGone disarms rows whose tool the server stopped offering. Only
// servers that are CONNECTED are considered: a tool missing because its server
// is down is not a tool that was withdrawn, and disabling a capability every
// time a process restarts would be worse than the problem.
func (w *mcpWiring) withdrawGone(ctx context.Context, seen map[string]string, now time.Time) {
	live := map[string]bool{}
	for _, name := range w.mgr.Connected() {
		live[name] = true
	}
	if len(live) == 0 {
		return
	}
	rows, err := w.st.ListTools(ctx, "")
	if err != nil {
		log.Printf("mcp: list tools: %v", err)
		return
	}
	for _, row := range rows {
		server, remote, ok := mcp.ParseCmd(row.Cmd)
		if !ok || !live[server] {
			continue
		}
		if _, still := seen[row.Name]; still {
			continue
		}
		note := fmt.Sprintf("mcp %s/%s: no longer offered by the server", server, remote)
		wasEnabled, err := w.st.WithdrawTool(ctx, row.Name, note, now)
		if err != nil {
			log.Printf("mcp: withdraw %s: %v", row.Name, err)
			continue
		}
		if wasEnabled {
			log.Printf("mcp: %s was enabled but %s no longer offers it — disabled", row.Name, server)
		}
	}
}

// run re-dials downed servers and picks up tools that appeared, until ctx is
// canceled.
func (w *mcpWiring) run(ctx context.Context) {
	if w == nil {
		return
	}
	// The first dial happens here rather than at wiring time, so boot never
	// waits on someone else's process.
	w.connect(ctx)
	t := time.NewTicker(mcpRefreshInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			w.mgr.Refresh(ctx)
			w.propose(ctx)
		}
	}
}

// status maps the manager's view onto the wire type the web surface reads.
func (w *mcpWiring) status() []ipc.MCPServerStatus {
	if w == nil {
		return nil
	}
	in := w.mgr.Status()
	out := make([]ipc.MCPServerStatus, 0, len(in))
	for _, s := range in {
		out = append(out, ipc.MCPServerStatus{
			Name:      s.Name,
			Transport: s.Transport,
			Target:    s.Target,
			Connected: s.Connected,
			Server:    s.Server,
			Tools:     s.Tools,
			Err:       s.Err,
		})
	}
	return out
}

func (w *mcpWiring) close() {
	if w == nil {
		return
	}
	_ = w.mgr.Close()
}

// caller is the tool.MCPCaller the executor gets, or nil when MCP is off.
func (w *mcpWiring) caller() *mcp.Manager {
	if w == nil {
		return nil
	}
	return w.mgr
}