The bus module takes and restores a restore point of a module's kept data, beside the live data (milestone G, W14.2)

Seat verbs snapshot-kept, restore-kept, discard-aside and aside, handed to the keeper; state kept; tool nats_kept.

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

modules/nats/README.md

@@ -136,6 +136,18 @@ a put of the assigner (novox/hq ADR 0546, ADR 0565 clauses 8 and 9). In this ord
   tested) and the manifest's shape (programs mounted read-only, named in `restart-on`, no image built from
   a Dockerfile).
 
+## A module's kept data, backed up and restored with its delivery (milestone G)
+
+The keeper takes, on its own account, a restore point of one module's kept data on the bus before the module moves across a
+change of that kept data's `form`, and restores it beside the live data when a want takes the module back (design 47, *A
+module's kept data on the bus*). Its seat verbs, served by the tools and handed to the keeper as requests:
+`mesh-broker.snapshot-kept` (the data as the server's snapshot, the module's consumers' positions and the catch-up tool, in
+one directory under the keeper's state), `mesh-broker.restore-kept` (the live data is snapshotted into a set-aside
+directory under a name of its own and only then removed; the restore point is put in its place and the consumers on other
+streams are set back), `mesh-broker.discard-aside` (only on a call that says why, recorded first) and `mesh-broker.aside`;
+the tool `nats_kept` reads the restore points and the set-aside data from the keeper's state `kept`. The backup holder
+keeps the restore points; it calls these verbs through the seat.
+
 ## Files it does not know
 
 Each directory or archive this module declares says how it treats a file it did not write (ADR 0422, ADR 0416).

modules/nats/cmd/nats-tools/keep.go

@@ -236,8 +236,10 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer
 	defer nc.Close()
 	recorder := meshhandactsclient.Recorder{Ask: conn.Call}
 	k := &keeper.Keeper{
-		Bus:   keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
-		Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
+		Bus:     keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))},
+		Kept:    keeper.KeptJetStream{NC: nc, JS: js, Logf: logger.Printf},
+		KeptDir: getenv("MESH_NATS_SNAPSHOTS_DIR"),
+		Store:   processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn),
 		Act: func(what, why, cause string) error {
 			_, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
 			return err

modules/nats/cmd/nats-tools/kept_integration_test.go

@@ -0,0 +1,153 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"nats-tools/internal/bussnap"
+	"nats-tools/internal/keeper"
+)
+
+// mapStore is the keeper's own state in memory.
+type mapStore struct {
+	mu sync.Mutex
+	kv map[string][]byte
+}
+
+func (s *mapStore) Get(state, key string) ([]byte, uint64, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, ok := s.kv[state+"/"+key]
+	return v, 1, ok, nil
+}
+func (s *mapStore) Put(state, key string, v []byte) (uint64, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kv[state+"/"+key] = v
+	return 1, nil
+}
+func (s *mapStore) Delete(state, key string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.kv, state+"/"+key)
+	return nil
+}
+func (s *mapStore) Keys(string) ([]string, error) { return nil, nil }
+func (s *mapStore) Create(state, key string, v []byte) (uint64, error) {
+	return 0, errors.New("not used")
+}
+
+// A module's kept data on the real bus is backed up and restored beside the live data (design 47, "A module's kept data on
+// the bus, backed up and restored with its delivery"): the bucket as the server's own snapshot, the module's consumer on
+// another stream at its position; after a change, a restore brings the bucket and the position back, and the live bucket
+// is kept whole in the data set aside.
+func TestAKeptDataIsRestoredBesideTheLiveDataOnTheRealBus(t *testing.T) {
+	s, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(5 * time.Second) {
+		t.Fatal("the server did not start")
+	}
+	defer s.Shutdown()
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "idx_modules"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	forge, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "forge", Subjects: []string{"forge.>"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 10; i++ {
+		if _, err := js.Publish(ctx, "forge.merged", []byte("m")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cons, err := forge.CreateConsumer(ctx, jetstream.ConsumerConfig{Durable: "ace_idx", AckPolicy: jetstream.AckExplicitPolicy})
+	if err != nil {
+		t.Fatal(err)
+	}
+	batch, err := cons.Fetch(4)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for m := range batch.Messages() {
+		_ = m.DoubleAck(ctx)
+	}
+	if _, err := kv.PutString(ctx, "ace", "fingerprint-1"); err != nil {
+		t.Fatal(err)
+	}
+
+	dir := t.TempDir()
+	var acts []string
+	k := &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, dir)}, Store: &mapStore{kv: map[string][]byte{}},
+		Kept: keeper.KeptJetStream{NC: nc, JS: js}, KeptDir: dir,
+		Act: func(what, why, cause string) error { acts = append(acts, what); return nil }}
+	take := &keeper.Request{Verb: keeper.VerbSnapshotKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", CatchUp: "indexer_take_in", Why: "before form 1"}
+	if _, err := k.SnapshotKept(ctx, take); err != nil {
+		t.Fatal(err)
+	}
+	// The form-1 build rewrites the kept data, and its consumer moves on.
+	if _, err := kv.PutString(ctx, "ace", "fingerprint-2-form-1"); err != nil {
+		t.Fatal(err)
+	}
+	batch, err = cons.Fetch(5)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for m := range batch.Messages() {
+		_ = m.DoubleAck(ctx)
+	}
+
+	restore := &keeper.Request{Verb: keeper.VerbRestoreKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", From: take.Dir, Why: "revert"}
+	if _, err := k.RestoreKept(ctx, restore); err != nil {
+		t.Fatal(err)
+	}
+	kv, err = js.KeyValue(ctx, "idx_modules")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if e, err := kv.Get(ctx, "ace"); err != nil || string(e.Value()) != "fingerprint-1" {
+		t.Errorf("the restored bucket holds %v (%v), want fingerprint-1", e, err)
+	}
+	if restore.CatchUp != "indexer_take_in" {
+		t.Errorf("catch-up %q", restore.CatchUp)
+	}
+	forge, _ = js.Stream(ctx, "forge")
+	ci, err := forge.Consumer(ctx, "ace_idx")
+	if err != nil || ci.CachedInfo().Config.OptStartSeq != 5 {
+		t.Errorf("the consumer is %+v (%v): it starts after the 4 messages it had acknowledged", ci.CachedInfo().Config, err)
+	}
+	kept, err := os.ReadDir(filepath.Join(dir, "aside"))
+	if err != nil || len(kept) != 1 {
+		t.Fatalf("aside holds %v (%v)", kept, err)
+	}
+	if _, err := os.Stat(filepath.Join(dir, "aside", kept[0].Name(), keeper.ArchiveFile)); err != nil {
+		t.Error("the live data set aside is not kept whole")
+	}
+	if len(acts) != 2 {
+		t.Errorf("hand acts %v: the restore point and the restore", acts)
+	}
+}

modules/nats/cmd/nats-tools/keptools.go

@@ -0,0 +1,171 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+
+	"nats-tools/internal/keeper"
+)
+
+// keptWait is how long a call to take or restore a kept data waits for the keeper's end on the watch of its requests: under
+// the runtime's own bound on a call, so a long snapshot is answered as "handed, not ended" and read again with its request.
+const keptWait = 50 * time.Second
+
+// KeptHeld reads the keeper's state kept: the restore points it took and the data it set aside.
+type KeptHeld func() ([]keeper.Point, []keeper.Aside, error)
+
+// KeptInState is the state kept through the runtime that launched this bundle.
+func KeptInState() ([]keeper.Point, []keeper.Aside, error) {
+	st := stdio.State(keeper.StateKept)
+	keys, err := st.Keys()
+	if err != nil {
+		return nil, nil, err
+	}
+	sort.Strings(keys)
+	var points []keeper.Point
+	var aside []keeper.Aside
+	for _, key := range keys {
+		e, err := st.Get(key)
+		if err != nil || e == nil {
+			continue
+		}
+		switch {
+		case strings.HasPrefix(key, "point."):
+			var p keeper.Point
+			if json.Unmarshal(e.Value, &p) == nil {
+				points = append(points, p)
+			}
+		case strings.HasPrefix(key, "aside."):
+			var a keeper.Aside
+			if json.Unmarshal(e.Value, &a) == nil {
+				aside = append(aside, a)
+			}
+		}
+	}
+	return points, aside, nil
+}
+
+func formOf(a map[string]any) int {
+	f, _ := a["form"].(float64)
+	return int(f)
+}
+
+// keptTools are the seat mesh-broker's verbs for a module's kept data and the bus module's tool that reads what it holds:
+// each verb hands its request to the keeper, which acts on its own account, and waits for the end on the watch of its
+// requests (never a poll). Asked again with `request`, a verb reads the end of the request it made.
+func keptTools(rs requestStore, held KeptHeld) []stdio.Tool {
+	entry := func(extra map[string]any) map[string]any {
+		in := map[string]any{"module": str("the module whose kept data it is, e.g. mesh-delivery-indexer"), "local-name": str("the kept data's local name, e.g. modules"),
+			"form": num("the kept data's form (a manifest field; absent reads 0)"), "fingerprint": str("the fingerprint the restore point was taken before"),
+			"why": str("why, in plain words: kept with the act"), "request": str("a request this verb made before: its end is read, nothing new is asked")}
+		for k, v := range extra {
+			in[k] = v
+		}
+		return in
+	}
+	run := func(verb string, build func(a map[string]any, r *keeper.Request) error) func(a map[string]any) (any, error) {
+		return func(a map[string]any) (any, error) {
+			name := "mesh-broker." + verb
+			if id := arg(a, "request"); id != "" {
+				r, err := rs.Get(id)
+				if err != nil || r == nil {
+					return nil, fmt.Errorf("no request %s is held (%v)", id, err)
+				}
+				if r.Done.IsZero() && rs.Ended != nil {
+					if got := rs.Ended(id, keptWait); got != nil {
+						return got, nil
+					}
+					if again, err := rs.Get(id); err == nil && again != nil {
+						r = again
+					}
+				}
+				return r, nil
+			}
+			r := keeper.Request{Verb: verb, Why: strings.TrimSpace(arg(a, "why")), Caller: stdio.CallerOf(a), At: rs.Now()}
+			if err := build(a, &r); err != nil {
+				return nil, err
+			}
+			if r.Why == "" {
+				return nil, fmt.Errorf("%s says why", name)
+			}
+			r.ID = requestID(r.At)
+			if err := rs.Put(r.ID, r); err != nil {
+				return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err)
+			}
+			if rs.Ended != nil {
+				if got := rs.Ended(r.ID, keptWait); got != nil {
+					return got, nil
+				}
+			}
+			return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", name, r.ID)}, nil
+		}
+	}
+	point := func(a map[string]any, r *keeper.Request) error {
+		r.Module, r.Local, r.Form, r.Fingerprint = arg(a, "module"), arg(a, "local-name"), formOf(a), arg(a, "fingerprint")
+		if r.Module == "" || r.Local == "" || r.Fingerprint == "" {
+			return fmt.Errorf("a restore point is of one kept data: a module, its local name and the fingerprint it is taken before")
+		}
+		return nil
+	}
+	readAside := func() (any, error) {
+		_, aside, err := held()
+		if err != nil {
+			return nil, err
+		}
+		if aside == nil {
+			aside = []keeper.Aside{}
+		}
+		return map[string]any{"aside": aside}, nil
+	}
+	return []stdio.Tool{
+		{Name: "mesh-broker.snapshot-kept",
+			Description: "Take the restore point of one module's kept data on the bus (milestone G, design 47): the data whole as the server's own snapshot, the position of each durable consumer the module reads with, and the catch-up tool the newer manifest declares, kept by the bus module's keeper. Answers the directory the restore point is in. Handed to the keeper, recorded as a hand act.",
+			Input:       entry(map[string]any{"catch-up": str("the tool a restore is followed by, from the newer manifest, kept in the restore point")}),
+			Run: run(keeper.VerbSnapshotKept, func(a map[string]any, r *keeper.Request) error {
+				r.CatchUp = arg(a, "catch-up")
+				return point(a, r)
+			})},
+		{Name: "mesh-broker.restore-kept",
+			Description: "Restore one module's kept data on the bus from a restore point beside the live data, never over it (design 47): the live data is snapshotted into a set-aside directory under a name of its own and then removed from the bus, the restore point's streams are put in its place and the module's other consumers are set back to their positions. Answers what was set aside and the catch-up tool the restore point carries. Handed to the keeper, recorded as a hand act.",
+			Input:       entry(map[string]any{"from": str("the directory the restore point was restored to, an absolute path")}),
+			Run: run(keeper.VerbRestoreKept, func(a map[string]any, r *keeper.Request) error {
+				r.From = arg(a, "from")
+				if r.From == "" {
+					return fmt.Errorf("a restore names the directory the restore point is in")
+				}
+				return point(a, r)
+			})},
+		{Name: "mesh-broker.discard-aside",
+			Description: "Remove the data a restore set aside, by its id (from `aside`), only on a person's call: it says who and why, and the keeper records the hand act before it removes anything.",
+			Input:       map[string]any{"id": str("the id of the data set aside"), "why": str("why, in plain words: kept with the act"), "request": str("a request made before: its end is read")},
+			Run: run(keeper.VerbDiscardAside, func(a map[string]any, r *keeper.Request) error {
+				r.From = arg(a, "id")
+				if r.From == "" {
+					return fmt.Errorf("discard-aside names the id of the data set aside")
+				}
+				return nil
+			})},
+		{Name: "mesh-broker.aside", Description: "The data restores set aside, each with its module, kept data, size, when, who and why: what a person decides to discard.",
+			Run: func(map[string]any) (any, error) { return readAside() }},
+		{Name: "nats_kept",
+			Description: "What the bus module holds of its modules' kept data besides the live data: every restore point it took (module, kept data, form, the fingerprint it was taken before, its consumers' positions, the catch-up tool, its size) and the data a restore set aside. Read from the keeper's state kept.",
+			Run: func(map[string]any) (any, error) {
+				points, aside, err := held()
+				if err != nil {
+					return nil, err
+				}
+				if points == nil {
+					points = []keeper.Point{}
+				}
+				if aside == nil {
+					aside = []keeper.Aside{}
+				}
+				return map[string]any{"points": points, "aside": aside}, nil
+			}},
+	}
+}

modules/nats/cmd/nats-tools/keptools_test.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+
+	"nats-tools/internal/keeper"
+)
+
+// The seat mesh-broker's verbs for a module's kept data hand their request to the keeper as the proved caller, wait for
+// its end on the watch of the requests, and say what a restore set aside (design 47, "A module's kept data on the bus").
+func TestTheKeptDataVerbsAreHandedToTheKeeperAndReadBack(t *testing.T) {
+	held := map[string]keeper.Request{}
+	rs := requestStore{
+		Put: func(key string, r keeper.Request) error { held[key] = r; return nil },
+		Get: func(key string) (*keeper.Request, error) {
+			r, ok := held[key]
+			if !ok {
+				return nil, nil
+			}
+			return &r, nil
+		},
+		Now: func() time.Time { return time.Date(2026, 11, 9, 20, 0, 0, 0, time.UTC) },
+		// The keeper's end, as its watch hands it.
+		Ended: func(id string, within time.Duration) *keeper.Request {
+			r := held[id]
+			r.Done, r.Outcome, r.Dir = time.Unix(1, 0), "done", "/keeper/points/idx/modules/f0-abc"
+			held[id] = r
+			return &r
+		},
+	}
+	aside := []keeper.Aside{{ID: "idx.modules.1", Module: "idx", Local: "modules", Bytes: 12}}
+	tools := map[string]stdio.Tool{}
+	for _, tl := range keptTools(rs, func() ([]keeper.Point, []keeper.Aside, error) { return nil, aside, nil }) {
+		tools[tl.Name] = tl
+	}
+	for _, verb := range KeptVerbs {
+		if _, ok := tools["mesh-broker."+verb]; !ok {
+			t.Errorf("the verb mesh-broker.%s is not served", verb)
+		}
+	}
+	out, err := tools["mesh-broker.snapshot-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "form": float64(1),
+		"fingerprint": "abc", "catch-up": "indexer_take_in", "why": "before form 1", "caller": "person.mallory"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if r := out.(*keeper.Request); r.Verb != keeper.VerbSnapshotKept || r.Form != 1 || r.Fingerprint != "abc" || r.CatchUp != "indexer_take_in" ||
+		r.Caller != stdio.Unproven || r.Dir == "" {
+		t.Errorf("answered %+v: the request for form 1 before abc, for the caller the runtime proved, with the directory the keeper answered", r)
+	}
+	for name, args := range map[string]map[string]any{
+		"mesh-broker.snapshot-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc"},
+		"mesh-broker.restore-kept":  {"module": "idx", "local-name": "modules", "fingerprint": "abc", "why": "revert"},
+		"mesh-broker.discard-aside": {"why": "x"},
+	} {
+		if _, err := tools[name].Run(args); err == nil {
+			t.Errorf("%s with %v was handed to the keeper", name, args)
+		}
+	}
+	if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "fingerprint": "abc",
+		"from": "/restored/f0-abc", "why": "revert"}); err != nil || out.(*keeper.Request).From != "/restored/f0-abc" {
+		t.Errorf("restore answered %+v (%v)", out, err)
+	}
+	read, err := tools["mesh-broker.aside"].Run(nil)
+	if err != nil || !strings.Contains(strings.ReplaceAll(strings.TrimSpace(toJSON(read)), " ", ""), `"id":"idx.modules.1"`) {
+		t.Errorf("aside answered %v (%v)", read, err)
+	}
+}
+
+func toJSON(v any) string {
+	b, _ := json.Marshal(v)
+	return string(b)
+}

modules/nats/cmd/nats-tools/main.go

@@ -166,7 +166,10 @@ func main() {
 // bus holds against what was asked, and what no manifest asks for. Its acting verbs (create, apply, recreate, drop) are
 // the keeper's own work, on its own account alone (keeper.go), so no other account, the tool runner's included, holds
 // them.
-var ReadingVerbs = []string{"list", "show", "check", "orphans"}
+var ReadingVerbs = []string{"list", "show", "check", "orphans", "aside"}
+
+// KeptVerbs are the seat mesh-broker's verbs for a module's kept data (keptools.go): each handed to the keeper.
+var KeptVerbs = []string{"snapshot-kept", "restore-kept", "discard-aside"}
 
 // Events are the seat mesh-broker's events (ADR 0595 §6, ADR 0620 §2): said on the seat's subjects on the shared events stream
 // and kept in full on the bus module's own stream broker-events (BROKER_EVENTS), which `history` reads (keep.go).
@@ -190,5 +193,7 @@ func allTools() []stdio.Tool {
 	out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...)
 	out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz))
 	out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...)
-	return append(out, requestTools(RequestsInState())...)
+	requests := RequestsInState()
+	out = append(out, requestTools(requests)...)
+	return append(out, keptTools(requests, KeptInState)...)
 }

modules/nats/cmd/nats-tools/manifest_test.go

@@ -189,8 +189,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	if err := json.Unmarshal(raw, &nm); err != nil {
 		t.Fatal(err)
 	}
-	if len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, ReadingVerbs) {
-		t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v alone", nm.Claims, ReadingVerbs)
+	if len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, append(slices.Clone(ReadingVerbs), KeptVerbs...)) {
+		t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v and the keeper-handed verbs %v alone", nm.Claims, ReadingVerbs, KeptVerbs)
 	}
 	if !slices.Equal(nm.Consumes, []string{"mesh-catalog.manifest-accepted"}) {
 		t.Errorf("consumes %v: the seat mesh-catalog's manifest-accepted alone (ADR 0595 §4)", nm.Consumes)
@@ -224,8 +224,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	}
 	// spent, the warrants the keeper spent, is written by the keeper alone and only read by the tools (review round 2,
 	// security F2: the tool runner's account, which writes requests, may never delete a spend record).
-	if !slices.Equal(states, []string{"asked", "requests", "spent"}) || !slices.Equal(readOnly, []string{"spent"}) {
-		t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), and spent, read-only", states, readOnly)
+	if !slices.Equal(states, []string{"asked", "requests", "spent", "kept"}) || !slices.Equal(readOnly, []string{"spent"}) {
+		t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), spent, read-only, and kept, the restore points taken", states, readOnly)
 	}
 	if _, ok := nm.OwnSecrets["broker-keeper"]; !ok {
 		t.Error("own-secrets lack broker-keeper, the keeper's own account (ADR 0436 §1)")
@@ -234,9 +234,9 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	bus, _ := keeper["bus"].(map[string]any)
 	health, _ := keeper["health"].(map[string]any)
 	if keeper["type"] != "process" || keeper["artifact"] != "tools" || fmt.Sprint(keeper["run"]) != "[./nats-tools keep]" ||
-		fmt.Sprint(bus["state"]) != "[asked requests spent]" ||
+		fmt.Sprint(bus["state"]) != "[asked requests spent kept]" ||
 		health["kind"] != "tool" || health["tool"] != "nats_keeper_health" {
-		t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests and spent, judged by nats_keeper_health", keeper)
+		t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests, spent and kept, judged by nats_keeper_health", keeper)
 	}
 	// Its bus declares keeps, the one key more ADR 0626 §1 allows it, read through the SDK as the grant composition reads it.
 	var whole catalogue.Manifest
@@ -263,7 +263,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 			t.Errorf("the manifest lists the tool %s, which the bundle does not serve", name)
 		}
 	}
-	for _, verb := range ReadingVerbs {
+	for _, verb := range append(slices.Clone(ReadingVerbs), KeptVerbs...) {
 		if !served["mesh-broker."+verb] {
 			t.Errorf("the bundle does not serve the seat verb mesh-broker.%s", verb)
 		}

modules/nats/internal/keeper/keeper.go

@@ -219,6 +219,9 @@ type Keeper struct {
 	Warrant func(askID string) (*oc.Ask, *oc.Warrant, error)
 	Now     func() time.Time
 	Logf    func(string, ...any)
+	// Kept takes and restores a module's kept data, in KeptDir (kept.go); nil where the process has no such bus.
+	Kept    KeptBus
+	KeptDir string
 	// Raise and Clear are the fill marks in percent (ADR 0595 §6): settings of the bus module, 75 and 70 by default.
 	Raise, Clear int64
 

modules/nats/internal/keeper/kept.go

@@ -0,0 +1,382 @@
+package keeper
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"time"
+
+	"nats-tools/internal/busarchive"
+)
+
+// A module's kept data on the bus, backed up before the module moves and restored beside the live data when a want takes
+// it back (milestone G, design 47 "A module's kept data on the bus, backed up and restored with its delivery"). The
+// keeper takes it on its own account, as it takes a snapshot before a recreate (ADR 0235 §5): the streams of one kept
+// data as the server's own snapshot, the position of each durable consumer the module reads with, and the catch-up tool
+// the newer manifest declares, in one directory the backup holder keeps as a restore point. A restore never goes over
+// the live data: the live data is first snapshotted into a set-aside directory under a name of its own, and only then
+// removed from the bus; only discard-aside removes the set-aside copy, on a person's call.
+
+const (
+	// VerbSnapshotKept takes a restore point of one kept data; VerbRestoreKept restores one beside the live data.
+	VerbSnapshotKept = "snapshot-kept"
+	VerbRestoreKept  = "restore-kept"
+	// VerbDiscardAside removes the data a restore set aside, by its id.
+	VerbDiscardAside = "discard-aside"
+)
+
+// StateKept is the keeper's own state of the restore points it took and the data it set aside (never their bytes, which
+// are files in the keeper's directory): `point.<module>.<local>.<form>.<fingerprint>` and `aside.<id>`. Read by the
+// bus module's tool nats_kept and by the seat mesh-broker's `aside`.
+const StateKept = "kept"
+
+// PointFile and ArchiveFile are the two files of a restore point's directory; AsideFile of a set-aside directory.
+const (
+	PointFile   = "point.json"
+	ArchiveFile = "bus.tar"
+	AsideFile   = "aside.json"
+)
+
+// Position is where one durable consumer of the module stood on a stream: the stream sequence it had acknowledged up to,
+// and its configuration as the server held it, so a restore can make it again at that place.
+type Position struct {
+	Stream   string          `json:"stream"`
+	Consumer string          `json:"consumer"`
+	AckFloor uint64          `json:"ack-floor"`
+	Config   json.RawMessage `json:"config"`
+}
+
+// KeptBus is what the keeper needs of the bus beyond Bus to take and restore a kept data.
+type KeptBus interface {
+	// Existing answers which of the stream names the bus holds.
+	Existing(ctx context.Context, names []string) ([]string, error)
+	// Positions are the module's durable consumers on every stream (a consumer is the module's when it is named for it).
+	Positions(ctx context.Context, module string) ([]Position, error)
+	// SetPositions makes each consumer again from the place it stood.
+	SetPositions(ctx context.Context, ps []Position) error
+	// RestoreArchive puts every stream of an archive into the bus, which must hold none of them.
+	RestoreArchive(ctx context.Context, a *busarchive.Archive) error
+	// DeleteStreams removes streams by name.
+	DeleteStreams(ctx context.Context, names []string) error
+}
+
+// Point is a restore point's description, its directory's point.json.
+type Point struct {
+	Module      string     `json:"module"`
+	Local       string     `json:"local"`
+	Form        int        `json:"form"`
+	Fingerprint string     `json:"fingerprint"`
+	CatchUp     string     `json:"catch-up,omitempty"`
+	Streams     []string   `json:"streams"`
+	Positions   []Position `json:"positions"`
+	At          time.Time  `json:"at"`
+	Bytes       int64      `json:"bytes"`
+}
+
+// Aside is data a restore set aside: the live kept data as it was when the restore removed it.
+type Aside struct {
+	ID      string    `json:"id"`
+	Module  string    `json:"module"`
+	Local   string    `json:"local"`
+	Streams []string  `json:"streams"`
+	At      time.Time `json:"at"`
+	Bytes   int64     `json:"bytes"`
+	By      string    `json:"by"`
+	Why     string    `json:"why"`
+	Dir     string    `json:"dir"`
+}
+
+// PointKey is a restore point's key in StateKept.
+func PointKey(module, local string, form int, fingerprint string) string {
+	return fmt.Sprintf("point.%s.%s.%d.%s", module, local, form, fingerprint)
+}
+
+// safeName is one path component made of what a module, a kept data and a fingerprint are made of.
+func safeName(s string) bool {
+	return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00 *>.")
+}
+
+func (k *Keeper) pointDir(module, local string, form int, fingerprint string) string {
+	return filepath.Join(k.KeptDir, "points", module, local, fmt.Sprintf("f%d-%s", form, fingerprint))
+}
+
+// streamsOf is the streams one kept data is: its plan's objects where the keeper made it, else the streams the bus holds
+// under its bus name (a foundation part's own section, made by the grants part).
+func (k *Keeper) streamsOf(ctx context.Context, module, local string) ([]string, error) {
+	if held, _, found := k.asked(module); found {
+		if m, made := held.Made[local]; made && len(m.Plan.Objects) > 0 {
+			var out []string
+			for _, o := range m.Plan.Objects {
+				out = append(out, o.Stream())
+			}
+			return out, nil
+		}
+	}
+	name := BusName(module, local)
+	got, err := k.Kept.Existing(ctx, []string{"KV_" + name, "OBJ_" + name, name})
+	if err != nil {
+		return nil, err
+	}
+	if len(got) == 0 {
+		return nil, fmt.Errorf("the bus holds no kept data %s of %s (looked for KV_%s, OBJ_%s and %s)", local, module, name, name, name)
+	}
+	return got, nil
+}
+
+func (k *Keeper) keptReady(r *Request) error {
+	if k.Kept == nil || k.KeptDir == "" {
+		return errors.New("no directory for restore points is set (MESH_NATS_SNAPSHOTS_DIR), so no kept data is taken or restored")
+	}
+	if !safeName(r.Module) || !safeName(r.Local) || !safeName(r.Fingerprint) {
+		return fmt.Errorf("%q, %q and %q are not a module, a kept data and a fingerprint: each is one plain name", r.Module, r.Local, r.Fingerprint)
+	}
+	return nil
+}
+
+// SnapshotKept takes the restore point of one kept data: the file pair in a directory named for its form and fingerprint,
+// written whole before its name is given, and its record in StateKept. It answers the directory.
+func (k *Keeper) SnapshotKept(ctx context.Context, r *Request) (string, error) {
+	if err := k.keptReady(r); err != nil {
+		return "", err
+	}
+	streams, err := k.streamsOf(ctx, r.Module, r.Local)
+	if err != nil {
+		return "", err
+	}
+	// The positions first: a consumer that moves while the snapshot is taken is then a little behind its archive, and a
+	// restored consumer that is behind hears a message twice, never misses one.
+	positions, err := k.Kept.Positions(ctx, r.Module)
+	if err != nil {
+		return "", fmt.Errorf("the consumers' positions of %s cannot be read, so no restore point is taken: %w", r.Module, err)
+	}
+	tar, err := k.Bus.Snapshot(ctx, streams)
+	if err != nil {
+		return "", fmt.Errorf("the snapshot of %s's %s failed, so no restore point is taken: %w", r.Module, r.Local, err)
+	}
+	dir := k.pointDir(r.Module, r.Local, r.Form, r.Fingerprint)
+	if err := os.MkdirAll(filepath.Dir(dir), 0o700); err != nil {
+		return "", err
+	}
+	partial := dir + ".partial"
+	_ = os.RemoveAll(partial)
+	if err := os.Mkdir(partial, 0o700); err != nil {
+		return "", err
+	}
+	if err := os.Rename(tar, filepath.Join(partial, ArchiveFile)); err != nil {
+		_ = os.RemoveAll(partial)
+		return "", fmt.Errorf("the snapshot cannot be kept as a restore point: %w", err)
+	}
+	st, _ := os.Stat(filepath.Join(partial, ArchiveFile))
+	p := Point{Module: r.Module, Local: r.Local, Form: r.Form, Fingerprint: r.Fingerprint, CatchUp: r.CatchUp,
+		Streams: streams, Positions: positions, At: k.now()}
+	if st != nil {
+		p.Bytes = st.Size()
+	}
+	raw, _ := json.MarshalIndent(p, "", "  ")
+	if err := os.WriteFile(filepath.Join(partial, PointFile), append(raw, '\n'), 0o600); err != nil {
+		_ = os.RemoveAll(partial)
+		return "", err
+	}
+	// A restore point taken again at the same form and fingerprint replaces the earlier one, whole.
+	_ = os.RemoveAll(dir)
+	if err := os.Rename(partial, dir); err != nil {
+		return "", err
+	}
+	val, _ := json.Marshal(p)
+	if _, err := k.Store.Put(StateKept, PointKey(r.Module, r.Local, r.Form, r.Fingerprint), val); err != nil {
+		k.logf("the record of the restore point %s cannot be written to its own %s: %v", dir, StateKept, err)
+	}
+	r.Dir = dir
+	k.record(r, fmt.Sprintf("took the restore point of %s's %s at form %d before %s (%d stream(s), %d consumer position(s), %d bytes)",
+		r.Module, r.Local, r.Form, r.Fingerprint, len(streams), len(positions), p.Bytes))
+	return fmt.Sprintf("restore point of %s's %s at form %d before %s in %s", r.Module, r.Local, r.Form, r.Fingerprint, dir), nil
+}
+
+// readPoint reads and checks a restore point's directory: its point.json names what the request names, its files are
+// plain files (never a link), and its archive matches its own manifest.
+func readPoint(dir string, r *Request) (*Point, *busarchive.Archive, error) {
+	for _, name := range []string{PointFile, ArchiveFile} {
+		fi, err := os.Lstat(filepath.Join(dir, name))
+		if err != nil {
+			return nil, nil, fmt.Errorf("the restore point %s holds no %s: %w", dir, name, err)
+		}
+		if !fi.Mode().IsRegular() {
+			return nil, nil, fmt.Errorf("%s of the restore point %s is not a plain file", name, dir)
+		}
+	}
+	raw, err := os.ReadFile(filepath.Join(dir, PointFile))
+	if err != nil {
+		return nil, nil, err
+	}
+	var p Point
+	if err := json.Unmarshal(raw, &p); err != nil {
+		return nil, nil, fmt.Errorf("%s of %s does not read: %w", PointFile, dir, err)
+	}
+	if p.Module != r.Module || p.Local != r.Local || p.Form != r.Form || p.Fingerprint != r.Fingerprint {
+		return nil, nil, fmt.Errorf("the directory %s holds the restore point of %s's %s at form %d before %s, not of %s's %s at form %d before %s",
+			dir, p.Module, p.Local, p.Form, p.Fingerprint, r.Module, r.Local, r.Form, r.Fingerprint)
+	}
+	f, err := os.Open(filepath.Join(dir, ArchiveFile))
+	if err != nil {
+		return nil, nil, err
+	}
+	defer f.Close()
+	a, err := busarchive.ReadArchive(f, "")
+	if err != nil {
+		return nil, nil, fmt.Errorf("the archive of the restore point %s does not read: %w", dir, err)
+	}
+	if err := a.Verify(); err != nil {
+		a.Close()
+		return nil, nil, fmt.Errorf("the archive of the restore point %s does not match its own manifest, so nothing is restored: %w", dir, err)
+	}
+	return &p, a, nil
+}
+
+// RestoreKept restores one kept data beside the live data: the live data is snapshotted into a set-aside directory first
+// and only then removed from the bus, and the restore point's streams are put in its place; the consumers the module
+// reads with on other streams are set back to their positions. It answers what was set aside, where.
+func (k *Keeper) RestoreKept(ctx context.Context, r *Request) (string, error) {
+	if err := k.keptReady(r); err != nil {
+		return "", err
+	}
+	if !filepath.IsAbs(r.From) {
+		return "", fmt.Errorf("the restore point's directory is given as an absolute path, not %q", r.From)
+	}
+	p, a, err := readPoint(filepath.Clean(r.From), r)
+	if err != nil {
+		return "", err
+	}
+	defer a.Close()
+	live, err := k.Kept.Existing(ctx, p.Streams)
+	if err != nil {
+		return "", err
+	}
+	id := fmt.Sprintf("%s.%s.%s", r.Module, r.Local, k.now().Format("20060102t150405"))
+	var aside *Aside
+	if len(live) > 0 {
+		// Aside before anything is removed: a snapshot that fails leaves the live data where it is.
+		tar, err := k.Bus.Snapshot(ctx, live)
+		if err != nil {
+			return "", fmt.Errorf("the live %s of %s could not be set aside, so nothing was removed: %w", r.Local, r.Module, err)
+		}
+		dir := filepath.Join(k.KeptDir, "aside", id)
+		if err := os.MkdirAll(dir, 0o700); err != nil {
+			return "", err
+		}
+		if err := os.Rename(tar, filepath.Join(dir, ArchiveFile)); err != nil {
+			return "", fmt.Errorf("the live data's snapshot cannot be kept: %w", err)
+		}
+		st, _ := os.Stat(filepath.Join(dir, ArchiveFile))
+		aside = &Aside{ID: id, Module: r.Module, Local: r.Local, Streams: live, At: k.now(), By: r.Caller, Why: r.Why, Dir: dir}
+		if st != nil {
+			aside.Bytes = st.Size()
+		}
+		raw, _ := json.MarshalIndent(aside, "", "  ")
+		if err := os.WriteFile(filepath.Join(dir, AsideFile), append(raw, '\n'), 0o600); err != nil {
+			return "", err
+		}
+		val, _ := json.Marshal(aside)
+		if _, err := k.Store.Put(StateKept, "aside."+id, val); err != nil {
+			return "", fmt.Errorf("what is set aside cannot be recorded, so nothing was removed: %w", err)
+		}
+		if err := k.Kept.DeleteStreams(ctx, live); err != nil {
+			return "", fmt.Errorf("the live data was set aside in %s and could not be removed from the bus: %w", dir, err)
+		}
+	}
+	if err := k.Kept.RestoreArchive(ctx, a); err != nil {
+		msg := fmt.Sprintf("the restore point could not be put on the bus: %v", err)
+		if aside != nil {
+			if back, rerr := k.putBack(ctx, aside); rerr != nil {
+				msg += fmt.Sprintf("; the live data, set aside in %s, could not be put back either: %v", aside.Dir, rerr)
+			} else {
+				msg += fmt.Sprintf("; the live data is back as it was (%s)", back)
+			}
+		}
+		return "", errors.New(msg)
+	}
+	// The consumers on streams the restore point did not carry (the snapshot of a stream carries its own consumers).
+	var apply []Position
+	for _, pos := range p.Positions {
+		if !slices.Contains(p.Streams, pos.Stream) {
+			apply = append(apply, pos)
+		}
+	}
+	if len(apply) > 0 {
+		if err := k.Kept.SetPositions(ctx, apply); err != nil {
+			return "", fmt.Errorf("the kept data is restored, but the consumers' positions could not be set back: %w", err)
+		}
+	}
+	r.Dir, r.CatchUp = "", p.CatchUp
+	where := "nothing was live, so nothing was set aside"
+	if aside != nil {
+		r.Dir = aside.Dir
+		where = fmt.Sprintf("the live data is set aside as %s (%d bytes) in %s", aside.ID, aside.Bytes, aside.Dir)
+	}
+	k.record(r, fmt.Sprintf("restored %s's %s from the restore point at form %d before %s beside the live data; %s", r.Module, r.Local, r.Form, r.Fingerprint, where))
+	return fmt.Sprintf("restored %s's %s from its restore point; %s; %d consumer position(s) set back; catch-up tool %q", r.Module, r.Local, where, len(apply), p.CatchUp), nil
+}
+
+// putBack restores set-aside data to the bus, after a restore failed.
+func (k *Keeper) putBack(ctx context.Context, a *Aside) (string, error) {
+	f, err := os.Open(filepath.Join(a.Dir, ArchiveFile))
+	if err != nil {
+		return "", err
+	}
+	defer f.Close()
+	arch, err := busarchive.ReadArchive(f, "")
+	if err != nil {
+		return "", err
+	}
+	defer arch.Close()
+	if err := k.Kept.DeleteStreams(ctx, a.Streams); err != nil {
+		return "", err
+	}
+	if err := k.Kept.RestoreArchive(ctx, arch); err != nil {
+		return "", err
+	}
+	return "from " + a.Dir, nil
+}
+
+// DiscardAside removes the data a restore set aside, by its id (Request.From), on a person's call: the request says who
+// and why, and is recorded as a hand act before the data goes.
+func (k *Keeper) DiscardAside(r *Request) (string, error) {
+	if k.KeptDir == "" {
+		return "", errors.New("no directory for restore points is set (MESH_NATS_SNAPSHOTS_DIR)")
+	}
+	id := strings.TrimSpace(r.From)
+	if !safeAsideID(id) {
+		return "", fmt.Errorf("%q is not the id of data set aside", id)
+	}
+	if strings.TrimSpace(r.Why) == "" {
+		return "", errors.New("discarding set-aside data says why")
+	}
+	raw, _, found, err := k.Store.Get(StateKept, "aside."+id)
+	if err != nil {
+		return "", err
+	}
+	if !found {
+		return "nothing done", fmt.Errorf("no data is set aside as %s", id)
+	}
+	var a Aside
+	_ = json.Unmarshal(raw, &a)
+	k.record(r, fmt.Sprintf("discarded the data set aside as %s (%s's %s, %d bytes)", id, a.Module, a.Local, a.Bytes))
+	if r.Unrecorded != "" {
+		return "nothing done", fmt.Errorf("the hand act cannot be recorded, so the set-aside data stays: %s", r.Unrecorded)
+	}
+	if err := os.RemoveAll(filepath.Join(k.KeptDir, "aside", id)); err != nil {
+		return "partly done", err
+	}
+	if err := k.Store.Delete(StateKept, "aside."+id); err != nil {
+		return "partly done", err
+	}
+	return fmt.Sprintf("discarded the data set aside as %s", id), nil
+}
+
+func safeAsideID(id string) bool {
+	return id != "" && !strings.ContainsAny(id, "/\\\x00 *>") && !strings.HasPrefix(id, ".")
+}

modules/nats/internal/keeper/kept_jetstream.go

@@ -0,0 +1,128 @@
+package keeper
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"nats-tools/internal/busarchive"
+	"nats-tools/internal/bussnap"
+)
+
+// KeptJetStream is KeptBus on the keeper's own connection.
+type KeptJetStream struct {
+	NC   *nats.Conn
+	JS   jetstream.JetStream
+	Logf func(string, ...any)
+}
+
+// Existing answers which of the names are streams the bus holds.
+func (b KeptJetStream) Existing(ctx context.Context, names []string) ([]string, error) {
+	var out []string
+	for _, n := range names {
+		_, err := b.JS.Stream(ctx, n)
+		switch {
+		case err == nil:
+			out = append(out, n)
+		case errors.Is(err, jetstream.ErrStreamNotFound):
+		default:
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// ConsumerOf is whether a durable consumer is the module's: named for it, alone or after the machine that runs it
+// (`<node>_<module>`, process.Account.QueueConsumerName).
+func ConsumerOf(consumer, module string) bool {
+	return consumer == module || strings.HasSuffix(consumer, "_"+module)
+}
+
+// Positions are the module's durable consumers on every stream, each with the stream sequence it has acknowledged up to.
+func (b KeptJetStream) Positions(ctx context.Context, module string) ([]Position, error) {
+	var out []Position
+	streams := b.JS.ListStreams(ctx)
+	var names []string
+	for s := range streams.Info() {
+		names = append(names, s.Config.Name)
+	}
+	if err := streams.Err(); err != nil {
+		return nil, err
+	}
+	for _, name := range names {
+		st, err := b.JS.Stream(ctx, name)
+		if err != nil {
+			return nil, err
+		}
+		list := st.ListConsumers(ctx)
+		for ci := range list.Info() {
+			if !ConsumerOf(ci.Name, module) {
+				continue
+			}
+			cfg, err := json.Marshal(ci.Config)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, Position{Stream: name, Consumer: ci.Name, AckFloor: ci.AckFloor.Stream, Config: cfg})
+		}
+		if err := list.Err(); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// SetPositions makes each consumer again, from the sequence after the one it had acknowledged. A consumer's place cannot
+// be edited, so it is removed and made with its own configuration and a new start.
+func (b KeptJetStream) SetPositions(ctx context.Context, ps []Position) error {
+	for _, p := range ps {
+		st, err := b.JS.Stream(ctx, p.Stream)
+		if err != nil {
+			return fmt.Errorf("the stream %s of %s's position cannot be read: %w", p.Stream, p.Consumer, err)
+		}
+		var cfg jetstream.ConsumerConfig
+		if err := json.Unmarshal(p.Config, &cfg); err != nil {
+			return fmt.Errorf("the configuration of %s on %s does not read: %w", p.Consumer, p.Stream, err)
+		}
+		info, err := st.Info(ctx)
+		if err != nil {
+			return err
+		}
+		start := p.AckFloor + 1
+		if start > info.State.LastSeq+1 {
+			start = info.State.LastSeq + 1
+		}
+		cfg.DeliverPolicy, cfg.OptStartSeq, cfg.OptStartTime = jetstream.DeliverByStartSequencePolicy, start, nil
+		if err := st.DeleteConsumer(ctx, p.Consumer); err != nil && !errors.Is(err, jetstream.ErrConsumerNotFound) {
+			return fmt.Errorf("%s on %s cannot be removed to be made again: %w", p.Consumer, p.Stream, err)
+		}
+		if _, err := st.CreateConsumer(ctx, cfg); err != nil {
+			return fmt.Errorf("%s on %s cannot be made again at %d: %w", p.Consumer, p.Stream, start, err)
+		}
+	}
+	return nil
+}
+
+// RestoreArchive puts every stream of the archive into the bus, which must hold none of them.
+func (b KeptJetStream) RestoreArchive(ctx context.Context, a *busarchive.Archive) error {
+	log := b.Logf
+	if log == nil {
+		log = func(string, ...any) {}
+	}
+	return bussnap.RestoreAll(ctx, b.NC, a, log)
+}
+
+// DeleteStreams removes streams by name.
+func (b KeptJetStream) DeleteStreams(ctx context.Context, names []string) error {
+	for _, n := range names {
+		if err := b.JS.DeleteStream(ctx, n); err != nil && !errors.Is(err, jetstream.ErrStreamNotFound) {
+			return fmt.Errorf("%s: %w", n, err)
+		}
+	}
+	return nil
+}

modules/nats/internal/keeper/kept_test.go

@@ -0,0 +1,298 @@
+package keeper
+
+import (
+	"archive/tar"
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/klauspost/compress/s2"
+
+	"nats-tools/internal/busarchive"
+)
+
+// tarBus is the fake bus whose snapshot is a real archive of the streams asked for, as the server's is.
+type tarBus struct {
+	*fakeBus
+	dir string
+	n   int
+}
+
+func (b *tarBus) Snapshot(ctx context.Context, streams []string) (string, error) {
+	if _, err := b.fakeBus.Snapshot(ctx, streams); err != nil {
+		return "", err
+	}
+	b.n++
+	path := filepath.Join(b.dir, fmt.Sprintf(".entry-%d.tar", b.n))
+	return path, os.WriteFile(path, archiveOf(streams), 0o600)
+}
+
+func sha(b []byte) string { h := sha256.Sum256(b); return hex.EncodeToString(h[:]) }
+
+// archiveOf is a snapshot archive of the streams, each holding one block.
+func archiveOf(streams []string) []byte {
+	var out bytes.Buffer
+	tw := tar.NewWriter(&out)
+	put := func(name string, b []byte) {
+		_ = tw.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(b))})
+		_, _ = tw.Write(b)
+	}
+	m := busarchive.Manifest{Format: 1, Finished: time.Date(2026, 11, 9, 3, 0, 0, 0, time.UTC)}
+	files := map[string][]byte{}
+	for _, name := range streams {
+		var raw bytes.Buffer
+		itw := tar.NewWriter(&raw)
+		_ = itw.WriteHeader(&tar.Header{Name: "msgs.blk", Mode: 0o600, Size: 3})
+		_, _ = itw.Write([]byte("abc"))
+		itw.Close()
+		var arc bytes.Buffer
+		w := s2.NewWriter(&arc)
+		_, _ = w.Write(raw.Bytes())
+		w.Close()
+		meta := []byte(`{"config":{"name":"` + name + `"}}`)
+		e := busarchive.StreamEntry{Name: name, Meta: "streams/" + name + "/backup.json", MetaSHA256: sha(meta),
+			Archive: "streams/" + name + "/stream.tar.s2", ArchiveBytes: int64(arc.Len()), ArchiveSHA256: sha(arc.Bytes())}
+		m.Streams = append(m.Streams, e)
+		files[e.Meta], files[e.Archive] = meta, arc.Bytes()
+	}
+	raw, _ := json.Marshal(m)
+	put(busarchive.ManifestName, raw)
+	for _, s := range m.Streams {
+		put(s.Meta, files[s.Meta])
+		put(s.Archive, files[s.Archive])
+	}
+	tw.Close()
+	return out.Bytes()
+}
+
+// keptBus is the fake bus's streams and consumers, as the server holds them.
+type keptBus struct {
+	streams   map[string]bool
+	positions []Position
+	set       []Position
+	restored  [][]string
+	refuse    error
+}
+
+func (b *keptBus) Existing(_ context.Context, names []string) ([]string, error) {
+	var out []string
+	for _, n := range names {
+		if b.streams[n] {
+			out = append(out, n)
+		}
+	}
+	return out, nil
+}
+func (b *keptBus) Positions(context.Context, string) ([]Position, error) { return b.positions, nil }
+func (b *keptBus) SetPositions(_ context.Context, ps []Position) error {
+	b.set = append(b.set, ps...)
+	return nil
+}
+func (b *keptBus) RestoreArchive(_ context.Context, a *busarchive.Archive) error {
+	if err := b.refuse; err != nil {
+		b.refuse = nil // refused once, as a server that was busy
+		return err
+	}
+	var names []string
+	for _, s := range a.Manifest.Streams {
+		if b.streams[s.Name] {
+			return fmt.Errorf("the server already holds %s", s.Name)
+		}
+		names = append(names, s.Name)
+	}
+	for _, n := range names {
+		b.streams[n] = true
+	}
+	b.restored = append(b.restored, names)
+	return nil
+}
+func (b *keptBus) DeleteStreams(_ context.Context, names []string) error {
+	for _, n := range names {
+		delete(b.streams, n)
+	}
+	return nil
+}
+
+func keptKeeper(t *testing.T) (*Keeper, *keptBus, *memStore, *[]string) {
+	t.Helper()
+	k, fb, store, _ := newKeeper(FirstView{})
+	dir := t.TempDir()
+	kb := &keptBus{streams: map[string]bool{"KV_idx_modules": true},
+		positions: []Position{{Stream: "forge", Consumer: "ace_idx", AckFloor: 41, Config: json.RawMessage(`{"durable_name":"ace_idx"}`)},
+			{Stream: "KV_idx_modules", Consumer: "ace_idx", AckFloor: 7, Config: json.RawMessage(`{"durable_name":"ace_idx"}`)}}}
+	acts := &[]string{}
+	k.Bus, k.Kept, k.KeptDir = &tarBus{fakeBus: fb, dir: dir}, kb, dir
+	k.Act = func(what, why, cause string) error { *acts = append(*acts, what); return nil }
+	return k, kb, store, acts
+}
+
+func take(t *testing.T, k *Keeper) *Request {
+	t.Helper()
+	r := &Request{Verb: VerbSnapshotKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", CatchUp: "indexer_take_in", Caller: "ace.restic", Why: "before form 1"}
+	if _, err := k.SnapshotKept(context.Background(), r); err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+// A module's kept data on the bus has a restore point before its module moves, and a restore never goes over the live
+// data (design 47 "A module's kept data on the bus, backed up and restored with its delivery").
+func TestAKeptDataIsBackedUpAndRestoredBesideTheLiveData(t *testing.T) {
+	for _, row := range []struct {
+		name string
+		run  func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string)
+	}{
+		{"a restore point holds the archive, the positions and the catch-up tool, keyed by form and fingerprint", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			r := take(t, k)
+			want := filepath.Join(k.KeptDir, "points", "idx", "modules", "f0-abc123")
+			if r.Dir != want {
+				t.Fatalf("restore point in %s, want %s", r.Dir, want)
+			}
+			raw, err := os.ReadFile(filepath.Join(want, PointFile))
+			if err != nil {
+				t.Fatal(err)
+			}
+			var p Point
+			if err := json.Unmarshal(raw, &p); err != nil || p.CatchUp != "indexer_take_in" || len(p.Positions) != 2 || p.Streams[0] != "KV_idx_modules" || p.Bytes == 0 {
+				t.Errorf("point.json = %s (%v)", raw, err)
+			}
+			if _, err := os.Stat(filepath.Join(want, ArchiveFile)); err != nil {
+				t.Error(err)
+			}
+			if _, _, found, _ := store.Get(StateKept, PointKey("idx", "modules", 0, "abc123")); !found {
+				t.Error("the restore point is not recorded in the state kept")
+			}
+			if len(*acts) != 1 || !strings.Contains((*acts)[0], "took the restore point") {
+				t.Errorf("hand acts %v", *acts)
+			}
+		}},
+		{"a kept data the bus does not hold is refused naming what was looked for, and nothing is written", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			r := &Request{Module: "idx", Local: "gone", Fingerprint: "abc123"}
+			_, err := k.SnapshotKept(context.Background(), r)
+			if err == nil || !strings.Contains(err.Error(), "KV_idx_gone") {
+				t.Fatalf("err = %v", err)
+			}
+			if _, err := os.Stat(filepath.Join(k.KeptDir, "points")); err == nil {
+				t.Error("a directory was made for a snapshot that was not taken")
+			}
+		}},
+		{"a name that is not one plain name is refused", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			for _, r := range []*Request{{Module: "../x", Local: "modules", Fingerprint: "a"}, {Module: "idx", Local: "mod/ules", Fingerprint: "a"}, {Module: "idx", Local: "modules", Fingerprint: ""}} {
+				if _, err := k.SnapshotKept(context.Background(), r); err == nil {
+					t.Errorf("%+v was taken", r)
+				}
+			}
+		}},
+		{"a restore sets the live data aside whole, removes it, restores the archive and sets back the other consumers", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			p := take(t, k)
+			r := &Request{Verb: VerbRestoreKept, Module: "idx", Local: "modules", Form: 0, Fingerprint: "abc123", From: p.Dir, Caller: "ace.restic", Why: "revert"}
+			if _, err := k.RestoreKept(context.Background(), r); err != nil {
+				t.Fatal(err)
+			}
+			if !kb.streams["KV_idx_modules"] || len(kb.restored) != 1 {
+				t.Errorf("streams %v restored %v", kb.streams, kb.restored)
+			}
+			if r.CatchUp != "indexer_take_in" {
+				t.Errorf("the catch-up tool is %q", r.CatchUp)
+			}
+			aside, err := os.ReadDir(filepath.Join(k.KeptDir, "aside"))
+			if err != nil || len(aside) != 1 {
+				t.Fatalf("aside: %v %v", aside, err)
+			}
+			if _, err := os.Stat(filepath.Join(k.KeptDir, "aside", aside[0].Name(), ArchiveFile)); err != nil {
+				t.Error("the live data's archive is not kept")
+			}
+			if _, _, found, _ := store.Get(StateKept, "aside."+aside[0].Name()); !found {
+				t.Error("what is set aside is not recorded")
+			}
+			// Only the consumer on a stream the point did not carry is made again; a stream's own comes with its snapshot.
+			if len(kb.set) != 1 || kb.set[0].Stream != "forge" || kb.set[0].AckFloor != 41 {
+				t.Errorf("positions set back: %+v", kb.set)
+			}
+		}},
+		{"a restore point of another fingerprint or a linked file is refused before anything is removed", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			p := take(t, k)
+			bad := *p
+			bad.Fingerprint = "other"
+			if _, err := k.RestoreKept(context.Background(), &bad); err == nil || !kb.streams["KV_idx_modules"] {
+				t.Errorf("err = %v, live %v", err, kb.streams)
+			}
+			link := filepath.Join(p.Dir, ArchiveFile)
+			real := filepath.Join(p.Dir, "real.tar")
+			if err := os.Rename(link, real); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.Symlink(real, link); err != nil {
+				t.Fatal(err)
+			}
+			r := &Request{Module: "idx", Local: "modules", Fingerprint: "abc123", From: p.Dir}
+			if _, err := k.RestoreKept(context.Background(), r); err == nil || !strings.Contains(err.Error(), "plain file") || !kb.streams["KV_idx_modules"] {
+				t.Errorf("err = %v, live %v", err, kb.streams)
+			}
+		}},
+		{"a live data that cannot be set aside stays where it is", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			p := take(t, k)
+			k.Bus.(*tarBus).fakeBus.refuse["snapshot"] = errors.New("the server refused")
+			r := &Request{Module: "idx", Local: "modules", Fingerprint: "abc123", From: p.Dir}
+			if _, err := k.RestoreKept(context.Background(), r); err == nil || !kb.streams["KV_idx_modules"] || !strings.Contains(err.Error(), "nothing was removed") {
+				t.Errorf("err = %v, live %v", err, kb.streams)
+			}
+		}},
+		{"a restore that fails puts the live data back and says so", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			p := take(t, k)
+			kb.refuse = errors.New("the server refused the archive")
+			r := &Request{Module: "idx", Local: "modules", Fingerprint: "abc123", From: p.Dir}
+			_, err := k.RestoreKept(context.Background(), r)
+			if err == nil || !strings.Contains(err.Error(), "the live data is back as it was") {
+				t.Fatalf("err = %v", err)
+			}
+		}},
+		{"set-aside data goes only on a call that says why and is recorded first", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) {
+			p := take(t, k)
+			if _, err := k.RestoreKept(context.Background(), &Request{Module: "idx", Local: "modules", Fingerprint: "abc123", From: p.Dir}); err != nil {
+				t.Fatal(err)
+			}
+			entries, _ := os.ReadDir(filepath.Join(k.KeptDir, "aside"))
+			id := entries[0].Name()
+			if _, err := k.DiscardAside(&Request{From: id}); err == nil {
+				t.Error("discarded with no why")
+			}
+			if _, err := k.DiscardAside(&Request{From: "../points", Why: "x"}); err == nil {
+				t.Error("discarded a path")
+			}
+			k.Act = func(string, string, string) error { return errors.New("the log is down") }
+			r := &Request{From: id, Why: "the revert held", Caller: "ace.operator"}
+			k.record(r, "x") // sets Unrecorded as a failing log does
+			if _, err := k.DiscardAside(r); err == nil {
+				t.Error("discarded with the act unrecorded")
+			}
+			if _, err := os.Stat(filepath.Join(k.KeptDir, "aside", id)); err != nil {
+				t.Error("the data went with its act unrecorded")
+			}
+			k.Act = func(what, why, cause string) error { *acts = append(*acts, what); return nil }
+			if _, err := k.DiscardAside(&Request{From: id, Why: "the revert held", Caller: "ace.operator"}); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := os.Stat(filepath.Join(k.KeptDir, "aside", id)); err == nil {
+				t.Error("still there")
+			}
+			if _, _, found, _ := store.Get(StateKept, "aside."+id); found {
+				t.Error("its record stays")
+			}
+		}},
+	} {
+		t.Run(row.name, func(t *testing.T) {
+			k, kb, store, acts := keptKeeper(t)
+			row.run(t, k, kb, store, acts)
+		})
+	}
+}

modules/nats/internal/keeper/requests.go

@@ -46,6 +46,13 @@ type Request struct {
 	Snapshot string    `json:"snapshot,omitempty"`
 	// Waiting is why it waits, while it waits: the operator's warrant, or the keeper's first view.
 	Waiting string `json:"waiting,omitempty"`
+	// Form, Fingerprint and CatchUp name a restore point (kept.go); From is the directory a restore reads, or the id of the
+	// data a discard removes; Dir is what the keeper answers: the restore point's directory, or the data set aside.
+	Form        int    `json:"form,omitempty"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+	CatchUp     string `json:"catch-up,omitempty"`
+	From        string `json:"from,omitempty"`
+	Dir         string `json:"dir,omitempty"`
 	// Unrecorded is why the hand act of what it did could not be recorded, when it could not: said, never only logged.
 	Unrecorded string `json:"unrecorded,omitempty"`
 }
@@ -136,6 +143,10 @@ func (k *Keeper) act(r *Request) (string, error) {
 		}
 		return fmt.Sprintf("the first view answered at revision %d", s.Revision), nil
 	}
+	switch r.Verb {
+	case VerbSnapshotKept, VerbRestoreKept, VerbDiscardAside:
+		return k.actKept(r)
+	}
 	if !k.Viewed() {
 		return "", fmt.Errorf("%w: the keeper holds no first view yet, so it acts on nothing", ErrNoWarrant)
 	}
@@ -295,3 +306,24 @@ func (k *Keeper) record(r *Request, what string) {
 		k.logf("the hand act %q could not be recorded: %v", what, err)
 	}
 }
+
+// actKept acts on a request about a kept data's restore point. These need no first view or warrant: a snapshot changes
+// nothing, a restore sets the live data aside whole before it removes it, and a discard names who and why (kept.go).
+func (k *Keeper) actKept(r *Request) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+	var out string
+	var err error
+	switch r.Verb {
+	case VerbSnapshotKept:
+		out, err = k.SnapshotKept(ctx, r)
+	case VerbRestoreKept:
+		out, err = k.RestoreKept(ctx, r)
+	default:
+		out, err = k.DiscardAside(r)
+	}
+	if err != nil {
+		return "nothing done", err
+	}
+	return out, nil
+}

modules/nats/module.json

@@ -32,7 +32,11 @@
         "list",
         "show",
         "check",
-        "orphans"
+        "orphans",
+        "aside",
+        "snapshot-kept",
+        "restore-kept",
+        "discard-aside"
       ]
     }
   ],
@@ -98,6 +102,10 @@
       "history": 1,
       "ttl-seconds": 86400,
       "read-only": true
+    },
+    {
+      "name": "kept",
+      "history": 1
     }
   ],
   "listens": [
@@ -133,7 +141,8 @@
     "nats_drop",
     "nats_keeper_health",
     "nats_first_view_again",
-    "nats_requests"
+    "nats_requests",
+    "nats_kept"
   ],
   "guards": [
     8222
@@ -257,7 +266,8 @@
         "state": [
           "asked",
           "requests",
-          "spent"
+          "spent",
+          "kept"
         ],
         "asks": true,
         "keeps": true,