From 9cee2111c0975721b5eb0c4589f3a605fcb1faff Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 13:57:05 +0100 Subject: [PATCH] T-3: a Go bundle's acknowledged publish and its subscription for new messages (EmitOptions.Ack, SubscribeNew) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §559 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/stdio.go | 40 +++++++++++++++++++++++++++++---- go/stdio_test.go | 58 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+), 4 deletions(-) diff --git a/go/stdio.go b/go/stdio.go index f1cf9ea9..c3084d2c 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -499,6 +499,17 @@ func live() (*channel, error) { // answer in time is kept, as the runtime may have bound it.) Call it from a goroutine // started before Serve, or from a tool, once Serve runs. func Subscribe(pattern string, handle func(Envelope) error) error { + return subscribe(pattern, handle, false) +} + +// SubscribeNew is Subscribe for a handler that is given only the messages published after the subscription is made: +// the runtime makes an ordered consumer on the stream that carries the pattern, starting at the next message +// (`mesh/subscribe` with `deliver: new`; design 47 G §559). +func SubscribeNew(pattern string, handle func(Envelope) error) error { + return subscribe(pattern, handle, true) +} + +func subscribe(pattern string, handle func(Envelope) error, deliverNew bool) error { c, err := live() if err != nil { return err @@ -508,7 +519,11 @@ func Subscribe(pattern string, handle func(Envelope) error) error { n := c.subscribed c.listeners = append(c.listeners, listener{n: n, pattern: pattern, handle: handle}) c.mu.Unlock() - _, err = c.ask("mesh/subscribe", map[string]any{"pattern": pattern}, 30*time.Second) + params := map[string]any{"pattern": pattern} + if deliverNew { + params["deliver"] = "new" + } + _, err = c.ask("mesh/subscribe", params, 30*time.Second) // Removed only when the runtime answered that it refused: one that did not answer in time may // have bound the consumer, and a handler removed then would have its events taken unhandled. var no *refused @@ -599,6 +614,9 @@ type EmitOptions struct { // is kept to one, never generated once an id is given (novox/hq ADR 0436 clause 4). Empty is a random // one, as on main. MessageID string + // Ack asks the runtime to answer only once the bus acknowledged the event (`mesh/publish` with `ack: true`): Emit + // then returns an error where the runtime answered `{error}` (design 47 G §559). + Ack bool } // Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has @@ -616,11 +634,12 @@ func Emit(eventType string, body any, opts ...EmitOptions) error { if node == "" { node = "unknown" } - id := "" + id, ack := "", false for _, o := range opts { if o.MessageID != "" { id = o.MessageID } + ack = ack || o.Ack } if id == "" { id = eventID() @@ -637,8 +656,21 @@ func Emit(eventType string, body any, opts ...EmitOptions) error { 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 + params := map[string]any{"key": eventType, "node": node, "body": body, "headers": headers} + if ack { + params["ack"] = true + } + answer, err := c.ask("mesh/publish", params, 30*time.Second) + if err != nil || !ack { + return err + } + var said struct { + Error any `json:"error"` + } + if json.Unmarshal(answer, &said) == nil && said.Error != nil { + return fmt.Errorf("the bus did not acknowledge %s: %v", eventType, said.Error) + } + return nil } // eventID is a random UUID, version 4, as the TypeScript SDK's randomUUID. diff --git a/go/stdio_test.go b/go/stdio_test.go index 258c5c47..6f5458d1 100644 --- a/go/stdio_test.go +++ b/go/stdio_test.go @@ -343,3 +343,61 @@ func servesAlone(t *testing.T) { } }) } + +// Design 47 G §559: a bundle's acknowledged publish asks `ack: true` and fails on the runtime's `{error}`; its +// subscription for new messages asks `deliver: new`. +func TestABundleAsksAnAcknowledgedPublishAndASubscriptionForNewMessages(t *testing.T) { + servesAlone(t) + t.Setenv(ServedAs, "keeper") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + errs := make(chan error, 2) + go func() { + _ = ServeOn("", []Tool{{Name: "go", Run: func(map[string]any) (any, error) { + go func() { errs <- SubscribeNew("mesh-conditions.findings-asked", func(Envelope) error { return nil }) }() + errs <- Emit("finding-written", map[string]any{}, EmitOptions{Ack: true}) + return map[string]any{}, 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": "tools/call", "params": map[string]any{"name": "go"}}) + asked := map[string]map[string]any{} + for len(asked) < 2 { + m := next() + asked[m["method"].(string)] = m + } + sub, pub := asked["mesh/subscribe"]["params"].(map[string]any), asked["mesh/publish"]["params"].(map[string]any) + if sub["deliver"] != "new" || pub["ack"] != true { + t.Fatalf("asked subscribe %v, publish %v", sub, pub) + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["mesh/subscribe"]["id"], "result": map[string]any{}}) + send(map[string]any{"jsonrpc": "2.0", "id": asked["mesh/publish"]["id"], "result": map[string]any{"error": "no stream"}}) + var failed bool + for range 2 { + if err := <-errs; err != nil { + failed = failed || strings.Contains(err.Error(), "did not acknowledge finding-written: no stream") + } + } + if !failed { + t.Errorf("an {error} answer to an acknowledged publish was not returned as an error") + } + _ = inW.Close() +}