mesh-host/internal/selfcompose/candidate.go

package selfcompose

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"

	"github.com/novox/mesh-host/internal/agent"
	"github.com/novox/mesh-host/internal/compose/frombus"
	"github.com/novox/mesh-host/internal/composer"
)

// **A check's candidate, composed in shadow on a call** (novox/hq ADR 0593 §7): the node seat's verb compose-candidate.
// Given the check's composition key (`<group>.<heads hash>`), it reads the group's candidate from the indexer
// (delivery.CandidatesBucket, keyed by the group alone, delivery.CandidatesKey) at those heads, and 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")

// candidateWithin is the node's own bound on an answer to compose-candidate: shorter than the checker's
// (delivery.ComposeAnswerBound), because the bus lets the engine answer an ask only within a minute of it (the grants
// part's allow_responses ttl, mesh-grants ResponseTTL): an answer at the checker's 60 s bound was refused by the bus
// (issue 1126, 2026-11-03 01:17:04 on shanks, "Permissions Violation for Publish to _INBOX..."), so the checker heard
// nothing at all. A node not done by then answers unready, naming what it waits on.
var candidateWithin = delivery.ComposeAnswerBound - 10*time.Second

// maxSentInput is the most composition input an answer carries: the bus carries 1 MiB in one message (its max_payload),
// and the answer's own fields and envelope come beside it.
const maxSentInput = 900 << 10

// warmAtOnce is how many of the indexer's manifests the shadow reads at once before it composes a candidate.
const warmAtOnce = 8

// candidateRun is what one ask's compositions wait on: each scratch composer it made, the last the one composing now.
type candidateRun struct {
	mu    sync.Mutex
	parts []*Composer
}

func (r *candidateRun) add(c *Composer) *Composer {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.parts = append(r.parts, c)
	return c
}

// said is the run's waits in words: how many reads it waited on the bus or the vault for, and the longest; and, while
// one is still running (now), what it waits on and since when.
func (r *candidateRun) said(now time.Time) (waits string, on string) {
	r.mu.Lock()
	parts := slices.Clone(r.parts)
	r.mu.Unlock()
	n, longest, longestOn := 0, time.Duration(0), ""
	for _, c := range parts {
		c.doingMu.Lock()
		n += c.waits.n
		if c.waits.longest > longest {
			longest, longestOn = c.waits.longest, c.waits.longestOn
		}
		if c.doing != "" {
			on = fmt.Sprintf("%s (for %s)", c.doing, now.Sub(c.doingSince).Round(time.Millisecond))
		} else if on == "" && c.waits.lastOn != "" {
			on = c.waits.lastOn + " (the last read)"
		}
		c.doingMu.Unlock()
	}
	waits = fmt.Sprintf("%d read(s) waited on", n)
	if longestOn != "" {
		waits += fmt.Sprintf(", the longest %s on %s", longest.Round(time.Millisecond), longestOn)
	}
	if on == "" {
		on = "nothing it reads (composing)"
	}
	return waits, on
}

// ComposeCandidate is the node's answer to ask, within candidateWithin (or ctx's deadline, if sooner). Each answer is
// said in one line, with what it waited on and for how long (issue 1126: ace, g14 and shanks answered nothing for a
// minute while nothing said why). An answer not composed within the bound is unready, naming what the node waits on.
func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer {
	s.mu.Lock()
	c := s.c
	s.mu.Unlock()
	began := c.now()
	ctx, cancel := context.WithTimeout(ctx, candidateWithin)
	defer cancel()
	run := &candidateRun{}
	done := make(chan delivery.ComposeCandidateAnswer, 1)
	go func() { done <- s.composeCandidate(ctx, c, ask, began, run) }()
	var a delivery.ComposeCandidateAnswer
	select {
	case a = <-done:
	case <-ctx.Done():
		_, on := run.said(c.now())
		a = delivery.ComposeCandidateAnswer{Node: c.env.Node, Candidate: ask.Candidate, Verdict: delivery.ComposeCandidateUnready,
			Refused: fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on),
			Took:    c.now().Sub(began).Milliseconds()}
	}
	waits, _ := run.said(c.now())
	words := ""
	if a.Refused != "" {
		words = ": " + a.Refused
	}
	c.env.Say(fmt.Sprintf("compose-candidate %s answered %s in %s (%s)%s", ask.Candidate, a.Verdict,
		(time.Duration(a.Took) * time.Millisecond).String(), waits, words))
	return a
}

// composeCandidate composes ask on scratch copies of c, each added to run as it starts.
func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery.ComposeCandidateAsk, began time.Time, run *candidateRun) delivery.ComposeCandidateAnswer {
	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 = c.now().Sub(began).Milliseconds()
		return a
	}
	// A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers
	// unready: the checker takes it as no answer (ADR 0593 §9) by the verdict, the words only saying why.
	unready := func(why string) delivery.ComposeCandidateAnswer {
		a = refuse(why)
		a.Verdict = delivery.ComposeCandidateUnready
		return a
	}
	group, heads, err := delivery.ParseCompositionKey(ask.Candidate)
	if err != nil {
		return refuse(err.Error())
	}
	// The ask names the check (the composition key, `<group>.<heads hash>`); the indexer keys the group's candidate by
	// the group alone (delivery.CandidatesKey) and holds its heads hash in the value.
	ckey, err := delivery.CandidatesKey(group)
	if err != nil {
		return refuse(err.Error())
	}
	if !s.ready.Load() {
		return unready(errNotReady.Error())
	}
	base := run.add(c.scratch())
	raw, err := base.get(ctx, delivery.CandidatesBucket, ckey)
	if err != nil {
		return unready(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))
	}
	if cand.HeadsHash != heads {
		// The group moved to other heads: the check asked is replaced, and this node composes not a candidate it was
		// not asked about. No answer (ADR 0593 §9), never a verdict on the wrong heads.
		return unready(fmt.Sprintf("the indexer's candidate of the group %s is at heads %s, not %s: the check asked is replaced", group, cand.HeadsHash, heads))
	}
	// Every module's manifest is read before either composition, several at once, into the shadow's own: a composition
	// reads every manifest in the mesh one by one, which off the control-node took about 35 s each (issue 1126).
	base.warm(ctx, c, cand)

	var running []delivery.Composed
	if c.env.Running != nil {
		running = c.env.Running()
	}
	baseComps, baseHeld, baseErr := base.composeShadow(ctx, running, nil, nil)
	// The candidate laid over a copy taken after the composition without it, so what that read is not read again.
	laid := run.add(c.scratch())
	touched, err := laid.layCandidate(ctx, cand)
	if err != nil {
		return unready(err.Error())
	}
	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
		}
		// As the sandbox reads it (composer.SentInput, the one shape): this node's own modules as taken in from their
		// builds, every other module's manifest left to the cases.
		if body, err := json.Marshal(composer.SentInput(in, laid.env.Node)); err == nil {
			a.Input = body
		}
	})
	if len(a.Input) > maxSentInput {
		// Past what the bus carries in one answer, the checker would hear nothing: refused, saying so.
		size := len(a.Input)
		a.Input = nil
		return refuse(fmt.Sprintf("the composition input this node composed from is %d bytes, more than the %d an answer on the bus carries", size, maxSentInput))
	}
	if ctx.Err() != nil {
		_, on := run.said(c.now())
		return unready(fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on))
	}
	switch {
	case laidErr != nil && BusFailed(laidErr):
		return unready("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 = c.now().Sub(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 = c.now().Sub(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
	out.origin = c
	return out
}

// warm reads, several at once, each manifest the indexer's modules and the candidate name that the shadow (origin) does
// not hold yet, keeping each in origin and in c; one not read is left to the composition, which reads it and says why.
// Its reads are not noted as the bus failing: the composition that reads again notes its own.
func (c *Composer) warm(ctx context.Context, origin *Composer, cand delivery.Candidate) {
	origin.mu.Lock()
	want := map[string]bool{}
	for _, m := range origin.modules {
		if _, ok := origin.manifests[m.Manifest]; m.Manifest != "" && !ok {
			want[m.Manifest] = true
		}
	}
	for _, pull := range cand.Pulls {
		for _, cm := range pull.Fps {
			if _, ok := origin.manifests[cm.Manifest]; cm.Manifest != "" && !ok {
				want[cm.Manifest] = true
			}
		}
	}
	origin.mu.Unlock()
	if len(want) == 0 {
		return
	}
	defer c.waiting(fmt.Sprintf("%d of the indexer's manifests (%s)", len(want), delivery.ManifestsBucket))()
	slots := make(chan struct{}, warmAtOnce)
	var wg sync.WaitGroup
	for digest := range want {
		key, err := delivery.ManifestsKey(digest)
		if err != nil {
			continue
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			select {
			case slots <- struct{}{}:
			case <-ctx.Done():
				return
			}
			defer func() { <-slots }()
			value, err := c.env.Get(ctx, delivery.ManifestsBucket, key)
			if err != nil || value == nil {
				return
			}
			m, err := delivery.DecodeManifest(value)
			if err != nil {
				return
			}
			c.keepRead(func(k *Composer) { k.manifests[digest] = []byte(m.JSON) })
		}()
	}
	wg.Wait()
}

// 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
}