mesh-tools/node-tools/internal/launch/launch.go

// Package launch starts a bundle the runtime serves and speaks MCP over stdio to it (novox/hq ADR
// 0188, ADR 0193): `initialize`, `tools/list` once, `tools/call` per call. A Go binary, a Python
// script and a Node launcher are the same thing here: an executable that answers those. The runtime
// knows no language; it starts the path it is given.
package launch

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"

	"golang.org/x/sys/unix"

	"github.com/novox/mesh-tools/node-tools/internal/wire"
)

// Protocol is the MCP version spoken to a bundle.
const Protocol = "2025-03-26"

// How long a child has for its handshake, and a call before the caller is told it is slow.
var (
	HandshakeTimeout = 10 * time.Second
	CallTimeout      = 30 * time.Second
)

// Tool is one tool a launched bundle listed, and how to call it.
type Tool struct {
	Name        string
	Description string
	Input       json.RawMessage
	Run         func(args json.RawMessage) (json.RawMessage, error)
}

// Registration is the tools a bundle listed under one name: its module's, or a seat's.
type Registration struct {
	Module string
	Tools  []Tool
}

// Publisher publishes an event a bundle asked the runtime to emit, as the bundle's module.
type Publisher func(params json.RawMessage) error

// Launched is a running bundle: what it registered, and how to stop it.
type Launched struct {
	Registrations []Registration
	Stop          func()
}

// Executable says whether an entrypoint can be started: a bundle the runtime serves is executable,
// and one that is not was not built to be served (ADR 0193).
func Executable(path string) bool {
	info, err := os.Stat(path)
	if err != nil || info.IsDir() {
		return false
	}
	return info.Mode().Perm()&0o111 != 0
}

type message struct {
	JSONRPC string          `json:"jsonrpc,omitempty"`
	ID      json.RawMessage `json:"id,omitempty"`
	Method  string          `json:"method,omitempty"`
	Params  json.RawMessage `json:"params,omitempty"`
	Result  json.RawMessage `json:"result,omitempty"`
	Error   *struct {
		Code    int    `json:"code"`
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

type child struct {
	cmd     *exec.Cmd
	stdin   io.WriteCloser
	writeMu sync.Mutex
	mu      sync.Mutex
	pending map[int64]chan message
	next    int64
	dead    chan struct{}
	why     error
}

func (c *child) write(m any) error {
	b, err := wire.Marshal(m)
	if err != nil {
		return err
	}
	c.writeMu.Lock()
	defer c.writeMu.Unlock()
	_, err = c.stdin.Write(append(b, '\n'))
	return err
}

var stackLine = regexp.MustCompile(`^\s+at\s`)

// Start launches one bundle and learns its tools. It fails when the child cannot be started or does
// not complete the handshake. A child that exits later is started again on its next call.
func Start(module, entry string, env []string, publish Publisher, logf func(string, ...any)) (*Launched, error) {
	var mu sync.Mutex
	var current *child
	stopped := false

	start := func() (*child, error) {
		cmd := exec.Command(entry)
		cmd.Env = env
		stdin, err := cmd.StdinPipe()
		if err != nil {
			return nil, err
		}
		stdout, err := cmd.StdoutPipe()
		if err != nil {
			return nil, err
		}
		stderr, err := cmd.StderrPipe()
		if err != nil {
			return nil, err
		}
		if err := cmd.Start(); err != nil {
			return nil, err
		}
		c := &child{cmd: cmd, stdin: stdin, pending: map[int64]chan message{}, next: 1, dead: make(chan struct{})}
		var lastSaid string
		var saidMu sync.Mutex
		stderrDone := make(chan struct{})
		go func() {
			defer close(stderrDone)
			scan := bufio.NewScanner(stderr)
			scan.Buffer(make([]byte, 64*1024), 1<<20)
			for scan.Scan() {
				line := scan.Text()
				if strings.TrimSpace(line) == "" {
					continue
				}
				logf("[%s] %s", module, line)
				if !stackLine.MatchString(line) && !strings.HasPrefix(line, "Node.js v") {
					saidMu.Lock()
					lastSaid = strings.TrimSpace(line)
					saidMu.Unlock()
				}
			}
		}()
		go func() {
			scan := bufio.NewScanner(stdout)
			scan.Buffer(make([]byte, 64*1024), 16<<20)
			for scan.Scan() {
				line := strings.TrimSpace(scan.Text())
				if line == "" {
					continue
				}
				var m message
				if err := json.Unmarshal([]byte(line), &m); err != nil {
					if len(line) > 120 {
						line = line[:120]
					}
					logf("[mesh-tools] %s's bundle said something that is not a reply: %s", module, line)
					continue
				}
				// The bundle asks the runtime to emit (ADR 0193): published as this module, answered
				// once the bus has accepted it. Nothing else a bundle may ask.
				if m.Method != "" {
					go func(m message) {
						if m.Method != "mesh/publish" {
							if len(m.ID) > 0 {
								_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID,
									"error": map[string]any{"code": -32601, "message": "the runtime answers no " + m.Method + " from a bundle"}})
							}
							return
						}
						err := publish(m.Params)
						if len(m.ID) == 0 {
							return
						}
						if err != nil {
							_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID,
								"error": map[string]any{"code": -32000, "message": err.Error()}})
							return
						}
						_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID, "result": map[string]any{}})
					}(m)
					continue
				}
				var id int64
				if json.Unmarshal(m.ID, &id) != nil {
					continue
				}
				c.mu.Lock()
				ch := c.pending[id]
				delete(c.pending, id)
				c.mu.Unlock()
				if ch != nil {
					ch <- m
				}
			}
			<-stderrDone
			err := cmd.Wait()
			code := "0"
			if exit := (*exec.ExitError)(nil); errors.As(err, &exit) {
				if status, ok := exit.Sys().(syscall.WaitStatus); ok && status.Signaled() {
					code = unix.SignalName(status.Signal())
				} else {
					code = fmt.Sprint(exit.ExitCode())
				}
			}
			saidMu.Lock()
			why := fmt.Sprintf("%s's bundle exited (%s)", module, code)
			if lastSaid != "" {
				why += ": " + lastSaid
			}
			saidMu.Unlock()
			c.mu.Lock()
			c.why = errors.New(why)
			c.mu.Unlock()
			close(c.dead)
			mu.Lock()
			if current == c {
				current = nil
			}
			wasStopped := stopped
			mu.Unlock()
			if !wasStopped {
				logf("[mesh-tools] %s; started again on its next call", why)
			}
		}()
		if _, err := c.ask(module, "initialize", map[string]any{"protocolVersion": Protocol, "capabilities": map[string]any{},
			"clientInfo": map[string]any{"name": "node-tools", "version": "1"}}, HandshakeTimeout); err != nil {
			_ = cmd.Process.Kill()
			return nil, err
		}
		_ = c.write(map[string]any{"jsonrpc": "2.0", "method": "notifications/initialized"})
		return c, nil
	}

	asking := func(method string, params any, timeout time.Duration) (json.RawMessage, error) {
		mu.Lock()
		c := current
		mu.Unlock()
		if c == nil {
			fresh, err := start()
			if err != nil {
				return nil, err
			}
			mu.Lock()
			current = fresh
			c = fresh
			mu.Unlock()
		}
		return c.ask(module, method, params, timeout)
	}

	first, err := start()
	if err != nil {
		return nil, err
	}
	mu.Lock()
	current = first
	mu.Unlock()

	raw, err := asking("tools/list", map[string]any{}, HandshakeTimeout)
	if err != nil {
		return nil, err
	}
	var listed struct {
		Tools []struct {
			Name        string          `json:"name"`
			Description string          `json:"description"`
			InputSchema json.RawMessage `json:"inputSchema"`
		} `json:"tools"`
	}
	if err := json.Unmarshal(raw, &listed); err != nil {
		return nil, fmt.Errorf("%s's bundle listed its tools in a shape that is not MCP's: %w", module, err)
	}
	order := []string{}
	groups := map[string][]Tool{}
	for _, t := range listed.Tools {
		under, name := module, t.Name
		if dot := strings.Index(t.Name, "."); dot >= 0 {
			under, name = t.Name[:dot], t.Name[dot+1:]
		}
		full := t.Name
		input := t.InputSchema
		if len(input) == 0 || string(input) == "null" {
			input = json.RawMessage("{}")
		}
		tool := Tool{Name: name, Description: t.Description, Input: input,
			Run: func(args json.RawMessage) (json.RawMessage, error) {
				if len(args) == 0 || string(args) == "null" {
					args = json.RawMessage("{}")
				}
				res, err := asking("tools/call", map[string]any{"name": full, "arguments": args}, CallTimeout)
				if err != nil {
					return nil, err
				}
				var called struct {
					Content []struct {
						Type string `json:"type"`
						Text string `json:"text"`
					} `json:"content"`
					IsError bool `json:"isError"`
				}
				_ = json.Unmarshal(res, &called)
				text := ""
				for _, c := range called.Content {
					if c.Type == "text" {
						text = c.Text
						break
					}
				}
				if called.IsError {
					if text == "" {
						text = module + "." + full + " failed"
					}
					return nil, errors.New(text)
				}
				// The bundle's answer is JSON as text; handed back as the value it encodes.
				if json.Valid([]byte(text)) && text != "" {
					return json.RawMessage(text), nil
				}
				b, _ := json.Marshal(text)
				return b, nil
			}}
		if _, seen := groups[under]; !seen {
			order = append(order, under)
		}
		groups[under] = append(groups[under], tool)
	}
	out := &Launched{Stop: func() {
		mu.Lock()
		stopped = true
		c := current
		current = nil
		mu.Unlock()
		if c != nil && c.cmd.Process != nil {
			_ = c.cmd.Process.Signal(syscall.SIGTERM)
		}
	}}
	for _, under := range order {
		out.Registrations = append(out.Registrations, Registration{Module: under, Tools: groups[under]})
	}
	return out, nil
}

func (c *child) ask(module, method string, params any, timeout time.Duration) (json.RawMessage, error) {
	c.mu.Lock()
	if c.why != nil {
		c.mu.Unlock()
		return nil, c.why
	}
	id := c.next
	c.next++
	ch := make(chan message, 1)
	c.pending[id] = ch
	c.mu.Unlock()
	if err := c.write(map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}); err != nil {
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()
		select {
		case <-c.dead:
			return nil, c.deathReason()
		case <-time.After(100 * time.Millisecond):
			return nil, err
		}
	}
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	select {
	case m := <-ch:
		if m.Error != nil {
			msg := m.Error.Message
			if msg == "" {
				msg = "the bundle refused the request"
			}
			return nil, errors.New(msg)
		}
		return m.Result, nil
	case <-c.dead:
		return nil, c.deathReason()
	case <-timer.C:
		c.mu.Lock()
		delete(c.pending, id)
		c.mu.Unlock()
		return nil, fmt.Errorf("%s's bundle did not answer %s in %ds", module, method, int(timeout/time.Second))
	}
}

func (c *child) deathReason() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.why != nil {
		return c.why
	}
	return errors.New("the bundle exited")
}