Merge pull request 'The Go SDK subscribes and asks through the runtime (hq ADR 0198)' (#17) from feat/go-sdk-subscribes-and-asks into main

go/README.md

@@ -3,7 +3,7 @@
 `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 `<seat>.<verb>` is the seat's implementation.
-`stdio.Emit(type, body)` publishes an event as the module, through the runtime. Standard library only.
+`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (acknowledged only after the handler returns nil) and `stdio.Ask(key, body)` calls a tool — all through the runtime (novox/hq ADR 0193, 0198). Standard library only.
 
 ## Installing
 

go/stdio.go

@@ -21,6 +21,7 @@ import (
 	"io"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 )
@@ -54,10 +55,25 @@ type message struct {
 // 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
+	mu        sync.Mutex
+	enc       *json.Encoder
+	next      int
+	waiting   map[string]chan message
+	listeners []listener
+}
+
+type listener struct {
+	pattern string
+	handle  func(Envelope) error
+}
+
+// Envelope is one event as the mesh carries it: its type, the machine it came from, its body as JSON,
+// and the ADR 0042 headers (x-event-id, x-source, x-node, x-time …).
+type Envelope struct {
+	Key     string            `json:"key"`
+	Node    string            `json:"node"`
+	Body    json.RawMessage   `json:"body"`
+	Headers map[string]string `json:"headers,omitempty"`
 }
 
 func (c *channel) say(m map[string]any) {
@@ -189,6 +205,34 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
 				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)
+		case "mesh/event":
+			// One event the runtime read from this module's consumer (ADR 0198), answered once every
+			// handler it is for has taken it; an error is answered as one, so the runtime does not
+			// acknowledge it and the bus delivers it again. Beside the loop, like a tool call.
+			var p struct {
+				Envelope *Envelope `json:"envelope"`
+			}
+			if json.Unmarshal(m.Params, &p) != nil || p.Envelope == nil || p.Envelope.Key == "" {
+				c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/event carries no envelope"}})
+				continue
+			}
+			c.mu.Lock()
+			matching := []listener{}
+			for _, l := range c.listeners {
+				if TopicMatches(l.pattern, p.Envelope.Key) {
+					matching = append(matching, l)
+				}
+			}
+			c.mu.Unlock()
+			go func(id any, env Envelope, ls []listener) {
+				for _, l := range ls {
+					if err := l.handle(env); err != nil {
+						c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
+						return
+					}
+				}
+				c.say(map[string]any{"id": id, "result": map[string]any{}})
+			}(id, *p.Envelope, matching)
 		default:
 			if id != nil {
 				c.say(map[string]any{"id": id, "error": map[string]any{"code": -32601, "message": "no " + m.Method}})
@@ -198,6 +242,65 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
 	return scanner.Err()
 }
 
+func live() (*channel, error) {
+	currentMu.Lock()
+	defer currentMu.Unlock()
+	if current == nil {
+		return nil, errors.New("before Serve: a bundle reaches the bus through the runtime that launched it")
+	}
+	return current, nil
+}
+
+// Subscribe asks the runtime for this module's events whose type matches the pattern (`*` one
+// segment, `#` any), and hands each to handle (ADR 0198). The runtime acknowledges an event on the bus
+// only after every matching handler returned nil; an error has it delivered again. Call it from a
+// goroutine started before Serve, or from a tool, once Serve runs.
+func Subscribe(pattern string, handle func(Envelope) error) error {
+	c, err := live()
+	if err != nil {
+		return err
+	}
+	c.mu.Lock()
+	c.listeners = append(c.listeners, listener{pattern: pattern, handle: handle})
+	c.mu.Unlock()
+	_, err = c.ask("mesh/subscribe", map[string]any{"pattern": pattern}, 30*time.Second)
+	return err
+}
+
+// Ask calls a tool on the mesh as this bundle's module, through the runtime (ADR 0198), and returns
+// its result as JSON. The key is `<module>.<tool>` as the runtime calls it.
+func Ask(key string, body any) (json.RawMessage, error) {
+	c, err := live()
+	if err != nil {
+		return nil, err
+	}
+	return c.ask("mesh/ask", map[string]any{"key": key, "body": body}, 60*time.Second)
+}
+
+// TopicMatches says whether an event's type matches a pattern: `*` is one dot-separated segment, `#`
+// (or `**`) any number, as the TypeScript SDK matches.
+func TopicMatches(pattern, key string) bool {
+	return matchFrom(strings.Split(pattern, "."), 0, strings.Split(key, "."), 0)
+}
+
+func matchFrom(p []string, pi int, k []string, ki int) bool {
+	if pi == len(p) {
+		return ki == len(k)
+	}
+	if p[pi] == "#" || p[pi] == "**" {
+		for skip := ki; skip <= len(k); skip++ {
+			if matchFrom(p, pi+1, k, skip) {
+				return true
+			}
+		}
+		return false
+	}
+	if ki == len(k) || (p[pi] != "*" && p[pi] != k[ki]) {
+		return false
+	}
+	return matchFrom(p, pi+1, k, ki+1)
+}
+
 // EmitOptions is the provenance an emitter may attach (ADR 0042).
 type EmitOptions struct {
 	CausationID string

go/stdio_test.go

@@ -3,6 +3,7 @@ package stdio
 import (
 	"bufio"
 	"encoding/json"
+	"errors"
 	"io"
 	"testing"
 	"time"
@@ -66,3 +67,75 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
 	}
 	_ = inW.Close()
 }
+
+// A Go bundle subscribes and asks through the runtime (novox/hq ADR 0198): the event is answered only
+// after the handler took it, a handler's error is answered as one, and a tool call is asked as the module.
+func TestABundleSubscribesAndAsksThroughTheRuntime(t *testing.T) {
+	t.Setenv(ServedAs, "watcher")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	ready := make(chan struct{})
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) {
+			go func() {
+				_ = Subscribe("database.*", func(e Envelope) error {
+					if e.Key == "database.bad" {
+						return errors.New("not taken")
+					}
+					return nil
+				})
+				close(ready)
+			}()
+			answer, err := Ask("postgres.list", map[string]any{})
+			return json.RawMessage(answer), err
+		}}}, 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": "tools/call", "params": map[string]any{"name": "start"}})
+	// Two asks arrive, in either order: the subscription and the tool call.
+	asked := map[string]map[string]any{}
+	for len(asked) < 2 {
+		m := next()
+		asked[m["method"].(string)] = m
+	}
+	sub, ask := asked["mesh/subscribe"], asked["mesh/ask"]
+	if sub["params"].(map[string]any)["pattern"] != "database.*" || ask["params"].(map[string]any)["key"] != "postgres.list" {
+		t.Fatalf("asked %v", asked)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": sub["id"], "result": map[string]any{}})
+	send(map[string]any{"jsonrpc": "2.0", "id": ask["id"], "result": []string{"a", "b"}})
+	if r := next(); r["id"] != float64(1) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != `["a","b"]` {
+		t.Fatalf("the call answered %v", r)
+	}
+	<-ready
+	send(map[string]any{"jsonrpc": "2.0", "id": 7, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{"key": "database.provisioned", "node": "anchor", "body": map[string]any{}}}})
+	if r := next(); r["id"] != float64(7) || r["error"] != nil {
+		t.Errorf("a taken event was answered %v", r)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": 8, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{"key": "database.bad", "node": "anchor", "body": map[string]any{}}}})
+	if r := next(); r["id"] != float64(8) || r["error"] == nil {
+		t.Errorf("a refused event was answered %v, want an error so the runtime does not acknowledge it", r)
+	}
+	_ = inW.Close()
+	if !TopicMatches("a.#", "a.b.c") || TopicMatches("a.*", "a.b.c") || !TopicMatches("#", "x") {
+		t.Error("topic matching differs from the TypeScript SDK's")
+	}
+}