Merge pull request 'The node's shadow says which input woke each answer (issue 1126)' (#163) from fix/1126-vault-own into main

internal/selfcompose/selfcompose_test.go

@@ -836,6 +836,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 	if got := answering(); got != settled+1 {
 		t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled)
 	}
+	// Issue 1126, 2026-10-28 (seed-9's probe): each answer says which input woke it, so a loop is told from inputs moving.
+	lastSaid := func(prefix string) string {
+		readOnly.mu.Lock()
+		defer readOnly.mu.Unlock()
+		for i := len(readOnly.said) - 1; i >= 0; i-- {
+			if strings.HasPrefix(readOnly.said[i], prefix) {
+				return readOnly.said[i]
+			}
+		}
+		return ""
+	}
+	if got := lastSaid("answering placement candidate same "); got != "answering placement candidate same (revision 8; woken by the candidate)" {
+		t.Errorf("a new revision's answer said %q", got)
+	}
 	settled = answering()
 	hk, _ := delivery.HaveKey("n1", "lemurs")
 	moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning})
@@ -844,6 +858,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 	if got := answering(); got != settled+len(answers) {
 		t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers))
 	}
+	if got := lastSaid("answering placement candidate drops "); got != "answering placement candidate drops (revision 5; woken by haves)" {
+		t.Errorf("an answer woken by what runs moving said %q", got)
+	}
 	// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
 	setFor(t, &laterAgain, []time.Duration{20 * time.Millisecond})
 	ownFails.Store(true)
@@ -856,6 +873,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 		return a["adds"].Revision == 9 && a["adds"].Verdict == delivery.CandidateDiffers
 	})
 	drain()
+	if got := lastSaid("answering placement candidate adds "); got != "answering placement candidate adds (revision 9; woken by a retry after a read the bus did not answer)" {
+		t.Errorf("an answer composed again after the bus did not answer said %q", got)
+	}
 	k, _ := delivery.PlacementCandidateKey("same", "n1")
 	sb.del(delivery.PlacementCandidatesBucket, k)
 	waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 })
@@ -973,8 +993,8 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 		// Issue 1126, T2: an answer says when it starts and when it ends (and, every minute between, what it waits on).
 		nmu.Lock()
 		said := strings.Join(nsaid, "\n")
-		for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3)",
+		for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3; woken by its first answer since the shadow began)",
-			"answered placement candidate same in", "answering placement candidate older (revision 2)"} {
+			"answered placement candidate same in", "answering placement candidate older (revision 2;"} {
 			if !strings.Contains(said, line) {
 				t.Errorf("%q was not said: %q", line, nsaid)
 			}

internal/selfcompose/shadow.go

@@ -40,22 +40,52 @@ type Shadow struct {
 	// ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy
 	// from the declaration kept before the switch (issue 1126, probe #532 on shanks).
 	ran map[string]string
-	// inputs counts the changes of the node's own inputs (everything but the candidates), and answered is, by group, the
+	// answered is, by group, the candidate revisions and the input changes (moved) each answer was composed at: a group
-	// candidate revisions and the count each answer was composed at: a group is answered again only when either moved
+	// is answered again only when either moved (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84
-	// (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, woken by its own modules'
+	// every 13-17 s, woken by its own modules' haves whose health words changed at every look).
-	// haves whose health words changed at every look).
-	inputs   uint64
 	answered map[string]answeredAt
+	// moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the groups
+	// answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126,
+	// 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving).
+	moved [len(inputWords)]uint64
+	retry map[string]bool
 	// owed says an answer met a read the bus did not answer: it is no answer to keep, and its group is answered again
 	// after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not
 	// yet ... no responders" in its answer to seed-7).
 	owed bool
 }
 
-// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the count of input changes.
+// answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the changes of each input.
 type answeredAt struct {
-	revs   string
+	revs  string
-	inputs uint64
+	moved [len(inputWords)]uint64
+}
+
+// inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it.
+var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"}
+
+// wokenBy says what moved between the answer a group stands at (had, when done) and the one about to be composed (at):
+// its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words.
+func wokenBy(had answeredAt, done, retry bool, at answeredAt) string {
+	switch {
+	case retry:
+		return "a retry after a read the bus did not answer"
+	case !done:
+		return "its first answer since the shadow began"
+	}
+	var words []string
+	if had.revs != at.revs {
+		words = append(words, inputWords[0])
+	}
+	for i := 1; i < len(inputWords); i++ {
+		if had.moved[i] != at.moved[i] {
+			words = append(words, inputWords[i])
+		}
+	}
+	if len(words) == 0 {
+		return "nothing that moved"
+	}
+	return strings.Join(words, ", ")
 }
 
 // candidateKey is one candidate key as last read: its value and revision.
@@ -112,10 +142,13 @@ func (s *Shadow) Run(ctx context.Context) error {
 	s.wants = map[string]delivery.Want{}
 	s.ran = map[string]string{}
 	s.answered = map[string]answeredAt{}
+	s.moved = [len(inputWords)]uint64{}
+	s.retry = map[string]bool{}
 	s.c = New(s.c.env)
 	s.mu.Unlock()
 	c := s.c
 	node := c.env.Node
+	// In inputWords' order.
 	ins := []input{
 		{delivery.PlacementCandidatesBucket, ">", nil, nil},
 		{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
@@ -197,7 +230,7 @@ func (s *Shadow) Run(ctx context.Context) error {
 			}
 			if changed {
 				if t.in != 0 {
-					s.inputs++ // the candidates' own changes are their revisions
+					s.moved[t.in]++ // the candidates' own changes are their revisions
 				}
 				dirty = true
 			}
@@ -388,7 +421,7 @@ func (s *Shadow) answerAll(ctx context.Context) {
 	}
 	var groups []listed
 	for g, keys := range s.candidates {
-		l := listed{group: g, at: answeredAt{revs: revsOf(keys), inputs: s.inputs}}
+		l := listed{group: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}}
 		for _, k := range keys {
 			l.rev = max(l.rev, k.rev)
 		}
@@ -447,12 +480,13 @@ func (s *Shadow) answerAll(ctx context.Context) {
 		group := l.group
 		s.mu.Lock()
 		at, done := s.answered[group]
+		retry := s.retry[group]
 		s.mu.Unlock()
 		if done && at == l.at {
 			continue
 		}
 		began := time.Now()
-		c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev))
+		c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", group, l.rev, wokenBy(at, done, retry, l.at)))
 		stop := make(chan struct{})
 		go func() {
 			tick := time.NewTicker(progressEvery)
@@ -491,10 +525,12 @@ func (s *Shadow) answerAll(ctx context.Context) {
 		if provisional {
 			// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
 			s.owed = true
+			s.retry[group] = true
 			delete(s.answered, group)
 			c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group))
 		} else {
 			s.answered[group] = l.at
+			delete(s.retry, group)
 		}
 		s.mu.Unlock()
 		hand()