A follower joining a watched state hears its current values, and a templates archive is fetched once

A read could start a state's watch before its follower joined, and a change landing in that gap woke nothing (review
of ADR 0580 §2): a joining follower now runs over the copy's current values before any later change. The watched
reader keeps each archive once per URL, and both caches are bounded.

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

go/compose/frombus/frombustest/frombustest_test.go

@@ -5,7 +5,9 @@ import (
 	"strings"
 	"testing"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
+	"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
@@ -135,6 +137,36 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
 			t.Errorf("%s: data items %v held %v: %v", row.name, items, iheld, err)
 		}
 	}
+
+	// ADR 0580 clause 1: the watched reader reads a state it composes from out of the copy, a build by key each time it is
+	// asked, and a manifest once per digest.
+	byKey := map[string]int{}
+	ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
+		byKey[state+" watched"]++
+		return handle(stdio.StateChange{Key: "placement." + Node, Op: stdio.OpPut, Value: []byte(`{}`), Current: true})
+	}, func(state, key string) ([]byte, bool, error) { byKey[state]++; return []byte(key), true, nil })
+	w := fromruntime.Watched{WatchedStates: ws}
+	for range 2 {
+		_, _, _ = w.Get(frombus.BuildsRead, "restic."+fp)
+		_, _, _ = w.Get(frombus.ManifestsRead, "sha256-d")
+		if _, found, _ := w.Get(frombus.PlacementRead, "placement."+Node); !found {
+			t.Error("the watched reader does not answer the placement from its copy")
+		}
+	}
+	if byKey[frombus.BuildsRead] != 2 || byKey[frombus.ManifestsRead] != 1 || byKey[frombus.PlacementRead] != 0 ||
+		byKey[frombus.PlacementRead+" watched"] != 1 {
+		t.Errorf("the watched reader read %v, want a build by key each time, a manifest once, the placement watched once", byKey)
+	}
+	fetched := 0
+	keep := fromruntime.KeptFetch(func(context.Context, string) ([]byte, error) { fetched++; return []byte("tar"), nil })
+	for range 2 {
+		if body, err := keep(context.Background(), "http://store/restic/files/blobs/sha256:a"); err != nil || string(body) != "tar" {
+			t.Fatalf("the kept fetch answered %q, %v", body, err)
+		}
+	}
+	if fetched != 1 {
+		t.Errorf("a templates archive was fetched %d times, want once", fetched)
+	}
 }
 
 func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok }

go/compose/fromruntime/fromruntime.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"sync"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
@@ -28,9 +29,60 @@ func (States) Get(state, key string) ([]byte, bool, error) {
 // Keys are a state's keys.
 func (States) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
 
-// Reader is frombus's Reader over this bundle's read states, fetching archives over HTTP.
+// Reader is frombus's Reader over this bundle's read states key by key, fetching archives over HTTP. A holder that
+// composes again and again reads through Watched instead (novox/hq ADR 0580).
 func Reader() frombus.Reader { return frombus.StatesReader(States{}, HTTPFetch) }
 
+// Watched is frombus's States over a watched copy (novox/hq ADR 0580 clause 1): the states a composition reads whole
+// (placement, haves, modules, resolved values) from the copy, a build by key when asked, a manifest once per digest.
+type Watched struct{ *stdio.WatchedStates }
+
+// Get is a key's value: a build read now, a manifest kept once read, any other state's from the copy.
+func (w Watched) Get(state, key string) ([]byte, bool, error) {
+	switch state {
+	case frombus.BuildsRead:
+		return w.Direct(state, key)
+	case frombus.ManifestsRead:
+		return w.Once(state, key)
+	}
+	return w.WatchedStates.Get(state, key)
+}
+
+// WatchedReader is frombus's Reader over w, fetching each templates archive over HTTP once (KeptFetch): a composition
+// made again on every change reads it again otherwise.
+func WatchedReader(w *stdio.WatchedStates) frombus.Reader {
+	return frombus.StatesReader(Watched{w}, KeptFetch(HTTPFetch))
+}
+
+// archivesAtMost bounds the archives KeptFetch keeps; past it they are emptied, which costs only a fetch again.
+const archivesAtMost = 64
+
+// KeptFetch is fetch kept once per URL: an archive's URL names its blob by digest, so its body never changes under it
+// (frombus checks the digest on every read). A failed fetch is not kept.
+func KeptFetch(fetch frombus.Fetch) frombus.Fetch {
+	var mu sync.Mutex
+	kept := map[string][]byte{}
+	return func(ctx context.Context, url string) ([]byte, error) {
+		mu.Lock()
+		body, ok := kept[url]
+		mu.Unlock()
+		if ok {
+			return body, nil
+		}
+		body, err := fetch(ctx, url)
+		if err != nil {
+			return nil, err
+		}
+		mu.Lock()
+		if len(kept) >= archivesAtMost {
+			clear(kept)
+		}
+		kept[url] = body
+		mu.Unlock()
+		return body, nil
+	}
+}
+
 // HTTPFetch reads an archive over HTTP, at most 256 MiB.
 func HTTPFetch(ctx context.Context, url string) ([]byte, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

go/handacts/handacts.go

@@ -1,8 +1,7 @@
 // Package handacts is the hand-act log's contract (novox/hq ADR 0565 clauses 10 and 11, issue 1953): the layout of an act,
 // moved unchanged from the controller's (mesh-controller internal/link HandAct) so one reader reads both logs alike; the
 // names of the seat mesh-hand-acts and of what its holder keeps; and the recorder every module records through, which
-// asks the seat and, while no holder answers, the controller's verb until the controller hands its lease away, saying
-// which log took the act.
+// asks the seat alone: the controller's hand-act and warranted are gone (issue 1953 L5b) and the bus refuses them.
 package handacts
 
 import (
@@ -95,13 +94,13 @@ type Taken struct {
 	Answer json.RawMessage `json:"answer,omitempty"`
 }
 
-// Recorder records acts for a module (ADR 0565 clause 11): through the seat mesh-hand-acts, and while the seat has no
-// holder that answers, through the controller's verb until the controller hands its lease away (the release's order
-// gives no seat's holder before its callers; 00-plan.md, "Missing" 1).
+// Recorder records acts for a module (ADR 0565 clause 11) through the seat mesh-hand-acts alone. An act the seat does
+// not take is an error that says nothing was recorded, never a quiet success.
 type Recorder struct {
 	// Ask calls a seat verb by its address (`seat:<seat>.<verb>`): stdio.Ask through the runtime.
 	Ask func(key string, body any) (json.RawMessage, error)
-	// LeaseAway says the controller's lease is handed away: then nothing falls back to it. Nil is never handed away.
+	// LeaseAway is no longer read: nothing falls back to the controller. Kept so a caller that sets it builds; it goes
+	// once no module sets it.
 	LeaseAway func() bool
 }
 
@@ -121,38 +120,22 @@ func (r Recorder) Record(q Request) (Taken, error) {
 			body[k] = v
 		}
 	}
-	controller := map[string]any{"what": q.What, "why": q.Why, "cause": q.Cause}
-	if q.Condition != "" {
-		controller["condition"] = q.Condition
-	}
-	if q.Cause == "" {
-		controller["cause"] = q.Verb
-	}
-	return r.either("hand-act", body, "hand-act", controller)
+	return r.record("hand-act", body)
 }
 
 // RecordWarrant records what the asking module did on the operator's warrant for its ask.
 func (r Recorder) RecordWarrant(asker, ask string) (Taken, error) {
 	body := map[string]any{"asker": asker, "ask": ask}
-	return r.either("warranted", body, "warranted", body)
+	return r.record("warranted", body)
 }
 
-func (r Recorder) either(verb string, body any, controllerVerb string, controllerBody any) (Taken, error) {
+func (r Recorder) record(verb string, body any) (Taken, error) {
 	answer, err := r.Ask("seat:"+Seat+"."+verb, body)
 	if err == nil {
 		return Taken{Log: LogSeat, Answer: answer}, nil
 	}
-	if !NoResponder(err) {
-		return Taken{}, fmt.Errorf("%s.%s refused: %w", Seat, verb, err)
+	if NoResponder(err) {
+		return Taken{}, fmt.Errorf("%s has no holder that answers: nothing was recorded (%w)", Seat, err)
 	}
-	if r.LeaseAway != nil && r.LeaseAway() {
-		return Taken{}, fmt.Errorf("%s has no holder that answers, and the controller handed its lease away: nothing "+
-			"was recorded (%w)", Seat, err)
-	}
-	answer, cerr := r.Ask("seat:mesh-controller."+controllerVerb, controllerBody)
-	if cerr != nil {
-		return Taken{}, fmt.Errorf("%s has no holder that answers (%v), and the controller's %s refused too: %w", Seat, err,
-			controllerVerb, cerr)
-	}
-	return Taken{Log: LogController, Answer: answer}, nil
+	return Taken{}, fmt.Errorf("%s.%s refused, nothing was recorded: %w", Seat, verb, err)
 }

go/handacts/handacts_test.go

@@ -8,9 +8,8 @@ import (
 	"testing"
 )
 
-// ADR 0565 clauses 10 and 11 (issue 1953 L5-S2, S4): an act reads the controller's entries unchanged, and a recorder
-// records through the seat, through the controller's verb only while the seat has no holder and the lease is the
-// controller's, saying which log took it, and refuses to record nowhere.
+// ADR 0565 clauses 10 and 11 (issue 1953 L5-S2, S4, L5b): an act reads the controller's entries unchanged, and a
+// recorder records through the seat alone, the controller's verbs being gone, and says loudly when nothing was recorded.
 func TestAnActIsRecordedInTheLogThatAnswers(t *testing.T) {
 	// The shape of mesh-controller internal/link HandAct, as its bucket keeps an entry (handacts.go at 0fbb5ef3).
 	raw := `{"id":"act-1","at":"2026-10-28T09:00:00Z","by":"person.jochen","verb":"hand-act record","why":"a stuck consumer",` +
@@ -39,14 +38,13 @@ func TestAnActIsRecordedInTheLogThatAnswers(t *testing.T) {
 		warrants bool
 	}{
 		{name: "the seat answers", log: LogSeat, asked: "seat:mesh-hand-acts.hand-act"},
-		{name: "no holder yet", seat: noResponders, log: LogController, asked: "seat:mesh-hand-acts.hand-act,seat:mesh-controller.hand-act"},
-		{name: "no holder, the lease handed away", seat: noResponders, away: true, refused: "handed its lease away",
+		{name: "no holder: nothing recorded, the controller never asked", seat: noResponders, refused: "nothing was recorded",
 			asked: "seat:mesh-hand-acts.hand-act"},
-		{name: "the seat refuses", seat: errors.New("why is required"), refused: "refused", asked: "seat:mesh-hand-acts.hand-act"},
-		{name: "neither answers", seat: noResponders, ctrl: noResponders, refused: "refused too",
-			asked: "seat:mesh-hand-acts.hand-act,seat:mesh-controller.hand-act"},
-		{name: "a warrant without a holder", seat: noResponders, warrants: true, log: LogController,
-			asked: "seat:mesh-hand-acts.warranted,seat:mesh-controller.warranted"},
+		{name: "the seat refuses", seat: errors.New("why is required"), refused: "refused, nothing was recorded",
+			asked: "seat:mesh-hand-acts.hand-act"},
+		{name: "a warrant without a holder", seat: noResponders, warrants: true, refused: "nothing was recorded",
+			asked: "seat:mesh-hand-acts.warranted"},
+		{name: "a warrant recorded", warrants: true, log: LogSeat, asked: "seat:mesh-hand-acts.warranted"},
 	} {
 		var asked []string
 		r := Recorder{LeaseAway: func() bool { return c.away }, Ask: func(key string, body any) (json.RawMessage, error) {

go/state_test.go

@@ -230,4 +230,57 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
 		t.Fatalf("the get answered %v", r)
 	}
 	_ = inW.Close()
+
+	// ADR 0580 clause 1: the watched copy answers keys and values after the watch's current values, a later put or delete
+	// changes its answer, a follower's then sees the copy that took its change, each state is watched once, and a
+	// value kept once is read by key once.
+	handles, watches, gets := map[string]func(StateChange) error{}, map[string]int{}, 0
+	w := WatchedStatesOver(func(state string, handle func(StateChange) error) error {
+		watches[state]++
+		handles[state] = handle
+		return handle(StateChange{State: state, Key: "a", Op: OpPut, Value: []byte("1"), Current: true})
+	}, func(state, key string) ([]byte, bool, error) { gets++; return []byte(key), key != "missing", nil })
+	var thenSaw []string
+	w.Follow("placement", func(c StateChange) error {
+		v, found, _ := w.Get("placement", c.Key)
+		thenSaw = append(thenSaw, fmt.Sprintf("%s %s %s %v", c.Op, c.Key, v, found))
+		return nil
+	})
+	if keys, err := w.Keys("placement"); err != nil || fmt.Sprint(keys) != "[a]" {
+		t.Fatalf("the copy's keys are %v (%v)", keys, err)
+	}
+	_ = handles["placement"](StateChange{Key: "b", Op: OpPut, Value: []byte("2")})
+	_ = handles["placement"](StateChange{Key: "a", Op: OpDelete})
+	vals, _ := w.Values("placement")
+	if v, found, _ := w.Get("placement", "a"); found || string(vals["b"]) != "2" || len(vals) != 1 || v != nil {
+		t.Fatalf("after a put and a delete the copy is %v", vals)
+	}
+	if fmt.Sprint(thenSaw) != "[put a 1 true put b 2 true delete a  false]" {
+		t.Fatalf("then saw %v", thenSaw)
+	}
+	_, _ = w.Keys("placement")
+	if watches["placement"] != 1 {
+		t.Fatalf("the state was watched %d times", watches["placement"])
+	}
+	// A read that starts a state's watch before its follower joins loses that follower no change: the follower is run over
+	// the copy's current values when it joins, then hears every later change (review of ADR 0580 §2).
+	if _, _, err := w.Get("haves", "n1.a"); err != nil {
+		t.Fatal(err)
+	}
+	_ = handles["haves"](StateChange{Key: "n1.b", Op: OpPut, Value: []byte("2")}) // in the gap: no follower yet
+	var joined []string
+	w.Follow("haves", func(c StateChange) error { joined = append(joined, c.Key); return nil })
+	_ = handles["haves"](StateChange{Key: "n2.c", Op: OpPut, Value: []byte("3")})
+	if fmt.Sprint(joined) != "[a n1.b n2.c]" || watches["haves"] != 1 {
+		t.Fatalf("a follower joining a watched state heard %v (watched %d times)", joined, watches["haves"])
+	}
+	for range 2 {
+		if v, found, _ := w.Once("manifests", "d1"); !found || string(v) != "d1" {
+			t.Fatalf("once read %q %v", v, found)
+		}
+		_, _, _ = w.Once("manifests", "missing")
+	}
+	if gets != 3 {
+		t.Fatalf("a kept value was read by key %d times, want once, and the missing one twice", gets)
+	}
 }

go/watched.go

@@ -0,0 +1,244 @@
+package stdio
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+)
+
+// WatchedStates is a copy of the states a module reads, each filled by its own watch and kept by its changes (novox/hq
+// ADR 0580 clause 1): a read of a whole state is one watch's current values, never a get per key, which took a
+// composition past 30 seconds and a compare past its call bound. A state is watched on its first read, or on Follow,
+// and the watch hands every current value over before the read is answered, so the copy is complete from the first
+// answer. A key whose value never changes once written (a manifest, by its digest) is read once with Once and kept.
+// A state helper, no rule: what a module reads is its own.
+type WatchedStates struct {
+	watch func(state string, handle func(StateChange) error) error
+	get   func(state, key string) ([]byte, bool, error)
+
+	mu     sync.Mutex
+	states map[string]*watchedState
+	// once holds each value read with Once, by state and key, for the process's life: a few kilobytes per manifest
+	// digest. Past onceAtMost keys it is emptied, which costs only a read again.
+	once map[string][]byte
+}
+
+// onceAtMost bounds the values Once keeps.
+const onceAtMost = 4096
+
+type watchedState struct {
+	ready chan struct{} // closed once the watch's current values are all in the copy, or it was refused
+	err   error
+
+	mu      sync.Mutex
+	values  map[string][]byte
+	then    []func(StateChange) error
+	started bool          // the current values are in, so a change runs then at once
+	pending []StateChange // the current values, run through then once all are in
+	thenMu  sync.Mutex    // then runs one change at a time
+}
+
+// NewWatchedStates is the copy over this bundle's runtime: each state by KeptState.Watch, Once by KeptState.Get.
+func NewWatchedStates() *WatchedStates {
+	return WatchedStatesOver(
+		func(state string, handle func(StateChange) error) error { return State(state).Watch("", handle) },
+		func(state, key string) ([]byte, bool, error) {
+			e, err := State(state).Get(key)
+			if err != nil || e == nil {
+				return nil, false, err
+			}
+			return e.Value, true, nil
+		})
+}
+
+// WatchedStatesOver is the copy over a watch and a get of one's own (a process's bus, a test's): watch hands handle
+// every current value and returns once they are handled, then hands every change; get reads one key.
+func WatchedStatesOver(watch func(state string, handle func(StateChange) error) error,
+	get func(state, key string) ([]byte, bool, error)) *WatchedStates {
+	return &WatchedStates{watch: watch, get: get, states: map[string]*watchedState{}, once: map[string][]byte{}}
+}
+
+// watched is the state's copy, its watch started now when it has none, the caller waiting until its current values
+// are in. then, when given, runs after the copy took each change: the current values once they are all in (a then
+// that reads the copy reads it whole), each later change as it comes. No lock is held while the watch starts, so a
+// then may read any state of the copy.
+func (w *WatchedStates) watched(state string, then func(StateChange) error) (*watchedState, error) {
+	w.mu.Lock()
+	s, ok := w.states[state]
+	if ok {
+		w.mu.Unlock()
+		<-s.ready
+		if s.err != nil {
+			return nil, s.err
+		}
+		if then != nil {
+			s.join(state, then)
+		}
+		return s, nil
+	}
+	s = &watchedState{ready: make(chan struct{}), values: map[string][]byte{}}
+	if then != nil {
+		s.then = []func(StateChange) error{then}
+	}
+	w.states[state] = s
+	w.mu.Unlock()
+	err := w.watch(state, func(c StateChange) error {
+		s.mu.Lock()
+		if c.Op == OpPut {
+			s.values[c.Key] = append([]byte(nil), c.Value...)
+		} else {
+			delete(s.values, c.Key)
+		}
+		if !s.started {
+			s.pending = append(s.pending, c)
+			s.mu.Unlock()
+			return nil
+		}
+		s.mu.Unlock()
+		return s.run(state, c)
+	})
+	if err != nil {
+		s.err = fmt.Errorf("the state %s is not watched: %w", state, err)
+		w.mu.Lock()
+		delete(w.states, state) // asked again on the next read
+		w.mu.Unlock()
+		close(s.ready)
+		return nil, s.err
+	}
+	s.mu.Lock()
+	s.started = true
+	pending := s.pending
+	s.pending = nil
+	s.mu.Unlock()
+	close(s.ready)
+	for _, c := range pending {
+		if err := s.run(state, c); err != nil {
+			sayf("%v", err)
+		}
+	}
+	return s, nil
+}
+
+// join adds then to a state already watched (a read started its watch first) and runs it over the copy's current values
+// before any later change, so no change that landed before it joined is lost to it (review of ADR 0580 §2: a read
+// beat the follower to the watch at start-up). Under thenMu, so a later change waits for these; a change the copy took
+// just before the snapshot may also be handed again, which a then that wakes work takes as a second wake.
+func (s *watchedState) join(state string, then func(StateChange) error) {
+	s.thenMu.Lock()
+	defer s.thenMu.Unlock()
+	s.mu.Lock()
+	s.then = append(s.then, then)
+	current := make([]StateChange, 0, len(s.values))
+	for k, v := range s.values {
+		current = append(current, StateChange{State: state, Key: k, Op: OpPut, Value: v, Current: true})
+	}
+	s.mu.Unlock()
+	sort.Slice(current, func(i, j int) bool { return current[i].Key < current[j].Key })
+	for _, c := range current {
+		if err := then(c); err != nil {
+			sayf("%s %s: %v", state, c.Key, err)
+		}
+	}
+}
+
+// run hands one change, already in the copy, to every then.
+func (s *watchedState) run(state string, c StateChange) error {
+	s.thenMu.Lock()
+	defer s.thenMu.Unlock()
+	s.mu.Lock()
+	then := append([]func(StateChange) error(nil), s.then...)
+	s.mu.Unlock()
+	var errs []error
+	for _, f := range then {
+		if err := f(c); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if len(errs) > 0 {
+		return fmt.Errorf("%s %s: %v", state, c.Key, errs)
+	}
+	return nil
+}
+
+// Follow watches the state into the copy and runs then on each change after the copy took it, so a reader then wakes
+// never reads a copy older than its change; it asks again until the watch is taken (2 s, doubled to a minute), as
+// KeptState.Follow does. On a state already watched then hears the changes from now on.
+func (w *WatchedStates) Follow(state string, then func(StateChange) error) {
+	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
+		if _, err := w.watched(state, then); err == nil {
+			return
+		} else {
+			sayf("%v; asking again in %s", err, wait)
+		}
+		time.Sleep(wait)
+	}
+}
+
+// Get is a key's value in the copy, false when it has none.
+func (w *WatchedStates) Get(state, key string) ([]byte, bool, error) {
+	s, err := w.watched(state, nil)
+	if err != nil {
+		return nil, false, err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, ok := s.values[key]
+	return v, ok, nil
+}
+
+// Keys are the state's keys in the copy, sorted.
+func (w *WatchedStates) Keys(state string) ([]string, error) {
+	s, err := w.watched(state, nil)
+	if err != nil {
+		return nil, err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	keys := make([]string, 0, len(s.values))
+	for k := range s.values {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys, nil
+}
+
+// Values is every key's value in the copy.
+func (w *WatchedStates) Values(state string) (map[string][]byte, error) {
+	s, err := w.watched(state, nil)
+	if err != nil {
+		return nil, err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make(map[string][]byte, len(s.values))
+	for k, v := range s.values {
+		out[k] = v
+	}
+	return out, nil
+}
+
+// Once is a key whose value never changes once written, read by key the first time it is found and kept after: a
+// manifest by its digest. A key not found is not kept, so it is read again on the next ask.
+func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) {
+	w.mu.Lock()
+	v, ok := w.once[state+" "+key]
+	w.mu.Unlock()
+	if ok {
+		return v, true, nil
+	}
+	v, found, err := w.get(state, key)
+	if err != nil || !found {
+		return nil, false, err
+	}
+	w.mu.Lock()
+	if len(w.once) >= onceAtMost {
+		clear(w.once)
+	}
+	w.once[state+" "+key] = v
+	w.mu.Unlock()
+	return v, true, nil
+}
+
+// Direct is a key read by key now, not from the copy: a value that changes and is read one at a time (a build).
+func (w *WatchedStates) Direct(state, key string) ([]byte, bool, error) { return w.get(state, key) }