Merge pull request 'The builder says it writes references only once its build modules proved it (issue 1886, ADR 0522)' (#23) from fix/1886-builder-says-references into main

Reviewed-on: #23

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go

@@ -7,6 +7,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -155,6 +156,8 @@ type bed struct {
 	tools      map[string]jetstream.KeyValue
 	presence   jetstream.KeyValue
 	have       jetstream.KeyValue
+	instances  jetstream.KeyValue
+	stop       func() // stops the builder started last
 	fake       *fakeBuildModules
 	srvURL     string
 	dir        string
@@ -227,6 +230,7 @@ func startBed(t *testing.T) *bed {
 	b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
 	presence := mk(delivery.PresenceBucket, delivery.PresenceHistory)
 	b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory)
+	b.instances = mk(instancesBucket, 1)
 	for _, node := range []string{"n1", "n2"} {
 		b.setOnline(t, node, true)
 		for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
@@ -344,7 +348,9 @@ func (b *bed) start(t *testing.T, node string, env map[string]string) {
 	runCtx, stop := context.WithCancel(b.ctx)
 	done := make(chan error, 1)
 	go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }()
-	t.Cleanup(func() { stop(); <-done })
+	var once sync.Once
+	b.stop = func() { once.Do(func() { stop(); <-done }) }
+	t.Cleanup(b.stop)
 }
 
 // await reads the build key until pred holds of it, within 20 s.
@@ -713,3 +719,26 @@ func TestAManifestFallingBackToAPredefinedKeyIsBuilt(t *testing.T) {
 		t.Errorf("build = %+v, want done: the manifest is read", v)
 	}
 }
+
+// 1886/AC5: a builder on n1 whose Go, Node and image build modules each answered a reference says under n1, in its bucket
+// of instances, that it writes references, and takes the say away when it stops: the controller delivers from its keys
+// only while it holds.
+func TestABuilderWhoseBuildModulesAnsweredReferencesSaysItAndStoppedSaysNothing(t *testing.T) {
+	b := startBed(t)
+	b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")              // a Go and a TypeScript bundle
+	b.moduleCommit(t, "imaged", "catalogue-a67f0196-imaged.module.json", "modules/imaged", nil, commit) // an image, for the trunk
+	b.start(t, "n1", nil)
+	var said string
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && said == ""; time.Sleep(50 * time.Millisecond) {
+		if e, err := b.instances.Get(b.ctx, "n1"); err == nil {
+			said = string(e.Value())
+		}
+	}
+	if !strings.Contains(said, `"writes":["references"]`) {
+		t.Fatalf("the builder said %q", said)
+	}
+	b.stop()
+	if _, err := b.instances.Get(b.ctx, "n1"); !errors.Is(err, jetstream.ErrKeyNotFound) {
+		t.Fatalf("the stopped builder still says: %v", err)
+	}
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

@@ -178,6 +178,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Manifests:  manifestReader{manifests},
 		Logf:       logger.Printf,
 	}
+	// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
+	// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
+	// yet (the manifest's state not made) is said, and the controller keeps asking build-agent.
+	switch kv, err := js.KeyValue(ctx, instancesBucket); {
+	case errors.Is(err, jetstream.ErrBucketNotFound):
+		logger.Printf("the bucket %s does not exist yet: the controller is not told that this builder writes references until it starts again", instancesBucket)
+	case err != nil:
+		return fmt.Errorf("the bucket %s: %w", instancesBucket, err)
+	default:
+		says := newSayer(ctx, kv, cred.Node, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, build.ImageBuildModule}, logger.Printf)
+		b.Answered = says.Answered
+		defer func() {
+			stopping, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+			defer cancel()
+			if err := says.unsay(stopping); err != nil {
+				logger.Printf("%v", err)
+			}
+		}()
+	}
 	// The toolchains first, so a build that starts from the first modules value finds the nodes.
 	watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
 	if err := watch.start(); err != nil {

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// What the builder says it writes (novox/hq issue 1886, ADR 0522): each running instance keeps a key, named by its node,
+// in its own bucket of instances (`mesh-delivery-builder_instances`, the manifest's state `instances`), saying that every
+// done key it writes names where its artifacts are (ADR 0498 §1). The references come from the build modules' own pushes,
+// so an instance says it only once a step of every build module it calls answered one since it started: a builder beside
+// a build module from before R4e says nothing. The controller delivers from the builder's keys only while an instance
+// says so, and asks build-agent as before R4e otherwise. The bucket is read by the controller alone, so its name and shape
+// are not in mesh-sdk.
+const (
+	instancesBucket  = "mesh-delivery-builder_instances"
+	writesReferences = `{"v":1,"writes":["references"]}`
+)
+
+// sayStore is the bucket of instances as the builder writes it.
+type sayStore interface {
+	Put(ctx context.Context, key string, value []byte) (uint64, error)
+	Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error
+}
+
+// sayer puts this instance's say once every build module in need answered a reference, and takes it away at the stop.
+type sayer struct {
+	ctx   context.Context
+	store sayStore
+	node  string
+	need  []string
+	logf  func(string, ...any)
+
+	mu       sync.Mutex
+	answered map[string]bool
+	said     bool
+}
+
+func newSayer(ctx context.Context, store sayStore, node string, need []string, logf func(string, ...any)) *sayer {
+	return &sayer{ctx: ctx, store: store, node: node, need: need, logf: logf, answered: map[string]bool{}}
+}
+
+// Answered is told the build module whose step answered a reference (build.Builder.Answered).
+func (s *sayer) Answered(module string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.answered[module] = true
+	if s.said {
+		return
+	}
+	for _, m := range s.need {
+		if !s.answered[m] {
+			return
+		}
+	}
+	if _, err := s.store.Put(s.ctx, s.node, []byte(writesReferences)); err != nil {
+		s.logf("saying in %s that this builder writes references: %v; said again at the next reference", instancesBucket, err)
+		return
+	}
+	s.said = true
+	s.logf("every build module answered a reference: said in %s that this builder writes references", instancesBucket)
+}
+
+// unsay takes the say away, if it was made: a builder replaced by one that does not write references leaves none.
+func (s *sayer) unsay(ctx context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if !s.said {
+		return nil
+	}
+	if err := s.store.Delete(ctx, s.node); err != nil {
+		return fmt.Errorf("taking away the say in %s: %w", instancesBucket, err)
+	}
+	s.said = false
+	return nil
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says_test.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+type saysOf map[string]string
+
+func (s saysOf) Put(_ context.Context, key string, value []byte) (uint64, error) {
+	s[key] = string(value)
+	return 1, nil
+}
+
+func (s saysOf) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
+	delete(s, key)
+	return nil
+}
+
+// 1886/AC5: a builder says, under its node, that it writes references only once every build module it calls answered
+// one; stopped, it says nothing any more.
+func TestTheBuilderSaysItWritesReferencesOnceEveryBuildModuleAnsweredOne(t *testing.T) {
+	s := saysOf{}
+	say := newSayer(t.Context(), s, "shanks", []string{"mesh-build-go", "mesh-build-node", "mesh-build-image"}, t.Logf)
+	say.Answered("mesh-build-go")
+	say.Answered("mesh-build-node")
+	if v, ok := s["shanks"]; ok {
+		t.Fatalf("said before the image build module answered a reference: %q", v)
+	}
+	say.Answered("mesh-build-image")
+	var v struct {
+		Writes []string `json:"writes"`
+	}
+	if err := json.Unmarshal([]byte(s["shanks"]), &v); err != nil || len(v.Writes) != 1 || v.Writes[0] != "references" {
+		t.Fatalf("the builder's say: %q %v", s["shanks"], err)
+	}
+	if err := say.unsay(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := s["shanks"]; ok {
+		t.Fatalf("the stopped builder still says: %v", s)
+	}
+}

modules/mesh-delivery-builder/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-builder
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-builder/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-builder/internal/build/r4e_test.go

@@ -383,3 +383,21 @@ func TestACandidatesFetchNamesTheNamedPathsTheIndexerGaveIt(t *testing.T) { // A
 		t.Errorf("fetch request = %+v, %v; want the candidate's base, the module's directory and its named paths", req, err)
 	}
 }
+
+// 1886/AC5: the builder tells who answered a reference: the build module whose step named one, and none for a step that
+// answered a digest alone (a build module from before R4e).
+func TestTheBuilderTellsWhichBuildModuleAnsweredAReference(t *testing.T) {
+	r := newRig(t, goBundle)
+	var answered []string
+	r.b.Answered = func(module string) { answered = append(answered, module) }
+	r.bm.noRef["ace/tool"] = true
+	r.consider(t, "modules/m")
+	if len(answered) != 0 {
+		t.Fatalf("a step that answered no reference was told as one: %v", answered)
+	}
+	delete(r.bm.noRef, "ace/tool")
+	r.consider(t, "modules/m")
+	if len(answered) != 1 || answered[0] != "mesh-build-go" {
+		t.Fatalf("the build module that answered a reference: %v, want mesh-build-go", answered)
+	}
+}

modules/mesh-delivery-builder/internal/build/run.go

@@ -45,6 +45,9 @@ type Builder struct {
 	// Limit is how many builds run at once; four when 0.
 	Limit int
 	Logf  func(format string, args ...any)
+	// Answered is told the build module whose step answered a reference (issue 1886): what the builder says it writes
+	// rests on it. Nil tells nobody.
+	Answered func(buildModule string)
 
 	once     sync.Once
 	sem      chan struct{}
@@ -749,6 +752,9 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del
 		if built.Reference != "" && (stored(t.Kind) || !candidate) {
 			l.references[t.Artifact] = built.Reference
 		}
+		if built.Reference != "" && b.Answered != nil {
+			b.Answered(l.module)
+		}
 	}
 }
 

modules/mesh-delivery-builder/manifest_test.go

@@ -83,7 +83,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",

modules/mesh-delivery-builder/module.json

@@ -7,7 +7,8 @@
     "mesh-build-node.toolchain.reported"
   ],
   "state": [
-    { "name": "builds", "history": 5 }
+    { "name": "builds", "history": 5 },
+    { "name": "instances", "history": 1 }
   ],
   "reads": [
     "mesh-build-go.steps",

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
+	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
 	"mesh-delivery-checker/internal/check"
@@ -651,3 +652,32 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
 		}
 	}
 }
+
+// TestTheCheckerSaysOwedReadableOnTheBus: started, the checker says on its own states subject what it keeps, owed among
+// the readable ones, which is what grants the forge module its read of owed (ADR 0495 §2-§3, ADR 0521; R4e finding R2-1).
+func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
+	b := startBed(t)
+	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	b.start(t, nil)
+	events, err := b.js.Stream(b.ctx, "EVENTS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var said announce.States
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		m, err := events.GetLastMsgForSubject(b.ctx, announce.Subject("mesh-delivery-checker"))
+		if err != nil {
+			continue
+		}
+		if err := json.Unmarshal(m.Data, &said); err != nil {
+			t.Fatal(err)
+		}
+		break
+	}
+	if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") ||
+		!said.Readable("previews") || !said.Readable("verdicts") || said.Readable("kept") || said.Readable("tries") || len(said.States) != 5 {
+		t.Fatalf("the checker said %+v, want its five states with owed, previews and verdicts readable", said)
+	}
+}

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/kept_test.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"sort"
+	"testing"
+)
+
+// The checker says on the bus which states it keeps and which others may read (novox/hq ADR 0495 clauses 2 and 3): a
+// reader is granted from that announcement. owed is read by the forge module, which writes mesh/check from it (ADR 0521,
+// R4e round 2 finding R2-1), and by the forge bridge; previews by the indexer; verdicts by the gate. The list names every
+// state of the manifest, so no state is refused at the call for want of being announced (clause 5).
+func TestTheCheckerAnnouncesEveryStateItKeepsAndOwedReadable(t *testing.T) { // R2-1, ADR 0495 §2-§3, ADR 0521
+	raw, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m struct {
+		State []struct{ Name string } `json:"state"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	var manifest, announced, readable []string
+	for _, s := range m.State {
+		manifest = append(manifest, s.Name)
+	}
+	for _, s := range keptStates {
+		announced = append(announced, s.Name)
+		if s.Readable {
+			readable = append(readable, s.Name)
+		}
+	}
+	sort.Strings(manifest)
+	sort.Strings(announced)
+	if !equal(manifest, announced) {
+		t.Errorf("announced %v, the manifest keeps %v", announced, manifest)
+	}
+	if want := []string{"owed", "previews", "verdicts"}; !equal(readable, want) {
+		t.Errorf("readable %v, want %v", readable, want)
+	}
+}
+
+func equal(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go

@@ -25,6 +25,7 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
+	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -113,6 +114,16 @@ func durationEnv(getenv func(string) string, name string, def time.Duration) (ti
 }
 
 // run is the program: it returns when ctx ends or the bus cannot be used.
+// keptStates is every state the checker keeps and which others may read (ADR 0495 §2-§3): owed by the forge module,
+// which writes mesh/check from it (ADR 0521), and by the forge bridge; previews by the indexer; verdicts by the gate.
+var keptStates = []announce.State{
+	{Name: "kept"},
+	{Name: "owed", Readable: true},
+	{Name: "previews", Readable: true},
+	{Name: "tries"},
+	{Name: "verdicts", Readable: true},
+}
+
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags)
 	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
@@ -167,6 +178,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	defer caller.Close()
+	// What the checker keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot be said
+	// now: process.Conn says it again at every reconnect.
+	if err := caller.Keeps(keptStates...); err != nil {
+		logger.Printf("what the checker keeps could not be said on the bus, so readers of owed, previews and verdicts are not granted until it is: %v", err)
+	}
 
 	stepBound, err := durationEnv(getenv, "MESH_CHECKER_STEP_BOUND", defaultStepBound)
 	if err != nil {

modules/mesh-delivery-checker/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-checker
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-checker/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go

@@ -6,9 +6,7 @@ import (
 	"fmt"
 	"log"
 	"strconv"
-	"time"
 
-	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -58,20 +56,8 @@ type performer struct {
 	pub    publisher
 	done   revisions
 	logger *log.Logger
-	// emit publishes the bridge's own events (check.status), and node is the machine said in their headers.
-	emit emitter
-	node string
 }
 
-// emitter publishes a message with its headers: the bus's JetStream.
-type emitter interface {
-	PublishMsg(ctx context.Context, m *nats.Msg, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error)
-}
-
-// checkStatusSubject is where the bridge emits check.status: under its own name, which the bus lets its account alone
-// publish (mesh-control internal/broker/nats.go, a module's own namespace).
-const checkStatusSubject = "mesh.mod." + delivery.ForgeBridgeModule + ".event." + bridge.CheckStatusEvent
-
 // seen says whether the revision of the key was acted on already. A revision that cannot be read is not seen, so the
 // act is done: a status set twice is the same status.
 func (p *performer) seen(ctx context.Context, doneKey string, rev uint64) bool {
@@ -110,46 +96,6 @@ func (p *performer) owed(ctx context.Context, key string, rev uint64, value []by
 	p.perform(ctx, "owed."+key, key, rev, value)
 }
 
-// checkOwed performs the status the checker owes on a head (mesh-delivery-checker_owed; novox/hq ADR 0484 clause 2) at
-// revision rev, once. Its keys are kept apart from the gate's, and a key that is not the checker's one act, the status
-// mesh/check, is left.
-func (p *performer) checkOwed(ctx context.Context, key string, rev uint64, value []byte) {
-	doneKey := "check." + key
-	if p.seen(ctx, doneKey, rev) {
-		return
-	}
-	if _, _, _, err := delivery.ParseCheckerOwedKey(key); err != nil {
-		p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
-		p.keep(ctx, doneKey, rev)
-		return
-	}
-	// mesh/check is emitted on the bridge's own subject, never written through a tool (R4e finding K1): the forge module
-	// consumes check.status and writes it. The event's id is the bus's dedup id, so a second emit of a revision is one.
-	id, body, err := bridge.CheckStatusMessage(key, rev, value)
-	if err != nil {
-		p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
-		p.keep(ctx, doneKey, rev)
-		return
-	}
-	if p.emit == nil {
-		p.logger.Printf("%s at revision %d: nothing to emit %s with, so it is not kept and a restart emits it", key, rev, bridge.CheckStatusEvent)
-		return
-	}
-	m := nats.NewMsg(checkStatusSubject)
-	m.Header.Set("x-event-id", id)
-	m.Header.Set("Nats-Msg-Id", id)
-	m.Header.Set("x-source", delivery.ForgeBridgeModule)
-	m.Header.Set("x-node", p.node)
-	m.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
-	m.Header.Set("content-type", "application/json")
-	m.Data = body
-	if _, err := p.emit.PublishMsg(ctx, m); err != nil {
-		p.logger.Printf("%s at revision %d: %s could not be emitted, and is emitted again when the key is seen again: %v", key, rev, bridge.CheckStatusEvent, err)
-		return
-	}
-	p.keep(ctx, doneKey, rev)
-}
-
 // perform does the act of an owed key once for the revision, under the bookkeeping key doneKey.
 func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) {
 	if p.seen(ctx, doneKey, rev) {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go

@@ -9,9 +9,6 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nats.go/jetstream"
-
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
@@ -211,104 +208,3 @@ func TestAMergeWithTwoHandedStopsAtTheFirstRefusal(t *testing.T) { // ADR 0441
 		t.Errorf("calls %d messages %d, want the first refusal to stop the pass", len(f.calls), len(pub.data))
 	}
 }
-
-// checkerOwed is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed).
-func checkerOwed(t *testing.T, number int, state, text string) (string, []byte) {
-	t.Helper()
-	key, err := delivery.CheckerOwedKey("novox/mesh-delivery", number, headA)
-	if err != nil {
-		t.Fatal(err)
-	}
-	v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
-	if err != nil {
-		t.Fatal(err)
-	}
-	return key, v
-}
-
-// fakeEmit keeps what the performer emits on its own subjects.
-type fakeEmit struct {
-	msgs []*nats.Msg
-	err  error
-}
-
-func (e *fakeEmit) PublishMsg(_ context.Context, m *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
-	if e.err != nil {
-		return nil, e.err
-	}
-	e.msgs = append(e.msgs, m)
-	return &jetstream.PubAck{}, nil
-}
-
-// R4e finding K1 (issue 1301 S4): the checker's status is emitted as check.status on the bridge's own subject, never a
-// tool call, once per revision of its owed key (1683/AC5, ADR 0484 clause 2).
-func TestTheCheckersStatusIsEmittedOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // K1, 1683/AC5
-	f := &forgeFake{}
-	p, _, done, _ := newPerformer(f)
-	em := &fakeEmit{}
-	p.emit, p.node = em, "n1"
-	key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb")
-	p.checkOwed(context.Background(), key, 3, v)
-	p.checkOwed(context.Background(), key, 3, v)
-	if len(f.calls) != 0 {
-		t.Fatalf("a tool was called for mesh/check: %+v", f.calls)
-	}
-	if len(em.msgs) != 1 {
-		t.Fatalf("emitted %d, want 1", len(em.msgs))
-	}
-	m := em.msgs[0]
-	if m.Subject != "mesh.mod.mesh-delivery-forge-bridge.event.check.status" {
-		t.Errorf("subject = %q", m.Subject)
-	}
-	if m.Header.Get("x-source") != "mesh-delivery-forge-bridge" || m.Header.Get("x-node") != "n1" || m.Header.Get("x-time") == "" ||
-		m.Header.Get("content-type") != "application/json" || m.Header.Get("x-event-id") != "check."+key+".3" || m.Header.Get("Nats-Msg-Id") != "check."+key+".3" {
-		t.Errorf("headers = %v", m.Header)
-	}
-	if want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","target_url":"https://view.example/checks/g.0123456789ab","key":"` + key + `","revision":3}`; string(m.Data) != want {
-		t.Errorf("body = %s", m.Data)
-	}
-	p.checkOwed(context.Background(), key, 4, v)
-	if len(em.msgs) != 2 {
-		t.Errorf("a new revision was not emitted: %d", len(em.msgs))
-	}
-	if _, ok := done["check."+key]; !ok {
-		t.Errorf("revision kept under %v, want check.%s", done, key)
-	}
-	// The gate's own owed bookkeeping is another key: the same owed key seen in the gate's bucket is its own act.
-	if _, ok := done["owed."+key]; ok {
-		t.Errorf("the checker's status was kept under the gate's key: %v", done)
-	}
-}
-
-func TestACheckStatusTheBusDidNotTakeIsNotKeptAsDone(t *testing.T) { // K1
-	p, _, done, logs := newPerformer(&forgeFake{})
-	em := &fakeEmit{err: errors.New("nats: timeout")}
-	p.emit = em
-	key, v := checkerOwed(t, 12, "pending", "checking")
-	p.checkOwed(context.Background(), key, 5, v)
-	if _, ok := done["check."+key]; ok || !strings.Contains(logs.String(), "nats: timeout") {
-		t.Fatalf("kept %v, log %q: a status the bus did not take is said and not kept", done, logs.String())
-	}
-	em.err = nil
-	p.checkOwed(context.Background(), key, 5, v)
-	if len(em.msgs) != 1 {
-		t.Errorf("the same revision was not emitted again after the failure: %d", len(em.msgs))
-	}
-}
-
-func TestAnythingButTheCheckersStatusInItsBucketIsLeft(t *testing.T) { // ADR 0484 clause 2: the checker owes one act
-	f := &forgeFake{}
-	p, _, done, logs := newPerformer(f)
-	// A comment, and a status of another context: the SDK refuses both to the checker's encoder, so they are written raw.
-	commentKey, commentValue := owedKV(t, "novox/mesh-delivery", 12, "comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"})
-	gateKey, gateValue := owedKV(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
-		delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"})
-	p.checkOwed(context.Background(), commentKey, 1, commentValue)
-	p.checkOwed(context.Background(), gateKey, 1, gateValue)
-	if len(f.calls) != 0 {
-		t.Errorf("the forge was called: %+v", f.calls)
-	}
-	if !strings.Contains(logs.String(), commentKey) || !strings.Contains(logs.String(), gateKey) || len(done) != 2 {
-		t.Errorf("log %q, kept %v: both keys are named and kept so they are not tried again", logs.String(), done)
-	}
-}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go

@@ -30,15 +30,8 @@ import (
 // The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the
 // bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342).
 //
-// The checker's owed bucket is read like the gate's (reads) and the checker's stream checks is heard like EVENTS is (consumes
-// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller
-// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file.
-//
-// The user n1.node-tools is the machine's tool runner, whose account every agent session on the machine shares, carrying the
-// forge module (mesh-control internal/broker users.go KindNodeTools): it invokes every tool (node-tools' invokes "*"),
-// publishes gitea's own events and hears gitea's consumer on EVENTS, n1_node-tools, which binds the bridge's check.status
-// (gitea's consumes). Written by analogy with what the controller composes, as the checker's entries are; it is never
-// granted the bridge's namespace, which is the point of R4e finding K1.
+// The bridge neither reads the checker's owed nor hears its stream (ADR 0521, R4e finding R2-1): the forge module writes
+// mesh/check from it. The stream checks is made here so a test can say the checker wrote a status.
 //
 // The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal
 // with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls,
@@ -114,15 +107,6 @@ accounts {
             "mesh.mod.gitea.tool.gitea_add_comment"
             "mesh.mod.gitea.tool.gitea_add_comment.*"
             "mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
-            "mesh.mod.mesh-delivery-forge-bridge.event.check.status"
-            "$JS.ACK.CHECKS.n1_mesh-delivery-forge-bridge.>"
-            "$JS.API.CONSUMER.INFO.CHECKS.n1_mesh-delivery-forge-bridge"
-            "$JS.API.CONSUMER.MSG.NEXT.CHECKS.n1_mesh-delivery-forge-bridge"
-            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-checker_owed.>"
-            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-checker_owed.>"
-            "$JS.API.DIRECT.GET.KV_mesh-delivery-checker_owed.>"
-            "$JS.API.STREAM.INFO.KV_mesh-delivery-checker_owed"
-            "$JS.FC.KV_mesh-delivery-checker_owed.>"
           ] }
           subscribe { allow: [
             "_INBOX.n1.mesh-delivery-forge-bridge.>"
@@ -136,24 +120,6 @@ accounts {
           allow_responses: true
         }
       }
-      { user: "n1.node-tools", password: "pw"
-        permissions {
-          publish { allow: [
-            "mesh.mod.*.tool.>"
-            "mesh.mod.*.call.>"
-            "mesh.mod.gitea.event.>"
-            "$JS.ACK.EVENTS.n1_node-tools.>"
-            "$JS.API.CONSUMER.INFO.EVENTS.n1_node-tools"
-            "$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_node-tools"
-          ] }
-          subscribe { allow: [
-            "_INBOX.n1.node-tools.>"
-            "mesh.mod.gitea.tool.>"
-            "mesh.mod.mesh-delivery-forge-bridge.event.check.status"
-          ] }
-          allow_responses: true
-        }
-      }
     ]
   }
 }
@@ -242,15 +208,10 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
 		t.Fatal(err)
 	}
-	// The checker's stream checks and the bridge's durable consumer on its owed.written event; the checker's owed bucket is made
-	// only by the test that needs it, as when the checker is assigned after the bridge started.
+	// The checker's stream checks, which the bridge no longer hears (ADR 0521): a test says on it that the checker wrote a status.
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := js.CreateOrUpdateConsumer(ctx, "CHECKS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy,
-		FilterSubjects: []string{"mesh.stream.mesh-delivery-checker.event.owed.written"}}); err != nil {
-		t.Fatal(err)
-	}
 	// The bridge's state bucket, made by the controller from the manifest's state entry before the process starts.
 	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
 		t.Fatal(err)
@@ -753,45 +714,13 @@ func forgeMerged(t *testing.T, b *bed) []string {
 
 func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }
 
-// checkStatusAt is the subject the bridge emits check.status on, and the one gitea's consumer binds.
-const checkStatusAt = "mesh.mod.mesh-delivery-forge-bridge.event.check.status"
-
-// gitesConsumer is the durable consumer of the tool runner carrying gitea on EVENTS, as the controller makes it from gitea's
-// consumes (nats.go consumerDurable <node>_<module>), filtered on the bridge's check.status.
-func giteaConsumer(ctx context.Context, t *testing.T, js jetstream.JetStream) {
-	t.Helper()
-	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_node-tools", AckPolicy: jetstream.AckExplicitPolicy,
-		FilterSubjects: []string{checkStatusAt}}); err != nil {
-		t.Fatal(err)
-	}
-}
-
-// asRunner connects as the machine's tool runner, the account agent sessions share.
-func asRunner(t *testing.T, b *bed) (*nats.Conn, jetstream.JetStream) {
-	t.Helper()
-	nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools"))
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(nc.Close)
-	js, err := jetstream.New(nc, jetstream.WithDefaultTimeout(2*time.Second))
-	if err != nil {
-		t.Fatal(err)
-	}
-	return nc, js
-}
-
-// TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started;
-// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker
-// itself writes with) and says so on its stream; the bridge emits check.status on its own subject within 10 s, with the
-// state, the line, the key and the revision; the same revision is not emitted again, a new revision is (ADR 0484 clause 2).
-// No tool is called for it (R4e finding K1). Both statuses are emitted before gitea's runner fetches anything, and its
-// durable consumer hands both over afterwards: a check.status emitted while gitea's process is not running is applied
-// when it runs.
-func TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // K1, 1683/AC5, ADR 0484 §2
-	b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) })
+// TestBridgePerformsNoMeshCheckAct: the checker owes mesh/check on a head and says so on its stream; the bridge calls no
+// tool with mesh/check and publishes nothing of its own on EVENTS for it. The forge module writes mesh/check from the
+// checker's owed state itself (ADR 0521; R4e round 2 finding R2-1).
+func TestBridgePerformsNoMeshCheckAct(t *testing.T) { // R2-1, ADR 0521
+	b := startBed(t, `{"result":{"pulls":[]}}`, nil)
 	calls := make(chan string, 8)
-	// Every tool call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
+	// Every forge call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
 	sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) {
 		if strings.Contains(string(m.Data), "mesh/check") {
 			calls <- m.Subject + " " + string(m.Data)
@@ -812,124 +741,26 @@ func TestBridgeEmitsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.0123456789ab"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := kv.Put(b.ctx, key, v); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := b.js.Publish(b.ctx, "mesh.stream.mesh-delivery-checker.event.owed.written", []byte(`{"key":"`+key+`"}`)); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case c := <-calls:
+		t.Errorf("a forge tool was called with mesh/check: %s", c)
+	case <-time.After(2 * time.Second):
+	}
 	events, err := b.js.Stream(b.ctx, "EVENTS")
 	if err != nil {
 		t.Fatal(err)
 	}
-	emitted := func() uint64 {
-		info, err := events.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt))
-		if err != nil {
-			t.Fatal(err)
-		}
-		return info.State.Msgs
-	}
-	put := func(state, text string, want uint64) uint64 {
-		v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
-		if err != nil {
-			t.Fatal(err)
-		}
-		rev, err := kv.Put(b.ctx, key, v)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil {
-			t.Fatal(err)
-		}
-		for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && emitted() < want; time.Sleep(50 * time.Millisecond) {
-		}
-		if got := emitted(); got != want {
-			t.Fatalf("EVENTS holds %d check.status within 10 s, want %d", got, want)
-		}
-		return rev
-	}
-	rev1 := put("pending", "checking 0123456789ab on aaaaaaaaaaaa", 1)
-	rev2 := put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept", 2)
-	time.Sleep(500 * time.Millisecond)
-	if got := emitted(); got != 2 {
-		t.Errorf("EVENTS holds %d check.status, want 2: a revision was emitted twice", got)
-	}
-	select {
-	case c := <-calls:
-		t.Errorf("a tool was called with mesh/check: %s", c)
-	default:
-	}
-
-	// gitea's runner starts fetching only now, and is handed both, in order.
-	_, rjs := asRunner(t, b)
-	cons, err := rjs.Consumer(b.ctx, "EVENTS", "n1_node-tools")
-	if err != nil {
-		t.Fatal(err)
-	}
-	batch, err := cons.Fetch(2, jetstream.FetchMaxWait(5*time.Second))
-	if err != nil {
-		t.Fatal(err)
-	}
-	var got []string
-	for m := range batch.Messages() {
-		var c struct {
-			Owner, Repo, Sha, State, Description, Key string
-			Revision                                  uint64
-		}
-		if err := json.Unmarshal(m.Data(), &c); err != nil {
-			t.Fatal(err)
-		}
-		if m.Headers().Get("x-source") != "mesh-delivery-forge-bridge" || m.Headers().Get("x-event-id") != fmt.Sprintf("check.%s.%d", c.Key, c.Revision) {
-			t.Errorf("headers = %v", m.Headers())
-		}
-		got = append(got, fmt.Sprintf("%s/%s@%s %s %d %v", c.Owner, c.Repo, c.Sha[:8], c.State, c.Revision, c.Key == key))
-		_ = m.Ack()
-	}
-	want := []string{fmt.Sprintf("novox/mesh-delivery@%s pending %d true", head[:8], rev1), fmt.Sprintf("novox/mesh-delivery@%s success %d true", head[:8], rev2)}
-	if strings.Join(got, "|") != strings.Join(want, "|") {
-		t.Errorf("gitea's consumer was handed %v, want %v", got, want)
-	}
-}
-
-// TestOnlyTheBridgesAccountMayEmitCheckStatus: the machine's tool runner, whose account every agent session there shares and
-// which invokes every tool, publishing on the bridge's check.status subject is refused by the bus, and gitea's consumer is
-// handed nothing (R4e finding K1).
-func TestOnlyTheBridgesAccountMayEmitCheckStatus(t *testing.T) { // K1
-	b := startBed(t, `{"result":{"pulls":[]}}`, func(ctx context.Context, js jetstream.JetStream) { giteaConsumer(ctx, t, js) })
-	refused := make(chan error, 4)
-	nc, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo("n1.node-tools", "pw"), nats.CustomInboxPrefix("_INBOX.n1.node-tools"),
-		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Cleanup(nc.Close)
-	js, err := jetstream.New(nc)
-	if err != nil {
-		t.Fatal(err)
-	}
-	forged, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "sha": head, "state": "success",
-		"description": "by a session", "key": "x", "revision": 1 << 40})
-	m := nats.NewMsg(checkStatusAt)
-	m.Header.Set("x-source", "mesh-delivery-forge-bridge")
-	m.Data = forged
-	ctx, cancel := context.WithTimeout(b.ctx, 2*time.Second)
-	defer cancel()
-	if _, err := js.PublishMsg(ctx, m); err == nil {
-		t.Fatal("the tool runner's account published check.status: any session could set mesh/check")
-	}
-	select {
-	case err := <-refused:
-		if !strings.Contains(err.Error(), "Permissions Violation for Publish") || !strings.Contains(err.Error(), checkStatusAt) {
-			t.Errorf("refused with %v", err)
-		}
-	case <-time.After(5 * time.Second):
-		t.Error("the bus said no permissions violation for the runner's publish")
-	}
-	// And the ordinary way in, a tool call, has no tool that writes mesh/check to call: the runner may call every tool, so
-	// the refusal is the bus's, not a manifest's.
-	info, err := b.js.Stream(b.ctx, "EVENTS")
-	if err != nil {
-		t.Fatal(err)
-	}
-	si, err := info.Info(b.ctx, jetstream.WithSubjectFilter(checkStatusAt))
-	if err != nil {
-		t.Fatal(err)
-	}
-	if si.State.Msgs != 0 {
-		t.Errorf("EVENTS holds %d check.status after the runner's publish, want 0", si.State.Msgs)
+	if _, err := events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-forge-bridge.event.>"); err == nil {
+		t.Error("the bridge published an event of its own for the checker's mesh/check")
 	}
 }

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go

@@ -1,82 +0,0 @@
-package main
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"log"
-	"strings"
-	"sync"
-
-	"github.com/nats-io/nats.go/jetstream"
-
-	"git.novox.be/novox/mesh-sdk/go/delivery"
-)
-
-// checksStream is the stream the checker carries its events on, as the bus names it (declared by the checker as the stream
-// checks, heard by this module through consumes; novox/hq ADR 0484 clause 2).
-const checksStream = "CHECKS"
-
-// owedWrittenSubject is the subject of the checker's owed.written event.
-const owedWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.owed.written"
-
-// lateWatch watches a bucket that may be made after the bridge started: the checker's owed bucket. The bridge holds no timer
-// (ADR 0437 §8), so it watches the bucket where it exists at start, and otherwise when the checker says it wrote a status
-// (owed.written, handled by Written).
-type lateWatch struct {
-	ctx context.Context
-	js  interface {
-		KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
-	}
-	bucket  string
-	react   func(ctx context.Context, key string, rev uint64, value []byte)
-	logger  *log.Logger
-	errs    chan error
-	mu      sync.Mutex
-	watched bool
-}
-
-// ensure starts the watch if the bucket exists and is not watched.
-func (w *lateWatch) ensure() error {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if w.watched {
-		return nil
-	}
-	kv, err := w.js.KeyValue(w.ctx, w.bucket)
-	switch {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a status", w.bucket)
-		return nil
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", w.bucket, err)
-	}
-	w.watched = true
-	go func() {
-		w.errs <- watch(w.ctx, kv, func(key string, rev uint64, value []byte) {
-			hctx, cancel := context.WithTimeout(w.ctx, syncDeadline)
-			defer cancel()
-			w.react(hctx, key, rev, value)
-		})
-	}()
-	return nil
-}
-
-// Written takes one owed.written event of the checker in: its bucket is watched if it is not. A subject that is not the
-// checker's is terminated; a failure stays unacknowledged, and the server delivers it again.
-func (w *lateWatch) Written(m interface {
-	Subject() string
-	Ack() error
-	Term() error
-}) {
-	if !strings.HasPrefix(m.Subject(), owedWrittenSubject) {
-		w.logger.Printf("%s: not the checker's owed.written, dropped", m.Subject())
-		_ = m.Term()
-		return
-	}
-	if err := w.ensure(); err != nil {
-		w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err)
-		return
-	}
-	_ = m.Ack()
-}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

@@ -201,7 +201,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return fmt.Errorf("the bucket %s: %w", performedBucket, err)
 	}
-	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger, emit: js, node: cred.Node}
+	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
 	watching := make(chan error, 3)
 	for _, w := range []struct {
 		bucket string
@@ -225,24 +225,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			})
 		}()
 	}
-	// The checker's statuses (mesh/check, unrequired until release R4f): its bucket is made when the checker is assigned, which
-	// may be after the bridge started, and the bridge holds no timer; the checker says on its stream each time it wrote a
-	// status, and the bucket is then watched (late.go).
-	late := &lateWatch{ctx: ctx, js: js, bucket: delivery.CheckerOwedBucket, react: perf.checkOwed, logger: logger, errs: watching}
-	if err := late.ensure(); err != nil {
-		return err
-	}
-	switch ccons, err := js.Consumer(ctx, checksStream, cred.Node+"_"+cred.Module); {
-	case err != nil:
-		logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's statuses are not heard until the bridge starts again: %v", cred.Node, cred.Module, checksStream, err)
-	default:
-		cc, err := ccons.Consume(func(m jetstream.Msg) { late.Written(m) },
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
-		if err != nil {
-			return err
-		}
-		defer cc.Stop()
-	}
 	if err := said(ctx); err != nil {
 		return err
 	}

modules/mesh-delivery-forge-bridge/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-forge-bridge/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-forge-bridge/internal/bridge/acts.go

@@ -1,9 +1,7 @@
 package bridge
 
 import (
-	"encoding/json"
 	"fmt"
-	"regexp"
 	"slices"
 	"strings"
 
@@ -39,54 +37,6 @@ const (
 	mergePullTool    = "gitea.gitea_merge_pull_request"
 )
 
-// CheckStatusEvent is the event the bridge writes mesh/check with (R4e finding K1; issue 1301 S4; ADR 0503 §3): no tool
-// writes mesh/check, since a tool is told no caller and node-tools invokes every tool. The bus grants its subject,
-// mesh.mod.mesh-delivery-forge-bridge.event.check.status, to this module's own account alone (its manifest's emits), and
-// the forge module consumes it and writes the status (mesh-catalog modules/gitea check.ts).
-const CheckStatusEvent = "check.status"
-
-// CheckStatus is the body of a check.status: the status the checker owes on a head, the owed key and the revision it was
-// owed at, by which the forge module drops a stale one delivered after a newer.
-type CheckStatus struct {
-	Owner       string `json:"owner"`
-	Repo        string `json:"repo"`
-	Sha         string `json:"sha"`
-	State       string `json:"state"`
-	Description string `json:"description"`
-	TargetURL   string `json:"target_url,omitempty"`
-	Key         string `json:"key"`
-	Revision    uint64 `json:"revision"`
-}
-
-var wholeSha = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
-
-// CheckStatusMessage turns the checker's owed status (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed) at revision rev
-// into the check.status it is emitted as: its id, `check.<key>.<rev>`, the bus's dedup id too, and its body. A key that is
-// not the checker's, a value that is not its status, and a head that is not a whole sha are errors, each saying why.
-func CheckStatusMessage(key string, rev uint64, value []byte) (id string, body []byte, err error) {
-	repository, _, head, err := delivery.ParseCheckerOwedKey(key)
-	if err != nil {
-		return "", nil, err
-	}
-	o, err := delivery.DecodeCheckerOwed(value)
-	if err != nil {
-		return "", nil, fmt.Errorf("the checker's owed status %s: %w", key, err)
-	}
-	if o.Act != delivery.OwedStatus || o.Context != delivery.StatusContextCheck {
-		return "", nil, fmt.Errorf("the checker's owed %s is %s %s, not the status %s", key, o.Act, o.Context, delivery.StatusContextCheck)
-	}
-	if !wholeSha.MatchString(head) {
-		return "", nil, fmt.Errorf("the checker's owed %s names the head %q, not a whole sha", key, head)
-	}
-	owner, repo := ownerRepo(repository)
-	body, err = json.Marshal(CheckStatus{Owner: owner, Repo: repo, Sha: head, State: o.State, Description: o.Text,
-		TargetURL: o.Target, Key: key, Revision: rev})
-	if err != nil {
-		return "", nil, err
-	}
-	return fmt.Sprintf("check.%s.%d", key, rev), body, nil
-}
-
 // MergeAct is the name of the act of merging a pull request.
 const MergeAct = "merge"
 
@@ -128,8 +78,10 @@ func OwedAct(key string, value []byte) (Act, error) {
 			return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context)
 		}
 		if o.Context == delivery.StatusContextCheck {
-			return Act{}, fmt.Errorf("the owed status %s is %s, which no tool writes: the bridge emits it as %s (CheckStatusMessage)",
-				key, delivery.StatusContextCheck, CheckStatusEvent)
+			// No tool writes mesh/check (R4e finding K1), and the bridge performs no act of it: the forge module writes it from
+			// the checker's state mesh-delivery-checker.owed, read by a watch (ADR 0521, finding R2-1).
+			return Act{}, fmt.Errorf("the owed status %s is %s, which the bridge does not perform: the forge module writes it from mesh-delivery-checker.owed",
+				key, delivery.StatusContextCheck)
 		}
 		a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context,
 			"state": o.State, "description": o.Text, "target_url": o.Target}}

modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go

@@ -58,49 +58,23 @@ func checkerKV(t *testing.T, o delivery.Owed) (string, []byte) {
 	return key, v
 }
 
-// R4e finding K1 (issue 1301 S4; ADR 0503 §3): no tool writes mesh/check, since a tool is told no caller and node-tools
-// invokes every tool. The bridge emits check.status, a subject only its own account may publish, and gitea writes it.
+// R4e findings K1 and R2-1 (issue 1301 S4; ADR 0503 §3, ADR 0521): no tool writes mesh/check, since a tool is told no
+// caller and node-tools invokes every tool; the forge module writes it from the checker's owed state, and the bridge
+// performs no mesh/check act.
 func TestOwedMeshCheckIsNoToolCall(t *testing.T) { // K1
 	key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
 	a, err := OwedAct(key, v)
-	if err == nil || !strings.Contains(err.Error(), "check.status") {
-		t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming check.status", a, err)
+	if err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") {
+		t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming where the forge module reads it", a, err)
 	}
 	// Nor through the gate's own owed bucket: the context is refused whichever key carries it.
 	gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check",
 		delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"})
-	if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "check.status") {
+	if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") {
 		t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err)
 	}
 }
 
-func TestCheckStatusMessageCarriesTheOwedStatusAndItsRevision(t *testing.T) { // K1
-	key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
-	id, body, err := CheckStatusMessage(key, 42, v)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if CheckStatusEvent != "check.status" {
-		t.Errorf("event = %q", CheckStatusEvent)
-	}
-	if id != "check."+key+".42" {
-		t.Errorf("id = %q", id)
-	}
-	want := `{"owner":"novox","repo":"mesh-delivery","sha":"` + headA + `","state":"success","description":"pass","target_url":"https://view.example/checks/g.1","key":"` + key + `","revision":42}`
-	if string(body) != want {
-		t.Errorf("body = %s\nwant   %s", body, want)
-	}
-	// A key that is not the checker's, and a value that is not its status, are no message.
-	gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
-		delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"})
-	if _, _, err := CheckStatusMessage(gk, 1, gv); err == nil {
-		t.Error("a key of the gate made a check.status")
-	}
-	if _, _, err := CheckStatusMessage(key, 1, []byte("{")); err == nil {
-		t.Error("a value that is no owed status made a check.status")
-	}
-}
-
 func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2
 	cases := []struct {
 		act  string

modules/mesh-delivery-forge-bridge/manifest_test.go

@@ -45,7 +45,6 @@ type manifest struct {
 	State        []state           `json:"state"`
 	Reads        []string          `json:"reads"`
 	Invokes      []string          `json:"invokes"`
-	Emits        []string          `json:"emits"`
 	OwnSecrets   map[string]string `json:"own-secrets"`
 	SecretsOwner string            `json:"secrets-owner"`
 	Resources    []map[string]any  `json:"resources"`
@@ -87,16 +86,13 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
 	if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) {
 		t.Errorf("streams = %+v", m.Streams)
 	}
-	if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated", "mesh-delivery-checker.owed.written"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
+	// The checker's owed is the forge module's to write (ADR 0521, R4e finding R2-1): the bridge neither hears nor reads it.
+	if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
 		t.Errorf("consumes = %v, want %v", m.Consumes, want)
 	}
-	if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) {
+	if want := []string{"mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) {
 		t.Errorf("reads = %v, want %v", m.Reads, want)
 	}
-	// No tool writes mesh/check (R4e finding K1): the bridge emits check.status, which only its account may publish.
-	if want := []string{"check.status"}; !reflect.DeepEqual(m.Emits, want) {
-		t.Errorf("emits = %v, want %v", m.Emits, want)
-	}
 	if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view",
 		"gitea.gitea_get_file", "gitea.gitea_merge_pull_request", "gitea.gitea_note_append", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)
@@ -113,8 +109,8 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
 	if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341
 		t.Errorf("state = %+v, want the buckets heads and performed", m.State)
 	}
-	// No other state, no bus, no seat, no tool (ADR 0437 §4); its one event is check.status, asserted above (R4e finding K1).
-	for _, k := range []string{"logs", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {
+	// No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status).
+	for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the bridge has none", k)
 		}

modules/mesh-delivery-forge-bridge/module.json

@@ -5,8 +5,7 @@
   "consumes": [
     "gitea.pull.updated",
     "gitea.pull.merged",
-    "gitea.pull.closed",
-    "mesh-delivery-checker.owed.written"
+    "gitea.pull.closed"
   ],
   "streams": [
     { "name": "forge", "events": ["forge.**"] }
@@ -16,13 +15,9 @@
     { "name": "performed", "history": 1 }
   ],
   "reads": [
-    "mesh-delivery-checker.owed",
     "mesh-delivery-gate.merges",
     "mesh-delivery-gate.owed"
   ],
-  "emits": [
-    "check.status"
-  ],
   "invokes": [
     "gitea.gitea_merge_pull_request",
     "gitea.gitea_commit_status",

modules/mesh-delivery-indexer/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-indexer
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-indexer/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-view/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-view
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-view/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=