Serve node-engine.compose-candidate from the node's shadow (H4)

ADR 0593 §7: a check's composition is the checker's, made from each
placed node's answer, since only a node can compose with its own sealed
values. The node composes the indexer's candidate in shadow on a copy of
its shadow's inputs, the candidate's modules laid over the indexer's
with their manifest verdicts, and answers per module composed (changed
or not), refused or held, and its own composition's verdict, within the
answer bound; it applies, writes and hands nothing. With input it also
answers the composition input, for the sandbox path.

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

internal/selfcompose/candidate.go

@@ -0,0 +1,289 @@
+package selfcompose
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"maps"
+	"sort"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"github.com/novox/mesh-host/internal/agent"
+	"github.com/novox/mesh-host/internal/compose/frombus"
+)
+
+// **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate.
+// Given the key of the indexer's candidate (delivery.CandidatesBucket, `<group>.<heads hash>`), the node composes itself
+// twice from what its shadow holds (placement, records, the indexer's modules, its config values), with its own sealed
+// values: once as it is, and once with the candidate's modules laid over the indexer's (each at the candidate's
+// fingerprint, its manifest by digest and the indexer's verdict on it, ADR 0578 §3; a module the candidate removes left
+// out). It answers per module placed here composed (its digest, and whether the candidate changes it), refused by the
+// candidate, or held without it too, and the verdict of its own composition. It applies nothing and writes nothing: it
+// composes on a copy of the shadow's inputs, so neither the shadow's answers nor the node's own composer move.
+
+// errNotReady is a shadow that has not read every input yet.
+var errNotReady = errors.New("this node's shadow has not read every input yet")
+
+// ComposeCandidate is the node's answer to ask, within delivery.ComposeAnswerBound (or ctx's deadline, if sooner).
+func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer {
+	began := time.Now()
+	s.mu.Lock()
+	c := s.c
+	s.mu.Unlock()
+	node := c.env.Node
+	a := delivery.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate}
+	refuse := func(why string) delivery.ComposeCandidateAnswer {
+		a.Verdict, a.Refused, a.Modules, a.Own = delivery.ComposeCandidateRefused, why, nil, delivery.ModuleComposed{}
+		a.Took = time.Since(began).Milliseconds()
+		return a
+	}
+	if _, _, err := delivery.ParseCompositionKey(ask.Candidate); err != nil {
+		return refuse(err.Error())
+	}
+	ctx, cancel := context.WithTimeout(ctx, delivery.ComposeAnswerBound)
+	defer cancel()
+	if !s.ready.Load() {
+		return refuse(errNotReady.Error())
+	}
+	base := c.scratch()
+	raw, err := base.get(ctx, delivery.CandidatesBucket, ask.Candidate)
+	if err != nil {
+		return refuse(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err))
+	}
+	if raw == nil {
+		return refuse("the indexer holds no candidate " + ask.Candidate)
+	}
+	cand, err := delivery.DecodeCandidate(raw)
+	if err != nil {
+		return refuse(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err))
+	}
+	laid := c.scratch()
+	touched, err := laid.layCandidate(ctx, cand)
+	if err != nil {
+		return refuse(err.Error())
+	}
+
+	var running []delivery.Composed
+	if c.env.Running != nil {
+		running = c.env.Running()
+	}
+	baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil)
+	laidComps, laidHeld, laidErr := laid.composeShadow(ctx, running, touched, func(scope frombus.Scope) {
+		if !ask.Input {
+			return
+		}
+		in, err := frombus.Input(ctx, reader{c: laid}, laid.env.Mesh, scope)
+		if err != nil {
+			return
+		}
+		// The request's input carries no catalogue (composer.Request.Input): the cases give the manifests.
+		in.Catalogue = nil
+		for i := range in.Nodes {
+			in.Nodes[i].Kept = nil
+		}
+		if body, err := json.Marshal(in); err == nil {
+			a.Input = body
+		}
+	})
+	if ctx.Err() != nil {
+		return refuse(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound))
+	}
+	switch {
+	case laidErr != nil && BusFailed(laidErr):
+		return refuse("a read on the bus did not answer: " + laidErr.Error())
+	case laidErr != nil && baseErr != nil:
+		// The node composes none of its own without the candidate either: the candidate changes nothing of that.
+		a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{}
+		a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, Words: laidErr.Error()}
+		a.Took = time.Since(began).Milliseconds()
+		return a
+	case laidErr != nil:
+		return refuse(laidErr.Error())
+	}
+	byModule := func(comps map[string]delivery.Composed) map[string]delivery.Composed {
+		out := map[string]delivery.Composed{}
+		for _, comp := range comps {
+			out[comp.Module] = comp
+		}
+		return out
+	}
+	was, now := byModule(baseComps), byModule(laidComps)
+	one := func(module string) (delivery.ModuleComposed, bool) {
+		m := delivery.ModuleComposed{Module: module}
+		if module == "" {
+			m.Module = delivery.HaveNodeWord
+		}
+		if why, held := laidHeld[module]; held {
+			m.State, m.Words = delivery.ModuleRefusedState, why
+			if _, before := baseHeld[module]; before || baseErr != nil {
+				m.State = delivery.ModuleHeldState
+			}
+			return m, true
+		}
+		comp, ok := now[module]
+		if !ok {
+			return m, false
+		}
+		dig, err := delivery.ComposedDigestOf(comp)
+		if err != nil {
+			m.State, m.Words = delivery.ModuleRefusedState, err.Error()
+			return m, true
+		}
+		m.State, m.FP, m.Digest = delivery.ModuleComposedState, comp.FP, dig
+		if before, ok := was[module]; !ok {
+			m.Changed = true
+		} else if d, err := delivery.ComposedDigestOf(before); err != nil || d != dig {
+			m.Changed = true
+		}
+		return m, true
+	}
+	modules := map[string]bool{}
+	for m := range now {
+		modules[m] = m != ""
+	}
+	for m := range laidHeld {
+		modules[m] = m != "" && m != agent.NodeModule
+	}
+	names := make([]string, 0, len(modules))
+	for m, isModule := range modules {
+		if isModule {
+			names = append(names, m)
+		}
+	}
+	sort.Strings(names)
+	a.Modules = []delivery.ModuleComposed{}
+	refused := false
+	for _, m := range names {
+		if mc, ok := one(m); ok {
+			a.Modules = append(a.Modules, mc)
+			refused = refused || mc.State == delivery.ModuleRefusedState
+		}
+	}
+	if own, ok := one(""); ok {
+		a.Own = own
+	} else {
+		a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState,
+			Words: "the node composes no composition of its own"}
+	}
+	refused = refused || a.Own.State == delivery.ModuleRefusedState
+	a.Verdict = delivery.ComposeCandidateComposed
+	if refused {
+		a.Verdict = delivery.ComposeCandidateRefused
+	}
+	a.Took = time.Since(began).Milliseconds()
+	return a
+}
+
+// scratch is a composer over a copy of what c holds of its inputs, for composing in shadow without moving c.
+func (c *Composer) scratch() *Composer {
+	out := New(c.env)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	out.placements = maps.Clone(c.placements)
+	out.records = maps.Clone(c.records)
+	out.inOwn = maps.Clone(c.inOwn)
+	out.modules = maps.Clone(c.modules)
+	out.values = maps.Clone(c.values)
+	out.manifests = maps.Clone(c.manifests)
+	out.archives = maps.Clone(c.archives)
+	out.builds = maps.Clone(c.builds)
+	out.sealedOwn = maps.Clone(c.sealedOwn)
+	out.ownHanded = maps.Clone(c.ownHanded)
+	out.certified = c.certified
+	out.placementKeys = c.placementKeys
+	return out
+}
+
+// layCandidate lays the candidate's modules over the indexer's: each at its fingerprint, with its manifest's digest and
+// the indexer's verdict on it (read from its verdicts, none read taken as accepted); a module the candidate removes goes.
+// It answers the modules laid.
+func (c *Composer) layCandidate(ctx context.Context, cand delivery.Candidate) (map[string]bool, error) {
+	touched := map[string]bool{}
+	for _, pull := range cand.Pulls {
+		for module, cm := range pull.Fps {
+			if cm.None2 == delivery.CandidateNoneNoModule {
+				c.mu.Lock()
+				delete(c.modules, module)
+				c.mu.Unlock()
+				continue
+			}
+			if cm.FP2 == "" {
+				continue // no fingerprint at the preview: composed as the trunk holds it
+			}
+			m := delivery.Module{FP2: cm.FP2, Manifest: cm.Manifest, Repository: pull.Repository, Commit: pull.Head.Head}
+			if cm.Manifest != "" {
+				key, err := delivery.VerdictsKey(cm.Manifest)
+				if err != nil {
+					return nil, err
+				}
+				raw, err := c.get(ctx, delivery.VerdictsBucket, key)
+				if err != nil {
+					return nil, fmt.Errorf("the indexer's verdict on %s's manifest cannot be read: %w", module, err)
+				}
+				if raw != nil {
+					v, err := delivery.DecodeManifestVerdict(raw)
+					if err != nil {
+						return nil, fmt.Errorf("the indexer's verdict on %s's manifest cannot be read: %w", module, err)
+					}
+					m.Verdict, m.Problems, m.Warnings = v.State, v.Problems, v.Warnings
+				}
+			}
+			c.mu.Lock()
+			c.modules[module] = m
+			c.mu.Unlock()
+			touched[module] = true
+		}
+	}
+	return touched, nil
+}
+
+// composeShadow composes this node from what c holds, as the shadow composes it (shadowBuilds): each module at the build
+// it runs, one placed and not run at the indexer's, and one the candidate laid (laid) at the candidate's. input, when
+// given, is handed the scope.
+func (c *Composer) composeShadow(ctx context.Context, running []delivery.Composed, laid map[string]bool, input func(frombus.Scope)) (map[string]delivery.Composed, map[string]string, error) {
+	node := c.env.Node
+	c.mu.Lock()
+	placement, placed := c.placements[node]
+	modules := maps.Clone(c.modules)
+	inOwn := c.inOwnOf(placement)
+	c.mu.Unlock()
+	if !placed {
+		return nil, nil, errors.New("this node's key in the seat's placement state is gone: it composes nothing of its own")
+	}
+	fps := shadowBuilds(running, placement, modules, inOwn)
+	// A module the laid candidate changes is composed at the candidate's build, though the node runs another.
+	for module := range fps {
+		if mod, ok := modules[module]; ok && laid[module] {
+			fps[module] = mod.FP2
+		}
+	}
+	kept := map[string]string{}
+	for module, fp := range fps {
+		if mod, ok := modules[module]; ok && mod.FP2 == fp {
+			continue
+		}
+		kept[module] = fp
+	}
+	wants := map[string]delivery.Want{agent.NodeModule: {}}
+	for module, fp := range fps {
+		wants[module] = delivery.Want{FP: fp}
+	}
+	c.mu.Lock()
+	c.wants = wants
+	c.mu.Unlock()
+	scope := frombus.Scope{Node: node, KeptAt: kept, InOwn: inOwn, Running: running, SameValue: c.sameValue}
+	if input != nil {
+		input(scope)
+	}
+	comps, _, held, err := frombus.Compose(ctx, reader{c: c}, c.env.Mesh, scope, fps)
+	if err != nil {
+		return nil, nil, err
+	}
+	for key, comp := range comps {
+		comps[key] = c.keepRunning(comp, running)
+	}
+	return comps, held, nil
+}

internal/selfcompose/candidate_test.go

@@ -0,0 +1,125 @@
+package selfcompose
+
+import (
+	"context"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"github.com/novox/mesh-host/internal/agent"
+)
+
+// novox/hq ADR 0593 §7: the verb compose-candidate composes a check's candidate on this node in shadow. A candidate that
+// changes a placed module composes it at the candidate's build, changed; one whose manifest the indexer refused is
+// answered refused, naming the module and the judge's words, and the verdict is refused; a candidate the indexer does not
+// hold is refused in words. Nothing is applied, nothing is written, no placement answer is handed, and the answer comes
+// within the bound (an ended bound answers at once, refused).
+func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
+	b := newFakeBus()
+	fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
+	fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
+	digest := delivery.ManifestDigest([]byte(lemurs))
+	lemurs2 := strings.Replace(lemurs, `"version":"1"`, `"version":"2"`, 1)
+	digest2 := delivery.ManifestDigest([]byte(lemurs2))
+	for d, body := range map[string]string{digest: lemurs, digest2: lemurs2} {
+		man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: body})
+		key, _ := delivery.ManifestsKey(d)
+		b.kv[delivery.ManifestsBucket+"/"+key] = man
+	}
+	candidateOf := func(manifest string) string {
+		head := delivery.Head{Repository: "novox/x", Number: 7, Head: "c0ffee" + manifest[len(manifest)-6:]}
+		heads, _ := delivery.HeadsHash([]delivery.Head{head})
+		pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"lemurs"},
+			Fps: map[string]delivery.CandidateModule{"lemurs": {FP2: fpNew, Manifest: manifest}}}}
+		raw, err := delivery.EncodeCandidate(delivery.Candidate{Group: "novox/x#7", HeadsHash: heads, Pulls: pulls,
+			FPSet: delivery.FingerprintSet(pulls)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		key, _ := delivery.CompositionKey("novox/x#7", heads)
+		b.kv[delivery.CandidatesBucket+"/"+key] = raw
+		return key
+	}
+	accepted, refusedKey := candidateOf(digest2), ""
+	{
+		// The same change, its manifest refused by the indexer's judge.
+		refusedMan := strings.Replace(lemurs2, `"version":"2"`, `"version":"3"`, 1)
+		d3 := delivery.ManifestDigest([]byte(refusedMan))
+		man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: refusedMan})
+		key, _ := delivery.ManifestsKey(d3)
+		b.kv[delivery.ManifestsBucket+"/"+key] = man
+		vkey, _ := delivery.VerdictsKey(d3)
+		v, err := delivery.EncodeManifestVerdict(delivery.ManifestVerdict{Module: "lemurs", State: delivery.VerdictRefused,
+			Problems: []string{"a field the judge refuses"}, Judge: "src2:j", At: time.Now()})
+		if err != nil {
+			t.Fatal(err)
+		}
+		b.kv[delivery.VerdictsBucket+"/"+vkey] = v
+		refusedKey = candidateOf(d3)
+	}
+
+	env := b.env()
+	state := &fakeComposed{kv: map[string]ComposedEntry{}}
+	env.ComposedState = state
+	var applied atomic.Int64
+	env.Refused = func(string, map[string]string) { applied.Add(1) }
+	var handed atomic.Int64
+	s := NewShadow(env, func(context.Context, map[string]delivery.CandidateAnswer) { handed.Add(1) })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() { _ = s.Run(ctx) }()
+	if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); !strings.Contains(a.Refused, "not read every input") {
+		t.Errorf("a shadow not ready answered %+v", a)
+	}
+	place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
+		Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
+	have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
+	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
+	deadline := time.Now().Add(5 * time.Second)
+	watching := func() int { b.mu.Lock(); defer b.mu.Unlock(); return len(b.watches) }
+	for watching() < 6 && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	b.put(t, delivery.PlacementBucket, "placement.n1", place)
+	b.put(t, delivery.HaveBucket, "n1.node", have)
+	b.put(t, delivery.ModulesBucket, "lemurs", mod)
+	for _, bucket := range []string{delivery.PlacementCandidatesBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
+		b.send(bucket, agent.Update{Caught: true})
+	}
+	for !s.ready.Load() && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted})
+	if _, err := delivery.EncodeComposeCandidateAnswer(a); err != nil {
+		t.Fatalf("the answer does not fit the seat's shape: %v (%+v)", err, a)
+	}
+	if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" ||
+		a.Modules[0].State != delivery.ModuleComposedState || a.Modules[0].FP != fpNew || !a.Modules[0].Changed ||
+		a.Own.Module != delivery.HaveNodeWord || a.Took > delivery.ComposeAnswerBound.Milliseconds() {
+		t.Errorf("an accepted candidate answered %+v", a)
+	}
+	r := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: refusedKey})
+	if r.Verdict != delivery.ComposeCandidateRefused || len(r.Modules) != 1 || r.Modules[0].State != delivery.ModuleRefusedState ||
+		!strings.Contains(r.Modules[0].Words, "a field the judge refuses") {
+		t.Errorf("a candidate whose manifest the indexer refused answered %+v", r)
+	}
+	if _, err := delivery.EncodeComposeCandidateAnswer(r); err != nil {
+		t.Errorf("the refused answer does not fit the seat's shape: %v", err)
+	}
+	missing, _ := delivery.CompositionKey("novox/x#8", "0123456789ab")
+	if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: missing}); !strings.Contains(m.Refused, "holds no candidate") {
+		t.Errorf("a candidate the indexer does not hold answered %+v", m)
+	}
+	ended, end := context.WithCancel(ctx)
+	end()
+	if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateRefused {
+		t.Errorf("an ended bound answered %+v", e)
+	}
+	if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
+		t.Errorf("composing candidates wrote %v, marked %d have(s), handed %d answer(s); want nothing", state.written(), applied.Load(), handed.Load())
+	}
+}

internal/selfcompose/shadow.go

@@ -7,6 +7,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -54,6 +55,8 @@ type Shadow struct {
 	// 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
+	// ready says this run read every input once: the verb compose-candidate composes only then (ComposeCandidate).
+	ready atomic.Bool
 }
 
 // answeredAt is what one proposal's answer was composed at: its keys' revisions, by node, and the changes of each input.
@@ -146,6 +149,8 @@ func (s *Shadow) Run(ctx context.Context) error {
 	s.retry = map[string]bool{}
 	s.c = New(s.c.env)
 	s.mu.Unlock()
+	s.ready.Store(false)
+	defer s.ready.Store(false)
 	c := s.c
 	node := c.env.Node
 	// In inputWords' order.
@@ -238,6 +243,7 @@ func (s *Shadow) Run(ctx context.Context) error {
 			if ready = all(caught); !ready {
 				continue
 			}
+			s.ready.Store(true)
 			dirty = true
 		}
 		if dirty {