diff --git a/go/README.md b/go/README.md index 4160a9f..0e02cca 100644 --- a/go/README.md +++ b/go/README.md @@ -1,8 +1,18 @@ # mesh-sdk for Go -The skeleton (novox/hq ADR 0188): `stdio.Serve(name, tools)` speaks MCP over stdio to the node's -tool runtime. A `Tool` is a name, a description, an argument schema and a function. A tool named -`.` is the seat's implementation. Standard library only. +`stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime +names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a +function; calls run concurrently. A tool named `.` is the seat's implementation. +`stdio.Emit(type, body)` publishes an event as the module, through the runtime. Standard library only. + +## Installing + +``` +GOPRIVATE=git.novox.be go get git.novox.be/novox/mesh-sdk/go@latest +``` + +Releases are tagged `go/vX.Y.Z` in this repository, the subdirectory form Go's module system expects, +at the same version as the TypeScript package. What it holds, under ADR 0039's test: the stdio loop and the tool type. Nothing else — no bus, no client, nothing of any module's. `go build ./example` and pipe a `tools/list` request into it. diff --git a/go/stdio.go b/go/stdio.go index 4bc6c84..9060b8c 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -1,4 +1,4 @@ -// Package stdio is the Novox Mesh SDK for Go — the skeleton (novox/hq ADR 0188). +// Package stdio is the Novox Mesh SDK for Go (novox/hq ADR 0188, ADR 0193). // // A tools bundle is a process the node's runtime launches and speaks MCP over stdio to: // `initialize`, `tools/list` once, `tools/call` per call, newline-framed JSON-RPC on stdin and @@ -7,20 +7,31 @@ // module's own. // // No transport here: the bus, the subjects and the memberships are the runtime's (ADR 0039, 0175). -// stdout is the protocol; log to stderr. +// What a bundle asks of the bus — emitting an event — it asks of the runtime over the same channel, +// and the runtime does it as the module (ADR 0193). stdout is the protocol; log to stderr. package stdio import ( "bufio" + "crypto/rand" + "encoding/hex" "encoding/json" + "errors" "fmt" "io" "os" + "strconv" + "sync" + "time" ) // Protocol is the MCP version this speaks; the runtime says the same. const Protocol = "2025-03-26" +// ServedAs is the word the runtime sets for a bundle it launches: the module it serves it as +// (ADR 0193). A bundle served under that name lists its own tools by their names. +const ServedAs = "MESH_SERVED_MODULE" + // Tool is one tool a bundle serves. type Tool struct { Name string @@ -30,48 +41,116 @@ type Tool struct { Run func(args map[string]any) (any, error) } -type request struct { - ID *json.RawMessage `json:"id"` - Method string `json:"method"` - Params struct { - Name string `json:"name"` - Arguments map[string]any `json:"arguments"` - } `json:"params"` +type message struct { + 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 { + Message string `json:"message"` + } `json:"error,omitempty"` } -// Serve answers the runtime on stdin and stdout until stdin closes. +// channel is the one conversation with the runtime: what this bundle writes, serialised, and what +// it asked and waits for. +type channel struct { + mu sync.Mutex + enc *json.Encoder + next int + waiting map[string]chan message +} + +func (c *channel) say(m map[string]any) { + m["jsonrpc"] = "2.0" + c.mu.Lock() + defer c.mu.Unlock() + _ = c.enc.Encode(m) +} + +// ask sends the runtime a request and waits for its answer. +func (c *channel) ask(method string, params any, timeout time.Duration) (json.RawMessage, error) { + c.mu.Lock() + c.next++ + id := "bundle-" + strconv.Itoa(c.next) + answer := make(chan message, 1) + c.waiting[id] = answer + c.mu.Unlock() + c.say(map[string]any{"id": id, "method": method, "params": params}) + select { + case m := <-answer: + if m.Error != nil { + return nil, errors.New(m.Error.Message) + } + return m.Result, nil + case <-time.After(timeout): + c.mu.Lock() + delete(c.waiting, id) + c.mu.Unlock() + return nil, fmt.Errorf("the runtime did not answer %s in %s", method, timeout) + } +} + +var ( + current *channel + currentMu sync.Mutex +) + +// Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the +// module the runtime named in MESH_SERVED_MODULE. func Serve(name string, tools []Tool) error { return ServeOn(name, tools, os.Stdin, os.Stdout) } // ServeOn is Serve over any pair of streams, for a test. func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { + if name == "" { + name = os.Getenv(ServedAs) + } + if name == "" { + name = "bundle" + } byName := map[string]Tool{} for _, t := range tools { byName[t.Name] = t } - enc := json.NewEncoder(out) - say := func(m map[string]any) { m["jsonrpc"] = "2.0"; _ = enc.Encode(m) } + c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}} + currentMu.Lock() + current = c + currentMu.Unlock() scanner := bufio.NewScanner(in) scanner.Buffer(make([]byte, 1<<20), 1<<24) for scanner.Scan() { - var req request - if err := json.Unmarshal(scanner.Bytes(), &req); err != nil { + var m message + if err := json.Unmarshal(scanner.Bytes(), &m); err != nil { + continue + } + // The runtime's answer to something this bundle asked. + if m.Method == "" && m.ID != nil { + var key string + if json.Unmarshal(*m.ID, &key) == nil { + c.mu.Lock() + answer, ok := c.waiting[key] + delete(c.waiting, key) + c.mu.Unlock() + if ok { + answer <- m + } + } continue } id := any(nil) - if req.ID != nil { - id = *req.ID + if m.ID != nil { + id = *m.ID } - switch req.Method { + switch m.Method { case "initialize": - say(map[string]any{"id": id, "result": map[string]any{"protocolVersion": Protocol, + c.say(map[string]any{"id": id, "result": map[string]any{"protocolVersion": Protocol, "capabilities": map[string]any{"tools": map[string]any{}}, "serverInfo": map[string]any{"name": name, "version": "1"}}}) case "notifications/initialized": case "ping": if id != nil { - say(map[string]any{"id": id, "result": map[string]any{}}) + c.say(map[string]any{"id": id, "result": map[string]any{}}) } case "tools/list": listed := make([]map[string]any, 0, len(tools)) @@ -84,25 +163,86 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { } listed = append(listed, map[string]any{"name": t.Name, "description": t.Description, "inputSchema": schema}) } - say(map[string]any{"id": id, "result": map[string]any{"tools": listed}}) + c.say(map[string]any{"id": id, "result": map[string]any{"tools": listed}}) case "tools/call": - t, ok := byName[req.Params.Name] + var p struct { + Name string `json:"name"` + Arguments map[string]any `json:"arguments"` + } + _ = json.Unmarshal(m.Params, &p) + t, ok := byName[p.Name] if !ok { - say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": fmt.Sprintf("%s has no tool %s", name, req.Params.Name)}}) + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": fmt.Sprintf("%s has no tool %s", name, p.Name)}}) continue } - result, err := t.Run(req.Params.Arguments) - if err != nil { - say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": err.Error()}}, "isError": true}}) - continue - } - text, _ := json.Marshal(result) - say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": string(text)}}}}) + // Beside the loop, never in it: a tool that emits waits for the runtime's answer, which + // arrives on the line this loop reads next — and calls do not queue behind a slow one. + go func(id any, t Tool, args map[string]any) { + if args == nil { + args = map[string]any{} + } + result, err := t.Run(args) + if err != nil { + c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": err.Error()}}, "isError": true}}) + return + } + text, _ := json.Marshal(result) + c.say(map[string]any{"id": id, "result": map[string]any{"content": []map[string]any{{"type": "text", "text": string(text)}}}}) + }(id, t, p.Arguments) default: if id != nil { - say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + req.Method}}) + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}}) } } } return scanner.Err() } + +// EmitOptions is the provenance an emitter may attach (ADR 0042). +type EmitOptions struct { + CausationID string + Schema string +} + +// Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has +// accepted it (ADR 0193). The envelope and headers are the TypeScript SDK's exactly: the source and +// node come from MESH_MODULE and MESH_NODE, which the runtime sets. +func Emit(eventType string, body any, opts ...EmitOptions) error { + currentMu.Lock() + c := current + currentMu.Unlock() + if c == nil { + return errors.New("emit before Serve: a bundle emits through the runtime that launched it") + } + source, node := os.Getenv("MESH_MODULE"), os.Getenv("MESH_NODE") + if source == "" { + source = "unknown" + } + if node == "" { + node = "unknown" + } + headers := map[string]string{ + "x-event-id": eventID(), "x-source": source, "x-node": node, + "x-time": time.Now().UTC().Format(time.RFC3339Nano), "content-type": "application/json", + } + for _, o := range opts { + if o.CausationID != "" { + headers["x-causation-id"] = o.CausationID + } + if o.Schema != "" { + headers["x-schema"] = o.Schema + } + } + _, err := c.ask("mesh/publish", map[string]any{"key": eventType, "node": node, "body": body, "headers": headers}, 30*time.Second) + return err +} + +// eventID is a random UUID, version 4, as the TypeScript SDK's randomUUID. +func eventID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + h := hex.EncodeToString(b[:]) + return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32] +} diff --git a/go/stdio_test.go b/go/stdio_test.go new file mode 100644 index 0000000..94bf2a3 --- /dev/null +++ b/go/stdio_test.go @@ -0,0 +1,68 @@ +package stdio + +import ( + "bufio" + "encoding/json" + "io" + "testing" + "time" +) + +// A Go bundle serves as the module the runtime names, answers calls concurrently, and emits through +// the runtime with the TypeScript SDK's envelope (novox/hq ADR 0193). +func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) { + t.Setenv(ServedAs, "chime") + t.Setenv("MESH_MODULE", "chime") + t.Setenv("MESH_NODE", "anchor") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + go func() { + _ = ServeOn("", []Tool{{Name: "ring", Description: "rings", Run: func(map[string]any) (any, error) { + if err := Emit("rang", map[string]any{"times": 1}); err != nil { + return nil, err + } + return map[string]any{"rang": true}, nil + }}}, inR, outW) + }() + lines := bufio.NewScanner(outR) + next := func() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + t.Fatal("the bundle said nothing") + return nil + } + } + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": map[string]any{}}) + if got := next()["result"].(map[string]any)["serverInfo"].(map[string]any)["name"]; got != "chime" { + t.Fatalf("served as %v, want the module the runtime named", got) + } + send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "ring", "arguments": map[string]any{}}}) + asked := next() + if asked["method"] != "mesh/publish" { + t.Fatalf("the tool's emit was not asked of the runtime: %v", asked) + } + p := asked["params"].(map[string]any) + h := p["headers"].(map[string]any) + if p["key"] != "rang" || p["node"] != "anchor" || h["x-source"] != "chime" || h["content-type"] != "application/json" || len(h["x-event-id"].(string)) != 36 { + t.Errorf("the envelope is not the SDK's: %v", p) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}}) + answered := next() + text := answered["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] + if answered["id"] != float64(2) || text != `{"rang":true}` { + t.Errorf("the call answered %v", answered) + } + _ = inW.Close() +}