RR2-5: the reader reads a seat-named view in consumes and a seat's views

G's builder and checker follow the assigner's views by the seat,
seat:mesh-module-assigner.views.assignments; SeatViewConsume splits that
form, and a seat compiled from its definition names the states it offers
as views, which the validator checks the form against.

Builds: design 47 G §621, §616, §699

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

go/busretry/busretry.go

@@ -8,10 +8,13 @@
 // have been delivered. The wait is bounded (Bound, about 30 s, backing off), and said on standard error when it ends
 // without an answer.
 //
-// This is the quick fix. The thorough one is a reload that keeps the account's imports, in the bus's holder.
+// DoRepeatable is the form for a call its caller keeps the answer of and marks safe to make again (a read, or a write
+// carrying its own message id): it also asks again on a timeout, and its blackout probe is optional (novox/hq ADR 0668,
+// extending ADR 0615 §6 to every kept call; issue 1966).
 package busretry
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -40,7 +43,8 @@ func Waits(bound, first, most time.Duration) (restore func()) {
 }
 
 // Undelivered says err means the request reached no one: no responders, JetStream not enabled (for the account), or
-// no response from the stream after the client's own retries on no responders; also said in words by a runtime that
+// no response from the stream after the client's own retries on no responders (either JetStream client's: the old
+// one's publish says it for no responders, review round 1 of issue 1966); also said in words by a runtime that
 // passed the error on.
 func Undelivered(err error) bool {
 	var answered *Answered
@@ -49,6 +53,7 @@ func Undelivered(err error) bool {
 	}
 	if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, jetstream.ErrJetStreamNotEnabled) ||
 		errors.Is(err, jetstream.ErrJetStreamNotEnabledForAccount) || errors.Is(err, jetstream.ErrNoStreamResponse) ||
+		errors.Is(err, nats.ErrNoStreamResponse) ||
 		errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) {
 		return true
 	}
@@ -66,8 +71,32 @@ func (a *Answered) Error() string { return a.Words }
 // given, is asked after each such error and Do stops waiting when it says the bus is not blacked out: a call nobody
 // answers because its callee is not running fails at once, as before. what names the request in the error.
 func Do[T any](what string, try func() (T, error), blackout func() bool) (T, error) {
+	return do(what, try, func(err error) bool { return Undelivered(err) && (blackout == nil || blackout()) })
+}
+
+// DoRepeatable is Do for a call the caller marks safe to make again (a read, or a write carrying its message id, which
+// the callee deduplicates): a timeout is asked again too, and with no blackout probe a request that reached no one is
+// asked again whether or not the bus is blacked out, since a callee that is restarting answers nothing for seconds
+// (novox/hq ADR 0668; issue 1966). Past Bound the error says which fault it last met.
+func DoRepeatable[T any](what string, try func() (T, error), blackout func() bool) (T, error) {
+	return do(what, try, func(err error) bool {
+		return TimedOut(err) || (Undelivered(err) && (blackout == nil || blackout()))
+	})
+}
+
+// TimedOut says err is a request's deadline passing without an answer: the request may have been delivered, so only a
+// call marked safe to make again (DoRepeatable) asks it again. A callee's own answer (Answered) never counts.
+func TimedOut(err error) bool {
+	var answered *Answered
+	if err == nil || errors.As(err, &answered) {
+		return false
+	}
+	return errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout)
+}
+
+func do[T any](what string, try func() (T, error), again func(error) bool) (T, error) {
 	v, err := try()
-	if !Undelivered(err) || (blackout != nil && !blackout()) {
+	if !again(err) {
 		return v, err
 	}
 	start, wait := time.Now(), FirstWait
@@ -75,12 +104,17 @@ func Do[T any](what string, try func() (T, error), blackout func() bool) (T, err
 		time.Sleep(wait)
 		wait = min(wait*2, MostWait)
 		v, err = try()
-		if !Undelivered(err) || (blackout != nil && !blackout()) {
+		if !again(err) {
 			return v, err
 		}
 	}
-	err = fmt.Errorf("%s: the bus answered nothing for %s, asked again until then (a user-list reload drops the "+
-		"account's imports for a few seconds; this was longer): %w", what, time.Since(start).Round(time.Second), err)
+	if TimedOut(err) {
+		err = fmt.Errorf("%s: no answer within its deadline, asked again for %s: %w", what, time.Since(start).Round(time.Second), err)
+	} else {
+		err = fmt.Errorf("%s: the bus answered nothing for %s, asked again until then (a user-list reload drops the "+
+			"account's imports for a few seconds, a callee that restarts answers nothing; this was longer): %w", what,
+			time.Since(start).Round(time.Second), err)
+	}
 	fmt.Fprintf(os.Stderr, "[mesh-sdk] %v\n", err)
 	return v, err
 }

go/busretry/busretry_test.go

@@ -16,30 +16,43 @@ import (
 // bound; a blackout probe that says the bus answers ends the wait at once.
 func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
 	for _, row := range []struct {
-		name      string
-		errs      []error // what each try answers; the last repeats
-		blackout  func() bool
-		tries     int // at least
-		atMost    int
-		delivered bool
+		name       string
+		errs       []error // what each try answers; the last repeats
+		blackout   func() bool
+		tries      int // at least
+		atMost     int
+		delivered  bool
+		repeatable bool // DoRepeatable, a call marked safe to make again (ADR 0668)
 	}{
-		{"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true},
-		{"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true},
-		{"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true},
-		{"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false},
-		{"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false},
-		{"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false},
-		{"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 2, 1000, false},
+		{"no responders, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, nil, 3, 3, true, false},
+		{"jetstream not enabled, then an answer", []error{jetstream.ErrJetStreamNotEnabled, nil}, nil, 2, 2, true, false},
+		{"said in words by a runtime", []error{errors.New("nats: no responders available for request"), nil}, nil, 2, 2, true, false},
+		{"a timeout may have been delivered: never asked again", []error{context.DeadlineExceeded}, nil, 1, 1, false, false},
+		{"an answer that refuses is an answer", []error{errors.New("refused")}, nil, 1, 1, false, false},
+		{"nobody answers on a bus that answers: the callee is not running", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, false},
+		{"blacked out past the bound: said", []error{jetstream.ErrJetStreamNotEnabledForAccount}, func() bool { return true }, 2, 1000, false, false},
+		// Issue 1966, ADR 0668: a call kept and safe to make again is asked again on a timeout, and on no responders
+		// from a callee that restarts on a bus that answers; Do still asks neither.
+		{"repeatable: timed out twice, then an answer", []error{context.DeadlineExceeded, nats.ErrTimeout, nil}, nil, 3, 3, true, true},
+		{"repeatable: no responders with no probe, then an answer", []error{nats.ErrNoResponders, nil}, nil, 2, 2, true, true},
+		{"repeatable: a callee's answer is an answer", []error{&Answered{Words: "context deadline exceeded"}}, nil, 1, 1, false, true},
+		{"repeatable: nobody answers and the probe says the bus answers", []error{nats.ErrNoResponders}, func() bool { return false }, 1, 1, false, true},
+		{"repeatable: no response from the stream, either client's, then an answer", []error{nats.ErrNoStreamResponse, jetstream.ErrNoStreamResponse, nil}, nil, 3, 3, true, true},
+		{"timed out past the bound: said", []error{context.DeadlineExceeded}, nil, 2, 1000, false, true},
 	} {
 		// A row that ends in an answer waits on a bound of seconds, which no slow machine reaches; only the row that
 		// runs past it has a short one, and asks only that it was asked again at least once.
 		bound := 10 * time.Second
-		if strings.HasPrefix(row.name, "blacked out") {
+		if strings.Contains(row.name, "past the bound") {
 			bound = 100 * time.Millisecond
 		}
 		t.Cleanup(Waits(bound, time.Millisecond, 4*time.Millisecond))
 		tries := 0
-		_, err := Do("the test", func() (int, error) {
+		do := Do[int]
+		if row.repeatable {
+			do = DoRepeatable[int]
+		}
+		_, err := do("the test", func() (int, error) {
 			e := row.errs[min(tries, len(row.errs)-1)]
 			tries++
 			return tries, e
@@ -50,5 +63,8 @@ func TestABusThatAnswersNothingIsWaitedOut(t *testing.T) {
 		if strings.HasPrefix(row.name, "blacked out") && (!Undelivered(err) || !strings.Contains(fmt.Sprint(err), "the bus answered nothing")) {
 			t.Errorf("%s: %v", row.name, err)
 		}
+		if strings.HasPrefix(row.name, "timed out") && (!TimedOut(err) || !strings.Contains(fmt.Sprint(err), "the test: no answer within its deadline")) {
+			t.Errorf("%s: %v", row.name, err)
+		}
 	}
 }

go/caller_test.go

@@ -3,6 +3,7 @@ package stdio
 import (
 	"bufio"
 	"encoding/json"
+	"errors"
 	"io"
 	"testing"
 	"time"
@@ -183,3 +184,44 @@ func TestArgumentsOfNoCallHaveNoCaller(t *testing.T) {
 		t.Fatalf("CallerOf(nil) answered %q", got)
 	}
 }
+
+// A holder's caller list is held (issue 1966): the runtime is asked once, by a watch; a later call never asks again, so
+// an ask that timed out refuses no verb, and a change that cannot be read keeps the list last read.
+func TestAHeldCallerListIsAskedOnceAndSurvivesAnUnreadableChange(t *testing.T) {
+	verbs, defaults := []string{"publish"}, VerbCallers{"publish": {"a"}}
+	var asks int
+	var push func(ProvisionChange) error
+	fail := true
+	read := holdVerbCallers(func(key string, h func(ProvisionChange) error) error {
+		asks++
+		if fail {
+			return errors.New("nats: timeout")
+		}
+		push = h
+		return h(ProvisionChange{Kind: "config", Key: key, Value: json.RawMessage(`{"publish":["b"]}`)})
+	}, "callers", verbs, defaults)
+	if _, err := read(); err == nil {
+		t.Fatal("a list never read was answered")
+	}
+	fail = false
+	c, err := read()
+	if err != nil || c.Allowed("publish", "ace.b") != nil || asks != 2 {
+		t.Fatalf("after the ask answered: %v, %v, asked %d times", c, err, asks)
+	}
+	for i := 0; i < 3; i++ {
+		if _, err := read(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if asks != 2 {
+		t.Errorf("a held list was asked again: %d asks", asks)
+	}
+	_ = push(ProvisionChange{Kind: "config", Key: "callers", Value: json.RawMessage(`{"nothing":["x"]}`)})
+	if c, err := read(); err != nil || c.Allowed("publish", "ace.b") != nil {
+		t.Errorf("an unreadable change dropped the list last read: %v, %v", c, err)
+	}
+	_ = push(ProvisionChange{Kind: "config", Key: "callers", Value: json.RawMessage(`{"publish":["c"]}`)})
+	if c, _ := read(); c.Allowed("publish", "ace.c") != nil || c.Allowed("publish", "ace.b") == nil {
+		t.Errorf("a change was not held: %v", c)
+	}
+}

go/callers.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"slices"
 	"strings"
+	"sync"
 )
 
 // Callers is a module's trusted list of who may call a verb of its (novox/hq ADR 0459 §2, ADR 0563 §8), as the mesh
@@ -122,6 +123,66 @@ func ConfiguredVerbCallers(key string, verbs []string, defaults VerbCallers) (Ve
 	return parseConfiguredVerbCallers(raw, verbs, defaults)
 }
 
+// HoldVerbCallers is ConfiguredVerbCallers held: the returned reader asks the runtime once, by a watch of the config
+// value key (WatchConfig), and from then on answers the list it holds, which each change of the value replaces. A call
+// never asks the runtime again, so a runtime that is slow or away for a moment (a timed-out ask) refuses no verb; a
+// change that cannot be read keeps the list last read and is said on the bundle's standard error, so a broken list is
+// said and never widened. Only a reader that has not yet read any value answers an error, and its next call asks again.
+func HoldVerbCallers(key string, verbs []string, defaults VerbCallers) func() (VerbCallers, error) {
+	return holdVerbCallers(WatchConfig, key, verbs, defaults)
+}
+
+func holdVerbCallers(watch func(string, func(ProvisionChange) error) error, key string, verbs []string, defaults VerbCallers) func() (VerbCallers, error) {
+	var start, mu sync.Mutex
+	var held VerbCallers
+	var bad error
+	handle := func(c ProvisionChange) error {
+		mu.Lock()
+		defer mu.Unlock()
+		if c.Refused != "" {
+			bad = fmt.Errorf("the config value %s is not answered: %s", key, c.Refused)
+		} else if l, err := parseConfiguredVerbCallers(c.Value, verbs, defaults); err != nil {
+			bad = fmt.Errorf("the config value %s cannot be read: %w", key, err)
+		} else {
+			held, bad = l, nil
+			return nil
+		}
+		if held != nil {
+			sayf("%v: the list last read stays", bad)
+		}
+		return nil
+	}
+	read := func() (VerbCallers, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if held != nil {
+			return held, nil
+		}
+		if bad != nil {
+			return nil, bad
+		}
+		return nil, nil
+	}
+	return func() (VerbCallers, error) {
+		if l, err := read(); l != nil || err != nil {
+			return l, err
+		}
+		start.Lock()
+		defer start.Unlock()
+		if l, err := read(); l != nil || err != nil {
+			return l, err
+		}
+		if err := watch(key, handle); err != nil {
+			return nil, fmt.Errorf("the config value %s cannot be read: %w", key, err)
+		}
+		l, err := read()
+		if l == nil && err == nil {
+			err = fmt.Errorf("the config value %s cannot be read: the runtime handed over no value", key)
+		}
+		return l, err
+	}
+}
+
 func parseConfiguredVerbCallers(raw json.RawMessage, verbs []string, defaults VerbCallers) (VerbCallers, error) {
 	var text string
 	if err := json.Unmarshal(raw, &text); err == nil || len(bytes.TrimSpace(raw)) == 0 || string(bytes.TrimSpace(raw)) == "null" {

go/catalogue/busstate.go

@@ -142,6 +142,8 @@ type BusStateEntry struct {
 	Configuration map[string]json.RawMessage `json:"configuration,omitempty"`
 	// Types are a ledger's types (ADR 0597 §1), for the storage type ledger only.
 	Types []LedgerType `json:"types,omitempty"`
+	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
+	KeptLayout
 }
 
 // isBusStateEntry is whether a `requires` item in its long form asks for bus state: it names a storage type.
@@ -177,6 +179,9 @@ func readBusStateEntry(raw json.RawMessage) (BusStateEntry, []string, error) {
 		var e BusStateEntry
 		err := strict(&e)
 		if err == nil {
+			if err := e.KeptLayout.refuse(fmt.Sprintf("requires entry %q", e.Name)); err != nil {
+				return BusStateEntry{}, nil, err
+			}
 			return e, aside, nil
 		}
 		unknown := asUnknownField(err)
@@ -205,3 +210,17 @@ func setAsideAlso(unknown string, paths []string) string {
 	}
 	return unknown + " (read without " + strings.Join(paths, ", ") + ")"
 }
+
+// KeptStates are the local names of the state a manifest keeps, each a key-value bucket of its own: the old `state`
+// section's and every key-value entry it asks for in `requires` (ADR 0595 §1, §11), sorted, once each. It is the one
+// rule every reader of what a module keeps calls (novox/hq issue 1966): a process's `bus.state` and a seat
+// contribution's state name one of these, in either form.
+func KeptStates(m Manifest) []string {
+	var out []string
+	for _, k := range KeptDeclarations(m) {
+		if k.Kind == KeptKindState {
+			out = append(out, k.Name)
+		}
+	}
+	return out
+}

go/catalogue/delivery_seat_test.go

@@ -4,6 +4,8 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
 )
 
 // The delivery seat answers "what did the mesh's checks say of this pull request?" as a verb of its own,
@@ -58,3 +60,18 @@ func TestTheDeliverySeatPromisesChecks(t *testing.T) {
 		t.Fatalf("a holder without %s: %v", before[0], err)
 	}
 }
+
+// An event whose subject carries parts is said by its subject with `*` for each part, so a consumer naming one
+// instance or every instance of it is heard as the seat's (design 47 *What each step says*).
+func TestAnEventWithPartsIsTheSeatsByItsSubject(t *testing.T) {
+	s := SeatFromDefinition(seatdef.Definition{Seat: "mesh-delivery", Scope: "mesh", Events: []seatdef.Message{
+		{Name: "owed", Subject: "owed.<repository>.<number>"}, {Name: "pull.updated"}}})
+	for _, c := range []struct {
+		event string
+		says  bool
+	}{{"owed.*.*", true}, {"owed.novox-mesh-sdk.12", true}, {"owed", false}, {"pull.updated", true}, {"pull.merged", false}} {
+		if got := SeatSays(s.Emits, c.event); got != c.says {
+			t.Errorf("%s: says %v, want %v (emits %v)", c.event, got, c.says, s.Emits)
+		}
+	}
+}

go/catalogue/kept_form.go

@@ -0,0 +1,146 @@
+package catalogue
+
+import (
+	"fmt"
+	"sort"
+)
+
+// A kept data's form (novox/hq design 47, *A module's kept data on the bus, backed up and restored with its delivery*):
+// every kept data a manifest declares, in an old section (`state`, `logs`, `streams`, `objects`) or in `requires` by
+// template (the five kinds: state, stream, objects, ledger, log), names a `form`, a number its holder raises whenever it
+// changes how it writes that data, or stops declaring it; a declaration naming none reads as form 0. Beside it, the
+// catch-up tool: the module's own tool that brings the data forward once a restore point of it is restored, recorded in
+// the restore point when it is taken. The declaration alone cannot tell a changed layout of the same bucket, so the holder
+// says it. Which form is right, and that the catch-up tool is the module's own, is the indexer's judge's.
+
+// KeptLayout is what a kept data's declaration says of the layout its holder writes: its form and its catch-up tool.
+type KeptLayout struct {
+	// Form is the form of the kept data's layout, 0 where none is said.
+	Form int `json:"form,omitempty"`
+	// CatchUp is the module's own tool a restore of this kept data is followed by, named by the manifest that raised its
+	// form; empty where none is.
+	CatchUp string `json:"catch-up,omitempty"`
+}
+
+// refuse is the one reading of a layout's form every kept kind's decoder calls: a form is a whole number from 0, and one
+// below it is refused here, naming the declaration (`what` is its kind and name, such as `state "seen"`).
+func (k KeptLayout) refuse(what string) error {
+	if k.Form < 0 {
+		return fmt.Errorf("%s: its form is %d; a form is a whole number from 0, which its holder raises when it changes how "+
+			"it writes it", what, k.Form)
+	}
+	return nil
+}
+
+// The kinds of kept data, by the template each is on (ADR 0595 §2): the five a restore point is taken of.
+const (
+	KeptKindState   = "state"
+	KeptKindStream  = "stream"
+	KeptKindObjects = "objects"
+	KeptKindLedger  = "ledger"
+	KeptKindLog     = "log"
+)
+
+// KeptDeclared is one kept data a manifest declares: its local name, its kind and its layout.
+type KeptDeclared struct {
+	Name string
+	Kind string
+	KeptLayout
+}
+
+// KeptDeclarations are every kept data a manifest declares, in the old sections and in `requires` by template, sorted by
+// name and kind, once each: the one reader of what a module keeps (KeptStates is its states). An entry in `requires` is
+// of its storage type's kind, a stream on the template log a log.
+func KeptDeclarations(m Manifest) []KeptDeclared {
+	var out []KeptDeclared
+	for _, s := range m.State {
+		out = append(out, KeptDeclared{s.Name, KeptKindState, s.KeptLayout})
+	}
+	for _, l := range m.Logs {
+		out = append(out, KeptDeclared{l.Name, KeptKindLog, l.KeptLayout})
+	}
+	for _, s := range m.Streams {
+		out = append(out, KeptDeclared{s.Name, KeptKindStream, s.KeptLayout})
+	}
+	for _, o := range m.Objects {
+		out = append(out, KeptDeclared{o.Name, KeptKindObjects, o.KeptLayout})
+	}
+	for _, e := range m.BusState {
+		kind := KeptKindStream
+		switch {
+		case e.Storage == StorageKeyValue:
+			kind = KeptKindState
+		case e.Storage == StorageObjectStore:
+			kind = KeptKindObjects
+		case e.Storage == StorageLedger:
+			kind = KeptKindLedger
+		case e.Template == TemplateLog:
+			kind = KeptKindLog
+		}
+		out = append(out, KeptDeclared{e.Name, kind, e.KeptLayout})
+	}
+	sort.SliceStable(out, func(i, j int) bool {
+		if out[i].Name != out[j].Name {
+			return out[i].Name < out[j].Name
+		}
+		return out[i].Kind < out[j].Kind
+	})
+	once := out[:0]
+	for i, k := range out {
+		if i > 0 && k.Name == out[i-1].Name && k.Kind == out[i-1].Kind {
+			continue
+		}
+		once = append(once, k)
+	}
+	return once
+}
+
+// FormChange is one kept data whose form differs between the build a node runs and the one a want moves it to, by its
+// name and kind: Running is its declaration in the running manifest and Applied in the one being applied, nil where that
+// manifest declares none.
+type FormChange struct {
+	Name    string
+	Kind    string
+	Running *KeptLayout
+	Applied *KeptLayout
+}
+
+// FormChanges are the kept data whose form differs between the running manifest and the one being applied, sorted by name
+// and kind: declared by both at another form, or by one of them only (a form-0 declaration of either side included, so
+// a revert to a manifest that declared it before is seen). Which of them is backed up before the move, and which restored
+// at a revert, is the node agent's applying it (design 47); a kept data whose form is unchanged is never one.
+func FormChanges(running, applied Manifest) []FormChange {
+	type key struct{ name, kind string }
+	before := map[key]KeptLayout{}
+	for _, k := range KeptDeclarations(running) {
+		before[key{k.Name, k.Kind}] = k.KeptLayout
+	}
+	after := map[key]KeptLayout{}
+	for _, k := range KeptDeclarations(applied) {
+		after[key{k.Name, k.Kind}] = k.KeptLayout
+	}
+	var out []FormChange
+	for k, r := range before {
+		a, has := after[k]
+		if has && a.Form == r.Form {
+			continue
+		}
+		c := FormChange{Name: k.name, Kind: k.kind, Running: &r}
+		if has {
+			c.Applied = &a
+		}
+		out = append(out, c)
+	}
+	for k, a := range after {
+		if _, has := before[k]; !has {
+			out = append(out, FormChange{Name: k.name, Kind: k.kind, Applied: &a})
+		}
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].Name != out[j].Name {
+			return out[i].Name < out[j].Name
+		}
+		return out[i].Kind < out[j].Kind
+	})
+	return out
+}

go/catalogue/kept_form_test.go

@@ -0,0 +1,89 @@
+package catalogue
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+// W1.5 (design 47, *Its form is declared*): every kept data, in an old section or in `requires` by template, says its
+// form and its catch-up tool; a declaration naming none reads as form 0. KeptDeclarations reads them in one walk, and
+// FormChanges is what differs between a running manifest and the one a want applies.
+func TestAKeptDataSaysItsFormAndItsCatchUp(t *testing.T) {
+	m, err := decodeForTest([]byte(`{"module":"x","version":"1","tools":["x_take_in"],
+		"state": ["seen", {"name":"modules","history":20,"form":1,"catch-up":"x_take_in"}],
+		"logs": [{"name":"acts","form":2}],
+		"streams": [{"name":"forge","events":["merged"],"form":1,"catch-up":"x_take_in"}],
+		"objects": [{"name":"revisions","max-mib":64,"form":1,"catch-up":"x_take_in"}],
+		"requires": [
+			{"name":"records","storage":"ledger","template":"ledger","form":3,"catch-up":"x_take_in"},
+			{"name":"holds","storage":"key-value","template":"state"},
+			{"name":"history","storage":"stream","template":"log","form":1},
+			{"name":"marks","storage":"stream","template":"events"},
+			{"name":"blobs","storage":"object-store","template":"objects","form":4}]}`))
+	if err != nil || m.SetAside() != "" {
+		t.Fatalf("refused or set aside: %v %q", err, m.SetAside())
+	}
+	want := []KeptDeclared{
+		{"acts", KeptKindLog, KeptLayout{Form: 2}},
+		{"blobs", KeptKindObjects, KeptLayout{Form: 4}},
+		{"forge", KeptKindStream, KeptLayout{1, "x_take_in"}},
+		{"history", KeptKindLog, KeptLayout{Form: 1}},
+		{"holds", KeptKindState, KeptLayout{}},
+		{"marks", KeptKindStream, KeptLayout{}},
+		{"modules", KeptKindState, KeptLayout{1, "x_take_in"}},
+		{"records", KeptKindLedger, KeptLayout{3, "x_take_in"}},
+		{"revisions", KeptKindObjects, KeptLayout{1, "x_take_in"}},
+		{"seen", KeptKindState, KeptLayout{}},
+	}
+	if got := KeptDeclarations(m); !reflect.DeepEqual(got, want) {
+		t.Errorf("kept data read as\n%+v\nwant\n%+v", got, want)
+	}
+	if got := KeptStates(m); !reflect.DeepEqual(got, []string{"holds", "modules", "seen"}) {
+		t.Errorf("kept states %v", got)
+	}
+	// Stored and read back, each form and catch-up tool stays; a state with only a form keeps its long form.
+	if back := roundTrip(t, m); !reflect.DeepEqual(KeptDeclarations(back), want) {
+		t.Errorf("read back as %+v", KeptDeclarations(back))
+	}
+
+	// What differs between the build a node runs and the one applied: a form raised, a kept data no longer declared, one
+	// newly declared at form 0 (as a revert to a manifest from before sees it); an unchanged form is none.
+	running, err := decodeForTest([]byte(`{"module":"x","version":"1","state":["modules","seen"],"logs":["acts"]}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	applied, err := decodeForTest([]byte(`{"module":"x","version":"1","state":["seen","fresh"],
+		"logs":[{"name":"acts","form":1,"catch-up":"x_take_in"}]}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	changes := FormChanges(running, applied)
+	wantChanges := []FormChange{
+		{Name: "acts", Kind: KeptKindLog, Running: &KeptLayout{}, Applied: &KeptLayout{1, "x_take_in"}},
+		{Name: "fresh", Kind: KeptKindState, Applied: &KeptLayout{}},
+		{Name: "modules", Kind: KeptKindState, Running: &KeptLayout{}},
+	}
+	if !reflect.DeepEqual(changes, wantChanges) {
+		t.Errorf("form changes %+v, want %+v", changes, wantChanges)
+	}
+	if got := FormChanges(applied, applied); len(got) != 0 {
+		t.Errorf("a manifest applied over itself changes %+v", got)
+	}
+}
+
+// A form below 0 is refused where it is read, in every kept kind's declaration, naming it.
+func TestAKeptDataFormBelowZeroIsRefused(t *testing.T) {
+	for _, c := range []struct{ name, raw, says string }{
+		{"state", `"state":[{"name":"seen","form":-3}]`, `state "seen": its form is -3`},
+		{"log", `"logs":[{"name":"acts","form":-1}]`, `log "acts": its form is -1`},
+		{"stream", `"streams":[{"name":"forge","events":["merged"],"form":-2}]`, `stream "forge": its form is -2`},
+		{"objects", `"objects":[{"name":"revisions","max-mib":64,"form":-1}]`, `object bucket "revisions": its form is -1`},
+		{"requires", `"requires":[{"name":"holds","storage":"key-value","template":"state","form":-1}]`, `requires entry "holds": its form is -1`},
+	} {
+		_, err := decodeForTest([]byte(`{"module":"x","version":"1",` + c.raw + `}`))
+		if err == nil || !strings.Contains(err.Error(), c.says) {
+			t.Errorf("%s: read as %v, want a refusal holding %q", c.name, err, c.says)
+		}
+	}
+}

go/catalogue/logs.go

@@ -23,6 +23,8 @@ type LogDeclaration struct {
 	// entry leaves a log: the server ages it out, never a delete or a purge (novox/hq ADR 0565 §10: the hand-act log
 	// keeps every act 400 days; issue 1953 S-age).
 	MaxAgeDays int `json:"max-age-days,omitempty"`
+	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
+	KeptLayout
 }
 
 // LogMostAgeDays is the longest a log may keep an entry by age: past it, a log keeps its entries as long as it is.
@@ -36,7 +38,7 @@ func (l LogDeclaration) Cap() int {
 	return l.MaxMiB
 }
 
-// UnmarshalJSON reads a log as its bare name, or as {name, max-mib, max-age-days}.
+// UnmarshalJSON reads a log as its bare name, or as {name, max-mib, max-age-days, form, catch-up}.
 func (l *LogDeclaration) UnmarshalJSON(raw []byte) error {
 	trimmed := bytes.TrimSpace(raw)
 	if len(trimmed) > 0 && trimmed[0] == '"' {
@@ -46,13 +48,18 @@ func (l *LogDeclaration) UnmarshalJSON(raw []byte) error {
 		Name       string `json:"name"`
 		MaxMiB     *int   `json:"max-mib"`
 		MaxAgeDays *int   `json:"max-age-days"`
+		KeptLayout
 	}
 	dec := json.NewDecoder(bytes.NewReader(trimmed))
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&full); err != nil {
-		return fmt.Errorf("a log is either a name or {name, max-mib, max-age-days}: %w", typedUnknown(err))
+		return fmt.Errorf("a log is either a name or {name, max-mib, max-age-days, form, catch-up}: %w", typedUnknown(err))
+	}
+	if err := full.KeptLayout.refuse(fmt.Sprintf("log %q", full.Name)); err != nil {
+		return err
 	}
 	l.Name = full.Name
+	l.KeptLayout = full.KeptLayout
 	l.MaxMiB = 0
 	if full.MaxMiB != nil {
 		// Said, and said as nothing: refused rather than read as the default, which it did not say.
@@ -73,7 +80,7 @@ func (l *LogDeclaration) UnmarshalJSON(raw []byte) error {
 
 // MarshalJSON writes back the short form when there is nothing else to say.
 func (l LogDeclaration) MarshalJSON() ([]byte, error) {
-	if l.MaxMiB == 0 && l.MaxAgeDays == 0 {
+	if l.MaxMiB == 0 && l.MaxAgeDays == 0 && l.KeptLayout == (KeptLayout{}) {
 		return json.Marshal(l.Name)
 	}
 	type plain LogDeclaration

go/catalogue/manifest.go

@@ -465,6 +465,11 @@ type Manifest struct {
 	// choosing `kc` for keycloak keeps the identity legible where a hash would not.
 	Slug string `json:"slug,omitempty"`
 
+	// NeedsApproval marks tools the module serves (`<tool>`), seat verbs it serves (`<seat>.<verb>`) and events it
+	// emits that need the operator's approval first (design 46 *The approval mark* §1; design 34 every-event §32). Absent
+	// is no mark.
+	NeedsApproval []string `json:"needs-approval,omitempty"`
+
 	// Provides are the names other modules may require. A module always provides its own name;
 	// this is for the rest — `zsh` provides `shell`, `xorg` provides `display-server`.
 	Provides []Offer `json:"provides,omitempty"`
@@ -582,6 +587,11 @@ type Manifest struct {
 	// person's account (design 25 §7) already had the same shape.
 	Invokes []string `json:"invokes,omitempty"`
 
+	// InvokeFields maps a verb of Invokes to the fields it names, read from an entry of `invokes` in its long form
+	// {verb, fields}, which may stand beside `*` (design 47 G §617): Invokes keeps the verb, so none of its readers
+	// changes. A long-form entry naming no field is a verb of Invokes alone.
+	InvokeFields map[string][]string `json:"-"`
+
 	// Waits are the event families a session may wait on through the mesh MCP server's `mesh_wait`, each
 	// `<emitter>.<event>` as `consumes` names one, a seat's or a module's (novox/hq ADR 0367 §2). Only the tool
 	// runner's module declares them, one by one and never by a wildcard. **A grant apart from `consumes`**: the
@@ -1236,6 +1246,21 @@ type Artifact struct {
 	// DefaultSeries (SeriesOf).
 	Series string `json:"series,omitempty"`
 
+	// Needs are the artifact references, `<module>:<artifact>`, of the catalog graph's edges this artifact has where no
+	// other line of the manifest or of a seat's definition says them (design 47 G §47, §432). Read here so a reader
+	// release keeps a manifest that names them whole (design 47 G §340, §409); the manifest validator checks their form.
+	Needs []string `json:"needs,omitempty"`
+
+	// Calls are the capability words (design 47 G §488, §605) a package's code calls of the tool runner: the delivery plan
+	// derives, for each module whose build resolved a version of such a package, an ordering edge to the module whose
+	// artifact States the word. Read here so the second reader release keeps a manifest that names them (§621).
+	Calls []string `json:"calls,omitempty"`
+
+	// States are the capability words (design 47 G §488) the tool runner's artifact states, the edge of a package's Calls
+	// running to the module that states each (§545). Read here so the second reader release keeps a manifest that names
+	// them (§621).
+	States []string `json:"states,omitempty"`
+
 	// GeneratedFrom marks a package of a contracts module (novox/hq ADR 0608 §5, design 26 *Contracts modules*): the
 	// directory, relative to the module's, of the seat definitions every file of the package is generated from. A
 	// module carrying it holds contracts only and claims no seat, which the judge refuses naming the seat; the build
@@ -1562,6 +1587,10 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
 	if err := consumesEntryProblem(keys["consumes"]); err != nil {
 		return err
 	}
+	invokeFields, err := readInvokes(keys)
+	if err != nil {
+		return err
+	}
 	plain := map[string]string{}
 	many := map[string]map[string]string{}
 	if secrets, ok := keys["secrets"]; ok && string(secrets) != "null" {
@@ -1716,6 +1745,7 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
 	m.forms = forms
 	m.OptionalRequires = optionalRequires
 	m.RequireShares = requireShares
+	m.InvokeFields = invokeFields
 	if len(plain) > 0 {
 		m.Secrets = plain
 	}
@@ -1734,6 +1764,61 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error {
 	return nil
 }
 
+// invokeLongForm is an entry of `invokes` in its long form: the verb, and the fields of its answer the caller relies on
+// (design 47 G §617).
+type invokeLongForm struct {
+	Verb   string   `json:"verb"`
+	Fields []string `json:"fields,omitempty"`
+}
+
+// readInvokes reads `invokes` whose entries are verbs or, in their long form, {verb, fields}: it puts the verbs back in
+// keys as plain names and answers the fields each long-form entry names.
+func readInvokes(keys map[string]json.RawMessage) (map[string][]string, error) {
+	raw, ok := keys["invokes"]
+	if !ok || string(raw) == "null" {
+		return nil, nil
+	}
+	var entries []json.RawMessage
+	if err := json.Unmarshal(raw, &entries); err != nil {
+		return nil, nil // not a list: the field's own decoder says so
+	}
+	var verbs []string
+	var fields map[string][]string
+	for i, e := range entries {
+		trimmed := bytes.TrimSpace(e)
+		if len(trimmed) == 0 || trimmed[0] != '{' {
+			var verb string
+			if err := json.Unmarshal(trimmed, &verb); err != nil {
+				return nil, fmt.Errorf("invokes[%d]: a verb, or {verb, fields}: %w", i, err)
+			}
+			verbs = append(verbs, verb)
+			continue
+		}
+		var long invokeLongForm
+		dec := json.NewDecoder(bytes.NewReader(trimmed))
+		dec.DisallowUnknownFields()
+		if err := dec.Decode(&long); err != nil {
+			return nil, fmt.Errorf("invokes[%d]: a verb, or {verb, fields}: %w", i, err)
+		}
+		if long.Verb == "" {
+			return nil, fmt.Errorf("invokes[%d]: a long-form entry names its verb", i)
+		}
+		verbs = append(verbs, long.Verb)
+		if len(long.Fields) > 0 {
+			if fields == nil {
+				fields = map[string][]string{}
+			}
+			fields[long.Verb] = long.Fields
+		}
+	}
+	plain, err := json.Marshal(verbs)
+	if err != nil {
+		return nil, err
+	}
+	keys["invokes"] = plain
+	return fields, nil
+}
+
 // checkSharesNotContributed refuses a module whose own `contributes` to s3-bucket names `shares` (ADR 0468 §8).
 func (m Manifest) checkSharesNotContributed() error {
 	if len(m.RequireShares[BucketRequirement]) == 0 {
@@ -1787,13 +1872,30 @@ func (m Manifest) marshalFields() ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-	if len(m.SecretsMany) == 0 && len(m.ContributesMany) == 0 && len(m.OptionalRequires) == 0 && len(m.RequireShares) == 0 {
+	if len(m.SecretsMany) == 0 && len(m.ContributesMany) == 0 && len(m.OptionalRequires) == 0 && len(m.RequireShares) == 0 &&
+		len(m.InvokeFields) == 0 {
 		return raw, nil
 	}
 	var keys map[string]json.RawMessage
 	if err := json.Unmarshal(raw, &keys); err != nil {
 		return nil, err
 	}
+	if len(m.InvokeFields) > 0 {
+		// Each verb that names fields is written back in its long form (design 47 G §617).
+		entries := make([]any, 0, len(m.Invokes))
+		for _, verb := range m.Invokes {
+			if fields, named := m.InvokeFields[verb]; named {
+				entries = append(entries, invokeLongForm{Verb: verb, Fields: fields})
+				continue
+			}
+			entries = append(entries, verb)
+		}
+		listed, err := json.Marshal(entries)
+		if err != nil {
+			return nil, err
+		}
+		keys["invokes"] = listed
+	}
 	if optional := optionalRequiresList(m); len(optional) > 0 {
 		// Beside `requires`, never inside it: an older controller reads `requires` as plain names, and one
 		// manifest it cannot read stops the whole catalogue (OptionalRequiresKey).

go/catalogue/objects.go

@@ -26,6 +26,8 @@ type ObjectsDeclaration struct {
 	// Grant names the trusted config value of the module whose entries name the modules that may read it; absent,
 	// no other module reads it.
 	Grant *ObjectsGrant `json:"grant,omitempty"`
+	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
+	KeptLayout
 }
 
 // ObjectsGrant says which trusted config value of the owning module grants another module the read of a bucket
@@ -45,7 +47,10 @@ func (o *ObjectsDeclaration) UnmarshalJSON(raw []byte) error {
 	dec := json.NewDecoder(bytes.NewReader(raw))
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&p); err != nil {
-		return fmt.Errorf("an object bucket is {name, max-mib, grant}: %w", typedUnknown(err))
+		return fmt.Errorf("an object bucket is {name, max-mib, grant, form, catch-up}: %w", typedUnknown(err))
+	}
+	if err := p.KeptLayout.refuse(fmt.Sprintf("object bucket %q", p.Name)); err != nil {
+		return err
 	}
 	*o = ObjectsDeclaration(p)
 	return nil

go/catalogue/optional_requirements_test.go

@@ -78,6 +78,9 @@ func TestARequirementsLongForm(t *testing.T) {
 		// 1966/AC3: a ledger type with each of the four effects round-trips through JSON unchanged.
 		{"a ledger with its type", `[{"name":"records","storage":"ledger","template":"ledger","options":{"max-mib":1024},
 			"types":[` + string(typeJSON) + `]}]`, "", []BusStateEntry{ledger}, nil, ""},
+		// Issue 1966: a key-value entry of `requires` is kept state beside the old section's, by its local name.
+		{"a key-value entry beside an old state", `[{"name":"holds","storage":"key-value","template":"state"}], "state": ["seen"]`,
+			"", []BusStateEntry{{Name: "holds", Storage: StorageKeyValue, Template: TemplateState}}, nil, ""},
 		// A field given twice at any depth is refused, named by its path: a decoder would keep the last without a word.
 		{"a field given twice", `[], "requires": ["postgres-database"]`, "", nil, nil, "the field requires is given twice"},
 		{"a field given twice deeper", `[{"name":"records","storage":"ledger","name":"other","template":"ledger"}]`,
@@ -100,6 +103,14 @@ func TestARequirementsLongForm(t *testing.T) {
 		if !reflect.DeepEqual(got.BusState, c.busState) {
 			t.Errorf("%s: bus state read as %+v, want %+v", c.name, got.BusState, c.busState)
 		}
+		// Issue 1966: a ledger is no kept state; KeptStates names the old section's and the key-value entries' alone.
+		var keeps []string
+		if c.name == "a key-value entry beside an old state" {
+			keeps = []string{"holds", "seen"}
+		}
+		if kept := KeptStates(got); !reflect.DeepEqual(kept, keeps) {
+			t.Errorf("%s: kept states %v, want %v", c.name, kept, keeps)
+		}
 		for _, path := range c.aside {
 			if !strings.Contains(got.SetAside(), path) {
 				t.Errorf("%s: %s is not named among what was set aside: %q", c.name, path, got.SetAside())

go/catalogue/reader_release_test.go

@@ -0,0 +1,101 @@
+package catalogue
+
+import (
+	"encoding/json"
+	"slices"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
+)
+
+// A manifest carrying the words design 47 G writes is read whole by the reader release (design 47 G §340, §409):
+// `version` as MAJOR.MINOR (§368), a `*` emitter in `consumes` (§386), an observer's finding stream (§387) and an
+// artifact's `needs` (§47, §432).
+func TestTheReaderReleaseReadsTheManifestWordsGWrites(t *testing.T) {
+	m, err := decodeForTest([]byte(`{"module":"mesh-watcher","version":"1.4","needs-approval":["restart","node-engine.composition","finding-written"],` +
+		`"consumes":["*.finding-written","mesh-conditions.findings-asked"],` +
+		`"streams":[{"name":"mesh-watcher-findings","events":["finding-written","finding-cleared","findings-stated"],"max-age-days":7,"max-mib":16}],` +
+		`"build":{"artifacts":[{"name":"program","kind":"bundle","language":"go","needs":["mesh-grants:program","node-tools:runtime"]}]}}`))
+	if err != nil {
+		t.Fatalf("a manifest with G's words was refused: %v", err)
+	}
+	if len(m.UnknownField()) != 0 || len(m.NeedsApproval) != 3 {
+		t.Fatalf("a word was set aside: %v (needs-approval %v)", m.UnknownField(), m.NeedsApproval)
+	}
+	if m.Version != "1.4" || m.Consumes[0] != "*.finding-written" || m.Streams[0].Name != "mesh-watcher-findings" ||
+		len(m.Build.Artifacts[0].Needs) != 2 || m.Build.Artifacts[0].Needs[0] != "mesh-grants:program" {
+		t.Fatalf("read as %+v", m)
+	}
+}
+
+// The second reader release reads the words G's manifests write beyond the first's (design 47 G §494, §621): each row a
+// manifest read whole, or refused naming what is wrong, and what the reader gives of it.
+func TestTheSecondReaderReleaseReadsGsManifestWords(t *testing.T) {
+	for _, c := range []struct {
+		name, raw, refused string
+		check              func(Manifest) bool
+	}{
+		{name: "a state's cap (§469)", raw: `{"module":"m","version":"1.0","state":["plain",{"name":"pipelines","max-mib":256}]}`,
+			check: func(m Manifest) bool { return m.State[0].MaxMiB == 0 && m.State[1].MaxMiB == 256 }},
+		{name: "a state's cap out of the objects section's bounds (§469)",
+			raw: `{"module":"m","version":"1.0","state":[{"name":"pipelines","max-mib":0}]}`, refused: "max-mib is between 1 and 8192"},
+		{name: "a package's calls and the tool runner's states (§605, §545)",
+			raw: `{"module":"m","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","calls":["put-sessions"]},` +
+				`{"name":"runtime","kind":"bundle","language":"go","states":["put-sessions","finding-calls"]}]}}`,
+			check: func(m Manifest) bool {
+				a := m.Build.Artifacts
+				return len(a) == 2 && len(a[0].Calls) == 1 && a[0].Calls[0] == "put-sessions" && len(a[1].States) == 2
+			}},
+		{name: "* with two long-form invokes (§617)",
+			raw: `{"module":"m","version":"1.0","invokes":["*",{"verb":"seat:mesh-catalog.revision","fields":["none-yet"]},` +
+				`{"verb":"seat:mesh-catalog.entry"}]}`,
+			check: func(m Manifest) bool {
+				out, _ := json.Marshal(m)
+				return slices.Equal(m.Invokes, []string{"*", "seat:mesh-catalog.revision", "seat:mesh-catalog.entry"}) &&
+					len(m.InvokeFields) == 1 && slices.Equal(m.InvokeFields["seat:mesh-catalog.revision"], []string{"none-yet"}) &&
+					strings.Contains(string(out), `"invokes":["*",{"verb":"seat:mesh-catalog.revision","fields":["none-yet"]},"seat:mesh-catalog.entry"]`)
+			}},
+		{name: "a long-form invoke with a word it does not know (§617)",
+			raw:     `{"module":"m","version":"1.0","invokes":[{"verb":"seat:mesh-catalog.revision","field":["none-yet"]}]}`,
+			refused: `invokes[0]: a verb, or {verb, fields}`},
+	} {
+		m, err := decodeForTest([]byte(c.raw))
+		switch {
+		case c.refused != "" && (err == nil || !strings.Contains(err.Error(), c.refused)):
+			t.Errorf("%s: read as %v, want refused naming %q", c.name, err, c.refused)
+		case c.refused == "" && err != nil:
+			t.Errorf("%s: refused: %v", c.name, err)
+		case c.refused == "" && (len(m.UnknownField()) != 0 || !c.check(m)):
+			t.Errorf("%s: read as %+v (set aside %v)", c.name, m, m.UnknownField())
+		}
+	}
+}
+
+// A consumed view of a seat, `seat:<seat>.views.<state>`, is read apart from a consumed event, and a seat's compiled form
+// names the states its definition offers as views (design 47 G §616, §699, §621).
+func TestTheSecondReaderReleaseReadsASeatNamedView(t *testing.T) {
+	for _, row := range []struct {
+		consume, seat, state string
+		ok, bad              bool
+	}{
+		{"seat:mesh-module-assigner.views.assignments", "mesh-module-assigner", "assignments", true, false},
+		{"mesh-delivery-assigner.views.assignments", "", "", false, false},
+		{"seat:mesh-module-assigner.assignments", "", "", true, true},
+		{"seat:.views.assignments", "", "", true, true},
+	} {
+		seat, state, ok, err := SeatViewConsume(row.consume)
+		if seat != row.seat || state != row.state || ok != row.ok || (err != nil) != row.bad {
+			t.Errorf("SeatViewConsume(%q) = %q, %q, %v, %v", row.consume, seat, state, ok, err)
+		}
+	}
+	m, err := decodeForTest([]byte(`{"module":"m","version":"1.0","consumes":["seat:mesh-module-assigner.views.assignments"]}`))
+	if err != nil || len(m.UnknownField()) != 0 || m.Consumes[0] != "seat:mesh-module-assigner.views.assignments" {
+		t.Errorf("a seat-named view in consumes read as %v, %v", m.Consumes, err)
+	}
+	s := SeatFromDefinition(seatdef.Definition{Seat: "mesh-module-assigner", Scope: "mesh",
+		States: []seatdef.State{{Name: "assignments", View: true}, {Name: "placement"}}})
+	if !slices.Equal(s.Views, []string{"assignments"}) {
+		t.Errorf("the seat's views = %v", s.Views)
+	}
+}

go/catalogue/resolver_manifests_test.go

@@ -5,83 +5,16 @@ import (
 	"testing"
 )
 
-// The catalogue's resolver modules as they are, parsed by the real parser and composed as a
-// machine would receive them (hal dnsmasq-app conversion, novox/hq 08-connectivity).
-//
-// The predecessor's resolver answered every name on a machine: the mesh's own itself, the rest
-// forwarded to two fixed upstreams, with the machine's resolv.conf naming it alone and the
-// container runtime pointed at its private-network address. These hold the mesh's modules to the
-// same arrangement, and to the two things a resolver here must never do — read resolv.conf for
-// its upstreams, or take an address systemd-resolved holds.
+// The resolver module's own configuration is held by the module's tests in the catalogue (novox/hq issue 1966): a
+// file moves with its settings, and a literal here broke on every pull request when the upstreams became settings.
+// What stays is what the uplink modules write, which names the mesh's resolvers and is the catalogue's across
+// modules.
 
-// Its configuration forwards to the upstreams the predecessor's module shipped, and gets them from
-// nowhere else: `no-resolv` is what makes the documented loop — the resolver finding its own
-// address in resolv.conf and becoming its own upstream — impossible.
-func TestTheResolverForwardsToFixedUpstreamsAndNeverReadsResolvConf(t *testing.T) {
-	m := catalogueManifest(t, "dnsmasq")
-	var config string
-	for _, r := range m.Resources {
-		if r["id"] == "config" {
-			config, _ = r["content"].(string)
-		}
-	}
-	if config == "" {
-		t.Fatal("the resolver has no configuration file")
-	}
-	for _, want := range []string{
-		"\nno-resolv\n", "\nserver=1.1.1.1\n", "\nserver=8.8.8.8\n",
-		// The private address and loopback, never a LAN's (novox/hq ADR 0194): a device that is not a
-		// member cannot reach what the mesh's names point at.
-		"\nlisten-address=127.0.0.1\n", "\nlisten-address=${machine:address}\n",
-
-		// No hosts file and no operator's files: the mesh's resolver answers every node (ADR 0199).
-		"\nno-hosts\n",
-		"\nconf-file=" + m.Facts["zones"].Path + "\n",
-		"\ndomain-needed\n", "\nbogus-priv\n",
-		"\nconf-file=" + m.Facts["node-zones"].Path + "\n",
-	} {
-		if !strings.Contains(config, want) {
-			t.Errorf("the resolver's configuration lacks %q:\n%s", strings.TrimSpace(want), config)
-		}
-	}
-	// Bound to each address once, at start, and never by bind-dynamic (novox/hq issue 348): bind-dynamic
-	// re-reads the machine's addresses on every change of any link and closes each listener that reading
-	// did not find, and a reading made while the container runtime tears down a bridge's virtual links
-	// fails, so the private address was left unbound for twenty minutes, the daemon running and silent.
-	// The catalogue's main binds once since mesh-catalog PR 161 (105cc7c0).
-	if !strings.Contains(config, "\nbind-interfaces\n") {
-		t.Errorf("the resolver's configuration does not bind its addresses once (bind-interfaces):\n%s", config)
-	}
-	if strings.Contains(config, "\nbind-dynamic\n") {
-		t.Errorf("the resolver's configuration says bind-dynamic, which closes its listeners when a bridge is "+
-			"torn down (novox/hq issue 348):\n%s", config)
-	}
-	// By address and never by interface: dnsmasq admits a query by the interface it arrives on
-	// when told one, and a container's query to the private address arrives on the runtime's
-	// bridge — `interface=mesh0` dropped every such query, silently (novox/hq issue 110).
-	for _, line := range strings.Split(config, "\n") {
-		if strings.HasPrefix(line, "interface=") {
-			t.Errorf("the resolver answers by interface, so a container's query on a bridge is dropped: %s", line)
-		}
-	}
-	// Not .53 or .54, which systemd-resolved holds; and not .55 any more, which was a convention
-	// beside the one every machine already followed — the predecessor's resolv.conf says .1.
-	for _, taken := range []string{"127.0.0.53", "127.0.0.54", "127.0.0.55"} {
-		if strings.Contains(config, "listen-address="+taken) {
-			t.Errorf("the resolver listens on %s", taken)
-		}
-	}
-	// Never a directory or a file the operator keeps: a line written for one machine's programs would
-	// become an answer for every node (ADR 0199).
-	for _, never := range []string{"conf-dir=", "addn-hosts=", "listen-address=${setting:"} {
-		if strings.Contains(config, never) {
-			t.Errorf("the mesh's resolver still reads or listens on %q", never)
-		}
-	}
-	// And the file that decides what the machine asks names every one of the mesh's resolvers, by
-	// address, from the seat's holders, and no public one (ADR 0223): a resolver library that asks
-	// every listed server at once takes the first reply, and a public "no such name" for a mesh name
-	// won it. Written by every uplink module, since the uplink's holder owns the file.
+// The file that decides what the machine asks names every one of the mesh's resolvers, by
+// address, from the seat's holders, and no public one (ADR 0223): a resolver library that asks
+// every listed server at once takes the first reply, and a public "no such name" for a mesh name
+// won it. Written by every uplink module, since the uplink's holder owns the file.
+func TestEveryUplinkNamesTheMeshsResolversAndNoOther(t *testing.T) {
 	for _, uplink := range uplinks {
 		fact, ok := catalogueManifest(t, uplink).Facts["resolvers"]
 		if !ok || fact.Path != "/etc/resolv.conf" {

go/catalogue/seat_from_definition.go

@@ -12,7 +12,7 @@ import (
 // contribution kind given as data gets its checks (withShapeCode), which are code and never a definition's.
 func SeatFromDefinition(d seatdef.Definition) Seat {
 	s := Seat{Name: d.Seat, Scope: d.Scope, Replicated: d.Replicated, Delivers: d.Delivers,
-		Accepts: seatdef.MessageNames(d.Accepts), Emits: seatdef.MessageNames(d.Events), PulledByKind: d.PulledByKind,
+		Accepts: seatdef.MessageNames(d.Accepts), Emits: localEvents(d.Events), PulledByKind: d.PulledByKind,
 		Yields: d.Yields, HeldByTheEngine: d.HeldByTheEngine, Decision: d.Decision}
 	if len(s.Accepts) == 0 {
 		s.Accepts = nil
@@ -25,6 +25,11 @@ func SeatFromDefinition(d seatdef.Definition) Seat {
 			Input: seatdef.ResolveRefs(v.Input, d.Defs), Output: seatdef.ResolveRefs(v.Answer, d.Defs),
 			Optional: v.Optional, Replaces: v.Replaces})
 	}
+	for _, st := range d.States {
+		if st.View {
+			s.Views = append(s.Views, st.Name)
+		}
+	}
 	for _, r := range d.Receives {
 		rv := Receivable{Kind: r.Kind, Comment: r.Comment, Dirs: r.Dirs, Offered: r.Offered}
 		if r.Shape != nil {
@@ -62,6 +67,16 @@ func SeatsFromDefinitions(defs []seatdef.Definition) []Seat {
 	return out
 }
 
+// localEvents are the events as a manifest names them: an event whose subject carries parts by its subject with `*`
+// for each part (owed.*.*), so SeatSays matches a consumer's name against it.
+func localEvents(es []seatdef.Message) []string {
+	out := make([]string, 0, len(es))
+	for _, e := range es {
+		out = append(out, e.Local())
+	}
+	return out
+}
+
 func examples(in []map[string]any) []map[string]any {
 	if in == nil {
 		return nil

go/catalogue/seat_pass.go

@@ -1,14 +1,97 @@
 package catalogue
 
-// An assignment as the passing of a seat reads it: a module on a machine, with its manifest.
+import (
+	"fmt"
+	"sort"
+)
+
+// An assignment as the passing of a seat reads it: a module on a machine, with the manifest of the build that machine
+// runs, else the trunk's (novox/hq ADR 0660 §2).
 type Assignment struct {
 	Node     string
 	Manifest Manifest
+	// Unread is an assignment whose manifest could not be read at all: it is still assigned, so its holdings stand,
+	// and it claims nothing.
+	Unread bool
 }
 
-// SeatPass is one seat whose holder on record is assigned nowhere, and the only assignment that can take it.
+// SeatPass is one seat whose holder on record has left it, and the only assignment that can take it.
 type SeatPass struct {
 	Seat Seat
 	From Held
 	To   Held
 }
+
+// SeatsToPassOn reads the mesh for seats held once whose holder on record has left (novox/hq ADR 0449 §4, ADR 0660).
+//
+// **A holder on record that is assigned on no machine, or whose every assignment's manifest no longer claims the seat
+// (or claims it and cannot hold it, by CanHold), holds nothing.** Of such a seat, exactly one assignment in the mesh
+// able to hold it takes it over: that is a pass, and the caller applies it and says it. An assignment is a module on
+// one node, so one module placed on two nodes is two claimants. With no such assignment the record stays; with two or
+// more nothing is passed and the answer says so, naming two and the `seat --to` that settles it. A holder with an
+// assignment whose manifest still claims the seat changes nothing (ADR 0131), and so does one whose manifest could not
+// be read. Only mesh seats held once are read: a replicated seat's holders are added by an act (ADR 0223), a kinded
+// bench is held by its claims, and a seat with several holders on record is a store that disagrees with the mesh.
+// Nothing here records a pass (ADR 0660 §4).
+func SeatsToPassOn(holdings []Held, assignments []Assignment) ([]SeatPass, []string) {
+	bySeat := map[string][]Held{}
+	var order []string
+	for _, h := range holdings {
+		s, ok := SeatNamed(h.Claim)
+		if !ok || s.Replicated || s.Scope != ScopeMesh || h.Scope != ScopeMesh || KindedBenches[s.Name] {
+			continue
+		}
+		if _, seen := bySeat[s.Name]; !seen {
+			order = append(order, s.Name)
+		}
+		bySeat[s.Name] = append(bySeat[s.Name], h)
+	}
+	sort.Strings(order)
+
+	var passes []SeatPass
+	var said []string
+	for _, name := range order {
+		recs := bySeat[name]
+		if len(recs) != 1 {
+			continue
+		}
+		seat, _ := SeatNamed(name)
+		holds := false
+		for _, a := range assignments {
+			if a.Manifest.Module != recs[0].Module {
+				continue
+			}
+			if a.Unread || CanHold(a.Manifest, seat) == nil {
+				holds = true
+			}
+		}
+		if holds {
+			continue
+		}
+		var claimants []Held
+		for _, a := range assignments {
+			if a.Unread || CanHold(a.Manifest, seat) != nil {
+				continue
+			}
+			claimants = append(claimants, Held{Claim: seat.Name, Scope: seat.Scope, Node: a.Node, Module: a.Manifest.Module})
+		}
+		sort.Slice(claimants, func(i, j int) bool {
+			if claimants[i].Node != claimants[j].Node {
+				return claimants[i].Node < claimants[j].Node
+			}
+			return claimants[i].Module < claimants[j].Module
+		})
+		switch len(claimants) {
+		case 0:
+		case 1:
+			passes = append(passes, SeatPass{Seat: seat, From: recs[0], To: claimants[0]})
+		default:
+			said = append(said, fmt.Sprintf(
+				"%s on %s is on record as the holder of %q and holds it nowhere, and %s on %s and %s on %s both claim it "+
+					"— nothing is passed; `seat %s --to <node>/<module>` records the one that holds it",
+				recs[0].Module, recs[0].Node, seat.Name, claimants[0].Module, claimants[0].Node,
+				claimants[1].Module, claimants[1].Node, seat.Name))
+		}
+	}
+	return passes, said
+}

go/catalogue/seat_pass_test.go

@@ -0,0 +1,96 @@
+package catalogue
+
+import (
+	"strings"
+	"testing"
+)
+
+func passSeat(t *testing.T) {
+	t.Helper()
+	was := Seats()
+	t.Cleanup(func() { UseSeats(was) })
+	UseSeats([]Seat{
+		{Name: "mesh-broker", Scope: ScopeMesh, Delivers: "mesh-bus", Decision: "test"},
+		{Name: "mesh-dns-resolver", Scope: ScopeMesh, Replicated: true, Decision: "test"},
+	})
+}
+
+func broker(module string) Manifest {
+	return Manifest{Module: module, Provides: []Offer{{Name: "mesh-bus", Scope: ScopeMesh}},
+		Claims: []Claim{{Name: "mesh-broker", Scope: ScopeMesh}}}
+}
+
+// A module that still runs but no longer claims the seat.
+func dropped(module string) Manifest { return Manifest{Module: module} }
+
+func recorded(node, module string) []Held {
+	return []Held{{Claim: "mesh-broker", Scope: ScopeMesh, Node: node, Module: module}}
+}
+
+// novox/hq ADR 0660 "How it is checked", and ADR 0449 §4's original case.
+func TestSeatsToPassOn(t *testing.T) {
+	passSeat(t)
+	rec := recorded("novox", "old")
+	cases := []struct {
+		name  string
+		rec   []Held
+		as    []Assignment
+		to    string // "node/module" of the one pass, "" for none
+		saidW []string
+	}{
+		{"holder assigned nowhere, one claimant (ADR 0449 §4)", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("new")}}, "novox/new", nil},
+		{"holder assigned but its manifest dropped the claim", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "novox", Manifest: broker("new")}}, "novox/new", nil},
+		{"holder claims but cannot hold the seat", rec,
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "old", Claims: []Claim{{Name: "mesh-broker", Scope: ScopeMesh}}}},
+				{Node: "ace", Manifest: broker("new")}}, "ace/new", nil},
+		{"the running build still claims: the holder keeps the seat", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("old")}, {Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"one of the holder's two nodes still claims it", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: broker("old")},
+				{Node: "g14", Manifest: broker("new")}}, "", nil},
+		{"holder's manifest unread: still assigned", rec,
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "old"}, Unread: true}, {Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"an unread assignment is no claimant", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: Manifest{Module: "new"}, Unread: true}}, "", nil},
+		{"a claimant whose running manifest does not claim is not counted", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: dropped("new")}}, "", nil},
+		{"no claimant keeps the record", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "novox", Manifest: dropped("other")}}, "", nil},
+		{"holder gone, no claimant keeps the record", rec, nil, "", nil},
+		{"one module on two nodes is two claimants: none passed, said", rec,
+			[]Assignment{{Node: "novox", Manifest: dropped("old")}, {Node: "ace", Manifest: broker("new")}, {Node: "novox", Manifest: broker("new")}},
+			"", []string{"old", "new on ace", "new on novox", "seat mesh-broker --to <node>/<module>"}},
+		{"two claimants, holder gone: said", rec,
+			[]Assignment{{Node: "novox", Manifest: broker("a")}, {Node: "ace", Manifest: broker("b")}},
+			"", []string{"a on novox", "b on ace", "seat mesh-broker --to <node>/<module>"}},
+		{"a seat with two holders on record is not read", append(rec[:1:1], Held{Claim: "mesh-broker", Scope: ScopeMesh, Node: "ace", Module: "x"}),
+			[]Assignment{{Node: "novox", Manifest: broker("new")}}, "", nil},
+		{"a replicated seat is never passed",
+			[]Held{{Claim: "mesh-dns-resolver", Scope: ScopeMesh, Node: "novox", Module: "gone"}},
+			[]Assignment{{Node: "novox", Manifest: Manifest{Module: "r", Claims: []Claim{{Name: "mesh-dns-resolver", Scope: ScopeMesh}}}}}, "", nil},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			passes, said := SeatsToPassOn(c.rec, c.as)
+			if c.to == "" && len(passes) != 0 || c.to != "" && len(passes) != 1 {
+				t.Fatalf("passes=%+v said=%v", passes, said)
+			}
+			if c.to != "" {
+				p := passes[0]
+				if p.Seat.Name != "mesh-broker" || p.From.Module != "old" || p.To.Node+"/"+p.To.Module != c.to {
+					t.Fatalf("the pass is %+v", p)
+				}
+			}
+			if (len(c.saidW) == 0) != (len(said) == 0) {
+				t.Fatalf("said=%v, want %v", said, c.saidW)
+			}
+			for _, w := range c.saidW {
+				if !strings.Contains(said[0], w) {
+					t.Fatalf("the answer does not say %q: %s", w, said[0])
+				}
+			}
+		})
+	}
+}

go/catalogue/seats.go

@@ -45,6 +45,9 @@ type Seat struct {
 	// Serves carries each verb in full — name, description, schema — because a role's tools are the
 	// mesh's to define and an agent's to call (novox/hq ADR 0132, design 33 §2).
 	Serves []Verb
+	// Views are the names of the states the seat offers as views, which a module follows by `consumes`
+	// `seat:<seat>.views.<state>` (design 47 G §616, §699). Compiled from the definition's states marked `view`.
+	Views []string
 	// Receives is what other modules may contribute to the seat's holder, by kind (novox/hq ADR
 	// 0212): each kind is text in the tool's own grammar, placed by the holder with
 	// ${contribution:<seat>:<kind>}. Compiled, never stored: like the protocol, it is the mesh's

go/catalogue/state.go

@@ -35,9 +35,14 @@ type StateDeclaration struct {
 	// the tool runner's account, read it and never write it (the lead's ruling, 2026-11-04: nats' warrant spend record
 	// `spent`). The judge's rule and the grants follow from it.
 	ReadOnly bool `json:"read-only,omitempty"`
+	// MaxMiB is the bucket's cap in MiB, between ObjectsLeastMiB and ObjectsMostMiB, as an old objects section's
+	// is (design 47 G §469, §494); zero is no cap, as before.
+	MaxMiB int `json:"max-mib,omitempty"`
+	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
+	KeptLayout
 }
 
-// UnmarshalJSON reads a bucket as its bare name, or as {name, history, ttl-seconds, per-machine, key-ttl, read-only}.
+// UnmarshalJSON reads a bucket as its bare name, or as {name, history, ttl-seconds, per-machine, key-ttl, read-only, max-mib, form, catch-up}.
 func (s *StateDeclaration) UnmarshalJSON(raw []byte) error {
 	trimmed := bytes.TrimSpace(raw)
 	if len(trimmed) > 0 && trimmed[0] == '"' {
@@ -45,10 +50,20 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error {
 	}
 	type plain StateDeclaration
 	var full plain
+	var said struct {
+		MaxMiB *int `json:"max-mib"`
+	}
+	_ = json.Unmarshal(trimmed, &said)
 	dec := json.NewDecoder(bytes.NewReader(trimmed))
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&full); err != nil {
-		return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine, key-ttl, read-only}: %w", typedUnknown(err))
+		return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine, key-ttl, read-only, max-mib, form, catch-up}: %w", typedUnknown(err))
+	}
+	if err := full.KeptLayout.refuse(fmt.Sprintf("state %q", full.Name)); err != nil {
+		return err
+	}
+	if said.MaxMiB != nil && (*said.MaxMiB < ObjectsLeastMiB || *said.MaxMiB > ObjectsMostMiB) {
+		return fmt.Errorf("state %q: max-mib is between %d and %d, not %d", full.Name, ObjectsLeastMiB, ObjectsMostMiB, *said.MaxMiB)
 	}
 	*s = StateDeclaration(full)
 	return nil
@@ -56,13 +71,32 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error {
 
 // MarshalJSON writes back the short form when there is nothing else to say.
 func (s StateDeclaration) MarshalJSON() ([]byte, error) {
-	if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine && !s.KeyTTL && !s.ReadOnly {
+	if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine && !s.KeyTTL && !s.ReadOnly && s.MaxMiB == 0 && s.KeptLayout == (KeptLayout{}) {
 		return json.Marshal(s.Name)
 	}
 	type plain StateDeclaration
 	return json.Marshal(plain(s))
 }
 
+// SeatViewPrefix opens a consumed view of a seat, `seat:<seat>.views.<state>` (design 47 G §616).
+const SeatViewPrefix = "seat:"
+
+// SeatViewConsume splits a consumed view of a seat, `seat:<seat>.views.<state>`, into the seat and the state; ok is false for a
+// consumed entry that is not of this kind (an event, `<emitter>.<event>`), and err says why a `seat:` entry is malformed. The
+// holder is not in it: it is resolved where the view is followed, so a new holder is followed with no change of the
+// manifest (design 47 G §616, §621).
+func SeatViewConsume(consume string) (seat, state string, ok bool, err error) {
+	rest, isSeat := strings.CutPrefix(consume, SeatViewPrefix)
+	if !isSeat {
+		return "", "", false, nil
+	}
+	seat, view, named := strings.Cut(rest, ".views.")
+	if !named || !stateName.MatchString(seat) || !stateName.MatchString(view) {
+		return "", "", true, fmt.Errorf("%q does not name a seat and its view: a consumed view of a seat is seat:<seat>.views.<state>", consume)
+	}
+	return seat, view, true, nil
+}
+
 // ReadState splits a read into the owning module and the local name, or says why it is not one.
 func ReadState(read string) (module, local string, err error) {
 	at := strings.LastIndex(read, ".")

go/catalogue/streams.go

@@ -64,6 +64,8 @@ type StreamDeclaration struct {
 	// Duplicates is a work-queue stream's duplicate window in hours, at most its maximum age; refused
 	// without retention work-queue (novox/hq ADR 0436 §6).
 	Duplicates int `json:"duplicates,omitempty"`
+	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
+	KeptLayout
 }
 
 // StreamGrant says which trusted config value of the module grants a consumer a part of the stream, and
@@ -84,7 +86,10 @@ func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error {
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&p); err != nil {
 		return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant, retention, max-deliveries, "+
-			"duplicates}: %w", typedUnknown(err))
+			"duplicates, form, catch-up}: %w", typedUnknown(err))
+	}
+	if err := p.KeptLayout.refuse(fmt.Sprintf("stream %q", p.Name)); err != nil {
+		return err
 	}
 	*s = StreamDeclaration(p)
 	return nil

go/membership/membership.go

@@ -6,6 +6,8 @@ package membership
 
 import (
 	"encoding/json"
+	"slices"
+	"sort"
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 )
@@ -171,3 +173,38 @@ type WaitIssued struct {
 	Event   string `json:"event"`
 	Subject string `json:"subject"`
 }
+
+// StateIssued is the state this membership issues its module's code by a local name, and whether it does: what every
+// runtime that serves a module's state calls reads, so no reader keeps a second copy of the rule (novox/hq issue 1966).
+// An entry of the old `state` list wins (ADR 0201); otherwise a binding of a key-value entry the module asks for in
+// `requires` is issued by its local name, as its own and written (ADR 0595 §1, §11: a moved module keeps its local
+// names), on the bucket the bus module named.
+func (m Membership) StateIssued(name string) (StateIssued, bool) {
+	for _, s := range m.State {
+		if s.Name == name {
+			return s, true
+		}
+	}
+	for _, b := range m.Bindings {
+		if b.Local == name && b.Storage == catalogue.StorageKeyValue && b.Bucket != "" {
+			return StateIssued{Name: b.Local, Bucket: b.Bucket, Writes: true}, true
+		}
+	}
+	return StateIssued{}, false
+}
+
+// StateNames are the local names StateIssued answers, the old list's and the key-value bindings', sorted, once each:
+// what a refusal names as issued.
+func (m Membership) StateNames() []string {
+	var out []string
+	for _, s := range m.State {
+		out = append(out, s.Name)
+	}
+	for _, b := range m.Bindings {
+		if b.Storage == catalogue.StorageKeyValue && b.Bucket != "" {
+			out = append(out, b.Local)
+		}
+	}
+	sort.Strings(out)
+	return slices.Compact(out)
+}

go/membership/membership_test.go

@@ -59,6 +59,31 @@ func TestAMembershipRoundTripsAsIssued(t *testing.T) {
 	if string(before) != `{"node":"ace","module":"forge","serves":[{"subject":"s"}],"emits":"e","tools":"t","state":[{"name":"repos","bucket":"forge_repos","writes":true}]}` {
 		t.Fatalf("a membership without bindings encodes otherwise than before: %s", before)
 	}
+	// Issue 1966 (F1): what a membership issues by a local name is one rule. An old `state` entry wins; a key-value
+	// binding of `requires` is issued by its local name, own and written; any other binding, and a name of neither, is not.
+	issuing := Membership{State: []StateIssued{{Name: "repos", Bucket: "forge_repos"}},
+		Bindings: []BusBinding{{Local: "holds", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState, Bucket: "forge_holds"},
+			{Local: "repos", Storage: catalogue.StorageKeyValue, Bucket: "forge_repos2"},
+			{Local: "blobs", Storage: catalogue.StorageObjectStore, ObjectStore: "forge_blobs"},
+			{Local: "events", Storage: catalogue.StorageStream, Stream: "forge_events"}}}
+	for _, c := range []struct {
+		name string
+		want StateIssued
+		ok   bool
+	}{
+		{"holds", StateIssued{Name: "holds", Bucket: "forge_holds", Writes: true}, true},
+		{"repos", StateIssued{Name: "repos", Bucket: "forge_repos"}, true},
+		{"blobs", StateIssued{}, false},
+		{"events", StateIssued{}, false},
+		{"other", StateIssued{}, false},
+	} {
+		if got, ok := issuing.StateIssued(c.name); ok != c.ok || got != c.want {
+			t.Errorf("StateIssued(%q) = %+v, %v; want %+v, %v", c.name, got, ok, c.want, c.ok)
+		}
+	}
+	if got := issuing.StateNames(); !reflect.DeepEqual(got, []string{"holds", "repos"}) {
+		t.Errorf("StateNames = %v", got)
+	}
 	var wire map[string]json.RawMessage
 	_ = json.Unmarshal(raw, &wire)
 	for _, k := range []string{"receives", "mesh", "seat-traffic"} {

go/process/call.go

@@ -51,6 +51,57 @@ func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) {
 	return c.callOn(called, fallback, body)
 }
 
+// CallKept is Call for a call whose answer the caller keeps and which is safe to make again (a read, or a write the
+// callee deduplicates): it is asked again through a transport fault, a timeout or no responders, whether or not the bus
+// is blacked out, until it is answered or busretry.Bound has passed (novox/hq ADR 0668, extending ADR 0615 §6; issue
+// 1966). A callee that restarts answers no one for seconds; Call fails that at once. The caller keeps a call still
+// unanswered past the bound as a wait of its own, and makes it again on the callee's event or its consumer's bind.
+func (c *Conn) CallKept(key string, body any) (json.RawMessage, error) {
+	called, fallback, err := callSubjectsOn(key, "", c.callerToken())
+	if err != nil {
+		return nil, err
+	}
+	return c.callKeptOn(called, fallback, body)
+}
+
+// CallKeptOn is CallKept to the tool or seat verb of one named machine, as CallOn is Call's.
+func (c *Conn) CallKeptOn(node, key string, body any) (json.RawMessage, error) {
+	if err := calladdress.CheckNode(node); err != nil {
+		return nil, err
+	}
+	called, fallback, err := callSubjectsOn(key, node, c.callerToken())
+	if err != nil {
+		return nil, err
+	}
+	return c.callKeptOn(called, fallback, body)
+}
+
+func (c *Conn) callKeptOn(called, fallback string, body any) (json.RawMessage, error) {
+	data, err := json.Marshal(body)
+	if err != nil {
+		return nil, err
+	}
+	return callKeptWith(c.request, called, fallback, data)
+}
+
+// callKeptWith is callWith for a kept call: busretry.DoRepeatable with no blackout probe.
+func callKeptWith(request func(string, []byte) (json.RawMessage, error), called, fallback string,
+	data []byte) (json.RawMessage, error) {
+	return busretry.DoRepeatable("the call on "+fallback, askBoth(request, called, fallback, data), nil)
+}
+
+// askBoth asks called, then fallback when nothing answers it.
+func askBoth(request func(string, []byte) (json.RawMessage, error), called, fallback string,
+	data []byte) func() (json.RawMessage, error) {
+	return func() (json.RawMessage, error) {
+		result, err := request(called, data)
+		if errors.Is(err, nats.ErrNoResponders) {
+			result, err = request(fallback, data)
+		}
+		return result, err
+	}
+}
+
 // callOn asks called, then fallback when nothing answers it; while the bus itself answers nothing (a user-list reload
 // drops the account's imports for a few seconds, busretry) both are asked again, within busretry.Bound. A call nobody
 // answers on a bus that answers is not asked again: its callee is not running. A call that may have been delivered (a
@@ -67,13 +118,7 @@ func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error
 // callee's answer is *busretry.Answered, whatever its words, so a call that reached its callee is never asked again.
 func callWith(request func(string, []byte) (json.RawMessage, error), blackout func() bool, called, fallback string,
 	data []byte) (json.RawMessage, error) {
-	return busretry.Do("the call on "+fallback, func() (json.RawMessage, error) {
-		result, err := request(called, data)
-		if errors.Is(err, nats.ErrNoResponders) {
-			result, err = request(fallback, data)
-		}
-		return result, err
-	}, blackout)
+	return busretry.Do("the call on "+fallback, askBoth(request, called, fallback, data), blackout)
 }
 
 // answered is the error a callee answered with: delivered, never asked again (busretry.Answered).

go/process/call_test.go

@@ -1,6 +1,7 @@
 package process
 
 import (
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
@@ -108,17 +109,28 @@ func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) {
 		name   string
 		answer []error // per ask; the last repeats
 		asks   int
+		kept   bool // CallKept (novox/hq ADR 0668; issue 1966)
 	}{
-		{"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1},
-		{"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1},
-		{"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3},
+		{"a callee answers jetstream not enabled: asked once", []error{answered("jetstream not enabled")}, 1, false},
+		{"a callee answers no responders in its words: asked once", []error{answered("nats: no responders available for request")}, 1, false},
+		{"no responders on either subject, then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, false},
+		{"a timeout may have been delivered: Call asks once", []error{context.DeadlineExceeded, nil}, 1, false},
+		{"kept: no responders on a bus that answers (a callee restarting), then an answer", []error{nats.ErrNoResponders, nats.ErrNoResponders, nil}, 3, true},
+		{"kept: a timeout, then an answer", []error{context.DeadlineExceeded, nil}, 2, true},
+		{"kept: a callee's answer is asked once", []error{answered("nats: timeout")}, 1, true},
 	} {
 		asks := 0
-		_, err := callWith(func(string, []byte) (json.RawMessage, error) {
+		request := func(string, []byte) (json.RawMessage, error) {
 			e := row.answer[min(asks, len(row.answer)-1)]
 			asks++
 			return json.RawMessage(`{}`), e
-		}, func() bool { return true }, "called", "fallback", nil)
+		}
+		var err error
+		if row.kept {
+			_, err = callKeptWith(request, "called", "fallback", nil)
+		} else {
+			_, err = callWith(request, func() bool { return true }, "called", "fallback", nil)
+		}
 		if asks != row.asks {
 			t.Errorf("%s: asked %d times (%v), want %d", row.name, asks, err, row.asks)
 		}

go/process/event_hub.go

@@ -23,6 +23,7 @@ type EventHub struct {
 	mu     sync.RWMutex
 	routes []hubRoute
 	cons   jetstream.ConsumeContext
+	bound  []func(created time.Time)
 }
 
 type hubRoute struct {
@@ -56,6 +57,17 @@ func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
 	return nil
 }
 
+// OnBound registers f, told each time the hub binds the module's consumer, with when the bus made it. A consumer is made
+// from now (ADR 0443), so an event said before it was made is not delivered to it: a part that took a picture before
+// that must ask its owner again (views.Follower.ConsumerMade). So every handler of an event that ends a kept call's wait
+// (a callee's "my state changed") also registers here the call it makes on that event, made again once at each bind
+// (novox/hq ADR 0668; issue 1966). Register before Start.
+func (h *EventHub) OnBound(f func(created time.Time)) {
+	h.mu.Lock()
+	h.bound = append(h.bound, f)
+	h.mu.Unlock()
+}
+
 // Patterns are the registered subject patterns, in order.
 func (h *EventHub) Patterns() []string {
 	h.mu.RLock()
@@ -83,6 +95,14 @@ func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Accou
 	if err != nil {
 		return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
 	}
+	if info, ierr := cons.Info(ctx); ierr != nil {
+		h.logf("events: the consumer %s on %s is heard, but its information is not read (%v): a first view taken before it was made is not asked again", cred.EventConsumerName(), EventsStream, ierr)
+	} else {
+		h.logf("events: the consumer %s on %s is heard; the bus made it at %s", cred.EventConsumerName(), EventsStream, info.Created.UTC().Format(time.RFC3339))
+		for _, f := range h.bound {
+			f(info.Created.UTC())
+		}
+	}
 	return nil
 }
 

go/seat.go

@@ -86,8 +86,15 @@ const (
 	SettleNak      = "nak"
 	SettleTerm     = "term"
 	SettleProgress = "progress"
+	// SettleNakAfter hands the message back to be delivered again after a delay (SeatSettleAfter): the bus's own
+	// delayed redelivery, so no part keeps a timer for it.
+	SettleNakAfter = "nak-after"
 )
 
+// CapSeatSettleDelay is the capability a tool runner states when it applies SettleNakAfter; a module that settles with
+// it asks it in its manifest's capabilities, so its want waits on a node whose runner does not state it.
+const CapSeatSettleDelay = "seat-settle-delay"
+
 // 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
@@ -124,6 +131,17 @@ func SeatSettle(id, fate string) error {
 	return err
 }
 
+// SeatSettleAfter hands the message pulled under id back to be delivered again after delay (SettleNakAfter). The
+// runtime refuses a delay of 0 or below; a runtime that does not state CapSeatSettleDelay refuses the fate.
+func SeatSettleAfter(id string, delay time.Duration) error {
+	c, err := live()
+	if err != nil {
+		return err
+	}
+	_, err = c.ask("mesh/seat.settle", map[string]any{"id": id, "fate": SettleNakAfter, "delay_ms": delay.Milliseconds()}, 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()

go/seat_pull_test.go

@@ -23,7 +23,10 @@ func TestABundlePullsAndSettlesThroughTheRuntime(t *testing.T) {
 				return nil, err
 			}
 			got <- p
-			return map[string]any{}, SeatSettle(p.ID, SettleAck)
+			if err := SeatSettle(p.ID, SettleAck); err != nil {
+				return nil, err
+			}
+			return map[string]any{}, SeatSettleAfter("p2", time.Minute)
 		}}}, inR, outW)
 	}()
 	lines := bufio.NewScanner(outR)
@@ -59,8 +62,15 @@ func TestABundlePullsAndSettlesThroughTheRuntime(t *testing.T) {
 	}
 	asked = next()
 	params, _ = asked["params"].(map[string]any)
-	if asked["method"] != "mesh/seat.settle" || params["id"] != "p1" || params["fate"] != "ack" {
+	if asked["method"] != "mesh/seat.settle" || params["id"] != "p1" || params["fate"] != "ack" || params["delay_ms"] != nil {
 		t.Fatalf("settled %v", asked)
 	}
 	send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}})
+	// S1 (issue 1748): a delayed nak names its fate and its delay in milliseconds.
+	asked = next()
+	params, _ = asked["params"].(map[string]any)
+	if asked["method"] != "mesh/seat.settle" || params["id"] != "p2" || params["fate"] != "nak-after" || params["delay_ms"] != 60000.0 {
+		t.Fatalf("settled after a delay %v", asked)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}})
 }

go/seatdef/callers.go

@@ -0,0 +1,167 @@
+package seatdef
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// Callers says who may call a verb or send an accepted kind. It is read as text (the form every definition on main
+// has) or as a list (the form design 47 G §336 writes); the reader release carries both so that a definition in
+// either form is read (design 47 G §340), and G's own reader refuses the text.
+type Callers struct {
+	// Text is the callers in words, when the definition gave them so.
+	Text string
+	// List is the callers as a list of entries, when the definition gave them so.
+	List []string
+}
+
+// IsText says whether the definition gave its callers in words.
+func (c Callers) IsText() bool { return c.List == nil }
+
+// UnmarshalJSON reads a string or a list of strings.
+func (c *Callers) UnmarshalJSON(raw []byte) error {
+	raw = bytes.TrimSpace(raw)
+	switch {
+	case len(raw) > 0 && raw[0] == '"':
+		return json.Unmarshal(raw, &c.Text)
+	case len(raw) > 0 && raw[0] == '[':
+		c.List = []string{}
+		return json.Unmarshal(raw, &c.List)
+	}
+	return errors.New("callers is a text or a list of entries")
+}
+
+// MarshalJSON writes the form the definition was read in.
+func (c Callers) MarshalJSON() ([]byte, error) {
+	if c.IsText() {
+		return json.Marshal(c.Text)
+	}
+	return json.Marshal(c.List)
+}
+
+func checkCallers(where string, c *Callers) Problems {
+	if c == nil {
+		return nil
+	}
+	if c.IsText() {
+		if c.Text == "" {
+			return Problems{{where, "is empty"}}
+		}
+		return nil
+	}
+	var ps Problems
+	for i, e := range c.List {
+		if why := CallerEntryProblem(e); why != "" {
+			ps = append(ps, Problem{fmt.Sprintf("%s[%d]", where, i), why})
+		}
+	}
+	return ps
+}
+
+// The entries of a list of callers that are not a name (design 47 G §336).
+const (
+	CallerOperator = "operator"
+	CallerHealers  = "healers"
+	CallerPersons  = "persons"
+	CallerSessions = "sessions"
+	CallerInvokers = "invokers"
+)
+
+// The prefixes of the entries that name a seat's holder or a module (design 47 G §392, §336).
+const (
+	CallerSeatPrefix   = "seat:"
+	CallerModulePrefix = "module:"
+	CallerHolderPrefix = "holder:"
+	CallerSameNode     = "@same-node"
+)
+
+// CallerEntryProblem says why an entry of a list of callers is none of the forms of design 47 G §336, §392; empty
+// when it is one: `operator`, `healers`, `persons`, `sessions`, `invokers`, `seat:<seat>`, `module:<module>` and
+// `holder:<seat>@same-node`. A bare name is refused, since a name such as `mesh-grants` is both a seat and a module.
+func CallerEntryProblem(e string) string {
+	switch e {
+	case CallerOperator, CallerHealers, CallerPersons, CallerSessions, CallerInvokers:
+		return ""
+	}
+	if n, ok := strings.CutPrefix(e, CallerSeatPrefix); ok && wordName.MatchString(n) {
+		return ""
+	}
+	if n, ok := strings.CutPrefix(e, CallerModulePrefix); ok && wordName.MatchString(n) {
+		return ""
+	}
+	if n, ok := strings.CutPrefix(e, CallerHolderPrefix); ok {
+		if seat, ok := strings.CutSuffix(n, CallerSameNode); ok && wordName.MatchString(seat) {
+			return ""
+		}
+	}
+	return fmt.Sprintf("%q is none of operator, healers, persons, sessions, invokers, seat:<seat>, module:<module> and "+
+		"holder:<seat>@same-node", e)
+}
+
+// The classes of a verb (design 47 G §391).
+const (
+	ClassRead    = "read"
+	ClassAct     = "act"
+	ClassWarrant = "warrant"
+)
+
+// Classes are the classes a verb's `class` may name.
+var Classes = []string{ClassRead, ClassAct, ClassWarrant}
+
+// Keep is an event's `keep`: the string of today's definitions (KeepLastPerSubject or KeepStream, with the message's
+// max-age-days and max-mib beside it) or the object of design 47 G §396 (`max-age-days`, `max-mib`,
+// `last-per-subject`). The reader release reads both (design 47 G §340, §409).
+type Keep struct {
+	// Mode is the string form; empty when the object form was given.
+	Mode string
+	// MaxAgeDays, MaxMiB and LastPerSubject are the object form's fields; MaxAgeDays and MaxMiB are nil when unsaid.
+	MaxAgeDays     *int
+	MaxMiB         *int
+	LastPerSubject bool
+	// Object says the object form was given.
+	Object bool
+}
+
+type keepObject struct {
+	MaxAgeDays     *int `json:"max-age-days"`
+	MaxMiB         *int `json:"max-mib"`
+	LastPerSubject bool `json:"last-per-subject"`
+}
+
+// UnmarshalJSON reads the string or the object, the object strictly.
+func (k *Keep) UnmarshalJSON(raw []byte) error {
+	raw = bytes.TrimSpace(raw)
+	if len(raw) > 0 && raw[0] == '"' {
+		return json.Unmarshal(raw, &k.Mode)
+	}
+	var o keepObject
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&o); err != nil {
+		return fmt.Errorf("keep is a string or an object of max-age-days, max-mib and last-per-subject: %w", err)
+	}
+	*k = Keep{MaxAgeDays: o.MaxAgeDays, MaxMiB: o.MaxMiB, LastPerSubject: o.LastPerSubject, Object: true}
+	return nil
+}
+
+// MarshalJSON writes the form the definition was read in.
+func (k Keep) MarshalJSON() ([]byte, error) {
+	if !k.Object {
+		return json.Marshal(k.Mode)
+	}
+	return json.Marshal(keepObject{k.MaxAgeDays, k.MaxMiB, k.LastPerSubject})
+}
+
+func checkKeepObject(where string, k Keep) Problems {
+	var ps Problems
+	if k.MaxAgeDays != nil && *k.MaxAgeDays < 0 {
+		ps = append(ps, Problem{where + ".max-age-days", "is negative (0 is no age limit)"})
+	}
+	if k.MaxMiB != nil && *k.MaxMiB <= 0 {
+		ps = append(ps, Problem{where + ".max-mib", "is not above zero"})
+	}
+	return ps
+}

go/seatdef/refs.go

@@ -1,18 +1,28 @@
 package seatdef
 
-import "strings"
-
 // ResolveRefs is a schema (its all-string lists as []string, StringLists) with every `{"$ref": "#/defs/<name>"}` replaced by the definition's named type, as the seat
 // list's readers need it (a verb's input and answer are one schema each, not a schema and its types). A reference to a
 // name the definition does not define stays as it is; nesting deeper than 32 levels stops there.
-func ResolveRefs(schema map[string]any, defs map[string]map[string]any) map[string]any {
+//
+// A reference `#/bodies/<name>` resolves against the definition's bodies, given after defs (design 47 G §439).
+func ResolveRefs(schema map[string]any, defs map[string]map[string]any, bodies ...map[string]map[string]any) map[string]any {
 	if schema == nil {
 		return nil
 	}
-	m, _ := StringLists(resolveValue(schema, defs, 0)).(map[string]any)
+	all := make(map[string]map[string]any, len(defs))
+	for n, d := range defs {
+		all["#/defs/"+n] = d
+	}
+	for _, b := range bodies {
+		for n, d := range b {
+			all["#/bodies/"+n] = d
+		}
+	}
+	m, _ := StringLists(resolveValue(schema, all, 0)).(map[string]any)
 	return m
 }
 
+// resolveValue resolves against refs, each reference as written (`#/defs/<name>`, `#/bodies/<name>`).
 func resolveValue(v any, defs map[string]map[string]any, depth int) any {
 	if depth > 32 {
 		return v
@@ -20,7 +30,7 @@ func resolveValue(v any, defs map[string]map[string]any, depth int) any {
 	switch x := v.(type) {
 	case map[string]any:
 		if r, ok := x["$ref"].(string); ok {
-			if d, ok := defs[strings.TrimPrefix(r, "#/defs/")]; ok {
+			if d, ok := defs[r]; ok {
 				return resolveValue(d, defs, depth+1)
 			}
 		}

go/seatdef/refs_test.go

@@ -11,6 +11,7 @@ func TestResolveRefsPutsNamedTypesInPlace(t *testing.T) {
 		"id":  {"type": "string"},
 		"row": {"type": "object", "properties": map[string]any{"id": map[string]any{"$ref": "#/defs/id"}}},
 	}
+	bodies := map[string]map[string]any{"Finding": {"type": "object", "properties": map[string]any{"id": map[string]any{"$ref": "#/defs/id"}}}}
 	cases := []struct {
 		name         string
 		schema, want map[string]any
@@ -20,11 +21,15 @@ func TestResolveRefsPutsNamedTypesInPlace(t *testing.T) {
 			map[string]any{"type": "array", "items": map[string]any{"type": "object",
 				"properties": map[string]any{"id": map[string]any{"type": "string"}}}}},
 		{"a name it does not define stays", map[string]any{"$ref": "#/defs/gone"}, map[string]any{"$ref": "#/defs/gone"}},
+		{"a body", map[string]any{"type": "array", "items": map[string]any{"$ref": "#/bodies/Finding"}},
+			map[string]any{"type": "array", "items": map[string]any{"type": "object",
+				"properties": map[string]any{"id": map[string]any{"type": "string"}}}}},
+		{"a body it does not define stays", map[string]any{"$ref": "#/bodies/Gone"}, map[string]any{"$ref": "#/bodies/Gone"}},
 		{"a list of schemas", map[string]any{"oneOf": []any{map[string]any{"$ref": "#/defs/id"}}},
 			map[string]any{"oneOf": []any{map[string]any{"type": "string"}}}},
 	}
 	for _, c := range cases {
-		if got := ResolveRefs(c.schema, defs); !reflect.DeepEqual(got, c.want) {
+		if got := ResolveRefs(c.schema, defs, bodies); !reflect.DeepEqual(got, c.want) {
 			t.Errorf("%s: %v, want %v", c.name, got, c.want)
 		}
 	}

go/seatdef/seatdef.go

@@ -20,6 +20,7 @@ import (
 	"slices"
 	"sort"
 	"strings"
+	"time"
 )
 
 // MaxSize is the largest definition, or part, the mesh keeps (design 26: at most 256 KiB a definition).
@@ -92,6 +93,115 @@ type Definition struct {
 	Paths    []Path       `json:"paths,omitempty"`
 	// Defs are named types a schema refers to as `{"$ref": "#/defs/<name>"}`.
 	Defs map[string]map[string]any `json:"defs,omitempty"`
+	// Bodies are the bodies other modules say on their own streams, named: each a JSON Schema, referred to as
+	// `{"$ref": "#/bodies/<name>"}`. The grants part makes no seat event or stream from them (design 47 G §439, §409).
+	Bodies map[string]map[string]any `json:"bodies,omitempty"`
+	// Acts are the seat's acting rules (novox/hq ADR 0656 §1; design 47 *What ADR 0656 leaves to this design*): each step
+	// of a delivery the seat's holder acts in, on which modules, and how long an entry waits there for the holder's new
+	// version. At either grain: a seat-list definition names its holder's step too.
+	Acts []ActingRule `json:"acts,omitempty"`
+}
+
+// ActingRule is one acting rule of a seat: the step its holder acts in, the modules it acts on (every module, or those
+// whose manifest says this language), and how long an entry waits in that step for the holder's new version, a
+// duration such as "30m"; none is the step's default (DefaultBound).
+type ActingRule struct {
+	Step  string `json:"step"`
+	On    string `json:"on,omitempty"`
+	Bound string `json:"bound,omitempty"`
+}
+
+// The steps of a delivery a seat's holder acts in (design 47): what the indexer, the checker, the build modules, the
+// node-engine, the grants part and the assigner each do for a change.
+const (
+	StepTakeIn   = "take-in"
+	StepValidate = "validate"
+	StepCheck    = "check"
+	StepBuild    = "build"
+	StepCompose  = "compose"
+	StepGrant    = "grant"
+	StepPlace    = "place"
+	// StepPipeline is the runner's step, in the place of StepCheck from G's delivery (design 47 G §203).
+	StepPipeline = "pipeline"
+)
+
+// Steps are the steps an acting rule names, in a delivery's order.
+var Steps = []string{StepTakeIn, StepValidate, StepCheck, StepBuild, StepCompose, StepGrant, StepPlace, StepPipeline}
+
+// OnEvery is an acting rule's `on` for every module; an absent `on` reads as it.
+const OnEvery = "every"
+
+// OnKindImage is an acting rule's `on` for the modules whose artifact entry's kind is image (design 47 G §324, §397).
+const OnKindImage = "kind:image"
+
+// DefaultBound is how long an entry waits in step for its acting part's new version when the rule names no bound
+// (design 47 *An acting part's new version acts first*): 60 minutes to build, 30 to compose, 20 to check (the first
+// tier's), 10 to grant and to place; the indexer's take-in and validate have no wait, so zero.
+func DefaultBound(step string) time.Duration {
+	switch step {
+	case StepBuild:
+		return 60 * time.Minute
+	case StepCompose:
+		return 30 * time.Minute
+	case StepCheck:
+		return 20 * time.Minute
+	case StepGrant, StepPlace:
+		return 10 * time.Minute
+	case StepPipeline:
+		return 30 * time.Minute
+	}
+	return 0
+}
+
+// Acting is the modules the rule acts on: OnEvery, or a language.
+func (r ActingRule) Acting() string {
+	if r.On == "" {
+		return OnEvery
+	}
+	return r.On
+}
+
+// ActsOn says whether the rule acts on a module whose manifest says language.
+func (r ActingRule) ActsOn(language string) bool {
+	on := r.Acting()
+	return on == OnEvery || strings.EqualFold(on, strings.TrimSpace(language))
+}
+
+// Wait is how long an entry waits in the rule's step for its acting part's new version: its bound, or the step's
+// default. A bound the reader refused never reaches here; one that does not parse reads as the default.
+func (r ActingRule) Wait() time.Duration {
+	if d, err := time.ParseDuration(r.Bound); err == nil && d > 0 {
+		return d
+	}
+	return DefaultBound(r.Step)
+}
+
+func checkActs(rs []ActingRule) Problems {
+	var ps Problems
+	seen := map[string]bool{}
+	for i, r := range rs {
+		where := fmt.Sprintf("acts[%d]", i)
+		switch {
+		case r.Step == "":
+			ps = append(ps, Problem{where + ".step", "missing"})
+		case !slices.Contains(Steps, r.Step):
+			ps = append(ps, Problem{where + ".step", fmt.Sprintf("%q is not a step (%s)", r.Step, strings.Join(Steps, ", "))})
+		}
+		if r.On != "" && r.On != OnEvery && r.On != OnKindImage && !wordName.MatchString(r.On) {
+			ps = append(ps, Problem{where + ".on", fmt.Sprintf("%q is neither every, kind:image nor a language's lower-case name", r.On)})
+		}
+		if r.Bound != "" {
+			if d, err := time.ParseDuration(r.Bound); err != nil || d <= 0 {
+				ps = append(ps, Problem{where + ".bound", fmt.Sprintf("%q is not a duration such as 30m", r.Bound)})
+			}
+		}
+		if k := r.Step + " " + r.Acting(); r.Step != "" && seen[k] {
+			ps = append(ps, Problem{where, fmt.Sprintf("the step %s on %s twice", r.Step, r.Acting())})
+		} else {
+			seen[k] = true
+		}
+	}
+	return ps
 }
 
 // Verb is one verb the seat serves: its input and, written whole, its answer, each a JSON Schema.
@@ -104,16 +214,142 @@ type Verb struct {
 	Replaces []string       `json:"replaces,omitempty"`
 	Input    map[string]any `json:"input,omitempty"`
 	Answer   map[string]any `json:"answer,omitempty"`
-	// Callers says who may call it, in words: the list itself is the holder's trusted config value (design 51).
-	Callers string `json:"callers,omitempty"`
+	// Callers says who may call it, as a list (design 47 G §336) or, in the definitions before G, in words.
+	Callers *Callers `json:"callers,omitempty"`
+	// Class is the verb's class, ClassRead, ClassAct or ClassWarrant (design 47 G §391).
+	Class string `json:"class,omitempty"`
+	// NeedsApproval marks the verb for every holder: the operator approves a call of it first (design 46 *The approval
+	// mark* §2).
+	NeedsApproval bool `json:"needs-approval,omitempty"`
+	// ByCaller says the holder's process answers the verb itself and tells callers apart by the account the bus proved
+	// (design 47 G §554, §621, §697).
+	ByCaller bool `json:"by-caller,omitempty"`
+	// Chunks says the holder's process sends the answer's file in chunks, each acknowledged by the caller, on the
+	// subjects mesh.seat.<seat>.chunk.<verb>.<call>.<n> and mesh.seat.<seat>.ack.<verb>.<call>.<n> (design 47 G §681).
+	Chunks bool `json:"chunks,omitempty"`
 }
 
+// The streams an accepted kind may ride (design 47 G §415).
+const (
+	StreamCheck    = "check"
+	StreamDelivery = "delivery"
+)
+
 // Message is one accepted message or one event of the seat, its body a JSON Schema.
 type Message struct {
+	// Name is the message's name; an event's may be lower-case words joined by dots (pull.updated), as a module's
+	// event keys are.
 	Name string         `json:"name"`
 	Body map[string]any `json:"body,omitempty"`
 	// ByCaller says the subject ends in the name of who sent it (or, for an event, of who it is for).
 	ByCaller bool `json:"by-caller,omitempty"`
+	// Subject is an event's subject after `mesh.seat.<seat>.event.` when it carries parts (design 47 *What each step
+	// says*: `owed.<repository>.<number>`, `running.<node>.<module>`, `milestone.<milestone>.asked`): its tokens are
+	// the name's words in order with parts between angle brackets among them, so a reader that knows only the name
+	// still knows the event. Empty, the subject is the name. A part `<node>` is the node that publishes it, which
+	// the grants part grants a node's account alone (ADR 0363 §2).
+	Subject string `json:"subject,omitempty"`
+	// Keep says how an event is kept when not as the family keeps the seat's others, in a stream of its own named
+	// KeptStream: KeepLastPerSubject, the last message of each subject and no age limit (design 47: the node-engine's
+	// running.<node>.<module>, so the last build of every module on every node stays on the bus however long ago it
+	// was said, and a durable consumer starting from the last message per subject holds them all); or KeepStream,
+	// every message within MaxAgeDays and MaxMiB.
+	Keep *Keep `json:"keep,omitempty"`
+	// Callers says who may send an accepted kind, as a verb's (design 47 G §338).
+	Callers *Callers `json:"callers,omitempty"`
+	// Stream is the stream an accepted kind rides, StreamCheck or StreamDelivery; its subject is `<stream>.<kind>`, and
+	// the same kind name may ride more than one stream (design 47 G §415).
+	Stream string `json:"stream,omitempty"`
+	// MaxAgeDays and MaxMiB bound an event kept KeepStream: how long a message is kept and how much the stream holds
+	// (design 47 *What the gate owes the forge*: the stream `owed`, 7 days and at most 64 MiB). Given with KeepStream
+	// alone.
+	MaxAgeDays int `json:"max-age-days,omitempty"`
+	MaxMiB     int `json:"max-mib,omitempty"`
+}
+
+// The ways an event is kept in a stream of its own, KeptStream, instead of its seat's family.
+const (
+	// KeepLastPerSubject keeps the last message of each subject, with no age limit.
+	KeepLastPerSubject = "last-per-subject"
+	// KeepStream keeps every message for MaxAgeDays, at most MaxMiB in all.
+	KeepStream = "stream"
+)
+
+// KeptStream is the stream an event with Keep is kept in: `SEAT_<seat>_<event>` (design 47 G §260, §396).
+func KeptStream(seat string, m Message) string { return KeptStreamPrefix + seat + "_" + m.Name }
+
+// KeptStreamPrefix begins the name of every kept event's stream.
+const KeptStreamPrefix = "SEAT_"
+
+// legacyMode is the string form of Keep, empty for none or for the object form.
+func (m Message) legacyMode() string {
+	if m.Keep == nil {
+		return ""
+	}
+	return m.Keep.Mode
+}
+
+// SubjectPattern is the event's subject after the seat's event prefix with its parts between angle brackets: the
+// name itself when the definition gives no subject.
+func (m Message) SubjectPattern() string {
+	if m.Subject == "" {
+		return m.Name
+	}
+	return m.Subject
+}
+
+// Parts are the names of the subject's parts, in order; none for a subject that is the name.
+func (m Message) Parts() []string {
+	var out []string
+	for _, tok := range strings.Split(m.SubjectPattern(), ".") {
+		if keyPartOK.MatchString(tok) {
+			out = append(out, tok[1:len(tok)-1])
+		}
+	}
+	return out
+}
+
+// Local is the subject after the seat's event prefix with `*` for each part: what a manifest's `emits` and
+// `consumes` name of the event, and the filter that hears every instance of it.
+func (m Message) Local() string {
+	toks := strings.Split(m.SubjectPattern(), ".")
+	for i, tok := range toks {
+		if keyPartOK.MatchString(tok) {
+			toks[i] = "*"
+		}
+	}
+	return strings.Join(toks, ".")
+}
+
+// checkSubject says what is wrong with an event's subject, or "".
+func checkSubject(m Message) string {
+	if m.Subject == "" {
+		return ""
+	}
+	if m.ByCaller {
+		return "an event named by its caller ends in the caller's name; it carries no subject of its own"
+	}
+	var words, parts []string
+	for _, tok := range strings.Split(m.Subject, ".") {
+		switch {
+		case keyPartOK.MatchString(tok):
+			if slices.Contains(parts, tok) {
+				return fmt.Sprintf("part %s twice", tok)
+			}
+			parts = append(parts, tok)
+		case wordName.MatchString(tok):
+			words = append(words, tok)
+		default:
+			return fmt.Sprintf("%q is neither a lower-case word nor a part between < and >", tok)
+		}
+	}
+	if len(parts) == 0 {
+		return "no part: an event whose subject is its name gives no subject"
+	}
+	if strings.Join(words, ".") != m.Name {
+		return fmt.Sprintf("its words are %q, not the event's name %q", strings.Join(words, "."), m.Name)
+	}
+	return ""
 }
 
 // HolderEvent is one of the holder's own events listed beside the seat.
@@ -172,6 +408,8 @@ type State struct {
 	History     int    `json:"history,omitempty"`
 	MaxAgeDays  int    `json:"max-age-days,omitempty"`
 	Description string `json:"description,omitempty"`
+	// View is true when the seat offers the state as a view for a module to follow by `consumes` (design 47 G §699).
+	View bool `json:"view,omitempty"`
 }
 
 // KeyForm is one form a state's key takes: a pattern of named parts (`<asker>.<id>`, a part between angle brackets,
@@ -461,6 +699,7 @@ func (d Definition) check() Problems {
 		}
 	}
 	ps = append(ps, checkStates(d.States)...)
+	ps = append(ps, checkActs(d.Acts)...)
 	for i, p := range d.Provisions {
 		where := fmt.Sprintf("provisions[%d]", i)
 		if p.Name == "" {
@@ -567,6 +806,10 @@ func checkVerbs(vs []Verb, whole bool) Problems {
 			ps = append(ps, Problem{where + ".name", fmt.Sprintf("%q twice", v.Name)})
 		}
 		seen[v.Name] = true
+		ps = append(ps, checkCallers(where+".callers", v.Callers)...)
+		if v.Class != "" && !slices.Contains(Classes, v.Class) {
+			ps = append(ps, Problem{where + ".class", fmt.Sprintf("%q is not a class (%s)", v.Class, strings.Join(Classes, ", "))})
+		}
 		if whole && v.Input == nil {
 			ps = append(ps, Problem{where + ".input", "missing: a whole definition gives every verb's input"})
 		}
@@ -582,12 +825,52 @@ func checkMessages(field string, ms []Message, whole bool) Problems {
 	seen := map[string]bool{}
 	for i, m := range ms {
 		where := fmt.Sprintf("%s[%d]", field, i)
-		if !wordName.MatchString(m.Name) {
+		named := wordName
+		if field == "events" {
+			named = eventName
+		}
+		if !named.MatchString(m.Name) {
 			ps = append(ps, Problem{where + ".name", "missing, or not a lower-case name"})
-		} else if seen[m.Name] {
+		} else if key := m.Stream + "/" + m.Name; seen[key] {
 			ps = append(ps, Problem{where + ".name", fmt.Sprintf("%q twice", m.Name)})
 		}
-		seen[m.Name] = true
+		seen[m.Stream+"/"+m.Name] = true
+		switch {
+		case m.Stream != "" && field != "accepts":
+			ps = append(ps, Problem{where + ".stream", "only an accepted kind rides a stream"})
+		case m.Stream != "" && m.Stream != StreamCheck && m.Stream != StreamDelivery:
+			ps = append(ps, Problem{where + ".stream", fmt.Sprintf("%q is not %s or %s", m.Stream, StreamCheck, StreamDelivery)})
+		}
+		mode := m.legacyMode()
+		switch {
+		case m.Keep != nil && field != "events":
+			ps = append(ps, Problem{where + ".keep", "only an event is kept"})
+		case m.Keep != nil && m.Keep.Object:
+			ps = append(ps, checkKeepObject(where+".keep", *m.Keep)...)
+			if m.MaxAgeDays != 0 || m.MaxMiB != 0 {
+				ps = append(ps, Problem{where + ".max-age-days", "an event whose keep is an object gives its max-age-days and max-mib there"})
+			}
+			if m.Keep.LastPerSubject && len(m.Parts()) == 0 && !m.ByCaller {
+				ps = append(ps, Problem{where + ".keep", "an event of one subject kept last per subject is one message: give it a subject with parts"})
+			}
+		case mode != "" && mode != KeepLastPerSubject && mode != KeepStream:
+			ps = append(ps, Problem{where + ".keep", fmt.Sprintf("%q is not %s or %s", mode, KeepLastPerSubject, KeepStream)})
+		case mode == KeepLastPerSubject && len(m.Parts()) == 0 && !m.ByCaller:
+			ps = append(ps, Problem{where + ".keep", "an event of one subject kept last per subject is one message: give it a subject with parts"})
+		case mode == KeepStream && (m.MaxAgeDays <= 0 || m.MaxMiB <= 0):
+			ps = append(ps, Problem{where + ".keep", "an event kept in a stream of its own gives its max-age-days and max-mib"})
+		case mode != KeepStream && (m.MaxAgeDays != 0 || m.MaxMiB != 0):
+			ps = append(ps, Problem{where + ".max-age-days", "only an event kept as a stream of its own gives max-age-days and max-mib"})
+		}
+		if m.Callers != nil && field != "accepts" {
+			ps = append(ps, Problem{where + ".callers", "only an accepted kind carries callers"})
+		}
+		ps = append(ps, checkCallers(where+".callers", m.Callers)...)
+		if m.Subject != "" && field != "events" {
+			ps = append(ps, Problem{where + ".subject", "only an event carries a subject of its own"})
+		} else if why := checkSubject(m); why != "" {
+			ps = append(ps, Problem{where + ".subject", why})
+		}
 		if whole && m.Body == nil {
 			ps = append(ps, Problem{where + ".body", "missing: a whole definition gives every message's body"})
 		}

go/seatdef/seatdef_test.go

@@ -52,6 +52,31 @@ func TestADefinitionIsReadStrictlyNamingWhatIsWrong(t *testing.T) {
 			`"subject":"mesh.mod.<holder>.event.verdict","family":"checker.asks"}],"family":`, 1), "a family names the holder's own stream"},
 		{"a holder's event on a subject of no module", strings.Replace(whole, `"family":`, `"holder-events":[{"name":"verdict",`+
 			`"subject":"mesh.event.verdict"}],"family":`, 1), "holder-events[0].subject: not mesh.mod.<holder>.event.verdict"},
+		{"an event with dotted words", strings.Replace(whole, `"name":"created"`, `"name":"pull.updated"`, 1), ""},
+		{"an event whose subject carries parts", strings.Replace(whole, `"name":"created"`,
+			`"name":"milestone.asked","subject":"milestone.<milestone>.asked"`, 1), ""},
+		{"a subject whose words are not the name", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","subject":"owing.<repository>"`, 1), `its words are "owing", not the event's name "owed"`},
+		{"a subject with no part", strings.Replace(whole, `"name":"created"`, `"name":"owed","subject":"owed"`, 1),
+			"events[0].subject: no part"},
+		{"a subject's part twice", strings.Replace(whole, `"name":"created"`,
+			`"name":"running","subject":"running.<node>.<node>"`, 1), "part <node> twice"},
+		{"a subject on an event named by its caller", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","by-caller":true,"subject":"owed.<number>"`, 1), "carries no subject of its own"},
+		{"an event kept last per subject", strings.Replace(whole, `"name":"created"`,
+			`"name":"running","subject":"running.<node>.<module>","keep":"last-per-subject"`, 1), ""},
+		{"an event kept otherwise", strings.Replace(whole, `"name":"created"`,
+			`"name":"running","subject":"running.<node>","keep":"forever"`, 1), `events[0].keep: "forever" is not last-per-subject or stream`},
+		{"an event kept in a stream of its own, bounded", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","subject":"owed.<repository>.<number>","keep":"stream","max-age-days":7,"max-mib":64`, 1), ""},
+		{"an event kept in a stream of its own, unbounded", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","subject":"owed.<repository>.<number>","keep":"stream","max-age-days":7`, 1), "gives its max-age-days and max-mib"},
+		{"a bound on an event kept by its family", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","max-mib":64`, 1), "events[0].max-age-days: only an event kept as a stream"},
+		{"an event of one subject kept last per subject", strings.Replace(whole, `"name":"created"`,
+			`"name":"running","keep":"last-per-subject"`, 1), "give it a subject with parts"},
+		{"an accept with dotted words", strings.Replace(whole, `"events":`,
+			`"accepts":[{"name":"run.now","body":{}}],"events":`, 1), "accepts[0].name"},
 		{"a verb twice", strings.Replace(listGrain, `]}`, `,{"name":"add"}]}`, 1), `"add" twice`},
 		{"states at the seat list's grain", strings.Replace(listGrain, `"domain":"Core","decision":"ADR 1",`,
 			`"domain":"Core","decision":"ADR 1","states":[{"name":"asks"}],`, 1), "grain: seat-list holds"},
@@ -59,6 +84,8 @@ func TestADefinitionIsReadStrictlyNamingWhatIsWrong(t *testing.T) {
 			`"domain":"Core","decision":"ADR 1","records":[{"name":"asks"}],`, 1), "records: a field the format does not know"},
 		{"a state with a key form and a value", withState(whole, `{"name":"asks","template":"state","writer":"caller",
 			"history":3,"key-forms":[{"key":"<asker>.<id>","value":{"type":"object"}}]}`), ""},
+		{"a state offered as a view", withState(whole, `{"name":"asks","template":"state","view":true,
+			"key-forms":[{"key":"<id>","value":{"type":"object"}}]}`), ""},
 		{"a stream state with an age", withState(whole, `{"name":"placement","template":"events","max-age-days":7,
 			"key-forms":[{"key":"changed.<node>","value":{"type":"object"}}]}`), ""},
 		{"a state of an unknown template", withState(whole, `{"name":"asks","template":"bucket",
@@ -74,6 +101,59 @@ func TestADefinitionIsReadStrictlyNamingWhatIsWrong(t *testing.T) {
 			"key-forms":[{"key":"<id>","value":{}}]}`), "states[0].history"},
 		{"a state twice", withState(withState(whole, `{"name":"asks","template":"state","key-forms":[{"key":"<id>","value":{}}]}`),
 			`{"name":"asks","template":"log","key-forms":[{"key":"<id>","value":{}}]}`), `"asks" twice`},
+		// W1.1: a seat's acting rules (design 47 *What ADR 0656 leaves to this design*), at either grain.
+		{"acting rules", strings.Replace(whole, `"family":`, `"acts":[{"step":"take-in"},{"step":"validate","on":"every"},`+
+			`{"step":"build","on":"go","bound":"60m"}],"family":`, 1), ""},
+		{"an acting rule at the seat list's grain", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"compose","bound":"30m"}],"verbs":`, 1), ""},
+		{"an acting rule with no step", strings.Replace(listGrain, `"verbs":`, `"acts":[{"on":"go"}],"verbs":`, 1), "acts[0].step: missing"},
+		{"an acting rule of an unknown step", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"deploy"}],"verbs":`, 1),
+			`acts[0].step: "deploy" is not a step (take-in, validate, check, build, compose, grant, place, pipeline)`},
+		{"an acting rule on no language", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"build","on":"Go Lang"}],"verbs":`, 1),
+			"acts[0].on"},
+		{"an acting rule's bound that is no duration", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"build","bound":"an hour"}],"verbs":`, 1),
+			`acts[0].bound: "an hour" is not a duration`},
+		{"an acting rule's field the format does not know", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"build","when":"now"}],"verbs":`, 1),
+			"when: a field the format does not know"},
+		{"one step on one language twice", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"grant"},{"step":"grant","on":"every"}],"verbs":`, 1),
+			"acts[1]: the step grant on every twice"},
+		// The reader release (design 47 G §340, §409): the forms G writes beside the ones on main.
+		{"callers as text", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":"a person",`, 1), ""},
+		{"callers as a list", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":["operator","seat:mesh-grants","invokers"],"class":"read",`, 1), ""},
+		{"callers as neither", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":7,`, 1), "callers is a text or a list"},
+		{"a list with an empty entry", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":["operator",""],`, 1), "verbs[0].callers[1]"},
+		{"a bare name among callers", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":["mesh-grants"],`, 1), `"mesh-grants" is none of operator`},
+		{"a class of the old branches", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":["holder","invokes","reviewers"],`, 1), `"holder" is none of operator`},
+		{"every form among callers", strings.Replace(whole, `"name":"list",`,
+			`"name":"list","callers":["operator","healers","persons","sessions","invokers","seat:mesh-grants","module:mesh-grants","holder:node-backup@same-node"],`, 1), ""},
+		{"a holder not on the same node", strings.Replace(whole, `"name":"list",`, `"name":"list","callers":["holder:node-backup"],`, 1), "none of operator"},
+		{"a class that is none", strings.Replace(whole, `"name":"list",`, `"name":"list","class":"free",`, 1), `verbs[0].class: "free" is not a class (read, act, warrant)`},
+		{"an accepted kind's callers", strings.Replace(whole, `"events":`, `"accepts":[{"name":"step","body":{},"callers":["seat:mesh-pipeline-runner"]}],"events":`, 1), ""},
+		{"an event's callers", strings.Replace(whole, `"name":"created"`, `"name":"created","callers":["operator"]`, 1), "only an accepted kind carries callers"},
+		{"keep as an object", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","keep":{"max-age-days":7,"max-mib":64}`, 1), ""},
+		{"keep as an object with no age limit", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","keep":{"max-age-days":0,"last-per-subject":true},"subject":"owed.<repository>"`, 1), ""},
+		{"keep as an object, last per subject, one subject", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","keep":{"last-per-subject":true}`, 1), "give it a subject with parts"},
+		{"keep as an object with a field it does not know", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","keep":{"days":7}`, 1), "keep is a string or an object"},
+		{"keep as an object with a negative age", strings.Replace(whole, `"name":"created"`,
+			`"name":"owed","keep":{"max-age-days":-1}`, 1), "events[0].keep.max-age-days: is negative"},
+		{"the step pipeline and an image rule", strings.Replace(listGrain, `"verbs":`,
+			`"acts":[{"step":"pipeline"},{"step":"build","on":"kind:image"},{"step":"check"}],"verbs":`, 1), ""},
+		{"an acting rule on another kind", strings.Replace(listGrain, `"verbs":`, `"acts":[{"step":"build","on":"kind:Image"}],"verbs":`, 1), "acts[0].on"},
+		{"a verb marked for approval", strings.Replace(whole, `"name":"list",`, `"name":"list","needs-approval":true,`, 1), ""},
+		{"a verb marked with a word that is no boolean", strings.Replace(whole, `"name":"list",`, `"name":"list","needs-approval":"yes",`, 1), "not a seat definition's JSON"},
+		{"a verb answered by its caller's account, in chunks", strings.Replace(whole, `"name":"list",`, `"name":"list","by-caller":true,"chunks":true,`, 1), ""},
+		{"a verb's by-caller that is no boolean", strings.Replace(whole, `"name":"list",`, `"name":"list","by-caller":"yes",`, 1), "not a seat definition's JSON"},
+		{"a bodies section", strings.Replace(whole, `"family":`, `"bodies":{"Finding":{"type":"object"}},"family":`, 1), ""},
+		{"one kind on two streams", strings.Replace(whole, `"events":`,
+			`"accepts":[{"name":"step","body":{},"stream":"check"},{"name":"step","body":{},"stream":"delivery"}],"events":`, 1), ""},
+		{"one kind twice on a stream", strings.Replace(whole, `"events":`,
+			`"accepts":[{"name":"step","body":{},"stream":"check"},{"name":"step","body":{},"stream":"check"}],"events":`, 1), `"step" twice`},
+		{"a stream that is none", strings.Replace(whole, `"events":`, `"accepts":[{"name":"step","body":{},"stream":"mail"}],"events":`, 1),
+			`accepts[0].stream: "mail" is not check or delivery`},
+		{"an event on a stream", strings.Replace(whole, `"name":"created"`, `"name":"created","stream":"check"`, 1), "only an accepted kind rides a stream"},
 		{"over 256 KiB", listGrain[:len(listGrain)-1] + `,"description":"` + strings.Repeat("x", MaxSize) + `"}`,
 			ErrTooLarge.Error()},
 	}
@@ -150,3 +230,60 @@ func withState(def, state string) string {
 	}
 	return strings.Replace(def, `"domain":"Core","decision":"ADR 1",`, `"domain":"Core","decision":"ADR 1","states":[`+state+`],`, 1)
 }
+
+// TestAnEventsSubjectNamesItsParts holds the three readings of an event's subject: its pattern, its parts in order and
+// the local name a manifest's emits and consumes give it.
+func TestAnEventsSubjectNamesItsParts(t *testing.T) {
+	for _, c := range []struct {
+		m                     Message
+		pattern, parts, local string
+	}{
+		{Message{Name: "created"}, "created", "", "created"},
+		{Message{Name: "pull.updated"}, "pull.updated", "", "pull.updated"},
+		{Message{Name: "running", Subject: "running.<node>.<module>"}, "running.<node>.<module>", "node,module", "running.*.*"},
+		{Message{Name: "milestone.asked", Subject: "milestone.<milestone>.asked"}, "milestone.<milestone>.asked", "milestone", "milestone.*.asked"},
+	} {
+		if got := c.m.SubjectPattern(); got != c.pattern {
+			t.Errorf("%s: pattern %q, want %q", c.m.Name, got, c.pattern)
+		}
+		if got := strings.Join(c.m.Parts(), ","); got != c.parts {
+			t.Errorf("%s: parts %q, want %q", c.m.Name, got, c.parts)
+		}
+		if got := c.m.Local(); got != c.local {
+			t.Errorf("%s: local %q, want %q", c.m.Name, got, c.local)
+		}
+	}
+}
+
+// W1.1: what an acting rule reads as where it says less (design 47 *An acting part's new version acts first*).
+func TestAnActingRuleReadsItsStepsDefaults(t *testing.T) {
+	for _, c := range []struct {
+		rule     ActingRule
+		language string
+		acts     bool
+		wait     string
+	}{
+		{ActingRule{Step: StepBuild, On: "go"}, "go", true, "1h0m0s"},
+		{ActingRule{Step: StepBuild, On: "go"}, "typescript", false, "1h0m0s"},
+		{ActingRule{Step: StepCompose}, "anything", true, "30m0s"},
+		{ActingRule{Step: StepCheck, On: OnEvery}, "", true, "20m0s"},
+		{ActingRule{Step: StepGrant, Bound: "5m"}, "go", true, "5m0s"},
+		{ActingRule{Step: StepPlace}, "go", true, "10m0s"},
+		{ActingRule{Step: StepTakeIn}, "go", true, "0s"},
+		{ActingRule{Step: StepPipeline}, "go", true, "30m0s"},
+	} {
+		if got := c.rule.ActsOn(c.language); got != c.acts {
+			t.Errorf("%+v acts on %q: %v, want %v", c.rule, c.language, got, c.acts)
+		}
+		if got := c.rule.Wait().String(); got != c.wait {
+			t.Errorf("%+v waits %s, want %s", c.rule, got, c.wait)
+		}
+	}
+}
+
+// Design 47 G §260, §396: a kept event's stream is SEAT_<seat>_<event>.
+func TestAKeptEventsStreamIsNamedFromTheSeatAndTheEvent(t *testing.T) {
+	if got := KeptStream("mesh-delivery", Message{Name: "owed"}); got != "SEAT_mesh-delivery_owed" {
+		t.Errorf("the stream is %s", got)
+	}
+}

go/seatlist/bootstrap/mesh-broker.json

@@ -403,6 +403,10 @@
         "why": {
           "type": "string",
           "description": "why it is in its state, in plain words; empty when held as asked"
+        },
+        "binding": {
+          "$ref": "#/defs/Binding",
+          "description": "the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2)"
         }
       },
       "required": [
@@ -540,31 +544,66 @@
     "Binding": {
       "type": "object",
       "properties": {
-        "bus-name": {
+        "local": {
           "type": "string",
-          "description": "the entry's name on the bus"
+          "description": "the entry's local name in the module's manifest"
         },
-        "storage-type": {
+        "storage": {
           "type": "string",
-          "description": "how the bus keeps it"
+          "description": "how the bus keeps it: the entry's storage type"
         },
         "template": {
           "type": "string",
           "description": "its template"
         },
-        "subjects": {
-          "type": "array",
-          "items": {
+        "stream": {
+          "type": "string",
+          "description": "a stream's name on the bus"
+        },
+        "subject": {
+          "type": "string",
+          "description": "the subject a stream's messages are on; also a ledger's history's"
+        },
+        "bucket": {
+          "type": "string",
+          "description": "a key-value bucket's name"
+        },
+        "object-store": {
+          "type": "string",
+          "description": "an object store's name"
+        },
+        "history": {
+          "type": "string",
+          "description": "a ledger's history stream"
+        },
+        "state": {
+          "type": "string",
+          "description": "a ledger's state bucket"
+        },
+        "indexes": {
+          "type": "object",
+          "additionalProperties": {
             "type": "string"
           },
-          "description": "its subjects, for a stream"
+          "description": "a ledger's index buckets, by index name"
+        },
+        "snapshot": {
+          "type": "string",
+          "description": "the subject of a ledger's instances' snapshots"
+        },
+        "types": {
+          "type": "array",
+          "items": {
+            "type": "object"
+          },
+          "description": "a ledger's types, as its manifest declares them"
         }
       },
       "required": [
-        "bus-name",
-        "storage-type"
+        "local",
+        "storage"
       ],
-      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7)"
+      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership"
     }
   }
 }

go/seatlist/bootstrap/mesh-delivery-checker.json

@@ -140,6 +140,40 @@
           "ended"
         ]
       }
+    },
+    {
+      "name": "input",
+      "description": "A checking program's input the checker keeps, by the sha256 of its exact bytes: the facts of a check its verdict is asked of, or the heads its seat list is asked of, as a run ask names it (its input's verb and digest). Answers the digest and the bytes, base64; a digest it keeps no input of is an error. Asked by the build module that runs the program, as its manifest's invokes grants it. Read-only.",
+      "optional": true,
+      "input": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the input's sha256, 64 lowercase hex digits, as the run ask names it"
+          }
+        },
+        "required": [
+          "digest"
+        ]
+      },
+      "answer": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the digest asked"
+          },
+          "input": {
+            "type": "string",
+            "description": "the input's exact bytes, base64, so no encoding on the way changes a byte the digest covers"
+          }
+        },
+        "required": [
+          "digest",
+          "input"
+        ]
+      }
     }
   ],
   "description": "The merge check's tries: start one, read one, end one. The checker's own events stay under its own subjects (ADR 0620 §2).",
@@ -455,6 +489,65 @@
                     "to"
                   ]
                 }
+              },
+              "judged-by": {
+                "type": "object",
+                "description": "the checking program that judged the check: where it was built, the key its build is kept under, whether this check's run built it, and the heads it read (ADR 0644, ADR 0656). Absent in a value written before it",
+                "properties": {
+                  "program": {
+                    "type": "object",
+                    "properties": {
+                      "repository": {
+                        "type": "string"
+                      },
+                      "commit": {
+                        "type": "string"
+                      },
+                      "dir": {
+                        "type": "string"
+                      },
+                      "package": {
+                        "type": "string"
+                      }
+                    },
+                    "required": [
+                      "repository",
+                      "commit",
+                      "dir",
+                      "package"
+                    ]
+                  },
+                  "key": {
+                    "type": "string",
+                    "description": "program- and 16 hex digits"
+                  },
+                  "built": {
+                    "type": "boolean"
+                  },
+                  "heads": {
+                    "type": "array",
+                    "items": {
+                      "type": "object",
+                      "properties": {
+                        "repository": {
+                          "type": "string"
+                        },
+                        "commit": {
+                          "type": "string"
+                        }
+                      },
+                      "required": [
+                        "repository",
+                        "commit"
+                      ]
+                    }
+                  }
+                },
+                "required": [
+                  "program",
+                  "key",
+                  "heads"
+                ]
               }
             },
             "required": [

go/seatlist/bootstrap/mesh-watchdogs.json

@@ -43,7 +43,7 @@
     },
     {
       "name": "presence",
-      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
+      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
       "input": {
         "properties": {
           "machine": {
@@ -55,7 +55,54 @@
       },
       "answer": {
         "type": "object",
-        "description": "who is connected and who is silent"
+        "description": "with machine, that machine: online and since; without, every key of the state presence",
+        "properties": {
+          "machine": {
+            "type": "string",
+            "description": "the machine asked, with machine"
+          },
+          "online": {
+            "type": "boolean",
+            "description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
+          },
+          "since": {
+            "type": "string",
+            "description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
+          },
+          "state": {
+            "type": "string",
+            "description": "without machine: the state read, presence"
+          },
+          "holds": {
+            "type": "integer",
+            "description": "without machine: how many keys the state holds"
+          },
+          "keys": {
+            "type": "array",
+            "description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<node> or part.<part>.<instance>"
+                },
+                "value": {
+                  "type": "object",
+                  "description": "the key's value, online and its layout's version"
+                },
+                "revision": {
+                  "type": "integer",
+                  "description": "the key's revision"
+                }
+              },
+              "required": [
+                "key",
+                "value"
+              ]
+            }
+          }
+        }
       }
     },
     {

go/seatlist/bootstrap/operator-channel.json

@@ -116,15 +116,65 @@
     },
     {
       "name": "asks",
-      "description": "The asks the router holds: open first, then the last month's, each with how it ended.",
+      "description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
       "input": {
         "type": "object",
-        "properties": {},
-        "description": "no arguments"
+        "properties": {
+          "key": {
+            "type": "string",
+            "description": "one ask, by its key (<asker>.<id>); none for the open asks"
+          },
+          "state": {
+            "type": "string",
+            "description": "open (open and held), ended or all; none is open when no key or since is given, else all"
+          },
+          "since": {
+            "type": "string",
+            "description": "only the asks opened at or after this time, RFC 3339"
+          },
+          "bound": {
+            "type": "integer",
+            "description": "the most bytes of the asks in the answer; said when it cuts the answer short"
+          }
+        }
       },
       "answer": {
         "type": "object",
-        "description": "the asks"
+        "description": "the asks, open first",
+        "properties": {
+          "asks": {
+            "type": "array",
+            "description": "each ask the router holds, as its state asks holds it",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<asker>.<id>"
+                },
+                "state": {
+                  "type": "string",
+                  "description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
+                }
+              },
+              "required": [
+                "key",
+                "state"
+              ]
+            }
+          },
+          "cut": {
+            "type": "string",
+            "description": "what the bound cut, and how to ask for more; empty when nothing was cut"
+          },
+          "matched": {
+            "type": "integer",
+            "description": "how many asks the arguments matched"
+          }
+        },
+        "required": [
+          "asks"
+        ]
       }
     }
   ],

go/seatlist/seatlist.go

@@ -7,7 +7,10 @@
 // **What it asks and hears.** The seat's verb with no seat (every trunk definition's name and digest, Listing) and with
 // one seat (that definition, Answer); the seat's two events, whose bodies name the seat (Changed). It calls the seat,
 // never the module that holds it, and only trunk definitions are held: a pull request's are the indexer's to judge.
-// Nothing polls and nothing waits on a timer: a reader that could not ask says so in its Status and reads again on the
+// An ask that reached no one (busretry.Undelivered: the bus's "no responders", as while the seat's holder re-serves) is
+// asked again within the bus's retry bound (busretry.Bound, about 30 s, backing off; issue 774's form, issue 1966), and so is one
+// that timed out: the ask is a read, safe to make again (busretry.DoRepeatable; novox/hq ADR 0668).
+// Nothing else polls or waits on a timer: a reader that could not ask says so in its Status and reads again on the
 // next event or on Reload, which the host offers as a tool (Tools).
 //
 // **A reader that has not heard acts on nothing new** (ADR 0636 §1): MayAct says what a host may start, compose, grant
@@ -27,6 +30,7 @@ import (
 	"sync"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/busretry"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/seatdef"
 )
@@ -284,7 +288,7 @@ func (r *Reader) seatsLocked() []catalogue.Seat {
 // cannot be read is returned and changes nothing; one definition that does not parse leaves the one held standing and is
 // said in Status.
 func (r *Reader) Reload() error {
-	raw, err := r.cfg.Bus.Ask(Key(), map[string]any{})
+	raw, err := r.ask(map[string]any{}, "the seat list")
 	if err != nil {
 		return r.failed(fmt.Errorf("seatlist: asking %s: %w", Key(), err))
 	}
@@ -358,9 +362,19 @@ func (r *Reader) ReloadIfUnread() <-chan struct{} {
 	return done
 }
 
+// ask asks the seat's verb, and again while the ask reached no one or timed out, within the bus's retry bound: a holder
+// that re-serves for a few seconds while the reader starts leaves no list unread (issue 1966). The ask is a read, so it
+// is safe to make again on a timeout (busretry.DoRepeatable; novox/hq ADR 0668). An answer or a refusal is never asked
+// again.
+func (r *Reader) ask(body map[string]any, what string) (json.RawMessage, error) {
+	return busretry.DoRepeatable(what, func() (json.RawMessage, error) {
+		return r.cfg.Bus.Ask(Key(), body)
+	}, nil)
+}
+
 // fetch asks the seat for one definition and holds it, or drops it when the seat holds none.
 func (r *Reader) fetch(seat string) error {
-	raw, err := r.cfg.Bus.Ask(Key(), map[string]any{"seat": seat})
+	raw, err := r.ask(map[string]any{"seat": seat}, "the seat "+seat)
 	if err != nil {
 		return fmt.Errorf("seatlist: asking %s for %s: %w", Key(), seat, err)
 	}

go/seatlist/seatlist_test.go

@@ -1,23 +1,29 @@
 package seatlist
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/busretry"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/seatdef"
 	"git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest"
+	"github.com/nats-io/nats.go"
 )
 
 // fakeOwner is the seat mesh-catalog: its verb answers from its table and the events are delivered by the test.
 type fakeOwner struct {
 	table    map[string]string // seat -> definition JSON, trunk
 	down     bool
+	silent   int           // when set, so many asks first reach no one (the bus's no responders), as while the holder re-serves
+	timeouts int           // when set, so many asks first time out (a deadline passed, as a callee that restarts mid-ask)
 	gate     chan struct{} // when set, an ask waits for it to be closed
 	asked    []string      // "key seat"
 	handlers map[string]func(string, json.RawMessage) error
@@ -30,6 +36,14 @@ func (f *fakeOwner) Ask(key string, body any) (json.RawMessage, error) {
 	args, _ := body.(map[string]any)
 	seat, _ := args["seat"].(string)
 	f.asked = append(f.asked, key+" "+seat)
+	if f.timeouts > 0 {
+		f.timeouts--
+		return nil, fmt.Errorf("asking %s: %w", key, context.DeadlineExceeded)
+	}
+	if f.silent > 0 {
+		f.silent--
+		return nil, fmt.Errorf("not asked: nothing on this machine answers it (%w)", nats.ErrNoResponders)
+	}
 	if f.down {
 		return nil, errors.New("no answer")
 	}
@@ -266,6 +280,30 @@ func TestStartReturnsAtOnceAndTheReaderSaysItIsReadingUntilTheSeatAnswers(t *tes
 	if st := r.Status(); st.Reading || !st.Heard || !r.MayAct("node-b") {
 		t.Fatalf("after the read: %+v", st)
 	}
+
+	// Issue 1966: the first read falls into the holder's re-serve (the bus answers no responders twice); it is asked again
+	// within the bus's retry bound, with no event and no hand act, and the reader is heard.
+	t.Cleanup(busretry.Waits(10*time.Second, time.Millisecond, 4*time.Millisecond))
+	owner = &fakeOwner{silent: 2, table: map[string]string{"node-b": def("node-b", "node", "put")}}
+	r, err = Start(Config{Bus: owner, Bootstrap: Bootstrapped()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-r.Ready()
+	if st := r.Status(); !st.Heard || st.LastError != "" || !r.Has("node-b") || len(owner.asked) != 4 {
+		t.Fatalf("after a holder that re-served: %+v, asked %q: want heard on the third ask of the list", st, owner.asked)
+	}
+
+	// Issue 1966, ADR 0668: a read of the list is safe to make again, so an ask that timed out is asked again too.
+	owner = &fakeOwner{timeouts: 1, table: map[string]string{"node-b": def("node-b", "node", "put")}}
+	r, err = Start(Config{Bus: owner, Bootstrap: Bootstrapped()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-r.Ready()
+	if st := r.Status(); !st.Heard || st.LastError != "" || !r.Has("node-b") || len(owner.asked) != 3 {
+		t.Fatalf("after an ask that timed out: %+v, asked %q: want heard on the second ask of the list", st, owner.asked)
+	}
 }
 
 // The bootstrap and the tests' seat list are copies of the same contracts files: where a seat is in both, they are equal,

go/seatlist/seatlisttest/seats/mesh-broker.json

@@ -403,6 +403,10 @@
         "why": {
           "type": "string",
           "description": "why it is in its state, in plain words; empty when held as asked"
+        },
+        "binding": {
+          "$ref": "#/defs/Binding",
+          "description": "the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2)"
         }
       },
       "required": [
@@ -540,31 +544,66 @@
     "Binding": {
       "type": "object",
       "properties": {
-        "bus-name": {
+        "local": {
           "type": "string",
-          "description": "the entry's name on the bus"
+          "description": "the entry's local name in the module's manifest"
         },
-        "storage-type": {
+        "storage": {
           "type": "string",
-          "description": "how the bus keeps it"
+          "description": "how the bus keeps it: the entry's storage type"
         },
         "template": {
           "type": "string",
           "description": "its template"
         },
-        "subjects": {
-          "type": "array",
-          "items": {
+        "stream": {
+          "type": "string",
+          "description": "a stream's name on the bus"
+        },
+        "subject": {
+          "type": "string",
+          "description": "the subject a stream's messages are on; also a ledger's history's"
+        },
+        "bucket": {
+          "type": "string",
+          "description": "a key-value bucket's name"
+        },
+        "object-store": {
+          "type": "string",
+          "description": "an object store's name"
+        },
+        "history": {
+          "type": "string",
+          "description": "a ledger's history stream"
+        },
+        "state": {
+          "type": "string",
+          "description": "a ledger's state bucket"
+        },
+        "indexes": {
+          "type": "object",
+          "additionalProperties": {
             "type": "string"
           },
-          "description": "its subjects, for a stream"
+          "description": "a ledger's index buckets, by index name"
+        },
+        "snapshot": {
+          "type": "string",
+          "description": "the subject of a ledger's instances' snapshots"
+        },
+        "types": {
+          "type": "array",
+          "items": {
+            "type": "object"
+          },
+          "description": "a ledger's types, as its manifest declares them"
         }
       },
       "required": [
-        "bus-name",
-        "storage-type"
+        "local",
+        "storage"
       ],
-      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7)"
+      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership"
     }
   }
 }

go/seatlist/seatlisttest/seats/mesh-delivery-checker.json

@@ -140,6 +140,40 @@
           "ended"
         ]
       }
+    },
+    {
+      "name": "input",
+      "description": "A checking program's input the checker keeps, by the sha256 of its exact bytes: the facts of a check its verdict is asked of, or the heads its seat list is asked of, as a run ask names it (its input's verb and digest). Answers the digest and the bytes, base64; a digest it keeps no input of is an error. Asked by the build module that runs the program, as its manifest's invokes grants it. Read-only.",
+      "optional": true,
+      "input": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the input's sha256, 64 lowercase hex digits, as the run ask names it"
+          }
+        },
+        "required": [
+          "digest"
+        ]
+      },
+      "answer": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the digest asked"
+          },
+          "input": {
+            "type": "string",
+            "description": "the input's exact bytes, base64, so no encoding on the way changes a byte the digest covers"
+          }
+        },
+        "required": [
+          "digest",
+          "input"
+        ]
+      }
     }
   ],
   "description": "The merge check's tries: start one, read one, end one. The checker's own events stay under its own subjects (ADR 0620 §2).",
@@ -455,6 +489,65 @@
                     "to"
                   ]
                 }
+              },
+              "judged-by": {
+                "type": "object",
+                "description": "the checking program that judged the check: where it was built, the key its build is kept under, whether this check's run built it, and the heads it read (ADR 0644, ADR 0656). Absent in a value written before it",
+                "properties": {
+                  "program": {
+                    "type": "object",
+                    "properties": {
+                      "repository": {
+                        "type": "string"
+                      },
+                      "commit": {
+                        "type": "string"
+                      },
+                      "dir": {
+                        "type": "string"
+                      },
+                      "package": {
+                        "type": "string"
+                      }
+                    },
+                    "required": [
+                      "repository",
+                      "commit",
+                      "dir",
+                      "package"
+                    ]
+                  },
+                  "key": {
+                    "type": "string",
+                    "description": "program- and 16 hex digits"
+                  },
+                  "built": {
+                    "type": "boolean"
+                  },
+                  "heads": {
+                    "type": "array",
+                    "items": {
+                      "type": "object",
+                      "properties": {
+                        "repository": {
+                          "type": "string"
+                        },
+                        "commit": {
+                          "type": "string"
+                        }
+                      },
+                      "required": [
+                        "repository",
+                        "commit"
+                      ]
+                    }
+                  }
+                },
+                "required": [
+                  "program",
+                  "key",
+                  "heads"
+                ]
               }
             },
             "required": [

go/seatlist/seatlisttest/seats/mesh-watchdogs.json

@@ -43,7 +43,7 @@
     },
     {
       "name": "presence",
-      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
+      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
       "input": {
         "properties": {
           "machine": {
@@ -55,7 +55,54 @@
       },
       "answer": {
         "type": "object",
-        "description": "who is connected and who is silent"
+        "description": "with machine, that machine: online and since; without, every key of the state presence",
+        "properties": {
+          "machine": {
+            "type": "string",
+            "description": "the machine asked, with machine"
+          },
+          "online": {
+            "type": "boolean",
+            "description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
+          },
+          "since": {
+            "type": "string",
+            "description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
+          },
+          "state": {
+            "type": "string",
+            "description": "without machine: the state read, presence"
+          },
+          "holds": {
+            "type": "integer",
+            "description": "without machine: how many keys the state holds"
+          },
+          "keys": {
+            "type": "array",
+            "description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<node> or part.<part>.<instance>"
+                },
+                "value": {
+                  "type": "object",
+                  "description": "the key's value, online and its layout's version"
+                },
+                "revision": {
+                  "type": "integer",
+                  "description": "the key's revision"
+                }
+              },
+              "required": [
+                "key",
+                "value"
+              ]
+            }
+          }
+        }
       }
     },
     {

go/seatlist/seatlisttest/seats/operator-channel.json

@@ -116,15 +116,65 @@
     },
     {
       "name": "asks",
-      "description": "The asks the router holds: open first, then the last month's, each with how it ended.",
+      "description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
       "input": {
         "type": "object",
-        "properties": {},
-        "description": "no arguments"
+        "properties": {
+          "key": {
+            "type": "string",
+            "description": "one ask, by its key (<asker>.<id>); none for the open asks"
+          },
+          "state": {
+            "type": "string",
+            "description": "open (open and held), ended or all; none is open when no key or since is given, else all"
+          },
+          "since": {
+            "type": "string",
+            "description": "only the asks opened at or after this time, RFC 3339"
+          },
+          "bound": {
+            "type": "integer",
+            "description": "the most bytes of the asks in the answer; said when it cuts the answer short"
+          }
+        }
       },
       "answer": {
         "type": "object",
-        "description": "the asks"
+        "description": "the asks, open first",
+        "properties": {
+          "asks": {
+            "type": "array",
+            "description": "each ask the router holds, as its state asks holds it",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<asker>.<id>"
+                },
+                "state": {
+                  "type": "string",
+                  "description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
+                }
+              },
+              "required": [
+                "key",
+                "state"
+              ]
+            }
+          },
+          "cut": {
+            "type": "string",
+            "description": "what the bound cut, and how to ask for more; empty when nothing was cut"
+          },
+          "matched": {
+            "type": "integer",
+            "description": "how many asks the arguments matched"
+          }
+        },
+        "required": [
+          "asks"
+        ]
       }
     }
   ],

go/state.go

@@ -107,6 +107,25 @@ func (s KeptState) Get(key string) (*StateEntry, error) {
 	return &e, json.Unmarshal(raw, &e)
 }
 
+// StatePage is the values of one bulk read: those after a key, in key order, and the key to ask after for the rest.
+type StatePage struct {
+	Entries []StateEntry `json:"entries"`
+	// Next is the last key of Entries when more follow it, else empty.
+	Next string `json:"next,omitempty"`
+}
+
+// Page is the current values in key order after the key `after`, as many as fit maxBytes (keys and values) or the time budget allows, at least
+// one: one request for many values, where Get is one request each. A runtime that does not answer it gives an error
+// satisfying errors.Is(err, errors.ErrUnsupported), and the reader asks Get per key instead.
+func (s KeptState) Page(after string, maxBytes int, budget time.Duration) (StatePage, error) {
+	raw, err := s.ask("page", map[string]any{"after": after, "max_bytes": maxBytes, "budget_ms": budget.Milliseconds()})
+	if err != nil {
+		return StatePage{}, err
+	}
+	var p StatePage
+	return p, json.Unmarshal(raw, &p)
+}
+
 // History is every value the bucket still keeps for key, oldest first — as many as the state declared it
 // keeps, deletes included — and empty for a key it holds nothing for (novox/hq ADR 0307). A read, like Get:
 // the owner's instances and its readers may ask it, and it writes nothing.

go/state_test.go

@@ -222,6 +222,9 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
 	if !errors.Is(&refused{code: CodeChanged, message: "x"}, ErrChanged) {
 		t.Fatal("a refusal coded -32010 is not ErrChanged")
 	}
+	if !errors.Is(&refused{code: CodeNoMethod, message: "no mesh/state.page"}, errors.ErrUnsupported) || errors.Is(&refused{code: -32000}, errors.ErrUnsupported) {
+		t.Fatal("only a refusal coded -32601 is ErrUnsupported")
+	}
 	// Get of a key with no value is nil.
 	send(map[string]any{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": map[string]any{"name": "recall",
 		"arguments": map[string]any{"key": "nothing"}}})

go/stdio.go

@@ -150,7 +150,12 @@ type refused struct {
 func (r *refused) Error() string { return r.message }
 
 // Is makes a compare-and-set the runtime refused because another writer was first ErrChanged.
-func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged }
+func (r *refused) Is(target error) bool {
+	return target == ErrChanged && r.code == CodeChanged || target == errors.ErrUnsupported && r.code == CodeNoMethod
+}
+
+// CodeNoMethod is the error code the runtime answers a verb it does not have with (JSON-RPC's method not found).
+const CodeNoMethod = -32601
 
 // ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq
 // ADR 0259). Told apart by the runtime's error code, never by its words.

go/testdata/beside/CAPTURED

@@ -12,7 +12,8 @@ repositories; before that it was mesh-controller's copy, which still held two mo
 
 mesh-catalog  b5dbfbb784507e211372a7c9640d91f7d6dadd84  modules/*/module.json, modules/*/files/**, modules/*/templates/**, predefined.json
 mesh-contracts 70fe186d7f5743e61f3ce2bbd97a8bfabc5b596e  the seat files of the eight bootstrapped seats held there (seats/, health-and-repair/seats/, operator-and-conversation/seats/), as seat.json.captured
-mesh-delivery 432df75997637542a441b3c8f855335c9070ebbd  contracts/seats/mesh-catalog and mesh-delivery-checker, as seat.json.captured (go/seatlist's bootstrap test)
+mesh-contracts bb16f99e2d5e854bb8f6552ce4b256e7c10e8ea0  of those, health-and-repair/seats/mesh-watchdogs and operator-and-conversation/seats/operator-channel (issue 1966)
+mesh-delivery 932e7ccf025513ec4fce38a534c2c38defbc6701  contracts/seats/mesh-catalog and mesh-delivery-checker, as seat.json.captured (go/seatlist's bootstrap test)
 mesh-host     cdcba9ca73755854d21e4e657d7e46c32b9b5265  internal/apply/placement_guard.go
 
 The two seat copies are put back the same way (git show <commit>:<file> > <file>.captured, the path kept). To move the others, from this repository's go/ directory, with the two repositories checked out beside mesh-sdk:

go/testdata/beside/mesh-contracts/health-and-repair/seats/mesh-watchdogs/seat.json.captured

@@ -43,7 +43,7 @@
     },
     {
       "name": "presence",
-      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent.",
+      "description": "Every machine and tool runner the watchman sees connected, with its last word, and those it says silent; a machine that said it sleeps is shown as sleeping, not silent. With machine, that machine's presence alone: whether the watchman hears it and since when.",
       "input": {
         "properties": {
           "machine": {
@@ -55,7 +55,54 @@
       },
       "answer": {
         "type": "object",
-        "description": "who is connected and who is silent"
+        "description": "with machine, that machine: online and since; without, every key of the state presence",
+        "properties": {
+          "machine": {
+            "type": "string",
+            "description": "the machine asked, with machine"
+          },
+          "online": {
+            "type": "boolean",
+            "description": "with machine: whether the watchman hears the machine now; false for a machine it holds no word of"
+          },
+          "since": {
+            "type": "string",
+            "description": "with machine: when its presence last changed, RFC 3339; empty for a machine the watchman holds no word of"
+          },
+          "state": {
+            "type": "string",
+            "description": "without machine: the state read, presence"
+          },
+          "holds": {
+            "type": "integer",
+            "description": "without machine: how many keys the state holds"
+          },
+          "keys": {
+            "type": "array",
+            "description": "without machine: each key of the state presence, <node> or part.<part>.<instance>, with its value",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<node> or part.<part>.<instance>"
+                },
+                "value": {
+                  "type": "object",
+                  "description": "the key's value, online and its layout's version"
+                },
+                "revision": {
+                  "type": "integer",
+                  "description": "the key's revision"
+                }
+              },
+              "required": [
+                "key",
+                "value"
+              ]
+            }
+          }
+        }
       }
     },
     {

go/testdata/beside/mesh-contracts/operator-and-conversation/seats/operator-channel/seat.json.captured

@@ -116,15 +116,65 @@
     },
     {
       "name": "asks",
-      "description": "The asks the router holds: open first, then the last month's, each with how it ended.",
+      "description": "The asks the router holds: open first, then the last month's, each with how it ended. With no argument only the open asks, within its default bound; key for one ask, state, since and bound narrow or widen it, and the answer says what it cut.",
       "input": {
         "type": "object",
-        "properties": {},
-        "description": "no arguments"
+        "properties": {
+          "key": {
+            "type": "string",
+            "description": "one ask, by its key (<asker>.<id>); none for the open asks"
+          },
+          "state": {
+            "type": "string",
+            "description": "open (open and held), ended or all; none is open when no key or since is given, else all"
+          },
+          "since": {
+            "type": "string",
+            "description": "only the asks opened at or after this time, RFC 3339"
+          },
+          "bound": {
+            "type": "integer",
+            "description": "the most bytes of the asks in the answer; said when it cuts the answer short"
+          }
+        }
       },
       "answer": {
         "type": "object",
-        "description": "the asks"
+        "description": "the asks, open first",
+        "properties": {
+          "asks": {
+            "type": "array",
+            "description": "each ask the router holds, as its state asks holds it",
+            "items": {
+              "type": "object",
+              "properties": {
+                "key": {
+                  "type": "string",
+                  "description": "<asker>.<id>"
+                },
+                "state": {
+                  "type": "string",
+                  "description": "how the ask stands: open, held, or how it ended (answered, expired, cancelled)"
+                }
+              },
+              "required": [
+                "key",
+                "state"
+              ]
+            }
+          },
+          "cut": {
+            "type": "string",
+            "description": "what the bound cut, and how to ask for more; empty when nothing was cut"
+          },
+          "matched": {
+            "type": "integer",
+            "description": "how many asks the arguments matched"
+          }
+        },
+        "required": [
+          "asks"
+        ]
       }
     }
   ],

go/testdata/beside/mesh-contracts/seats/mesh-broker/seat.json.captured

@@ -403,6 +403,10 @@
         "why": {
           "type": "string",
           "description": "why it is in its state, in plain words; empty when held as asked"
+        },
+        "binding": {
+          "$ref": "#/defs/Binding",
+          "description": "the binding a module receives for the entry, as created and recreated carry it; present on an entry whose objects the bus holds, absent on one it does not (design 25 bindings-first-view §2)"
         }
       },
       "required": [
@@ -540,31 +544,66 @@
     "Binding": {
       "type": "object",
       "properties": {
-        "bus-name": {
+        "local": {
           "type": "string",
-          "description": "the entry's name on the bus"
+          "description": "the entry's local name in the module's manifest"
         },
-        "storage-type": {
+        "storage": {
           "type": "string",
-          "description": "how the bus keeps it"
+          "description": "how the bus keeps it: the entry's storage type"
         },
         "template": {
           "type": "string",
           "description": "its template"
         },
-        "subjects": {
-          "type": "array",
-          "items": {
+        "stream": {
+          "type": "string",
+          "description": "a stream's name on the bus"
+        },
+        "subject": {
+          "type": "string",
+          "description": "the subject a stream's messages are on; also a ledger's history's"
+        },
+        "bucket": {
+          "type": "string",
+          "description": "a key-value bucket's name"
+        },
+        "object-store": {
+          "type": "string",
+          "description": "an object store's name"
+        },
+        "history": {
+          "type": "string",
+          "description": "a ledger's history stream"
+        },
+        "state": {
+          "type": "string",
+          "description": "a ledger's state bucket"
+        },
+        "indexes": {
+          "type": "object",
+          "additionalProperties": {
             "type": "string"
           },
-          "description": "its subjects, for a stream"
+          "description": "a ledger's index buckets, by index name"
+        },
+        "snapshot": {
+          "type": "string",
+          "description": "the subject of a ledger's instances' snapshots"
+        },
+        "types": {
+          "type": "array",
+          "items": {
+            "type": "object"
+          },
+          "description": "a ledger's types, as its manifest declares them"
         }
       },
       "required": [
-        "bus-name",
-        "storage-type"
+        "local",
+        "storage"
       ],
-      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7)"
+      "description": "the binding a module receives in its membership for one entry (ADR 0595 §1, §7): the names on the bus of the entry's objects, one form in the events, in list, show and orphans, and in the membership"
     }
   }
 }

go/testdata/beside/mesh-delivery/contracts/seats/mesh-delivery-checker/seat.json.captured

@@ -140,6 +140,40 @@
           "ended"
         ]
       }
+    },
+    {
+      "name": "input",
+      "description": "A checking program's input the checker keeps, by the sha256 of its exact bytes: the facts of a check its verdict is asked of, or the heads its seat list is asked of, as a run ask names it (its input's verb and digest). Answers the digest and the bytes, base64; a digest it keeps no input of is an error. Asked by the build module that runs the program, as its manifest's invokes grants it. Read-only.",
+      "optional": true,
+      "input": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the input's sha256, 64 lowercase hex digits, as the run ask names it"
+          }
+        },
+        "required": [
+          "digest"
+        ]
+      },
+      "answer": {
+        "type": "object",
+        "properties": {
+          "digest": {
+            "type": "string",
+            "description": "the digest asked"
+          },
+          "input": {
+            "type": "string",
+            "description": "the input's exact bytes, base64, so no encoding on the way changes a byte the digest covers"
+          }
+        },
+        "required": [
+          "digest",
+          "input"
+        ]
+      }
     }
   ],
   "description": "The merge check's tries: start one, read one, end one. The checker's own events stay under its own subjects (ADR 0620 §2).",
@@ -455,6 +489,65 @@
                     "to"
                   ]
                 }
+              },
+              "judged-by": {
+                "type": "object",
+                "description": "the checking program that judged the check: where it was built, the key its build is kept under, whether this check's run built it, and the heads it read (ADR 0644, ADR 0656). Absent in a value written before it",
+                "properties": {
+                  "program": {
+                    "type": "object",
+                    "properties": {
+                      "repository": {
+                        "type": "string"
+                      },
+                      "commit": {
+                        "type": "string"
+                      },
+                      "dir": {
+                        "type": "string"
+                      },
+                      "package": {
+                        "type": "string"
+                      }
+                    },
+                    "required": [
+                      "repository",
+                      "commit",
+                      "dir",
+                      "package"
+                    ]
+                  },
+                  "key": {
+                    "type": "string",
+                    "description": "program- and 16 hex digits"
+                  },
+                  "built": {
+                    "type": "boolean"
+                  },
+                  "heads": {
+                    "type": "array",
+                    "items": {
+                      "type": "object",
+                      "properties": {
+                        "repository": {
+                          "type": "string"
+                        },
+                        "commit": {
+                          "type": "string"
+                        }
+                      },
+                      "required": [
+                        "repository",
+                        "commit"
+                      ]
+                    }
+                  }
+                },
+                "required": [
+                  "program",
+                  "key",
+                  "heads"
+                ]
               }
             },
             "required": [

go/views/emit.go

@@ -89,6 +89,7 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue,
 	if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last {
 		from = m.Revision + 1
 	} else {
+		// Marked at the end with nothing said between: a reader behind it hears the mark and asks its first view.
 		e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill})
 	}
 	w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from))
@@ -101,15 +102,16 @@ func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue,
 	}
 	go func() {
 		defer w.Stop()
-		e.follow(ctx, state, made, w.Updates(), fillNow, errs)
+		e.follow(ctx, state, made, from-1, w.Updates(), fillNow, errs)
 	}()
 	return nil
 }
 
-// follow says each change handed on updates in order, and marks once every change handed so far is said. A change
-// that fails to be said is said again, the same change on a backoff (Wait), until it is said or ctx ends: the mark
-// never moves past a change not said, so no reader misses one (design 25 §12.2).
-func (e Emitter) follow(ctx context.Context, state string, made time.Time, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) {
+// follow says each change handed on updates in order, each naming the revision said before it (prev, from the mark it
+// starts after), and marks once every change handed so far is said. A change that fails to be said is said again, the
+// same change on a backoff (Wait), until it is said or ctx ends: the mark never moves past a change not said, so no
+// reader misses one (design 25 §12.2).
+func (e Emitter) follow(ctx context.Context, state string, made time.Time, prev uint64, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) {
 	said := uint64(0)
 	raised := false
 	// A finding a run before this one raised is not known to this one: the first change said clears it once, so a
@@ -129,7 +131,7 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat
 			if u != nil {
 				var first time.Time
 				for attempt := 1; ; attempt++ {
-					err := e.say(ctx, state, made, u)
+					err := e.say(ctx, state, made, prev, u)
 					if err == nil {
 						break
 					}
@@ -149,7 +151,7 @@ func (e Emitter) follow(ctx context.Context, state string, made time.Time, updat
 					raised = !e.clearUnsaid(state)
 					leftover = false
 				}
-				said = u.Revision()
+				said, prev = u.Revision(), u.Revision()
 			}
 			// Marked once every change handed so far is said, so a busy state writes one mark per burst.
 			if said > 0 && len(updates) == 0 {
@@ -236,8 +238,13 @@ func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry)
 	return ev
 }
 
-func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error {
+func (e Emitter) say(ctx context.Context, state string, made time.Time, prev uint64, u jetstream.KeyValueEntry) error {
 	ev := e.Event(state, made, u)
+	ev.Prev = prev
+	return e.publish(ctx, state, ev)
+}
+
+func (e Emitter) publish(ctx context.Context, state string, ev Event) error {
 	data, err := ev.Encode()
 	if err != nil {
 		return err
@@ -282,5 +289,16 @@ func (e Emitter) writeMark(ctx context.Context, state string, m Mark) {
 	}
 	if _, err := e.Marks.Put(ctx, state, data); err != nil {
 		e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err)
+		return
+	}
+	// The mark said as an event of its own, after it is written: a reader whose picture is short of it (a change's event
+	// that never reached it) asks its first view now, which answers at this mark or later. Not said again on a failure:
+	// the next mark says it. A mark at 0 (a state with no change yet) is no event.
+	if m.Revision == 0 {
+		return
+	}
+	ev := Event{V: Version, Owner: e.Owner, State: state, Revision: m.Revision, Created: m.Created, At: m.At, Mark: true}
+	if err := e.publish(ctx, state, ev); err != nil {
+		e.logf("views: the mark of %s at %d is written but not said (%v): a reader short of it asks at the next change", state, m.Revision, err)
 	}
 }

go/views/follow.go

@@ -51,7 +51,17 @@ type Follower struct {
 	// during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read
 	// before it was written.
 	during []Event
-	ready  chan struct{}
+	// heardRev is the highest revision of the owner's state (made at heardCreated) any event handed to Apply carried,
+	// whether it was taken into the picture, passed over or heard while an ask ran: what the picture must reach, so a
+	// reader that heard it knows it is behind without a further event.
+	heardRev     uint64
+	heardCreated time.Time
+	// consumerMade is when the bus made the reader's consumer of the owner's events (EventHub.OnBound): an event said
+	// before it is not delivered to it, so a picture asked before it is behind.
+	consumerMade time.Time
+	// catching is set while catchUp runs: one catch-up at a time.
+	catching bool
+	ready    chan struct{}
 }
 
 // Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision
@@ -69,6 +79,11 @@ type Status struct {
 	Asks     int       `json:"asks"`
 	ByHand   int       `json:"by-hand"`
 	Raised   bool      `json:"raised"`
+	// Heard is the highest revision of the owner's state any event heard carried (made at HeardCreated), taken into the
+	// picture or not; Behind says why the picture is short of it, or of its consumer's making, and an ask for it runs.
+	Heard        uint64    `json:"heard,omitempty"`
+	HeardCreated time.Time `json:"heard-created,omitzero"`
+	Behind       string    `json:"behind,omitempty"`
 	// Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed.
 	Why   string `json:"why,omitempty"`
 	Error string `json:"error,omitempty"`
@@ -98,20 +113,26 @@ func (f *Follower) init() {
 func (f *Follower) say(s Status) {
 	if f.Said != nil {
 		f.mu.Lock()
-		s.Keys = len(f.values)
+		s = f.statusLocked(s)
 		f.mu.Unlock()
 		f.Said(s)
 	}
 }
 
+// statusLocked is s with what the follower holds beside it: its keys, the highest revision heard and why it is behind.
+// The caller holds mu.
+func (f *Follower) statusLocked(s Status) Status {
+	s.Keys = len(f.values)
+	s.Heard, s.HeardCreated, s.Behind = f.heardRev, f.heardCreated, f.behindLocked()
+	return s
+}
+
 // Status is the follower's status now.
 func (f *Follower) Status() Status {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.init()
-	s := f.st
-	s.Keys = len(f.values)
-	return s
+	return f.statusLocked(f.st)
 }
 
 // Ready is closed once a first view was answered.
@@ -163,7 +184,14 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 		f.st.ByHand++
 	}
 	f.mu.Unlock()
-	defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }()
+	released := false
+	defer func() {
+		if !released {
+			f.mu.Lock()
+			f.asking = false
+			f.mu.Unlock()
+		}
+	}()
 	bound, wait := f.Bound, f.Wait
 	if bound == 0 {
 		bound = Bound
@@ -175,6 +203,23 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 	for {
 		err := f.askOnce(ctx)
 		if err == nil {
+			// An event heard that the answer does not hold yet (the owner answered before it wrote it, or the event came
+			// before the first view and was passed over, or it was a gap heard while this ask ran) leaves the picture
+			// behind. No further event is waited for: it is asked again on a goroutine of its own, decided in the same
+			// check that ends this ask, so an event heard after the check finds no ask running and asks itself (ADR 0615
+			// §5). The ask that returns has brought the picture as far as the owner answered; the module need not wait.
+			f.mu.Lock()
+			behind := f.behindLocked()
+			f.asking, released = false, true
+			start := behind != "" && !f.catching
+			if start {
+				f.catching = true
+			}
+			f.mu.Unlock()
+			if start {
+				f.logf("views: %s's %s is behind an event heard (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+				go f.catchUp(ctx)
+			}
 			return nil
 		}
 		f.mu.Lock()
@@ -190,10 +235,7 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 			f.Raise(s)
 		}
 		f.say(s)
-		pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own
-		<-pause.Done()
-		cancel()
-		if ctx.Err() != nil {
+		if !f.pause(ctx, wait) {
 			return ctx.Err()
 		}
 		if wait *= 2; wait > time.Minute {
@@ -202,6 +244,90 @@ func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error
 	}
 }
 
+// catchUp asks the first view once more, each time after a pause (the reader's Wait, doubling to a minute: an owner that
+// answers from before its last write is not asked in a hot loop), until the picture holds every revision heard. It
+// holds no ask while it waits, so a gap, an ask by hand or an event heard meanwhile is not held up by it.
+func (f *Follower) catchUp(ctx context.Context) {
+	wait := f.Wait
+	if wait == 0 {
+		wait = 5 * time.Second
+	}
+	for {
+		if !f.pause(ctx, wait) {
+			f.mu.Lock()
+			f.catching = false
+			f.mu.Unlock()
+			return
+		}
+		f.mu.Lock()
+		behind := f.behindLocked()
+		if behind == "" {
+			f.catching = false
+			f.mu.Unlock()
+			return
+		}
+		f.mu.Unlock()
+		f.logf("views: %s's %s is still behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+		_ = f.askUntil(ctx, "behind: "+behind, false)
+		if wait *= 2; wait > time.Minute {
+			wait = time.Minute
+		}
+	}
+}
+
+// pause waits a context's bound between two asks, no timer of its own; false once ctx ended.
+func (f *Follower) pause(ctx context.Context, wait time.Duration) bool {
+	p, cancel := context.WithTimeout(ctx, wait)
+	<-p.Done()
+	cancel()
+	return ctx.Err() == nil
+}
+
+// behindLocked says why the picture is behind an event heard, or "" when it holds every revision heard. The caller
+// holds mu.
+func (f *Follower) behindLocked() string {
+	switch {
+	case !f.consumerMade.IsZero() && f.consumerMade.After(f.st.Asked) && !f.st.Asked.IsZero():
+		return fmt.Sprintf("the consumer of its events was made at %s, after the first view was asked at %s: what was said between is not delivered", f.consumerMade.Format(time.RFC3339), f.st.Asked.Format(time.RFC3339))
+	case f.heardCreated.IsZero():
+		return ""
+	case f.heardCreated.After(f.st.Created):
+		return fmt.Sprintf("the state was made again at %s, the picture holds the one made at %s", f.heardCreated.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
+	case f.heardCreated.Equal(f.st.Created) && f.heardRev > f.st.Revision:
+		return fmt.Sprintf("revision %d heard, the picture holds %d", f.heardRev, f.st.Revision)
+	}
+	return ""
+}
+
+// ConsumerMade tells the follower when the bus made its consumer of the owner's events. A picture asked before that may
+// lack an event said between its ask and the consumer, which the consumer never delivers and no later event shows: it
+// asks again, by itself (ADR 0443: a consumer is made from now). It does nothing before a first view was asked.
+func (f *Follower) ConsumerMade(ctx context.Context, created time.Time) {
+	f.mu.Lock()
+	f.init()
+	if created.After(f.consumerMade) {
+		f.consumerMade = created
+	}
+	behind := f.behindLocked()
+	start := behind != "" && !f.asking && !f.catching
+	if start {
+		f.catching = true
+	}
+	f.mu.Unlock()
+	if start {
+		f.logf("views: %s's %s is behind (%s): its first view %s is asked once more", f.Owner, f.State, behind, f.Tool)
+		go f.catchUp(ctx)
+	}
+}
+
+// Behind says whether an event heard is not in the picture yet, and why: for a reader's tool and a test.
+func (f *Follower) Behind() string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.init()
+	return f.behindLocked()
+}
+
 // askOnce asks every page and, once whole, takes it as the picture.
 func (f *Follower) askOnce(ctx context.Context) error {
 	f.mu.Lock()
@@ -238,7 +364,7 @@ func (f *Follower) askOnce(ctx context.Context) error {
 	f.during = nil
 	slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) })
 	for _, ev := range during {
-		if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap {
+		if ev.Created.Equal(f.st.Created) && ev.Revision > f.st.Revision && ev.before() <= f.st.Revision && !ev.Gap && !ev.Mark {
 			if ev.Deleted {
 				delete(f.values, ev.Key)
 			} else {
@@ -282,6 +408,12 @@ func (f *Follower) askOnce(ctx context.Context) error {
 func (f *Follower) Apply(ev Event) (gap bool, why string) {
 	f.mu.Lock()
 	f.init()
+	switch {
+	case ev.Created.After(f.heardCreated):
+		f.heardCreated, f.heardRev = ev.Created, ev.Revision
+	case ev.Created.Equal(f.heardCreated) && ev.Revision > f.heardRev:
+		f.heardRev = ev.Revision
+	}
 	if f.asking {
 		f.during = append(f.during, ev)
 	}
@@ -289,16 +421,23 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
 	case <-f.ready:
 	default:
 		f.mu.Unlock()
+		f.logf("views: %s's %s at revision %d was heard before its first view was answered: passed over, the answer holds it or is asked again", f.Owner, f.State, ev.Revision)
 		return false, ""
 	}
 	switch {
 	case !ev.Created.Equal(f.st.Created):
 		why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339))
 	case ev.Revision <= f.st.Revision:
+		held := f.st.Revision
 		f.mu.Unlock()
+		if !ev.Mark {
+			f.logf("views: %s's %s at revision %d is in the picture already (it holds %d): passed over", f.Owner, f.State, ev.Revision, held)
+		}
 		return false, ""
-	case ev.Revision > f.st.Revision+1:
-		why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1)
+	case ev.Mark:
+		why = fmt.Sprintf("%s said its %s up to revision %d, the picture holds %d: a change's event was not heard", f.Owner, f.State, ev.Revision, f.st.Revision)
+	case ev.before() > f.st.Revision:
+		why = fmt.Sprintf("%s's %s at revision %d was said after %d, the picture holds %d: changes missed", f.Owner, f.State, ev.Revision, ev.before(), f.st.Revision)
 	case ev.Gap:
 		why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision)
 	}
@@ -312,7 +451,9 @@ func (f *Follower) Apply(ev Event) (gap bool, why string) {
 		f.values[ev.Key] = ev.Value
 	}
 	f.st.Revision = ev.Revision
+	s := f.statusLocked(f.st)
 	f.mu.Unlock()
+	f.say(s) // the revision the picture holds follows the events, not only the asks (a read tool says where it stands)
 	if f.Changed != nil {
 		f.Changed(ev.Key, ev.Value, ev.Deleted, ev.At)
 	}
@@ -343,7 +484,8 @@ func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte,
 	}
 	f := followers[ev.State]
 	if f == nil {
-		return nil // a state this reader does not follow
+		logf("views: %s's %s is a state this reader does not follow: event at revision %d passed over", ev.Owner, ev.State, ev.Revision)
+		return nil
 	}
 	if gap, why := f.Apply(ev); gap {
 		logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool)

go/views/process_reader.go

@@ -545,6 +545,11 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred
 	if hub == nil {
 		hub, pattern, own = process.NewEventHub(r.logf), ">", true
 	}
+	hub.OnBound(func(created time.Time) {
+		for _, f := range r.Followers() {
+			f.ConsumerMade(ctx, created)
+		}
+	})
 	err := hub.Handle(pattern, r.Hear(ctx))
 	if err == nil && own {
 		err = hub.Start(ctx, js, cred)

go/views/process_reader_integration_test.go

@@ -5,6 +5,7 @@ package views_test
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"testing"
 	"time"
 
@@ -77,6 +78,43 @@ func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) {
 	if got := <-other; got != "mesh.mod.o.event.something-else" {
 		t.Fatalf("other got %s", got)
 	}
+	// Across bursts that write the same keys over and over, the reader's copy stays its owner's, by the events alone: a
+	// revision written over before the owner's watch was handed it is no gap, and the owner's mark after each burst asks
+	// nothing of a reader that holds it (issue 1966).
+	all, err := pic.WatchAll(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var last uint64
+	for round := range 300 {
+		for range 3 {
+			if last, err = kv.Put(ctx, fmt.Sprintf("k%d", round%5), fmt.Appendf(nil, `{"round":%d}`, round)); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	f := r.Followers()[0]
+	wait, stop := context.WithTimeout(ctx, 20*time.Second)
+	defer stop()
+	for f.Status().Revision < last {
+		select {
+		case <-all.Updates():
+		case <-wait.Done():
+			t.Fatalf("the reader stayed behind its owner at %d: %+v", last, f.Status())
+		}
+	}
+	for i := range 5 {
+		e, err := kv.Get(ctx, fmt.Sprintf("k%d", i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if v, _ := f.Get(e.Key()); string(v) != string(e.Value()) {
+			t.Errorf("%s: the reader holds %s, its owner %s", e.Key(), v, e.Value())
+		}
+	}
+	if s := f.Status(); s.Asks != 1 {
+		t.Errorf("the reader asked its first view %d times across the bursts (%s): a revision written over was taken as a gap", s.Asks, s.Why)
+	}
 }
 
 // A module has one durable consumer on the shared events stream, so the loops that hear events in one module must share

go/views/status.go

@@ -99,8 +99,9 @@ func StatusTools(reader, read, again string, st stdio.KeptState, acts HandActs,
 	return []stdio.Tool{
 		{Name: read,
 			Description: fmt.Sprintf("%s's first views of the owners' states it follows (novox/hq ADR 0615): per owner and state, "+
-				"the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+
-				"(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader),
+				"the tool it asks, the revision its picture holds now (events taken in included), the highest revision it heard and why "+
+				"it is behind it, if it is, when it asked and was answered, how many asks (by hand among them) and whether its "+
+				"condition (unanswered past two minutes) is raised.", reader),
 			Input: map[string]any{"type": "object", "properties": map[string]any{}},
 			Run: func(map[string]any) (any, error) {
 				keys, err := st.Keys()

go/views/tools.go

@@ -25,6 +25,19 @@ func (k KeptSource) Get(key string) (*Entry, bool, error) {
 	return &Entry{Value: e.Value, Revision: e.Revision, Written: e.At}, true, nil
 }
 
+// Page is the values after a key in one request (stdio.KeptState.Page), as FirstView prefers them to Get per key.
+func (k KeptSource) Page(after string, maxBytes int, budget time.Duration) ([]KeyedEntry, string, error) {
+	p, err := k.State.Page(after, maxBytes, budget)
+	if err != nil {
+		return nil, "", err
+	}
+	out := make([]KeyedEntry, 0, len(p.Entries))
+	for _, e := range p.Entries {
+		out = append(out, KeyedEntry{Key: e.Key, Entry: Entry{Value: e.Value, Revision: e.Revision, Written: e.At}})
+	}
+	return out, p.Next, nil
+}
+
 // History is every value the state still keeps for key, oldest first (stdio.KeptState.History): as many as the
 // state declares it keeps, deletes included.
 func (k KeptSource) History(key string) ([]Revision, error) {

go/views/views.go

@@ -40,6 +40,14 @@ const MaxValueBytes = 960 * 1024
 // PageBytes bounds the values one answer carries, so the answer fits one bus message; Answer.Next continues it.
 const PageBytes = 768 * 1024
 
+// PageBudget bounds the time one answer spends reading values, so a slow source (kept state reads one value per bus
+// request) never makes a page outlast its reader's patience (30 s): the page stops when it is spent, always after at
+// least one value, and Answer.Next continues it. A variable so a test sets it.
+var PageBudget = 10 * time.Second
+
+// clock is the page budget's time, replaced by a test.
+var clock = time.Now
+
 // Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb
 // (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody.
 const Bound = 2 * time.Minute
@@ -88,20 +96,41 @@ type Event struct {
 	At time.Time `json:"at,omitzero"`
 	// Gap says the value was too large to carry (MaxValueBytes): the reader asks its first view once.
 	Gap bool `json:"gap,omitempty"`
+	// Prev is the revision of the change the owner said before this one, or of its mark when this is the first said
+	// since its start. A revision between the two was written over by a later change of the same key before the bus
+	// handed it to the owner's watch (a state keeps one value per key), so nothing of it is missed: a reader takes the
+	// event when its picture holds Prev or more. Zero says nothing, and a reader takes Revision-1 as the one before.
+	Prev uint64 `json:"prev,omitempty"`
+	// Mark says the event is the owner's mark (MarksState), said each time the mark is written: no key and no value,
+	// Revision the change the owner said its state up to. A reader whose picture is short of it asks its first view, so a
+	// change whose event never reached it is caught up without waiting for a later change.
+	Mark bool `json:"mark,omitempty"`
+}
+
+// before is the revision of the change said before e: Prev, or Revision-1 when the owner said none.
+func (e Event) before() uint64 {
+	if e.Prev != 0 {
+		return e.Prev
+	}
+	return e.Revision - 1
 }
 
 // ID is the event's id on the bus, so the same change said twice (an owner's restart) is kept once.
 func (e Event) ID() string {
-	return fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision)
+	id := fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision)
+	if e.Mark {
+		id += "-mark"
+	}
+	return id
 }
 
 // Encode writes an event.
 func (e Event) Encode() ([]byte, error) {
 	e.V = Version
-	if e.Owner == "" || e.State == "" || e.Key == "" || e.Revision == 0 {
-		return nil, errors.New("views: an event names its owner, state, key and revision")
+	if e.Owner == "" || e.State == "" || (e.Key == "" && !e.Mark) || e.Revision == 0 {
+		return nil, errors.New("views: an event names its owner, state, key (or is a mark) and revision")
 	}
-	if e.Deleted || e.Gap {
+	if e.Deleted || e.Gap || e.Mark {
 		e.Value = nil
 	}
 	return json.Marshal(e)
@@ -113,8 +142,8 @@ func DecodeEvent(data []byte) (Event, error) {
 	if err := decode(data, &e); err != nil {
 		return Event{}, err
 	}
-	if e.State == "" || e.Key == "" || e.Revision == 0 {
-		return Event{}, errors.New("views: an event names no state, key or revision")
+	if e.State == "" || (e.Key == "" && !e.Mark) || e.Revision == 0 {
+		return Event{}, errors.New("views: an event names no state, key (nor is a mark) or revision")
 	}
 	return e, nil
 }
@@ -186,8 +215,21 @@ type Source interface {
 	Get(key string) (*Entry, bool, error)
 }
 
+// KeyedEntry is an Entry with its key.
+type KeyedEntry struct {
+	Key string
+	Entry
+}
+
+// Pager is a Source that answers many values in one request: the values in key order after a key, as many as fit
+// maxBytes or the time budget and at least one, and the key to ask after for the rest. FirstView uses it when its source has it, unless
+// it answers errors.Is(err, errors.ErrUnsupported), when it reads per key.
+type Pager interface {
+	Page(after string, maxBytes int, budget time.Duration) ([]KeyedEntry, string, error)
+}
+
 // FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it
-// answers at is missed, then the values in key order after after, until PageBytes. A state not said yet (no mark)
+// answers at is missed, then the values in key order after after, until PageBytes or PageBudget. A state not said yet (no mark)
 // answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event.
 func FirstView(owner, state string, values, marks Source, after string, now time.Time) (Answer, error) {
 	a := Answer{V: Version, Owner: owner, State: state, Values: map[string]json.RawMessage{}, At: now.UTC()}
@@ -204,16 +246,34 @@ func FirstView(owner, state string, values, marks Source, after string, now time
 			a.Revision, a.Created = m.Revision, m.Created
 		}
 	}
+	if p, ok := values.(Pager); ok {
+		got, next, err := p.Page(after, PageBytes, PageBudget)
+		if err == nil {
+			for _, e := range got {
+				a.put(e.Key, e.Entry)
+			}
+			a.Next = next
+			return a, nil
+		}
+		if !errors.Is(err, errors.ErrUnsupported) {
+			return Answer{}, fmt.Errorf("the state %s cannot be read: %w", state, err)
+		}
+	}
 	keys, err := values.Keys()
 	if err != nil {
 		return Answer{}, fmt.Errorf("the state %s cannot be listed: %w", state, err)
 	}
 	slices.Sort(keys)
 	size := 0
+	started := clock()
 	for _, k := range keys {
 		if after != "" && k <= after {
 			continue
 		}
+		if size > 0 && clock().Sub(started) >= PageBudget {
+			a.Next = lastKey(a.Values)
+			break
+		}
 		e, ok, err := values.Get(k)
 		if err != nil {
 			return Answer{}, fmt.Errorf("%s.%s cannot be read: %w", state, k, err)
@@ -225,18 +285,22 @@ func FirstView(owner, state string, values, marks Source, after string, now time
 			a.Next = lastKey(a.Values)
 			break
 		}
-		a.Values[k] = e.Value
-		if !e.Written.IsZero() {
-			if a.Written == nil {
-				a.Written = map[string]time.Time{}
-			}
-			a.Written[k] = e.Written.UTC()
-		}
+		a.put(k, *e)
 		size += len(k) + len(e.Value)
 	}
 	return a, nil
 }
 
+func (a *Answer) put(k string, e Entry) {
+	a.Values[k] = e.Value
+	if !e.Written.IsZero() {
+		if a.Written == nil {
+			a.Written = map[string]time.Time{}
+		}
+		a.Written[k] = e.Written.UTC()
+	}
+}
+
 func lastKey(m map[string]json.RawMessage) string {
 	last := ""
 	for k := range m {

go/views/views_test.go

@@ -5,7 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -93,6 +96,43 @@ func TestViews(t *testing.T) {
 			t.Errorf("pages hold %v, want the four keys", got)
 		}
 	})
+	t.Run("a first view from a slow source stops at the page budget, with one value at least, and pages on to every key", func(t *testing.T) {
+		at := time.Unix(0, 0)
+		clock = func() time.Time { return at }
+		defer func() { clock = time.Now }()
+		slow := mapSource{}
+		for i := range 5 {
+			slow[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`{"n":1}`), Revision: uint64(i + 1)}
+		}
+		got, after := map[string]bool{}, ""
+		for pages := 0; ; pages++ {
+			a, err := FirstView("o", "s", &tickSource{slow, &at, PageBudget / 2}, nil, after, now())
+			if err != nil || pages > 5 || len(a.Values) == 0 || len(a.Values) > 3 {
+				t.Fatalf("page %d: %d values, %v", pages, len(a.Values), err)
+			}
+			for k := range a.Values {
+				got[k] = true
+			}
+			if a.Next == "" {
+				break
+			}
+			after = a.Next
+		}
+		if len(got) != 5 {
+			t.Errorf("pages hold %v, want the five keys", got)
+		}
+	})
+	t.Run("a first view from a source that pages reads it in one request, and from one that cannot, per key", func(t *testing.T) {
+		p := &pagedSource{mapSource: mapSource{"a": {Value: json.RawMessage(`1`), Revision: 1, Written: written}, "b": {Value: json.RawMessage(`2`), Revision: 2}, "c": {Value: json.RawMessage(`3`), Revision: 3}}}
+		a, err := FirstView("o", "s", p, nil, "", now())
+		if err != nil || len(a.Values) != 2 || a.Next != "b" || p.pages != 1 || p.gets != 0 || !a.Written["a"].Equal(written) {
+			t.Fatalf("answer = %+v, pages %d, gets %d, %v", a, p.pages, p.gets, err)
+		}
+		p.unsupported = true
+		if a, err = FirstView("o", "s", p, nil, "", now()); err != nil || len(a.Values) != 3 || a.Next != "" || p.gets != 3 {
+			t.Fatalf("fallback answer = %+v, gets %d, %v", a, p.gets, err)
+		}
+	})
 	t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) {
 		callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner")
 		for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true,
@@ -193,6 +233,56 @@ func TestViews(t *testing.T) {
 			}
 		}
 	})
+	t.Run("a change said after revisions written over is no gap; one said after a revision the picture lacks, or a mark past the picture, asks once more and changes nothing", func(t *testing.T) {
+		f, _ := follower(answer)
+		_ = f.Start(context.Background())
+		over := ev(12, "a", `{"n":5}`)
+		over.Prev = 9 // 10 and 11 were written over before the owner's watch was handed them
+		if gap, why := f.Apply(over); gap || f.Status().Revision != 12 {
+			t.Fatalf("a change said after 9 on a picture at 9: gap %v (%s), at %d", gap, why, f.Status().Revision)
+		}
+		missed := ev(14, "b", `{"n":6}`)
+		missed.Prev = 13
+		mark := func(rev uint64) Event {
+			return Event{V: Version, Owner: "o", State: "modules", Revision: rev, Created: made, Mark: true}
+		}
+		for _, c := range []struct {
+			e   Event
+			gap bool
+		}{{missed, true}, {mark(12), false}, {mark(11), false}, {mark(13), true}} {
+			if gap, why := f.Apply(c.e); gap != c.gap {
+				t.Errorf("revision %d (mark %v) on a picture at 12: gap %v, want %v (%s)", c.e.Revision, c.e.Mark, gap, c.gap, why)
+			}
+		}
+		if v, _ := f.Get("a"); string(v) != `{"n":5}` || f.Status().Revision != 12 || len(f.Values()) != 2 {
+			t.Errorf("a gap or a mark changed the picture: %v at %d", f.Values(), f.Status().Revision)
+		}
+	})
+	t.Run("a reader that never heard the owner's last change asks its first view on the owner's mark alone, and holds what it read", func(t *testing.T) {
+		later := mapSource{"a": values["a"], "b": values["b"], "c": {Value: json.RawMessage(`{"n":3}`), Revision: 10}}
+		asks := 0
+		f, _ := follower(func(ctx context.Context, after string) (Answer, error) {
+			asks++
+			if asks == 1 {
+				return answer(ctx, after)
+			}
+			return FirstView("o", "modules", later, mapSource{"modules": markOf("modules", 10, made)}, after, now())
+		})
+		_ = f.Start(context.Background())
+		got := make(chan string, 4)
+		f.Changed = func(key string, _ json.RawMessage, _ bool, _ time.Time) { got <- key }
+		// The change at 10 (c) never reached this reader; only the mark the owner said after it does.
+		data, _ := Event{V: Version, Owner: "o", State: "modules", Revision: 10, Created: made, Mark: true}.Encode()
+		if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, data, t.Logf); err != nil {
+			t.Fatal(err)
+		}
+		if k := <-got; k != "c" {
+			t.Fatalf("changed %s", k)
+		}
+		if v, _ := f.Get("c"); string(v) != `{"n":3}` || asks != 2 {
+			t.Errorf("c = %s after %d asks", v, asks)
+		}
+	})
 	t.Run("a reader whose owner is down keeps asking, raises past its bound, and the answer clears it", func(t *testing.T) {
 		down := 3
 		f, said := follower(func(ctx context.Context, after string) (Answer, error) {
@@ -224,6 +314,101 @@ func TestViews(t *testing.T) {
 			t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision)
 		}
 	})
+	t.Run("a reader that heard an event its answer does not hold asks again by itself on its Wait, with no further event", func(t *testing.T) {
+		// The owner's state: revision 9 at first, 12 once its writer is done. The picture a first view reads lags the events.
+		for name, drop := range map[string]func(f *Follower, owner *uint64){
+			// A gap heard while the ask runs: the running ask swallowed the gap's own ask and the event was not consecutive.
+			"a gap heard while an ask runs": func(f *Follower, owner *uint64) { *owner = 12; f.Apply(ev(12, "e", `{"n":5}`)) },
+			// An event heard before the first view, passed over, the owner answering from before its write.
+			"an event heard before the first view": func(f *Follower, owner *uint64) { f.Apply(ev(10, "e", `{"n":5}`)) },
+		} {
+			t.Run(name, func(t *testing.T) {
+				owner := uint64(9)
+				var f *Follower
+				asks := 0
+				f, _ = follower(func(context.Context, string) (Answer, error) {
+					asks++
+					a := Answer{V: Version, Owner: "o", State: "modules", Revision: owner, Created: made, Values: map[string]json.RawMessage{"e": json.RawMessage(`{"n":5}`)}}
+					if asks == 1 && name == "a gap heard while an ask runs" {
+						drop(f, &owner) // heard while this ask runs; the answer is the one read before it
+					}
+					if asks == 2 {
+						owner = 12 // the owner has written it by the second ask
+						a.Revision = owner
+					}
+					return a, nil
+				})
+				if name == "an event heard before the first view" {
+					drop(f, &owner)
+				}
+				caught := make(chan struct{})
+				var once sync.Once
+				f.Said = func(s Status) {
+					if s.Revision >= 10 {
+						once.Do(func() { close(caught) })
+					}
+				}
+				if err := f.Start(context.Background()); err != nil {
+					t.Fatal(err)
+				}
+				// No further event follows: the follower asks once more by itself and holds what was heard.
+				select {
+				case <-caught:
+				case <-time.After(10 * time.Second):
+					t.Fatal("the reader stayed behind the event it heard")
+				}
+				if s := f.Status(); s.Revision < 10 || s.Heard < 10 || s.Behind != "" || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind") {
+					t.Errorf("revision %d after %d asks (%q), behind: %q", s.Revision, s.Asks, s.Why, f.Behind())
+				}
+			})
+		}
+	})
+	t.Run("a consumer made after the first view was asked asks it again by itself, and the status says why until then", func(t *testing.T) {
+		asked := make(chan struct{}, 8)
+		var f *Follower
+		f, _ = follower(func(ctx context.Context, after string) (Answer, error) {
+			asked <- struct{}{}
+			return answer(ctx, after)
+		})
+		var said []Status
+		var mu sync.Mutex
+		var clock atomic.Int64
+		clock.Store(int64(time.Hour))
+		f.Now = func() time.Time { return made.Add(time.Duration(clock.Load())) }
+		f.Said = func(s Status) { mu.Lock(); said = append(said, s); mu.Unlock() }
+		_ = f.Start(context.Background())
+		<-asked
+		// The bus made the consumer after the ask: an event said between is never delivered to it.
+		f.ConsumerMade(context.Background(), made.Add(2*time.Hour))
+		if b := f.Status().Behind; !strings.Contains(b, "consumer") {
+			t.Errorf("behind = %q", b)
+		}
+		clock.Store(int64(3 * time.Hour)) // a later ask is after the consumer's making
+		select {
+		case <-asked:
+		case <-time.After(10 * time.Second):
+			t.Fatal("a consumer made after the first view asked nothing")
+		}
+	})
+	t.Run("an event heard but not in the picture is in the status as heard, with why it is behind", func(t *testing.T) {
+		f, _ := follower(answer)
+		_ = f.Start(context.Background())
+		f.Apply(ev(12, "a", `{}`)) // a gap: heard, not taken
+		if s := f.Status(); s.Heard != 12 || !strings.Contains(s.Behind, "12") {
+			t.Errorf("status = %+v", s)
+		}
+	})
+	t.Run("an event taken into the picture is said in the status, so a read tool shows the revision held", func(t *testing.T) {
+		var told []uint64
+		f, _ := follower(answer)
+		f.Said = func(s Status) { told = append(told, s.Revision) }
+		_ = f.Start(context.Background())
+		told = nil
+		f.Apply(ev(10, "c", `{"n":3}`))
+		if len(told) != 1 || told[0] != 10 {
+			t.Errorf("said revisions %v, want [10]", told)
+		}
+	})
 	t.Run("an event heard by its data asks a gap once more as a gap, never by hand", func(t *testing.T) {
 		asked := make(chan struct{}, 4)
 		f, _ := follower(func(ctx context.Context, after string) (Answer, error) {
@@ -406,7 +591,7 @@ func TestViews(t *testing.T) {
 		updates <- kvEntry{key: "b", rev: 2, at: made}
 		close(updates)
 		errs := make(chan error, 1)
-		em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, errs)
+		em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs)
 		if len(said) != 2 || !strings.HasSuffix(said[0], "-1") || !strings.HasSuffix(said[1], "-2") || fails != 0 {
 			t.Fatalf("said %v (fails left %d): the change at 1 must be said, again, before the one at 2", said, fails)
 		}
@@ -417,7 +602,7 @@ func TestViews(t *testing.T) {
 		updates = make(chan jetstream.KeyValueEntry, 2)
 		updates <- kvEntry{key: "a", rev: 1, at: made}
 		updates <- kvEntry{key: "b", rev: 2, at: made}
-		em.follow(ctx, "s", made, updates, func() (Fill, error) { return Fill{}, nil }, errs)
+		em.follow(ctx, "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs)
 		if len(said) != 0 {
 			t.Fatalf("said %v after the context ended on a change not said", said)
 		}
@@ -434,7 +619,7 @@ func TestViews(t *testing.T) {
 			updates = make(chan jetstream.KeyValueEntry, 1)
 			updates <- kvEntry{key: "a", rev: 1, at: made}
 			close(updates)
-			em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, nil)
+			em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil)
 			if fs.puts != c.puts || fs.deletes != 1 || len(fs.m) != 0 || len(said) != 1 {
 				t.Errorf("%d failures: %d findings written, %d cleared, %d left, said %v", c.fails, fs.puts, fs.deletes, len(fs.m), said)
 			}
@@ -454,11 +639,33 @@ func TestViews(t *testing.T) {
 		updates = make(chan jetstream.KeyValueEntry, 1)
 		updates <- kvEntry{key: "b", rev: 2, at: made}
 		close(updates)
-		em.follow(context.Background(), "s", made, updates, func() (Fill, error) { return Fill{}, nil }, nil)
+		em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil)
 		if _, left := fs.m[key]; left || len(said) != 1 {
 			t.Errorf("after a restart the leftover finding stands (%v), said %v", left, said)
 		}
 	})
+	t.Run("each change names the revision said before it, and the mark is said as an event once written", func(t *testing.T) {
+		var said []Event
+		marks := &marksKV{}
+		em := Emitter{Owner: "o", Subject: func(e string) string { return "s." + e }, Marks: marks,
+			Publish: func(_ context.Context, msg *nats.Msg, id string) error {
+				e, err := DecodeEvent(msg.Data)
+				if err != nil || e.ID() != id {
+					t.Errorf("said %s as %s: %v", msg.Data, id, err)
+				}
+				said = append(said, e)
+				return nil
+			}}
+		// 4 to 6 were written over before the watch was handed them: the watch hands 3, then 7.
+		updates := make(chan jetstream.KeyValueEntry, 2)
+		updates <- kvEntry{key: "a", rev: 3, at: made}
+		updates <- kvEntry{key: "a", rev: 7, at: made}
+		close(updates)
+		em.follow(context.Background(), "s", made, 2, updates, func() (Fill, error) { return Fill{}, nil }, nil)
+		if len(said) != 3 || said[0].Prev != 2 || said[1].Prev != 3 || !said[2].Mark || said[2].Revision != 7 || marks.puts != 1 {
+			t.Fatalf("said %+v with %d marks written: 3 after 2, 7 after 3, then the mark at 7", said, marks.puts)
+		}
+	})
 	t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) {
 		if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" {
 			t.Errorf("state of subject = %q", s)
@@ -495,6 +702,17 @@ func (f *findingsMap) Put(v Finding) error {
 	return nil
 }
 
+// marksKV is an owner's marks state that takes puts and nothing else.
+type marksKV struct {
+	jetstream.KeyValue
+	puts int
+}
+
+func (m *marksKV) Put(context.Context, string, []byte) (uint64, error) {
+	m.puts++
+	return uint64(m.puts), nil
+}
+
 func (f *findingsMap) Delete(v Finding) error { delete(f.m, v.Kind+"/"+v.ID); f.deletes++; return nil }
 
 // actsFunc records a hand act through a test's function.
@@ -516,3 +734,46 @@ func (k kvEntry) Revision() uint64                { return k.rev }
 func (k kvEntry) Created() time.Time              { return k.at }
 func (k kvEntry) Delta() uint64                   { return 0 }
 func (k kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
+
+// tickSource is a source whose every Get takes a step of the fake clock.
+type tickSource struct {
+	mapSource
+	at   *time.Time
+	step time.Duration
+}
+
+func (s *tickSource) Get(k string) (*Entry, bool, error) {
+	*s.at = s.at.Add(s.step)
+	return s.mapSource.Get(k)
+}
+
+// pagedSource is a source with a bulk read, two values a page.
+type pagedSource struct {
+	mapSource
+	pages, gets int
+	unsupported bool
+}
+
+func (s *pagedSource) Get(k string) (*Entry, bool, error) {
+	s.gets++
+	return s.mapSource.Get(k)
+}
+
+func (s *pagedSource) Page(after string, _ int, _ time.Duration) ([]KeyedEntry, string, error) {
+	if s.unsupported {
+		return nil, "", errors.ErrUnsupported
+	}
+	s.pages++
+	keys, _ := s.Keys()
+	slices.Sort(keys)
+	var out []KeyedEntry
+	for _, k := range keys {
+		if k > after && len(out) < 2 {
+			out = append(out, KeyedEntry{Key: k, Entry: s.mapSource[k]})
+		}
+	}
+	if len(out) == 2 && out[1].Key != keys[len(keys)-1] {
+		return out, out[1].Key, nil
+	}
+	return out, "", nil
+}