Pull a build seat's ask through the bundle's runtime one at a time and settle it by id, so a tools bundle pulls by its capacity and order (issue 1887 A11)

2026-10-23 15:29:05 +02:00
parent 8cd22892a3
commit 4b503f8d0d
4 changed files with 204 additions and 0 deletions
@@ -0,0 +1,44 @@
+package buildstep
+
+import (
+	"context"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// SeatAsks is a build seat's workers as a tools bundle reaches them: through its runtime's seat pull and settle
+// (stdio.SeatPull, SeatSettle; issue 1887 A11), one message per pull, only from the workers the module's membership
+// lists.
+type SeatAsks struct {
+	Language string
+	// Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own.
+	Pull   func(worker string, wait time.Duration) (*stdio.Pulled, error)
+	Settle func(id, fate string) error
+}
+
+// Next is AskSource's.
+func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
+	pull, settle := s.Pull, s.Settle
+	if pull == nil {
+		pull = stdio.SeatPull
+	}
+	if settle == nil {
+		settle = stdio.SeatSettle
+	}
+	worker, err := delivery.AskWorker(s.Language, kind)
+	if err != nil {
+		return nil, err
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	p, err := pull(worker, wait)
+	if err != nil || p == nil {
+		return nil, err
+	}
+	fate := func(f string) func() error { return func() error { return settle(p.ID, f) } }
+	return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress),
+		Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm)}, nil
+}
@@ -0,0 +1,40 @@
+package buildstep
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// 1887/A11: a bundle's source pulls the kind's worker by its seat's name and settles each ask by the runtime's id.
+func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
+	var pulled []string
+	var settled []string
+	src := SeatAsks{Language: delivery.AskGo,
+		Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) {
+			pulled = append(pulled, worker)
+			if worker != "SEAT_NODE_BUILD_GO_CHECK_worker" {
+				return nil, nil
+			}
+			return &stdio.Pulled{ID: "p9", Worker: worker, Body: []byte(`{"v":1}`), Delivered: 2}, nil
+		},
+		Settle: func(id, fate string) error { settled = append(settled, id+" "+fate); return nil }}
+	if p, err := src.Next(context.Background(), delivery.AskDelivery, 0); err != nil || p != nil {
+		t.Fatalf("delivery = %+v, %v", p, err)
+	}
+	p, err := src.Next(context.Background(), delivery.AskCheck, time.Second)
+	if err != nil || p == nil || string(p.Data) != `{"v":1}` || p.Delivered != 2 {
+		t.Fatalf("check = %+v, %v", p, err)
+	}
+	_ = p.InProgress()
+	_ = p.Ack()
+	if len(pulled) != 2 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" {
+		t.Errorf("pulled %v", pulled)
+	}
+	if len(settled) != 2 || settled[0] != "p9 progress" || settled[1] != "p9 ack" {
+		t.Errorf("settled %v", settled)
+	}
+}
@@ -69,6 +69,61 @@ func SeatTake(worker string, handle func(Work) error) error {
 	return err
 }

+// Pulled is one message a bundle pulled from a seat's worker (SeatPull), named by the runtime's id for it, which
+// SeatSettle gives its fate.
+type Pulled struct {
+	ID        string            `json:"id"`
+	Worker    string            `json:"worker"`
+	Subject   string            `json:"subject"`
+	Body      json.RawMessage   `json:"body"`
+	Headers   map[string]string `json:"headers,omitempty"`
+	Delivered uint64            `json:"delivered"`
+}
+
+// The fates SeatSettle gives a pulled message: done, handed back to be delivered again, dropped, or still being worked.
+const (
+	SettleAck      = "ack"
+	SettleNak      = "nak"
+	SettleTerm     = "term"
+	SettleProgress = "progress"
+)
+
+// SeatPull pulls at most one message from a held seat's worker the module's membership lists, waiting at most wait
+// (none when 0, at most 25 s): nil when none came. Unlike SeatTake nothing is fetched ahead: a build module pulls an
+// ask only when it has a free slot (novox/hq ADR 0525 §4; issue 1887 A11). A message never settled is delivered again
+// by the worker's acknowledgement wait.
+func SeatPull(worker string, wait time.Duration) (*Pulled, error) {
+	c, err := live()
+	if err != nil {
+		return nil, err
+	}
+	raw, err := c.ask("mesh/seat.pull", map[string]any{"worker": worker, "wait_ms": wait.Milliseconds()}, seatTimeout)
+	if err != nil {
+		return nil, err
+	}
+	if len(raw) == 0 || string(raw) == "null" {
+		return nil, nil
+	}
+	var p Pulled
+	if err := json.Unmarshal(raw, &p); err != nil {
+		return nil, err
+	}
+	if p.ID == "" {
+		return nil, nil
+	}
+	return &p, nil
+}
+
+// SeatSettle gives the message pulled under id its fate (SettleAck, SettleNak, SettleTerm, SettleProgress).
+func SeatSettle(id, fate string) error {
+	c, err := live()
+	if err != nil {
+		return err
+	}
+	_, err = c.ask("mesh/seat.settle", map[string]any{"id": id, "fate": fate}, seatTimeout)
+	return err
+}
+
 // SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil.
 func SeatRecord(bucket, key string) (*StateEntry, error) {
 	c, err := live()
@@ -0,0 +1,65 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+)
+
+// 1887/A11: a bundle pulls one message through its runtime, naming the worker and the wait, and settles it by the id the
+// runtime gave.
+func TestABundlePullsAndSettlesThroughTheRuntime(t *testing.T) {
+	t.Setenv(ServedAs, "mesh-build-go")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	got := make(chan *Pulled, 1)
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) {
+			p, err := SeatPull("SEAT_NODE_BUILD_GO_DELIVERY_worker", 1500*time.Millisecond)
+			if err != nil {
+				return nil, err
+			}
+			got <- p
+			return map[string]any{}, SeatSettle(p.ID, SettleAck)
+		}}}, 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"}})
+	asked := next()
+	params, _ := asked["params"].(map[string]any)
+	if asked["method"] != "mesh/seat.pull" || params["worker"] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" || params["wait_ms"] != 1500.0 {
+		t.Fatalf("asked %v", asked)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{"id": "p1", "worker": "SEAT_NODE_BUILD_GO_DELIVERY_worker",
+		"subject": "mesh.seat.node-build-go.accept.delivery", "body": map[string]any{"v": 1}, "delivered": 1}})
+	p := <-got
+	if p.ID != "p1" || string(p.Body) != `{"v":1}` {
+		t.Fatalf("pulled %+v", p)
+	}
+	asked = next()
+	params, _ = asked["params"].(map[string]any)
+	if asked["method"] != "mesh/seat.settle" || params["id"] != "p1" || params["fate"] != "ack" {
+		t.Fatalf("settled %v", asked)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}})
+}