mesh-hand-acts answers bounded, newest first; mesh-healers seeds from a partial answer (issue 1966 S8)

hand-acts read the whole log (84k entries, 47 MB) on every call and could never answer inside the call bound or
the bus payload. It now reads from the newest entry backwards, page by page, and stops at the window's start, at
900 KiB of acts or at 20 s, saying complete: false, cut and why. warrants is bounded the same way. mesh-healers'
start-up seed takes a partial answer and logs it, instead of retrying forever.

Quick fix: this (mesh-catalog only, about 1.5 h). Thorough fix: also the tool runner never freezing on an oversized
bundle line and logging unanswered calls (mesh-tools, a separate fix by another developer, lead's ruling Q3), and an
SDK-side bound. Chosen here: quick for this module, the thorough part running beside it; the byte cap is the
module's own (lead's ruling Q1, a `limit` argument is a later note).

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

modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go

@@ -62,21 +62,38 @@ type Store interface {
 	Get(state, key string) ([]byte, bool, error)
 	Create(state, key string, value any) error
 	Keys(state string) ([]string, error)
-	// Append adds an entry to the end of a log; Last is a key's newest entry; Entries every entry in order; Fill how
-	// many entries and bytes the log holds against its cap.
+	// Append adds an entry to the end of a log; Last is a key's newest entry; Page at most limit entries from a
+	// sequence on, oldest first; Fill how many entries and bytes the log holds against its cap, and its first and
+	// last sequence.
 	Append(log, key string, value any) error
 	Last(log, key string) ([]byte, bool, error)
-	Entries(log string) ([][]byte, error)
+	Page(log string, from uint64, limit int) ([]PageEntry, error)
 	Fill(log string) (Fill, error)
 }
 
-// Fill is how full a log is: its entries, its bytes and its cap in bytes.
+// PageEntry is one entry of a log read: its sequence and its value.
+type PageEntry struct {
+	Seq   uint64
+	Value []byte
+}
+
+// Fill is how full a log is: its entries, its bytes and its cap in bytes, and its first and last sequence.
 type Fill struct {
 	Entries  uint64 `json:"entries"`
 	Bytes    uint64 `json:"bytes"`
 	MaxBytes uint64 `json:"max-bytes"`
+	First    uint64 `json:"first"`
+	Last     uint64 `json:"last"`
 }
 
+// What bounds one answer, so it always comes, whatever the log holds (issue 1966): the bus carries at most 1 MiB, and
+// the runtime's call is bound at 30 s. A read stops at either and says so.
+const (
+	MaxAnswerBytes = 900 << 10
+	ReadBudget     = 20 * time.Second
+	pageSize       = 1000
+)
+
 // ErrExists is a create of a key another writer made first.
 var ErrExists = errors.New("the key was made by another writer first")
 
@@ -281,7 +298,13 @@ type Listing struct {
 	Days     int            `json:"days"`
 	// Unread says which log could not be read, so what is missing is said and never read as nothing.
 	Unread map[string]string `json:"unread,omitempty"`
-	Kept   map[string]int    `json:"kept"`
+	// Kept is how many entries of the log were read, newest first.
+	Kept map[string]int `json:"kept"`
+	// Complete is false when the read stopped before the window's start (Why says at what, Cut how many older entries
+	// of the log were not read): what is listed is the newest part, and Repeated counts that part alone.
+	Complete bool   `json:"complete"`
+	Cut      uint64 `json:"cut,omitempty"`
+	Why      string `json:"why,omitempty"`
 	// Fill is how full this seat's log is against its cap: full, it refuses the next act.
 	Fill *Fill `json:"fill,omitempty"`
 }
@@ -291,11 +314,12 @@ func (l *Log) HandActs(days int, cause, who string) Listing {
 	now := l.Now()
 	since := now.Add(-time.Duration(days) * 24 * time.Hour)
 	out := Listing{Acts: []Act{}, Repeated: map[string]int{}, Days: days, Kept: map[string]int{}}
-	all, kept, err := l.read(now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin)))
+	all, kept, cut, err := l.read(now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin)))
 	if err != nil {
 		out.Unread = map[string]string{ActsLog: err.Error()}
 	} else {
 		out.Kept[ActsLog] = kept
+		out.Complete, out.Cut, out.Why = cut.why == "", cut.unread, cut.why
 	}
 	for _, a := range all {
 		if now.Sub(a.At) <= RepeatedWithin && handactsclient.Repair(a) {
@@ -318,20 +342,57 @@ func (l *Log) HandActs(days int, cause, who string) Listing {
 	return out
 }
 
-// read is every act of this seat's log since a moment, entry by entry, and how many it keeps.
-func (l *Log) read(since time.Time) ([]Act, int, error) {
-	var out []Act
-	entries, err := l.Store.Entries(ActsLog)
+// cutRead says why a read stopped before the window's start, and how many older entries it left unread.
+type cutRead struct {
+	why    string
+	unread uint64
+}
+
+// read is the acts of this seat's log since a moment, from the newest entry backwards, page by page: it stops at the
+// first entry older than the moment (the log is in time order), at MaxAnswerBytes of acts or at ReadBudget, and says
+// which. It also answers how many entries it read.
+func (l *Log) read(since time.Time) ([]Act, int, cutRead, error) {
+	f, err := l.Store.Fill(ActsLog)
 	if err != nil {
-		return nil, 0, err
+		return nil, 0, cutRead{}, err
 	}
-	for _, raw := range entries {
-		var a Act
-		if json.Unmarshal(raw, &a) != nil || a.At.Before(since) {
-			continue
+	var out []Act
+	var size, read int
+	deadline := l.Now().Add(ReadBudget)
+	for to := f.Last; to > 0 && to >= f.First; {
+		if l.Now().After(deadline) {
+			return out, read, cutRead{"the read took more than " + ReadBudget.String(), to - f.First + 1}, nil
+		}
+		from := f.First
+		if to-from+1 > pageSize {
+			from = to - pageSize + 1
+		}
+		page, err := l.Store.Page(ActsLog, from, pageSize)
+		if err != nil {
+			return nil, 0, cutRead{}, err
+		}
+		for i := len(page) - 1; i >= 0; i-- {
+			if page[i].Seq > to {
+				continue
+			}
+			read++
+			var a Act
+			if json.Unmarshal(page[i].Value, &a) != nil {
+				continue
+			}
+			if a.At.Before(since) {
+				return out, read, cutRead{}, nil
+			}
+			if size += len(page[i].Value); size > MaxAnswerBytes {
+				return out, read, cutRead{"the answer reached " + strconv.Itoa(MaxAnswerBytes>>10) + " KiB", page[i].Seq - f.First + 1}, nil
+			}
+			a.Log = handacts.Seat
+			out = append(out, a)
+		}
+		if from <= f.First {
+			break
 		}
-		a.Log = handacts.Seat
-		out = append(out, a)
+		to = from - 1
 	}
-	return out, len(entries), nil
+	return out, read, cutRead{}, nil
 }

modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -20,6 +21,7 @@ type memStore struct {
 	kv      map[string]map[string][]byte
 	logs    map[string][]entry
 	keysErr map[string]error
+	pages   int
 }
 
 func newStore() *memStore {
@@ -51,15 +53,16 @@ func (s *memStore) Last(log, key string) ([]byte, bool, error) {
 	return nil, false, nil
 }
 
-func (s *memStore) Entries(log string) ([][]byte, error) {
+func (s *memStore) Page(log string, from uint64, limit int) ([]PageEntry, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if err := s.keysErr[log]; err != nil {
 		return nil, err
 	}
-	var out [][]byte
-	for _, e := range s.logs[log] {
-		out = append(out, e.raw)
+	s.pages++
+	var out []PageEntry
+	for i := max(int(from), 1) - 1; i < len(s.logs[log]) && len(out) < limit; i++ {
+		out = append(out, PageEntry{Seq: uint64(i + 1), Value: s.logs[log][i].raw})
 	}
 	return out, nil
 }
@@ -67,7 +70,8 @@ func (s *memStore) Entries(log string) ([][]byte, error) {
 func (s *memStore) Fill(log string) (Fill, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return Fill{Entries: uint64(len(s.logs[log])), MaxBytes: 1 << 30}, nil
+	n := uint64(len(s.logs[log]))
+	return Fill{Entries: n, MaxBytes: 1 << 30, First: min(n, 1), Last: n}, nil
 }
 
 // logKeys are a log's keys in order.
@@ -242,12 +246,13 @@ func TestHandActsReadsItsOwnLogAlone(t *testing.T) {
 		}
 		_ = store.Put(log, actKey(at), a)
 	}
-	put(ActsLog, VerbRecord, "relogin", "person.jochen", time.Hour)
+	// The log is in time order, oldest first, as the bus keeps it.
+	put(ActsLog, VerbRecord, "config", "ace.mesh-config", 30*24*time.Hour)
 	put(ActsLog, VerbRecord, "relogin", "shanks.claude-code", 3*24*time.Hour)
 	put(ActsLog, VerbDrill, CauseDrill, "person.jochen", 2*time.Hour)
-	put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour)
 	put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", 2*time.Hour)
-	put(ActsLog, VerbRecord, "config", "ace.mesh-config", 30*24*time.Hour)
+	put(ActsLog, VerbRecord, "relogin", "person.jochen", time.Hour)
+	put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour)
 	put(controllerActs, VerbRecord, "never-read", "ace.old", time.Hour)
 	store.keysErr[controllerActs] = errors.New("the controller's bucket must not be read")
 	cases := []struct {
@@ -277,10 +282,50 @@ func TestHandActsReadsItsOwnLogAlone(t *testing.T) {
 		got.Fill.Entries != 6 {
 		t.Errorf("the log of an act or the kept counts are not said: %+v", got.Kept)
 	}
+	if got := l.HandActs(14, "", ""); !got.Complete || got.Cut != 0 || got.Why != "" {
+		t.Errorf("a window within the log is read whole: %+v", got)
+	}
+	// An 84,000-entry log with 3 acts in the window reads one page and stops at the window's start (issue 1966).
+	big, bigStore, _, bnow := newLog()
+	for i := 0; i < 84000; i++ {
+		at := bnow.Add(-20*24*time.Hour + time.Duration(i)*time.Second)
+		if i >= 83997 {
+			at = bnow.Add(-time.Duration(84000-i) * time.Minute)
+		}
+		_ = bigStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, Cause: "flood", By: "x", Why: "x"})
+	}
+	bigStore.pages = 0
+	if got := big.HandActs(1, "", ""); len(got.Acts) != 3 || !got.Complete || bigStore.pages != 1 {
+		t.Errorf("the window's 3 acts of 84,000: %d acts, complete %v, %d pages", len(got.Acts), got.Complete, bigStore.pages)
+	}
+	// A fortnight of it is cut at the answer's cap: partial, said, and under the bus's 1 MiB.
+	for i := 0; i < 40; i++ {
+		at := bnow.Add(-time.Duration(40-i) * time.Second)
+		_ = bigStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, Cause: "flood", By: "x", Why: strings.Repeat("w", 40<<10)})
+	}
+	got := big.HandActs(14, "", "")
+	if raw, _ := json.Marshal(got); got.Complete || got.Cut == 0 || got.Why == "" || len(raw) > 1<<20 || len(got.Acts) < 20 ||
+		got.Acts[0].At.Before(got.Acts[1].At) {
+		t.Errorf("a window over the cap: complete %v, cut %d, why %q, %d acts, %d bytes", got.Complete, got.Cut, got.Why, len(got.Acts), len(raw))
+	}
+	// A read that takes longer than its budget stops and says so.
+	slow := &Log{Store: bigStore, Logf: func(string, ...any) {}}
+	slowNow := *bnow
+	slow.Now = func() time.Time { slowNow = slowNow.Add(ReadBudget + time.Second); return slowNow }
+	if got := slow.HandActs(14, "", ""); got.Complete || !strings.Contains(got.Why, "took more than") {
+		t.Errorf("a slow read: complete %v, why %q", got.Complete, got.Why)
+	}
 	store.keysErr[ActsLog] = errors.New("the bus is away")
 	if got := l.HandActs(14, "", ""); got.Unread[ActsLog] == "" || len(got.Acts) != 0 {
 		t.Errorf("an unreadable log: %+v", got.Unread)
 	}
+	for i := 0; i < 30; i++ {
+		_ = store.Put(WarrantsState, "asker.ask-"+strconv.Itoa(i), Warrant{Act: strings.Repeat("w", 40<<10)})
+	}
+	if w, err := l.Warrants(); err != nil || w.(map[string]any)["complete"] != false || w.(map[string]any)["cut"].(int) == 0 {
+		t.Errorf("warrants over the cap answered %v, %v", w, err)
+	}
+	store.kv[WarrantsState] = nil
 	w, err := l.Warrants()
 	if err != nil || w.(map[string]any)["count"] != 0 {
 		t.Errorf("warrants answered %v, %v", w, err)

modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go

@@ -49,21 +49,16 @@ func (runtimeStore) Last(log, key string) ([]byte, bool, error) {
 	return e.Value, true, nil
 }
 
-func (runtimeStore) Entries(log string) ([][]byte, error) {
-	var out [][]byte
-	for from := uint64(0); ; {
-		page, err := stdio.Log(log).Read(from, "", 1000)
-		if err != nil {
-			return nil, err
-		}
-		for _, e := range page.Entries {
-			out = append(out, e.Value)
-		}
-		if page.Next == 0 {
-			return out, nil
-		}
-		from = page.Next
+func (runtimeStore) Page(log string, from uint64, limit int) ([]PageEntry, error) {
+	page, err := stdio.Log(log).Read(from, "", limit)
+	if err != nil {
+		return nil, err
 	}
+	out := make([]PageEntry, 0, len(page.Entries))
+	for _, e := range page.Entries {
+		out = append(out, PageEntry{Seq: e.Sequence, Value: e.Value})
+	}
+	return out, nil
 }
 
 func (runtimeStore) Fill(log string) (Fill, error) {
@@ -71,7 +66,7 @@ func (runtimeStore) Fill(log string) (Fill, error) {
 	if err != nil {
 		return Fill{}, err
 	}
-	return Fill{Entries: i.Entries, Bytes: i.Bytes, MaxBytes: i.MaxBytes}, nil
+	return Fill{Entries: i.Entries, Bytes: i.Bytes, MaxBytes: i.MaxBytes, First: i.First, Last: i.Last}, nil
 }
 
 func main() {

modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go

@@ -96,12 +96,17 @@ func (l *Log) Warrants() (any, error) {
 	}
 	sort.Strings(keys)
 	rows := map[string]json.RawMessage{}
-	for _, k := range keys {
+	size := 0
+	for i, k := range keys {
 		if raw, found, err := l.Store.Get(WarrantsState, k); err == nil && found {
+			if size += len(raw) + len(k); size > MaxAnswerBytes {
+				// Bounded like hand-acts: the answer always comes, and says what it left out.
+				return map[string]any{"warrants": rows, "count": len(rows), "complete": false, "cut": len(keys) - i}, nil
+			}
 			rows[k] = raw
 		}
 	}
-	return map[string]any{"warrants": rows, "count": len(rows)}, nil
+	return map[string]any{"warrants": rows, "count": len(rows), "complete": true}, nil
 }
 
 func arg(a map[string]any, k string) string {

modules/mesh-healers/cmd/mesh-healers/healers_test.go

@@ -360,6 +360,14 @@ func TestACauseRepairedByHandTwiceIsAHealerWanted(t *testing.T) {
 	if n, err := w.h.Seeded([]byte(`{"acts":[` + string(seatAct) + `]}`)); err != nil || n != 1 {
 		t.Fatalf("the seat's listing was not taken: %d, %v", n, err)
 	}
+	// A partial answer (the seat bounds it) seeds all the same and says it is partial, so the start never retries forever (1966).
+	var said []string
+	w.h.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
+	if n, err := w.h.Seeded([]byte(`{"acts":[` + string(seatAct) + `],"complete":false,"why":"the answer reached 900 KiB","cut":83000}`)); err != nil ||
+		n != 1 || len(said) != 1 || !strings.Contains(said[0], "partial: the answer reached 900 KiB") {
+		t.Fatalf("a partial listing: %d, %v, said %v", n, err, said)
+	}
+	w.h.Logf = func(string, ...any) {}
 	// A drill in each log counts none.
 	act("a3", "hand-act drill", "drill", time.Hour)
 	drill, _ := json.Marshal(HandAct{ID: "a4", At: w.now.Add(-2 * time.Hour), Verb: "hand-act drill", Why: "x", Cause: "drill"})

modules/mesh-healers/cmd/mesh-healers/wanted.go

@@ -95,7 +95,10 @@ func (h *Healers) ActRecorded(body []byte) {
 // answers how many it took.
 func (h *Healers) Seeded(answer []byte) (int, error) {
 	var l struct {
-		Acts []json.RawMessage `json:"acts"`
+		Acts     []json.RawMessage `json:"acts"`
+		Complete *bool             `json:"complete"`
+		Why      string            `json:"why"`
+		Cut      uint64            `json:"cut"`
 	}
 	if err := json.Unmarshal(answer, &l); err != nil {
 		return 0, fmt.Errorf("the seat's hand-acts answer is not a list of acts: %w", err)
@@ -103,6 +106,11 @@ func (h *Healers) Seeded(answer []byte) (int, error) {
 	for _, raw := range l.Acts {
 		h.ActRecorded(raw)
 	}
+	// A partial answer (the seat bounds it, issue 1966) seeds all the same: the newest acts are what a fortnight's
+	// count needs most, and the older ones are heard no more, so asking again would only repeat it.
+	if l.Complete != nil && !*l.Complete {
+		h.Logf("[%s] took %d of the fortnight's hand acts, the newest (partial: %s; %d older entries not read)", Module, len(l.Acts), l.Why, l.Cut)
+	}
 	return len(l.Acts), nil
 }