G C-2, C-3: a value larger than one chunk is put and got by session through the tool runner

buscalls.ChunkBytes (512 KiB), mesh/objects.put-chunk and get-chunk with the store named in each chunk, the digest and size
of a get checked, the capability word put-sessions required, and a bundle's refusal of a value over 8 MiB gone.

Builds: design 47 G §518, §345, §490, §620, A-21
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

go/buscalls/buscalls.go

@@ -32,6 +32,11 @@ const (
 	// (design 47 G §347, §407), served under the standing grant the grants part derives from the manifests; the answer is
 	// ObjectData.
 	ObjectsSentGet = "mesh/objects.sent-get"
+	// ObjectsPutChunk and ObjectsGetChunk are the put session's and the get session's calls (design 47 G §518, §345, §490),
+	// under the capability word put-sessions (stdio.CapPutSessions): a value larger than ChunkBytes is put as chunks in order,
+	// each answered before the next, and a get that answers ObjectsChunked is read chunk by chunk.
+	ObjectsPutChunk = "mesh/objects.put-chunk"
+	ObjectsGetChunk = "mesh/objects.get-chunk"
 
 	StreamRead = "mesh/stream.read"
 	StreamFill = "mesh/stream.fill"
@@ -94,6 +99,52 @@ type ObjectsPutRequest struct {
 	Description string `json:"description,omitempty"`
 }
 
+// ChunkBytes is the size of one chunk of a put or get session, and so the largest value one call puts or gets (design 47 G
+// §518, A-21): it fits under the bus's default maximum payload, whatever the bus sets it to.
+const ChunkBytes = 512 << 10
+
+// ObjectsPutChunkRequest is one chunk of a put session (design 47 G §518, A-21): the object `name` in the store `objects`,
+// `call` the bundle's id for this put, `seq` from 1 and one more each chunk, `last` true on the last chunk alone, and the
+// chunk's bytes.
+type ObjectsPutChunkRequest struct {
+	Objects string `json:"objects"`
+	Name    string `json:"name"`
+	Call    string `json:"call"`
+	Seq     uint32 `json:"seq"`
+	Last    bool   `json:"last,omitempty"`
+	Data    []byte `json:"data"`
+}
+
+// ObjectsChunkTaken is the answer to a chunk that is not the last: the sequence taken, or the reason it was not.
+type ObjectsChunkTaken struct {
+	Seq   uint32 `json:"seq"`
+	Error string `json:"error,omitempty"`
+}
+
+// ObjectsChunked is the answer of a get of a value over one chunk (design 47 G §490, §518, §620): the count of chunks (or
+// true, for a sent object), the whole size, its `sha256:` digest and the session's `call`. Its Chunked is absent from the
+// answer of a value read in one call.
+type ObjectsChunked struct {
+	Chunked json.RawMessage `json:"chunked"`
+	Size    uint64          `json:"size"`
+	Digest  string          `json:"digest"`
+	Call    string          `json:"call"`
+}
+
+// ObjectsGetChunkRequest asks one chunk of a get session by its call and sequence.
+type ObjectsGetChunkRequest struct {
+	Call string `json:"call"`
+	Seq  uint32 `json:"seq"`
+}
+
+// ObjectsChunk is a get session's chunk: the sequence, whether it is the last, its bytes, or the reason it is none.
+type ObjectsChunk struct {
+	Seq   uint32 `json:"seq"`
+	Last  bool   `json:"last"`
+	Data  []byte `json:"data"`
+	Error string `json:"error,omitempty"`
+}
+
 // ObjectsNameRequest names one object of an object store: get, delete.
 type ObjectsNameRequest struct {
 	Objects string `json:"objects"`

go/busstate/busstate.go

@@ -31,14 +31,39 @@ func call[T any](method string, params any) (T, error) {
 }
 
 // Put stores one object whole under its name, replacing one of that name, and answers it as the store holds it.
+//
+// A value larger than buscalls.ChunkBytes is put as a session of chunks (design 47 G §518, §345): every object is one
+// object, never split over several.
 func (o Objects) Put(name string, data []byte, description string) (buscalls.ObjectInfo, error) {
+	if len(data) > buscalls.ChunkBytes {
+		return o.putSession(name, data)
+	}
 	return call[buscalls.ObjectInfo](buscalls.ObjectsPut, buscalls.ObjectsPutRequest{Objects: o.local, Name: name,
 		Data: data, Description: description})
 }
 
 // Get is one object, its info and its bytes.
+//
+// A value the runtime answers as a session (`chunked`) is read chunk by chunk and checked against its digest (design 47 G §490).
 func (o Objects) Get(name string) (buscalls.ObjectData, error) {
-	return call[buscalls.ObjectData](buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: o.local, Name: name})
+	raw, err := stdio.BusState(buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: o.local, Name: name})
+	if err != nil {
+		return buscalls.ObjectData{}, err
+	}
+	return ReadObject(name, raw)
+}
+
+// ReadObject is the object a runtime's answer to a get (or to a sent-get, design 47 G §620) holds: the value itself, or the
+// session of chunks it opened, read whole.
+func ReadObject(name string, raw json.RawMessage) (buscalls.ObjectData, error) {
+	if c, ok := chunkedAnswer(raw); ok {
+		return getSession(name, c)
+	}
+	var out buscalls.ObjectData
+	if err := json.Unmarshal(raw, &out); err != nil {
+		return out, fmt.Errorf("the runtime's answer for %s: %w", name, err)
+	}
+	return out, nil
 }
 
 // List is every object the store holds.

go/busstate/session.go

@@ -0,0 +1,110 @@
+package busstate
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// The put and get sessions of a value larger than one chunk (design 47 G §518, §345, §490, A-21): a bundle sends and asks the
+// chunks as calls on its channel to the tool runner, which holds the session; nothing of it is on the bus. The capability
+// word put-sessions names the two calls, so a tool runner that does not state it refuses the session at once.
+
+// callID is the bundle's id for one session: random, so two sessions never share one.
+func callID() string {
+	var b [12]byte
+	_, _ = rand.Read(b[:])
+	return hex.EncodeToString(b[:])
+}
+
+// put sends data as a put session: chunks of buscalls.ChunkBytes in order, each answered before the next is sent, the
+// answer to the last being the put's.
+func (o Objects) putSession(name string, data []byte) (buscalls.ObjectInfo, error) {
+	if err := stdio.RequireCapability(stdio.CapPutSessions); err != nil {
+		return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, larger than one call carries: %w", name, err)
+	}
+	id := callID()
+	for seq, at := uint32(1), 0; ; seq++ {
+		end := min(at+buscalls.ChunkBytes, len(data))
+		last := end == len(data)
+		raw, err := stdio.BusState(buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: o.local, Name: name, Call: id,
+			Seq: seq, Last: last, Data: data[at:end]})
+		if err != nil {
+			return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, chunk %d: %w", name, seq, err)
+		}
+		var taken buscalls.ObjectsChunkTaken
+		if err := json.Unmarshal(raw, &taken); err != nil {
+			return buscalls.ObjectInfo{}, fmt.Errorf("the runtime's answer to a chunk of %s: %w", name, err)
+		}
+		if taken.Error != "" {
+			return buscalls.ObjectInfo{}, fmt.Errorf("putting %s, chunk %d: %s", name, seq, taken.Error)
+		}
+		if last {
+			var info buscalls.ObjectInfo
+			if err := json.Unmarshal(raw, &info); err != nil {
+				return info, fmt.Errorf("the runtime's answer to the put of %s: %w", name, err)
+			}
+			return info, nil
+		}
+		if taken.Seq != seq {
+			return buscalls.ObjectInfo{}, fmt.Errorf("putting %s: chunk %d was answered as chunk %d", name, seq, taken.Seq)
+		}
+		at = end
+	}
+}
+
+// getSession reads the chunks of the session the runtime opened for name, and checks the whole against the digest and the
+// size it answered.
+func getSession(name string, c buscalls.ObjectsChunked) (buscalls.ObjectData, error) {
+	if err := stdio.RequireCapability(stdio.CapPutSessions); err != nil {
+		return buscalls.ObjectData{}, fmt.Errorf("getting %s, larger than one call carries: %w", name, err)
+	}
+	var whole []byte
+	for seq := uint32(1); ; seq++ {
+		raw, err := stdio.BusState(buscalls.ObjectsGetChunk, buscalls.ObjectsGetChunkRequest{Call: c.Call, Seq: seq})
+		if err != nil {
+			return buscalls.ObjectData{}, fmt.Errorf("getting %s, chunk %d: %w", name, seq, err)
+		}
+		var chunk buscalls.ObjectsChunk
+		if err := json.Unmarshal(raw, &chunk); err != nil {
+			return buscalls.ObjectData{}, fmt.Errorf("the runtime's chunk %d of %s: %w", seq, name, err)
+		}
+		if chunk.Error != "" {
+			return buscalls.ObjectData{}, fmt.Errorf("getting %s, chunk %d: %s", name, seq, chunk.Error)
+		}
+		if chunk.Seq != seq {
+			return buscalls.ObjectData{}, fmt.Errorf("getting %s: chunk %d was answered as chunk %d", name, seq, chunk.Seq)
+		}
+		whole = append(whole, chunk.Data...)
+		if chunk.Last {
+			break
+		}
+	}
+	sum := sha256.Sum256(whole)
+	if c.Digest != "" && !strings.EqualFold(c.Digest, "sha256:"+hex.EncodeToString(sum[:])) {
+		return buscalls.ObjectData{}, fmt.Errorf("getting %s: what was read is not the value the runtime named by %s", name, c.Digest)
+	}
+	if c.Size != 0 && c.Size != uint64(len(whole)) {
+		return buscalls.ObjectData{}, fmt.Errorf("getting %s: %d bytes were read, the runtime said %d", name, len(whole), c.Size)
+	}
+	return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: name, Size: uint64(len(whole)), Digest: c.Digest}, Data: whole}, nil
+}
+
+// chunkedAnswer is the session an answer to a get opens, if it is not the value itself.
+func chunkedAnswer(raw json.RawMessage) (buscalls.ObjectsChunked, bool) {
+	var c buscalls.ObjectsChunked
+	if json.Unmarshal(raw, &c) != nil {
+		return c, false
+	}
+	switch strings.TrimSpace(string(c.Chunked)) {
+	case "", "null", "0", "false":
+		return c, false
+	}
+	return c, c.Call != ""
+}

go/busstate/session_test.go

@@ -0,0 +1,123 @@
+package busstate
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// serveRuntime runs a bundle whose tool `go` runs fn in a goroutine and answers every call the bundle makes through handle,
+// until done is closed by fn returning.
+func serveRuntime(t *testing.T, fn func(), handle func(method string, params map[string]any) any) {
+	t.Helper()
+	t.Setenv(stdio.ServedAs, "distribution")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	done := make(chan struct{})
+	go func() {
+		_ = stdio.ServeOn("", []stdio.Tool{{Name: "go", Run: func(map[string]any) (any, error) {
+			go func() { fn(); close(done) }()
+			return "started", nil
+		}}}, inR, outW)
+	}()
+	t.Cleanup(func() { _ = inW.Close() })
+	lines := bufio.NewScanner(outR)
+	lines.Buffer(make([]byte, 0, 1<<20), 64<<20)
+	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"}})
+	deadline := time.After(10 * time.Second)
+	for {
+		got := make(chan map[string]any, 1)
+		go func() {
+			if lines.Scan() {
+				var m map[string]any
+				_ = json.Unmarshal(lines.Bytes(), &m)
+				got <- m
+			}
+		}()
+		select {
+		case <-done:
+			return
+		case <-deadline:
+			t.Fatal("the bundle did not finish")
+		case m := <-got:
+			method, _ := m["method"].(string)
+			if method == "" {
+				continue
+			}
+			p, _ := m["params"].(map[string]any)
+			send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "result": handle(method, p)})
+		}
+	}
+}
+
+// Design 47 G §518, §345, §490 (A-21): a value larger than one chunk is put as chunks in order under one call id, each
+// answered before the next, with the store named in each; a get the runtime answers as a session is read chunk by chunk and
+// checked against its digest.
+func TestAValueLargerThanOneChunkIsPutAndGotBySession(t *testing.T) {
+	value := bytes.Repeat([]byte("0123456789abcdef"), (2*buscalls.ChunkBytes+1000)/16)
+	sum := sha256.Sum256(value)
+	digest := "sha256:" + hex.EncodeToString(sum[:])
+	var stored []byte
+	var seqs []uint32
+	calls := map[string]bool{}
+	var gotValue []byte
+	var putErr, getErr, tamperErr error
+	var tampered bool
+	serveRuntime(t, func() {
+		o := ObjectsOf("incoming")
+		_, putErr = o.Put("big", value, "")
+		d, err := o.Get("big")
+		gotValue, getErr = d.Data, err
+		tampered = true
+		_, tamperErr = o.Get("big")
+	}, func(method string, p map[string]any) any {
+		switch method {
+		case stdio.CapabilitiesCall:
+			return map[string]any{"words": []string{stdio.CapPutSessions}}
+		case buscalls.ObjectsPutChunk:
+			calls[p["call"].(string)] = true
+			if p["objects"] != "incoming" || p["name"] != "big" {
+				t.Errorf("a chunk named %v in %v", p["name"], p["objects"])
+			}
+			raw, _ := json.Marshal(p["data"])
+			var b []byte
+			_ = json.Unmarshal(raw, &b)
+			stored = append(stored, b...)
+			seqs = append(seqs, uint32(p["seq"].(float64)))
+			if p["last"] == true {
+				return buscalls.ObjectInfo{Name: "big", Size: uint64(len(stored))}
+			}
+			return buscalls.ObjectsChunkTaken{Seq: uint32(p["seq"].(float64))}
+		case buscalls.ObjectsGet:
+			d := digest
+			if tampered {
+				d = "sha256:" + hex.EncodeToString(make([]byte, 32))
+			}
+			return map[string]any{"chunked": 3, "size": len(stored), "digest": d, "call": "g1"}
+		case buscalls.ObjectsGetChunk:
+			seq := int(p["seq"].(float64))
+			at := (seq - 1) * buscalls.ChunkBytes
+			return buscalls.ObjectsChunk{Seq: uint32(seq), Last: seq == 3, Data: stored[at:min(at+buscalls.ChunkBytes, len(stored))]}
+		}
+		t.Errorf("unexpected call %s", method)
+		return nil
+	})
+	if putErr != nil || getErr != nil {
+		t.Fatalf("put %v, get %v", putErr, getErr)
+	}
+	if len(seqs) != 3 || seqs[0] != 1 || seqs[1] != 2 || seqs[2] != 3 || len(calls) != 1 || !bytes.Equal(stored, value) || !bytes.Equal(gotValue, value) {
+		t.Errorf("chunks %v under %d call ids, stored %d bytes, got %d", seqs, len(calls), len(stored), len(gotValue))
+	}
+	if tamperErr == nil {
+		t.Error("a value that is not the one the runtime named by its digest was accepted")
+	}
+}

go/stateobjects/bundle.go

@@ -50,12 +50,9 @@ func (b BundleKV) Keys() ([]string, error) { return b.State.Keys() }
 // BundleObjects is an object store as a bundle reaches it.
 type BundleObjects struct{ Store busstate.Objects }
 
-// BundleObjectMost is the largest object a bundle puts in one call: the runtime's channel carries a message of 16 MiB, and
-// the bytes travel in it as base64.
-const BundleObjectMost = 8 << 20
-
-// MostBytes is BundleObjectMost: a larger value is refused, loudly, until the design decides how a bundle keeps one.
-func (BundleObjects) MostBytes() int { return BundleObjectMost }
+// MostBytes is zero: a bundle's value of any size is put by a session of chunks, bounded by its store's cap, which the tool
+// runner refuses at the chunk that passes it (design 47 G §518).
+func (BundleObjects) MostBytes() int { return 0 }
 
 func (b BundleObjects) Put(name string, data []byte) error {
 	_, err := b.Store.Put(name, data, "a value kept beside its key")

go/stateobjects/fakes_test.go

@@ -2,6 +2,8 @@ package stateobjects
 
 import (
 	"bufio"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -25,6 +27,7 @@ type memory struct {
 	failPuts bool
 	modified map[string]time.Time
 	most     int
+	puts     map[string][]byte // a put session's chunks so far, by call
 }
 
 func newMemory() *memory { return &memory{keys: map[string]Entry{}, objects: map[string][]byte{}} }
@@ -216,14 +219,47 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) {
 		return buscalls.Done{}, m.Delete(p.Key)
 	case "mesh/state.keys":
 		return m.Keys()
+	case stdio.CapabilitiesCall:
+		return map[string]any{"words": []string{stdio.CapPutSessions}}, nil
 	case buscalls.ObjectsPut:
 		return buscalls.ObjectInfo{Name: p.Name, Size: uint64(len(p.Data))}, m.Put(p.Name, p.Data)
-	case buscalls.ObjectsGet:
-		d, err := m.Object(p.Name)
-		return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Name}, Data: d}, err
-	case buscalls.ObjectsSentGet:
-		d, err := m.Object(p.Object)
-		return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Object}, Data: d}, err
+	case buscalls.ObjectsPutChunk:
+		// a put session (design 47 G §518): the chunks are kept by call and the object is put on the last
+		var c buscalls.ObjectsPutChunkRequest
+		_ = json.Unmarshal(raw, &c)
+		m.mu.Lock()
+		if m.puts == nil {
+			m.puts = map[string][]byte{}
+		}
+		m.puts[c.Call] = append(m.puts[c.Call], c.Data...)
+		whole := m.puts[c.Call]
+		if c.Last {
+			delete(m.puts, c.Call)
+		}
+		m.mu.Unlock()
+		if !c.Last {
+			return buscalls.ObjectsChunkTaken{Seq: c.Seq}, nil
+		}
+		return buscalls.ObjectInfo{Name: c.Name, Size: uint64(len(whole))}, m.Put(c.Name, whole)
+	case buscalls.ObjectsGet, buscalls.ObjectsSentGet:
+		name := p.Name
+		if method == buscalls.ObjectsSentGet {
+			name = p.Object
+		}
+		d, err := m.Object(name)
+		if err == nil && len(d) > buscalls.ChunkBytes {
+			sum := sha256.Sum256(d)
+			return map[string]any{"chunked": (len(d) + buscalls.ChunkBytes - 1) / buscalls.ChunkBytes, "size": len(d),
+				"digest": "sha256:" + hex.EncodeToString(sum[:]), "call": name}, nil
+		}
+		return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: name}, Data: d}, err
+	case buscalls.ObjectsGetChunk:
+		var c buscalls.ObjectsGetChunkRequest
+		_ = json.Unmarshal(raw, &c)
+		d, err := m.Object(c.Call)
+		at := int(c.Seq-1) * buscalls.ChunkBytes
+		end := min(at+buscalls.ChunkBytes, len(d))
+		return buscalls.ObjectsChunk{Seq: c.Seq, Last: end == len(d), Data: d[at:end]}, err
 	case buscalls.ObjectsDelete:
 		return buscalls.Done{}, memObjects{m}.Delete(p.Name)
 	case buscalls.ObjectsList:

go/stateobjects/send.go

@@ -8,6 +8,7 @@ import (
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/busstate"
 )
 
 // MessageLimit is the most a message of the bus carries when the sender does not know its connection's own
@@ -148,11 +149,8 @@ func BundleFetch(ref Ref) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-	var d buscalls.ObjectData
-	if err := json.Unmarshal(raw, &d); err != nil {
-		return nil, fmt.Errorf("the runtime's answer to %s: %w", buscalls.ObjectsSentGet, err)
-	}
-	return d.Data, nil
+	d, err := busstate.ReadObject(ref.Object, raw)
+	return d.Data, err
 }
 
 // ProcessFetch reads a sent object through the reader's own object store.

go/stateobjects/send_test.go

@@ -41,10 +41,9 @@ func TestABodyOverTheMessageLimitIsSentAsAReference(t *testing.T) {
 	if _, err := Receive(sent, fetch); err == nil {
 		t.Fatal("a tampered object was accepted")
 	}
-	// a bundle's store takes at most what one runtime message carries
-	bundle := Sender{Objects: BundleObjects{}, Store: "m_sent-objects"}
-	if _, err := bundle.Send("reply.huge", make([]byte, BundleObjectMost+1)); err == nil {
-		t.Fatal("a bundle sent a body its runtime cannot carry")
+	// a bundle's store is bounded by its cap alone, and the tool runner refuses at the chunk that passes it (design 47 G §518)
+	if (BundleObjects{}).MostBytes() != 0 {
+		t.Fatal("a bundle's value is refused by a number of its own")
 	}
 }
 

go/stateobjects/values_test.go

@@ -108,7 +108,7 @@ func TestValuesOfAnySizeThroughTheRuntime(t *testing.T) {
 	m := newMemory()
 	out := runtimeOf(t, m, func() (any, error) {
 		v := ForState("views")
-		valueRows(t, v, 1, 10) // a bundle puts at most 8 MiB in an object: 100x is a process's
+		valueRows(t, v, 1, 10, 100) // a bundle puts a value of any size by session (design 47 G §518)
 		return "done", nil
 	})
 	if !strings.Contains(out.(string), "done") {