C1c-4: the tool runner sends a bundle's chunked seat answer in chunks, and a node's chunked verb is asked of the node

A seat verb whose definition says chunks, served by a bundle that answers the whole file as the member `file`, is sent by the
runner with buscalls.SeatChunks.Send and answered with the chunk count (Conn.AnswerChunked, called from serveSeats). A chunked verb
asked of one machine (`@<node>`) is received in chunks as the verb is.

Builds: design 47 G A-257, A-249, A-254, §681

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

node-tools/internal/bus/chunked.go

@@ -25,9 +25,12 @@ const FileField = "file"
 // counts with buscalls.SeatChunks, each acknowledged before the next, and answers the verb's own answer with the whole
 // file added as FileField. The asking bundle receives no chunk itself.
 func (c *Conn) AskAsChunked(module, key string, body any) (Answered, error) {
-	seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
-	if !strings.HasPrefix(key, "seat:") || !ok || strings.Contains(verb, "@") {
-		return Answered{}, fmt.Errorf("%q is no seat:<seat>.<verb> of no one machine, which a chunked answer needs", key)
+	// The verb may be asked of the machine that ran the ask (`@<node>`, A-254): the chunk subjects name the seat, the verb and
+	// the call alone.
+	name, _, _ := strings.Cut(key, "@")
+	seat, verb, ok := strings.Cut(strings.TrimPrefix(name, "seat:"), ".")
+	if !strings.HasPrefix(key, "seat:") || !ok {
+		return Answered{}, fmt.Errorf("%q is no seat:<seat>.<verb>, which a chunked answer needs", key)
 	}
 	args := map[string]any{}
 	if raw, err := json.Marshal(body); err != nil {
@@ -93,3 +96,48 @@ func (c *Conn) AskAsChunked(module, key string, body any) (Answered, error) {
 	answered.Result = out
 	return answered, nil
 }
+
+// AnswerChunked is a seat verb's answer as its definition says `chunks` (design 47 G A-257, §681): a bundle answers the whole file
+// as the member FileField (base64), and the runner that serves the verb sends it in chunks with buscalls.SeatChunks on the call
+// token the body names, each acknowledged by the caller before the next, and answers the verb's own members with `chunks`, the
+// count, in place of the file. The bundle sends no chunk itself. An answer with no file is answered as it is. The send runs after
+// the answer is returned: the caller hears the chunks before it asks (AskAsChunked) and acknowledges each once it has the answer.
+func (c *Conn) AnswerChunked(seat, verb string, body json.RawMessage, answer any) (any, error) {
+	raw, err := json.Marshal(answer)
+	if err != nil {
+		return nil, err
+	}
+	fields := map[string]json.RawMessage{}
+	if err := json.Unmarshal(raw, &fields); err != nil {
+		return answer, nil
+	}
+	enc, has := fields[FileField]
+	if !has {
+		return answer, nil
+	}
+	var text string
+	if err := json.Unmarshal(enc, &text); err != nil {
+		return nil, fmt.Errorf("%s.%s answers a %s that is no string: %w", seat, verb, FileField, err)
+	}
+	file, err := base64.StdEncoding.DecodeString(text)
+	if err != nil {
+		return nil, fmt.Errorf("%s.%s answers a %s in no base64: %w", seat, verb, FileField, err)
+	}
+	var asked struct {
+		Call string `json:"call"`
+	}
+	if err := json.Unmarshal(body, &asked); err != nil || asked.Call == "" {
+		return nil, fmt.Errorf("%s.%s is asked in chunks with no call token", seat, verb)
+	}
+	delete(fields, FileField)
+	count := int64((len(file) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes)
+	fields["chunks"], _ = json.Marshal(count)
+	send := buscalls.SeatChunks{Seat: seat, Verb: verb, Call: asked.Call}
+	go func() { _ = send.Send(c.nc, file, buscalls.SeatAckWithin, c.Logf) }()
+	return json.RawMessage(mustMarshal(fields)), nil
+}
+
+func mustMarshal(v any) []byte {
+	b, _ := json.Marshal(v)
+	return b
+}

node-tools/internal/bus/chunked_test.go

@@ -73,3 +73,44 @@ func TestAChunkedSeatAnswerIsReceivedByTheRunnerAndHandedWhole(t *testing.T) {
 		t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err)
 	}
 }
+
+// Design 47 G A-257, A-254: a bundle answers a chunked verb with the whole file in the member `file`; the runner that serves the
+// verb sends the file in chunks (each acknowledged before the next) and answers the count, so the bundle sends none, and the asker
+// of one node's verb receives it whole.
+func TestABundlesChunkedAnswerIsSentByTheRunner(t *testing.T) {
+	url := mt.URL(t)
+	holder := dial(t, url, "mesh-build-go", "anchor", "")
+	asker := dial(t, url, "mesh-delivery-indexer", "anchor", "")
+	file := make([]byte, buscalls.SeatChunkBytes*2+7)
+	for i := range file {
+		file[i] = byte(i * 7)
+	}
+	stop, err := holder.Handle("seat:node-build-go.output@anchor", func(body json.RawMessage) (any, error) {
+		answer := map[string]any{"ask": "a1", "file": base64.StdEncoding.EncodeToString(file)}
+		return holder.AnswerChunked("node-build-go", "output", body, answer)
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	got, err := asker.AskAsChunked("mesh-delivery-indexer", "seat:node-build-go.output@anchor", map[string]any{"ask": "a1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var ans struct {
+		Ask    string `json:"ask"`
+		Chunks int    `json:"chunks"`
+		File   string `json:"file"`
+	}
+	if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Ask != "a1" || ans.Chunks != 3 {
+		t.Fatalf("the answer %s: %v", got.Result, err)
+	}
+	if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) {
+		t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err)
+	}
+	// an answer with no file is the verb's own: no chunks are sent
+	plain, err := holder.AnswerChunked("node-build-go", "output", json.RawMessage(`{"call":"c"}`), map[string]any{"ask": "x", "chunks": 0})
+	if m, _ := plain.(map[string]any); err != nil || m["chunks"] != 0 {
+		t.Fatalf("an answer with no file was %v, %v", plain, err)
+	}
+}

node-tools/internal/runtime/chunked_test.go

@@ -7,7 +7,7 @@ import (
 )
 
 // Design 47 G A-249: a bundle's seat call is received by the runner in chunks only for a verb whose definition says `chunks`,
-// of no one machine.
+// of any machine: a verb a node holds is asked of the node that ran the ask (A-254).
 func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) {
 	c := &consumers{definitions: func() map[string]seatdef.Definition {
 		return map[string]seatdef.Definition{"mesh-catalog": {Verbs: []seatdef.Verb{{Name: "input", Chunks: true}, {Name: "entry"}}}}
@@ -15,7 +15,8 @@ func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) {
 	for key, want := range map[string]bool{
 		"seat:mesh-catalog.input":       true,
 		"seat:mesh-catalog.entry":       false,
-		"seat:mesh-catalog.input@ace":   false,
+		"seat:mesh-catalog.input@ace":   true,
+		"seat:mesh-catalog.entry@ace":   false,
 		"seat:mesh-unread.input":        false,
 		"mesh-delivery-indexer.indexer": false,
 	} {

node-tools/internal/runtime/gate.go

@@ -94,8 +94,9 @@ func (c *consumers) chunked(key string) bool {
 	if c.definitions == nil || !strings.HasPrefix(key, "seat:") {
 		return false
 	}
-	seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
-	if !ok || strings.Contains(verb, "@") {
+	name, _, _ := strings.Cut(key, "@") // a verb asked of the machine that ran the ask is chunked as the verb is (A-254)
+	seat, verb, ok := strings.Cut(strings.TrimPrefix(name, "seat:"), ".")
+	if !ok {
 		return false
 	}
 	for _, v := range c.definitions()[seat].Verbs {

node-tools/internal/runtime/runtime.go

@@ -306,7 +306,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log
 	}
 	logf("%s", line)
 	conn.OnMembership(func(bus.Membership) { go func() { r.refreshRepairs(); conn.Flush() }() })
-	stopSeats, again := serveSeats(conn, r.modules, r.registrations, r.events.gated, r.events.admitted, r.fingerprintOf, logf)
+	stopSeats, again := serveSeats(conn, r.modules, r.registrations, r.events.gated, r.events.admitted, r.events.chunked, r.fingerprintOf, logf)
 	r.seatsAgain = again
 	r.stops = append(r.stops, stopSeats)
 
@@ -710,7 +710,7 @@ func subjectsOf(conn *bus.Conn, module, tool string) []string {
 // whenever a membership changes, and when again is called because a bundle was loaded again. Whether
 // this machine holds the seat is the bus's to decide. It answers its stop and again.
 func serveSeats(conn *bus.Conn, modules []string, current func() []registration, gated gatedRun,
-	admitted func(seat, verb, caller string) error, fingerprintOf func(string) string, logf func(string, ...any)) (func(), func()) {
+	admitted func(seat, verb, caller string) error, chunked func(key string) bool, fingerprintOf func(string) string, logf func(string, ...any)) (func(), func()) {
 	var mu sync.Mutex
 	var stops []func()
 	stopped := false
@@ -756,8 +756,13 @@ func serveSeats(conn *bus.Conn, modules []string, current func() []registration,
 					if err := admitted(v.Seat, v.Verb, caller); err != nil {
 						return nil, err
 					}
-					return gated(approval.Call{Address: v.Seat + "." + v.Verb, Module: module, Seat: v.Seat, Name: v.Verb,
+					answer, err := gated(approval.Call{Address: v.Seat + "." + v.Verb, Module: module, Seat: v.Seat, Name: v.Verb,
 						Fingerprint: fingerprintOf(module), Caller: caller, Args: argsOf(body), Description: t.Description}, t, caller)
+					if err != nil || !chunked("seat:"+v.Seat+"."+v.Verb) {
+						return answer, err
+					}
+					// A verb marked chunks: the bundle answers the whole file, the runner sends it in chunks (design 47 G A-257).
+					return conn.AnswerChunked(v.Seat, v.Verb, body, answer)
 				})
 				if err != nil {
 					logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)