T-3: a Go bundle's acknowledged publish and its subscription for new messages (EmitOptions.Ack, SubscribeNew)

Builds: design 47 G §559

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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.

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()
+}