diff --git a/go/README.md b/go/README.md index 0e02cca..6f8d4f8 100644 --- a/go/README.md +++ b/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 `.` 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 diff --git a/go/stdio.go b/go/stdio.go index 9060b8c..8e4995a 100644 --- a/go/stdio.go +++ b/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 `.` 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 diff --git a/go/stdio_test.go b/go/stdio_test.go index 94bf2a3..e05a16a 100644 --- a/go/stdio_test.go +++ b/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") + } +}