Hold a refused manifest in the bus reader instead of ending the whole read (ADR 0578 §3)

Issue 1585, S2: one refused manifest silenced disk-load for 45 hours. frombus decodes leniently, takes the
indexer's verdict, stands a refused module in as its stub and names it in Input.Refused; DataItemsHeld names it.

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

go/compose/frombus/frombus.go

@@ -33,6 +33,7 @@ package frombus
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"maps"
 	"net/netip"
@@ -110,8 +111,8 @@ type Mesh struct {
 	OverlayCIDR    string
 	InternalSuffix string
 	Authority      string
-	// Predefined is the list of predefined keys mesh-config published, which a manifest's values falling back to a
-	// predefined key are judged against; nil reads each manifest as the indexer does (catalogue.ParseManifestForInputs).
+	// Predefined is the list of predefined keys mesh-config published. No longer read: a reader decodes a manifest and
+	// takes the indexer's verdict on it, and judges nothing (novox/hq ADR 0578 §3); kept so a caller that sets it builds.
 	Predefined catalogue.PredefinedList
 }
 
@@ -186,15 +187,41 @@ func InOwn(running []delivery.Composed, p delivery.Placement) []string {
 }
 
 // Input is the composition input for s, read through r.
+//
+// **A refused manifest never ends the read** (novox/hq ADR 0578 §3): each manifest is decoded leniently
+// (catalogue.Decode) and the indexer's verdict on it taken from its modules value (delivery.TakeManifestVerdict, a missing
+// one accepted); a module whose manifest the indexer's judge refused, or whose bytes do not decode, stands in the
+// catalogue as its stub (unusable), so what it provides and requires still shapes the rest, and is named with why in
+// the input's Refused; a field set aside is named in its SetAside. Every other module of the read goes on.
 func Input(ctx context.Context, r Reader, mesh Mesh, s Scope) (compose.Input, error) {
 	return input(ctx, r, mesh, s, nil, nil)
 }
 
+// taken is a module's manifest as a reader takes it (novox/hq ADR 0578 §3): decoded leniently, with the indexer's verdict
+// from its modules value. A refused one, or bytes that do not decode, stands as its stub (unusable) and why says so; a
+// manifest not read past a field it set aside holds its name alone and is held the same.
+func taken(name string, raw []byte, m delivery.Module) (man catalogue.Manifest, why string) {
+	if v := m.VerdictOf(name); v != nil {
+		if refused, _ := delivery.TakeManifestVerdict(v); refused {
+			return unusable(name, raw), v.Why()
+		}
+	}
+	man, err := catalogue.Decode(raw)
+	switch {
+	case err != nil:
+		return unusable(name, raw), fmt.Sprintf("the manifest of %s does not decode: %v", name, err)
+	case man.ReadNoFurther():
+		return unusable(name, raw), fmt.Sprintf("the manifest of %s could not be read past a field this reader does not "+
+			"know: %s", name, man.SetAside())
+	}
+	return man, ""
+}
+
 // input is Input; with refuse set, a module placed on the composing node whose build cannot be read or taken in, or whose
-// manifest the mesh refuses, is handed to refuse and composed as a stub (compose.Stub, kept for that node alone)
-// rather than ending the read, so a node's answer names every module it cannot compose (issue 1126, issue 774); a
-// module placed elsewhere whose manifest the mesh refuses is read for its offer alone (unusable). A manifest the bus does
-// not answer for still ends the read: that is the bus's, not the module's.
+// manifest the indexer refused, is also handed to refuse and composed as a stub (compose.Stub, kept for that node alone),
+// so a node's answer names every module it cannot compose (issue 1126, issue 774); a module placed elsewhere whose
+// manifest is refused is read for its offer alone (unusable). A manifest the bus does not answer for still ends the
+// read: that is the bus's, not the module's.
 func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module string, err error), wanted map[string]string) (compose.Input, error) {
 	placements, err := r.Placements(ctx)
 	if err != nil {
@@ -258,7 +285,17 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module
 		if err != nil {
 			return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", module, digest, err)
 		}
-		return parse(raw)
+		man, err := parse(raw)
+		if err != nil {
+			return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s) does not decode: %w", module, digest, err)
+		}
+		return man, nil
+	}
+	named := func(field *map[string]string, module, why string) {
+		if *field == nil {
+			*field = map[string]string{}
+		}
+		(*field)[module] = why
 	}
 	ownPlaced := map[string]bool{}
 	for _, pm := range placements[s.Node].Modules {
@@ -274,19 +311,18 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module
 			// Not read is no refusal of the module: a bus that does not answer is waited out by the caller.
 			return compose.Input{}, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", name, m.Manifest, err)
 		}
-		man, err := parse(raw)
-		if err != nil {
-			if refuse == nil {
-				return compose.Input{}, err
-			}
-			// One module's manifest the mesh refuses refuses that module alone, by the rule a build that cannot be read
-			// follows (issue 774: power's bar item refused every node's composition, 2026-10-30). It stands in as its
-			// unjudged stub, so what it provides and requires still shapes the rest; it is held on the composing node
-			// where it is placed there, and elsewhere only its offer is read.
-			man = unusable(name, raw)
-			if ownPlaced[name] {
-				refuse(name, err)
+		// One module's manifest the indexer refused refuses that module alone, by the rule a build that cannot be read
+		// follows (issue 774: power's bar item refused every node's composition, 2026-10-30; ADR 0578 §3: one refused
+		// manifest silenced disk-load for 45 hours). It stands in as its stub, so what it provides and requires still
+		// shapes the rest; it is held on the composing node where it is placed there, and elsewhere only its offer is read.
+		man, why := taken(name, raw, m)
+		if why != "" {
+			named(&in.Refused, name, why)
+			if refuse != nil && ownPlaced[name] {
+				refuse(name, errors.New(why))
 			}
+		} else if aside := man.SetAside(); aside != "" {
+			named(&in.SetAside, name, aside)
 		}
 		in.Catalogue[name] = man
 	}
@@ -595,14 +631,9 @@ func unusable(module string, raw []byte) catalogue.Manifest {
 	return compose.Stub(m)
 }
 
-// parser reads a manifest as the indexer does, against the mesh's predefined keys when given.
-func parser(mesh Mesh) func([]byte) (catalogue.Manifest, error) {
-	return func(raw []byte) (catalogue.Manifest, error) {
-		if mesh.Predefined != nil {
-			return catalogue.ParseManifestWith(raw, mesh.Predefined)
-		}
-		return catalogue.ParseManifestForInputs(raw)
-	}
+// parser reads a manifest as every reader does (novox/hq ADR 0578 §3): decoded leniently, judged by nothing here.
+func parser(Mesh) func([]byte) (catalogue.Manifest, error) {
+	return catalogue.Decode
 }
 
 // ranOf is what a module last offered the others, read when it is held (composeEach, issue 1960): its manifest at the
@@ -723,6 +754,9 @@ func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p del
 			// it runs (Scope.InOwn).
 			continue
 		}
+		if _, refused := in.Refused[pm.Module]; refused {
+			continue // held as its stub: no build of a refused manifest is taken in (ADR 0578 §3)
+		}
 		if man, ok := in.Catalogue[pm.Module]; !ok || (man.Build == nil && !man.Templated()) {
 			continue // nothing is built of it: its manifest is what it composes from
 		}
@@ -1007,11 +1041,25 @@ func membershipOf(own delivery.Composed) string {
 }
 
 // DataItems is every data item the modules placed on scope's node declare, placed there, composed from the bus as Plans
-// composes (novox/hq ADR 0571 clause 3): disk-load measures these.
+// composes (novox/hq ADR 0571 clause 3): disk-load measures these. A module whose manifest the indexer refused is held
+// and its items left out; DataItemsHeld names it, and a reader that must say what it leaves out reads that.
 func DataItems(ctx context.Context, r Reader, mesh Mesh, s Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
-	in, err := Input(ctx, r, mesh, s)
+	items, unplaced, _, err := DataItemsHeld(ctx, r, mesh, s)
+	return items, unplaced, err
+}
+
+// DataItemsHeld is DataItems and, by module, why each module placed on scope's node is held: its manifest refused by
+// the indexer's judge or not decoded (compose.Input.Refused). One refused manifest never ends the read (novox/hq ADR
+// 0578 §3: disk-load measured nothing for 45 hours when one did).
+func DataItemsHeld(ctx context.Context, r Reader, mesh Mesh, s Scope) (items map[string][]catalogue.PlacedDataItem, unplaced []string, held map[string]string, err error) {
+	held = map[string]string{}
+	in, err := input(ctx, r, mesh, s, func(module string, err error) { held[module] = err.Error() }, nil)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, held, err
 	}
-	return compose.DataItems(in, s.Node)
+	items, unplaced, err = compose.DataItems(in, s.Node)
+	for m := range held {
+		delete(items, m)
+	}
+	return items, unplaced, held, err
 }

go/compose/frombus/frombus_test.go

@@ -142,6 +142,7 @@ func TestInput(t *testing.T) {
 	t.Run("no store reached says why", noStoreReachedSaysWhy)
 	// Issue 774: a node refused whole says what it held on the way.
 	t.Run("a node refused whole says what it held", aNodeRefusedWholeSaysWhatItHeld)
+	t.Run("a refused manifest never ends the read", aRefusedManifestNeverEndsTheRead)
 	t.Run("the runtime is blamed only where its stub composes", theRuntimeIsBlamedOnlyWhereItsStubComposes)
 	t.Run("a running module waits for its build", aRunningModuleWaitsForItsBuild)
 	t.Run("a held module's first reason leads", aHeldModulesFirstReasonLeads)
@@ -194,11 +195,11 @@ func TestInput(t *testing.T) {
 	if err := json.Unmarshal(raw, &private); err != nil {
 		t.Fatal(err)
 	}
-	parsed, err := catalogue.ParseManifestForInputs(xterm)
+	parsed, err := catalogue.Decode(xterm)
 	if err != nil {
 		t.Fatal(err)
 	}
-	parsedMedia, err := catalogue.ParseManifestForInputs(media)
+	parsedMedia, err := catalogue.Decode(media)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -549,7 +550,7 @@ func TestInput(t *testing.T) {
 	toolsRef := "artifact-store://xterm/tools/blobs/sha256:" + strings.Repeat("1", 64)
 	kept.builds = map[string]*delivery.BuilderBuild{"xterm " + oldFP: {State: delivery.BuildDone, Manifest: "old",
 		Artifacts: []delivery.BuiltArtifact{{Name: "tools", Kind: "bundle", Reference: toolsRef, Digest: "sha256:" + strings.Repeat("1", 64)}}}}
-	parsedOld, err := catalogue.ParseManifestForInputs(oldXterm)
+	parsedOld, err := catalogue.Decode(oldXterm)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -672,7 +673,7 @@ func TestInput(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	clientParsed, err := catalogue.ParseManifestForInputs(clientRaw)
+	clientParsed, err := catalogue.Decode(clientRaw)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -696,7 +697,7 @@ func TestInput(t *testing.T) {
 		}
 		// Kept as its bytes and read again, as the controller's store holds a build's manifest.
 		raw, _ := json.Marshal(resolved)
-		if storeIn.Catalogue[module], err = catalogue.ParseManifestForInputs(raw); err != nil {
+		if storeIn.Catalogue[module], err = catalogue.Decode(raw); err != nil {
 			t.Fatal(err)
 		}
 	}

go/compose/frombus/frombustest/frombustest_test.go

@@ -26,16 +26,21 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
 		why   string // words backup's hold says
 		// elsewhere leaves backup in the indexer's modules and places it on no node: nothing is held.
 		elsewhere bool
+		// refused is the indexer's verdict on bad, a problem its judge found (novox/hq ADR 0578 §1): Plans then holds
+		// backup too, and never ends the read.
+		refused string
 	}{
 		{name: "every module composes"},
 		{name: "a built manifest that cannot be used",
 			bad:   `{"module":"backup","version":"1","build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`,
-			built: `{"module":"backup","version":"1","requires":["amqp"]}`, why: "backup"},
+			built: `{"module":"backup","version":"1","requires":"amqp"}`, why: "backup"},
 		{name: "a plan that refuses",
 			bad: `{"module":"backup","version":"1","resources":[{"id":"conf","type":"file","path":"/etc/backup.conf",` +
 				`"mode":"0600","content":"key=${secret:nope}"}]}`, why: "${secret:nope}"},
-		{name: "an indexer's manifest the mesh refuses", bad: unusableItem, why: `"shows" is none of them`},
-		{name: "an indexer's manifest the mesh refuses, placed elsewhere", bad: unusableItem, elsewhere: true},
+		{name: "an indexer's manifest the mesh refuses", bad: unusableItem, why: `"shows" is none of them`,
+			refused: `"shows" is none of them`},
+		{name: "an indexer's manifest the mesh refuses, placed elsewhere", bad: unusableItem, elsewhere: true,
+			refused: `"shows" is none of them`},
 	} {
 		b, err := Provisioned()
 		if err != nil {
@@ -50,13 +55,21 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
 			digest := delivery.ManifestDigest([]byte(row.bad))
 			b.Manifests[digest] = []byte(row.bad)
 			b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest}
+			if row.refused != "" {
+				b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest, Verdict: delivery.VerdictRefused,
+					Problems: []string{row.refused}}
+			}
 			if row.built != "" {
 				builtDigest := delivery.ManifestDigest([]byte(row.built))
 				b.Manifests[builtDigest] = []byte(row.built)
 				b.Builds["backup "+fp] = &delivery.BuilderBuild{State: delivery.BuildDone, Manifest: builtDigest,
 					Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://backup/files/blobs/x", Digest: "sha256:x"}}}
 			}
-			if _, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}); err == nil && !row.elsewhere {
+			plans, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
+			switch {
+			case row.refused != "" && (err != nil || !planned(plans, "restic") || planned(plans, "backup")):
+				t.Errorf("%s: Plans did not hold the refused manifest alone: %v %v", row.name, plans, err)
+			case row.refused == "" && err == nil && !row.elsewhere:
 				t.Errorf("%s: Plans composed the node; the row does not refuse it", row.name)
 			}
 		}
@@ -101,6 +114,8 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
 	}
 }
 
+func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok }
+
 // unusableItem is a manifest the mesh refuses as the node-engine did power's at mesh-catalog ca8d6f33: a bar block item
 // with attributes its bar places by none of.
 const unusableItem = `{"module":"backup","version":"1","moved":true,` +

go/compose/frombus/held_rows_test.go

@@ -37,6 +37,7 @@ func aNodeRefusedWholeSaysWhatItHeld(t *testing.T) {
 	refused := `{"module":"power","version":"1","items":[{"collection":"contribution:node-bar:block","attributes":{"shows":"state"},"template":"files/a"}]}`
 	needs := `{"module":"lamp","version":"1","requires":["nothing-provides-this"]}`
 	b, fps := aNode(t, refused, needs)
+	refuse(b, "power", `"shows" is none of them`)
 	_, _, held, err := Compose(context.Background(), b, Mesh{}, Scope{Node: "n1"}, fps)
 	if err == nil || !strings.Contains(err.Error(), "nothing provides") || !strings.Contains(err.Error(), `held before: power: `) ||
 		!strings.Contains(err.Error(), `"shows" is none of them`) {
@@ -44,6 +45,48 @@ func aNodeRefusedWholeSaysWhatItHeld(t *testing.T) {
 	}
 }
 
+// refuse writes the indexer's refusal of module's manifest into its modules value (novox/hq ADR 0578 §1).
+func refuse(b *bus, module string, problems ...string) {
+	m := b.modules[module]
+	m.Verdict, m.Problems = delivery.VerdictRefused, problems
+	b.modules[module] = m
+}
+
+// ADR 0578 §3: the bus reader never ends a whole read on a refused manifest. Input, Plans and DataItems compose the rest
+// and hold the refused module, named with the indexer's words in Refused; a field the decoder does not know is set aside
+// and named, and its module composed.
+func aRefusedManifestNeverEndsTheRead(t *testing.T) {
+	ok := `{"module":"lamp","version":"1","resources":[{"id":"d","type":"directory","path":"/srv/lamp","mode":"0700"}],` +
+		`"data":{"own":[{"id":"d","path":"${dir:d}","class":"valuable"}]}}`
+	bad := `{"module":"power","version":"1","resources":[{"id":"d","type":"directory","path":"/srv/power","mode":"0700"}],` +
+		`"data":{"own":[{"id":"d","path":"${dir:d}","class":"valuable"}]}}`
+	later := `{"module":"later","version":"1","a-field-from-later":1}`
+	b, _ := aNode(t, ok, bad, later)
+	refuse(b, "power", "removes a stream it may not")
+	ctx := context.Background()
+	in, err := Input(ctx, b, Mesh{}, Scope{Node: "n1"})
+	if err != nil {
+		t.Fatalf("one refused manifest ended the read: %v", err)
+	}
+	if why := in.Refused["power"]; !strings.Contains(why, "refused the manifest of power: removes a stream it may not") ||
+		len(in.Refused) != 1 {
+		t.Fatalf("refused: %v", in.Refused)
+	}
+	if aside := in.SetAside["later"]; !strings.Contains(aside, "a-field-from-later") {
+		t.Fatalf("set aside: %v", in.SetAside)
+	}
+	if !in.Catalogue["power"].Stubbed || in.Catalogue["lamp"].Stubbed || in.Catalogue["later"].Module != "later" {
+		t.Fatalf("the catalogue: %+v", in.Catalogue)
+	}
+	if _, err := Plans(ctx, b, Mesh{}, Scope{Node: "n1"}); err != nil {
+		t.Fatalf("plans: %v", err)
+	}
+	items, _, held, err := DataItemsHeld(ctx, b, Mesh{}, Scope{Node: "n1"})
+	if err != nil || len(items["lamp"]) != 1 || items["power"] != nil || !strings.Contains(held["power"], "removes a stream") {
+		t.Fatalf("data items %v held %v: %v", items, held, err)
+	}
+}
+
 // Issue 774: the culprit rule takes a module whose stub composes or fails otherwise; the runtime only where its stub
 // composes, since its stub always fails otherwise.
 func theRuntimeIsBlamedOnlyWhereItsStubComposes(t *testing.T) {

go/compose/input.go

@@ -101,6 +101,15 @@ type Input struct {
 	// declaration carries to a node-engine that reads them.
 	CurrentBuilds map[string]string
 	BuildIDs      map[string]map[string]string
+
+	// Refused is each module whose manifest the indexer's judge refused, or whose bytes do not decode, by module, with why
+	// (novox/hq ADR 0578 §3, §5): its verdict, carried in the plain input. Such a module stands in Catalogue as its stub
+	// (compose.Stub), so it is held: nothing of its own is composed, while what it provides and requires still shapes the
+	// rest. A module not named is accepted, a missing verdict included.
+	Refused map[string]string `json:",omitempty"`
+	// SetAside is each module whose manifest has a field the reader's decoder does not know, by module, with what was set
+	// aside (catalogue.Manifest.SetAside): the rest of it is composed, and the reader names it (ADR 0578 §6).
+	SetAside map[string]string `json:",omitempty"`
 }
 
 // Issued is one pair credential made for a consumer of a provider (inv.Secret): ForProvider is its copy sealed to the