Move every delivery part's bus connection onto mesh-sdk's module helpers: credential, pinned dial, own-stream subjects, durable consumer, and refuse a hand-written one (issue 1242)

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

@@ -8,13 +8,11 @@ package main
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"os/signal"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -31,7 +29,7 @@ import (
 )
 
 func main() {
-	if os.Getenv("MESH_BROKER_FILE") == "" {
+	if os.Getenv(process.BrokerFileEnv) == "" {
 		// Started by the tool runner, with no credential of its own: the put's tools (ADR 0546), served through the runtime,
 		// since a process never serves (ADR 0436). The process the manifest runs names its credential.
 		if err := serveTools(); err != nil {
@@ -118,33 +116,15 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	account, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with")
-	}
-	account, err := process.ReadAccount(path)
 	if err != nil {
 		return err
 	}
-	opts := []nats.Option{
+	conn, js, err := account.Dial()
-		nats.Name(account.Node + "." + account.Module),
-		nats.UserInfo(account.User, account.Password),
-		nats.CustomInboxPrefix(account.InboxPrefix()),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(account.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint)))
-	}
-	conn, err := nats.Connect(account.URL, opts...)
 	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
+		return err
 	}
 	defer conn.Close()
-	js, err := jetstream.New(conn)
-	if err != nil {
-		return err
-	}
 	kvs, err := open(ctx, js, logger.Printf)
 	if err != nil {
 		return err

modules/mesh-delivery-assigner/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-assigner
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-assigner/go.sum

@@ -1,17 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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/cmd/mesh-delivery-builder/integration_test.go

@@ -24,6 +24,7 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 
 	"mesh-delivery-builder/internal/build"
 )
@@ -345,7 +346,7 @@ func (b *bed) archive(t *testing.T, name, path, at string) string {
 func (b *bed) start(t *testing.T, node string, env map[string]string) {
 	t.Helper()
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -6,12 +6,6 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -51,80 +45,18 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK).
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-builder: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
-	opts := []nats.Option{
+	conn, js, err := cred.Dial()
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
 	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
+		return err
 	}
 	defer conn.Close()
-	js, err := jetstream.New(conn)
-	if err != nil {
-		return err
-	}
 	builds, err := process.EssentialBucket(ctx, js, "builder", delivery.BuilderBuildsBucket, "claim or write a build key")
 	if err != nil {
 		return err
@@ -163,11 +95,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again() })
 	// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
 	// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
-	account, err := process.ReadAccount(path)
+	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
-	if err != nil {
-		return err
-	}
-	calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
 	if err != nil {
 		return err
 	}
@@ -232,13 +160,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// module.
 	for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} {
 		stream := toolchainStreams[module]
-		cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module)
+		cons, err := cred.EventConsumer(ctx, js, stream)
 		if err != nil {
 			logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err)
 			continue
 		}
-		cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) },
+		cc, err := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf)
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		if err != nil {
 			return err
 		}
@@ -247,13 +174,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// The controller says artifacts-gone when its take-in found a key's blob gone (issue 1937): the builder's durable
 	// consumer on EVENTS, made by the controller from the manifest's consumes. Where it is not there yet the builder says
 	// so; it still asks held at its start and at every module value.
-	switch cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module); {
+	switch cons, err := cred.EventConsumer(ctx, js, eventsStream); {
 	case err != nil:
 		logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
 			cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
 	default:
-		cc, err := cons.Consume(func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) },
+		cc, err := process.HearEvents(cons, func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, logger.Printf)
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		if err != nil {
 			return err
 		}

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

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -53,12 +54,12 @@ func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstre
 
 // emitAskedAgain says the builder's event build.asked-again (issue 1945): mesh.mod.<module>.event.build.asked-again,
 // with the SDK's envelope headers and the new ask's id as the bus's dedup id.
-func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred credential, e build.AskedAgain) error {
+func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred process.Account, e build.AskedAgain) error {
 	data, err := json.Marshal(e)
 	if err != nil {
 		return err
 	}
-	msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + build.AskedAgainEvent)
+	msg := nats.NewMsg(cred.EventSubject(build.AskedAgainEvent))
 	id := "asked-again-" + e.New
 	msg.Header.Set("x-event-id", id)
 	msg.Header.Set("x-source", cred.Module)

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.20261026142434-65ee43f07583
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	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.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/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-checker/cmd/mesh-delivery-checker/integration_test.go

@@ -25,6 +25,7 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 	"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"
 
 	"mesh-delivery-checker/internal/check"
 )
@@ -423,7 +424,7 @@ func pull(repository string, number int, head string) delivery.CandidatePull {
 func (b *bed) start(t *testing.T, env map[string]string) {
 	t.Helper()
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srvURL, Node: "n1", Module: "mesh-delivery-checker", User: "n1.mesh-delivery-checker", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srvURL, Node: "n1", Module: "mesh-delivery-checker", User: "n1.mesh-delivery-checker", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -7,12 +7,7 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -36,7 +31,7 @@ import (
 // eventSubject is the subject of an event of the checker's stream checks (the bus names the stream CHECKS): the indexer's
 // watch of the previews bucket and the forge bridge's watch of the owed bucket are woken by them.
 func eventSubject(event string) string {
-	return "mesh.stream." + delivery.CheckerModule + ".event." + event
+	return process.ModuleStreamSubject(delivery.CheckerModule, event)
 }
 
 const (
@@ -53,50 +48,6 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK).
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // durationEnv reads a duration setting from the environment, def when it is not set.
 func durationEnv(getenv func(string) string, name string, def time.Duration) (time.Duration, error) {
 	v := strings.TrimSpace(getenv(name))
@@ -128,33 +79,15 @@ var keptStates = []announce.State{
 
 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"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
-	opts := []nats.Option{
+	conn, js, err := cred.Dial()
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
 	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
+		return err
 	}
 	defer conn.Close()
-	js, err := jetstream.New(conn)
-	if err != nil {
-		return err
-	}
 	// Its own states, the candidates it checks, and the modules and manifests that say which build modules judge a repository:
 	// a check without one of them would be wrong or unwritten, so each stops the start, naming its grant (issue 1945).
 	buckets := map[string]jetstream.KeyValue{}
@@ -166,11 +99,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	// Calls to the build modules go out as this module's own account, naming the machine (mesh-sdk process.CallOn); the
 	// buckets are read and written on the connection above.
-	account, err := process.ReadAccount(path)
+	caller, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
-	if err != nil {
-		return err
-	}
-	caller, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
 	if err != nil {
 		return err
 	}
@@ -287,13 +216,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// module.
 	for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} {
 		stream := toolchainStreams[module]
-		cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module)
+		cons, err := cred.EventConsumer(ctx, js, stream)
 		if err != nil {
 			logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the checker starts again: %v", cred.Node, cred.Module, stream, module, err)
 			continue
 		}
-		cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) },
+		cc, err := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf)
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		if err != nil {
 			return err
 		}

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.20261027014310-cadd7b1b2f8f
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	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,11 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9 h1:MS7i4NVNQHMAMqRRFMdzbto31BHZY+BuEZY1ql3p4Rs=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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/integration_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream,
@@ -271,7 +272,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	}
 
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -6,18 +6,12 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"os/signal"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -39,7 +33,7 @@ const eventsStream = "EVENTS"
 // The bridge's own stream FORGE takes `mesh.stream.<module>.event.<event>` (mesh-control internal/broker
 // modstreams.go ModuleStreamSubject).
 func forgeSubject(event string) string {
-	return "mesh.stream." + delivery.ForgeBridgeModule + ".event." + event
+	return process.ModuleStreamSubject(delivery.ForgeBridgeModule, event)
 }
 
 func main() {
@@ -51,50 +45,6 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK).
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // delivered is the part of a JetStream message the bridge uses.
 type delivered interface {
 	Subject() string
@@ -152,40 +102,23 @@ func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees,
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-forge-bridge: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
 	again := make(chan struct{}, 1)
-	opts := []nats.Option{
+	conn, js, err := cred.Dial(
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
 		nats.ReconnectHandler(func(*nats.Conn) {
 			select {
 			case again <- struct{}{}:
 			default:
 			}
 		}),
-	}
+	)
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
-	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
-	}
-	defer conn.Close()
-	js, err := jetstream.New(conn)
 	if err != nil {
 		return err
 	}
+	defer conn.Close()
 	// What the forge holds open, said before any event is heard, and again at each reconnect (ADR 0429 §13).
 	call := callerOn(conn, cred.User)
 	// No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns.
@@ -227,20 +160,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	go onReconnect(ctx, again, said, logger)
-	cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module)
+	cons, err := cred.EventConsumer(ctx, js, eventsStream)
 	if err != nil {
 		// Essential: the forge's events are what the bridge carries.
 		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err)
 	}
-	cc, err := cons.Consume(func(m jetstream.Msg) {
+	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
 		hear(&turn, func() {
 			// Bounded: the turn is held, and no sync or other event goes on without it.
 			hctx, cancel := context.WithTimeout(ctx, syncDeadline)
 			defer cancel()
 			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger)
 		})
-	},
+	}, logger.Printf)
-		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 	if 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.20261027014310-cadd7b1b2f8f
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	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,17 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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-indexer/cmd/mesh-delivery-indexer/integration_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -171,7 +172,7 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), b
 	}
 
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -10,10 +10,6 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -21,7 +17,6 @@ import (
 	"log"
 	"os"
 	"os/signal"
-	"strings"
 	"syscall"
 
 	"github.com/nats-io/nats.go"
@@ -47,50 +42,6 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK).
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is
 // process.ErrNotYet.
 type awaitedStore struct{ b *process.AwaitedBucket }
@@ -272,33 +223,15 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-indexer: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
-	opts := []nats.Option{
+	conn, js, err := cred.Dial()
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
 	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
+		return err
 	}
 	defer conn.Close()
-	js, err := jetstream.New(conn)
-	if err != nil {
-		return err
-	}
 	candidates, err := process.EssentialBucket(ctx, js, "indexer", delivery.CandidatesBucket, "write a candidate's fingerprints")
 	if err != nil {
 		return err
@@ -344,11 +277,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 
 	// The indexer asks the forge for the objects of a manifest's named paths as its own account (ADR 0499 clause 3).
-	account, err := process.ReadAccount(path)
+	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
-	if err != nil {
-		return err
-	}
-	calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
 	if err != nil {
 		return err
 	}
@@ -449,30 +378,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// with the build module, as with the gate's source.
 	for _, module := range sortedModules() {
 		stream := index.ToolchainStreams[module]
-		tcons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module)
+		tcons, err := cred.EventConsumer(ctx, js, stream)
 		if err != nil {
 			logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the indexer starts again: %v", cred.Node, cred.Module, stream, module, err)
 			continue
 		}
-		tc, err := tcons.Consume(func(m jetstream.Msg) { tools.Reported(m) },
+		tc, err := process.HearEvents(tcons, func(m jetstream.Msg) { tools.Reported(m) }, logger.Printf)
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		if err != nil {
 			return err
 		}
 		defer tc.Stop()
 	}
-	switch pcons, err := js.Consumer(ctx, PreviewsStream, cred.Node+"_"+cred.Module); {
+	switch pcons, err := cred.EventConsumer(ctx, js, PreviewsStream); {
 	case err != nil:
 		logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, err)
 	default:
-		pc, err := pcons.Consume(func(m jetstream.Msg) { previews.Written(m) },
+		pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.Written(m) }, logger.Printf)
-			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 		if err != nil {
 			return err
 		}
 		defer pc.Stop()
 	}
-	cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module)
+	cons, err := cred.EventConsumer(ctx, js, forgeStream)
 	if err != nil {
 		// Essential: the forge's merges are what the indexer indexes.
 		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err)
@@ -487,8 +414,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return info.NumPending == 0 && info.NumAckPending == 0, nil
 	}
 	taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) })
-	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) },
+	cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, logger.Printf)
-		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
 	if err != nil {
 		return err
 	}

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go

@@ -3,11 +3,12 @@ package main
 import (
 	"context"
 	"errors"
-	"git.novox.be/novox/mesh-sdk/go/process"
 	"log"
 	"strings"
 	"sync"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"

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.20261027014310-cadd7b1b2f8f
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	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,17 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -19,6 +19,7 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -90,7 +91,7 @@ func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
 	defer cancel()
 
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -321,7 +322,7 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
 	b := startFull(t)
 	tools := b.as(t, "n1.node-tools")
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -411,7 +412,7 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
 	b := startFull(t)
 	tools := b.as(t, "n1.node-tools")
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -483,7 +484,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
 	b := startFull(t)
 	assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -556,7 +557,7 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
 	b := startFull(t)
 	assigner := b.as(t, "n1.mesh-delivery-assigner")
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -650,7 +651,7 @@ func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
 	b := startFull(t)
 	assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
 	broker := filepath.Join(t.TempDir(), "broker")
-	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -8,23 +8,16 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"log"
 	"os"
 	"os/signal"
-	"strings"
 	"syscall"
 
 	"git.novox.be/novox/mesh-sdk/go/process"
 
-	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/asker"
@@ -40,80 +33,18 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK).
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-teller: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to be on the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
-	opts := []nats.Option{
+	conn, js, err := cred.Dial()
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
 	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
+		return err
 	}
 	defer conn.Close()
-	js, err := jetstream.New(conn)
-	if err != nil {
-		return err
-	}
 	gate, err := js.KeyValue(ctx, delivery.AsksBucket)
 	if errors.Is(err, jetstream.ErrBucketNotFound) {
 		logger.Print(waiting)

modules/mesh-delivery-teller/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-teller
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-teller/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/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-teller/manifest_test.go

@@ -177,3 +177,28 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR
 		t.Fatalf("read %d source files, want the program's", files)
 	}
 }
+
+// Issue 1242: no delivery part hand-writes its bus connection. A part's own credential, its pinned dial, the subjects of its
+// streams and its durable consumer come from mesh-sdk's process package (ReadBrokerAccount, Account.Dial, Account.EventConsumer,
+// ModuleStreamSubject); a part whose main.go dials with nats.Connect, pins with its own certificate check or binds the
+// consumer name itself is refused here. mesh-grants is outside it until its own seats change (issue 1945) lands.
+func TestNoPartHandWritesItsBusConnection(t *testing.T) {
+	mains, err := filepath.Glob("../mesh-delivery-*/cmd/*/*.go")
+	if err != nil || len(mains) == 0 {
+		t.Fatalf("no delivery part found: %v", err)
+	}
+	for _, f := range mains {
+		if strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		raw, err := os.ReadFile(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, banned := range []string{"nats.Connect(", "VerifyPeerCertificate", `cred.Node+"_"+cred.Module`, `"MESH_BROKER_FILE"`} {
+			if bytes.Contains(raw, []byte(banned)) {
+				t.Errorf("%s hand-writes the bus connection (%s): use mesh-sdk's process.ReadBrokerAccount, Account.Dial, Account.EventConsumer or ModuleStreamSubject", f, banned)
+			}
+		}
+	}
+}

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

@@ -18,6 +18,7 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 
 	"mesh-delivery-view/internal/view"
 )
@@ -81,7 +82,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -180,7 +181,7 @@ func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -246,7 +247,7 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -332,7 +333,7 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { //
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -422,7 +423,7 @@ func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -519,7 +520,7 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -670,7 +671,7 @@ func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -738,7 +739,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -834,7 +835,7 @@ func TestProcessWholeSaysModulesInMotion(t *testing.T) {
 	}
 	t.Cleanup(func() { os.RemoveAll(dir) })
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -7,8 +7,6 @@ package main
 import (
 	"context"
 	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
@@ -26,6 +24,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 
 	"mesh-delivery-view/internal/view"
 )
@@ -44,58 +43,10 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password.
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // run is the program: it returns when ctx ends or a listener cannot be made.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-view: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to watch the buckets with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
@@ -118,25 +69,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	go func() { webUp <- srv.ListenAndServe() }()
 	go view.ServeSocket(ctx, l, model)
 
-	opts := []nats.Option{
+	conn, js, err := cred.Dial()
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
-	}
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
-	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
-	}
-	defer conn.Close()
-	js, err := jetstream.New(conn)
 	if err != nil {
 		return err
 	}
+	defer conn.Close()
 	model.SetWriter(kvWriter{ctx: ctx, js: js})
 	model.SetMotionEvent(func(e view.MotionEvent) {
 		if err := emitEvent(ctx, js, cred, e.Type, e.Body, e.ID); err != nil {
@@ -263,12 +200,12 @@ const DeliveredEventType = "delivered"
 // emitEvent publishes an event of the view's own namespace, mesh.mod.<module>.event.<type>, with the SDK's envelope headers
 // (as mesh-sdk's process.Conn.Emit does) and the id as the bus's own dedup id, so the same revision said twice in the stream's
 // duplicate window is kept once.
-func emitEvent(ctx context.Context, js jetstream.JetStream, cred credential, eventType string, body any, id string) error {
+func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account, eventType string, body any, id string) error {
 	data, err := json.Marshal(body)
 	if err != nil {
 		return err
 	}
-	msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + eventType)
+	msg := nats.NewMsg(cred.EventSubject(eventType))
 	msg.Header.Set("x-event-id", id)
 	msg.Header.Set("x-source", cred.Module)
 	msg.Header.Set("x-node", cred.Node)

modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go

@@ -3,29 +3,10 @@ package main
 import (
 	"context"
 	"io"
-	"os"
-	"path/filepath"
 	"strings"
 	"testing"
 )
 
-func TestCredentialIsTheBrokerFile(t *testing.T) {
-	path := filepath.Join(t.TempDir(), "broker")
-	os.WriteFile(path, []byte(`{"url":"nats://127.0.0.1:4222","fingerprint":"sha256:ab","node":"n","module":"mesh-delivery-view","user":"u","password":"p"}`), 0o600)
-	c, err := readCredential(path)
-	if err != nil || c.URL != "nats://127.0.0.1:4222" || c.User != "u" || c.Password != "p" || c.Fingerprint != "sha256:ab" {
-		t.Fatalf("credential = %+v, %v", c, err)
-	}
-	os.WriteFile(path, []byte(`{"url":"x"}`), 0o600)
-	if _, err := readCredential(path); err == nil {
-		t.Fatal("a credential naming no user must be refused")
-	}
-	os.WriteFile(path, []byte(`not json`), 0o600)
-	if _, err := readCredential(path); err == nil {
-		t.Fatal("a file that is not a credential must be refused")
-	}
-}
-
 func TestRunNeedsItsCredential(t *testing.T) {
 	err := run(context.Background(), func(string) string { return "" }, io.Discard)
 	if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {

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.20261027014310-cadd7b1b2f8f
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	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,17 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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-watchman/cmd/mesh-delivery-watchman/integration_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system
@@ -114,7 +115,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 	connect("node.n0")
 
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -286,7 +287,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
 		t.Fatal(err)
 	}
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}
@@ -388,7 +389,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
 	putHave("n2", "web", fpOld)
 	putWant("n2", "web", fpNew) // stood before the start
 	broker := filepath.Join(dir, "broker")
-	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
 	if err := os.WriteFile(broker, cred, 0o600); err != nil {
 		t.Fatal(err)
 	}

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

@@ -7,11 +7,6 @@ package main
 
 import (
 	"context"
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -49,50 +44,6 @@ func main() {
 	}
 }
 
-// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
-// fingerprint to pin, the principal and its password.
-type credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint"`
-	Node        string `json:"node"`
-	Module      string `json:"module"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
-
-func readCredential(path string) (credential, error) {
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		return credential{}, err
-	}
-	var c credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
-	}
-	if c.URL == "" || c.User == "" {
-		return credential{}, errors.New("the broker credential names no bus address or user")
-	}
-	return c, nil
-}
-
-// pinned accepts exactly the certificate with this SHA-256 and no other.
-func pinned(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
-			}
-			return nil
-		},
-	}
-}
-
 // kvOut writes the watchman's three buckets.
 type kvOut struct {
 	ctx                   context.Context
@@ -190,40 +141,23 @@ func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) {
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags)
-	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	cred, err := process.ReadBrokerAccount(getenv)
-	if path == "" {
-		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to watch the bus with")
-	}
-	cred, err := readCredential(path)
 	if err != nil {
 		return err
 	}
 	again := make(chan struct{}, 1)
-	opts := []nats.Option{
+	conn, js, err := cred.Dial(
-		nats.Name(cred.Node + "." + cred.Module),
-		nats.UserInfo(cred.User, cred.Password),
-		nats.CustomInboxPrefix("_INBOX." + cred.User),
-		nats.MaxReconnects(-1),
-		nats.RetryOnFailedConnect(true),
 		nats.ReconnectHandler(func(*nats.Conn) {
 			select {
 			case again <- struct{}{}:
 			default:
 			}
 		}),
-	}
+	)
-	if strings.TrimSpace(cred.Fingerprint) != "" {
-		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
-	}
-	conn, err := nats.Connect(cred.URL, opts...)
-	if err != nil {
-		return fmt.Errorf("connecting to the bus: %w", err)
-	}
-	defer conn.Close()
-	js, err := jetstream.New(conn)
 	if err != nil {
 		return err
 	}
+	defer conn.Close()
 	// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
 	// and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build
 	// seats' steps, and says that moves are not watched (issue 1945).

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go

@@ -5,8 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"os"
-	"path/filepath"
 	"strings"
 	"testing"
 
@@ -15,23 +13,6 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 )
 
-func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6
-	path := filepath.Join(t.TempDir(), "broker")
-	os.WriteFile(path, []byte(`{"url":"nats://127.0.0.1:4222","fingerprint":"sha256:ab","node":"n","module":"mesh-delivery-watchman","user":"u","password":"p"}`), 0o600)
-	c, err := readCredential(path)
-	if err != nil || c.URL != "nats://127.0.0.1:4222" || c.User != "u" || c.Password != "p" || c.Fingerprint != "sha256:ab" {
-		t.Fatalf("credential = %+v, %v", c, err)
-	}
-	os.WriteFile(path, []byte(`{"url":"x"}`), 0o600)
-	if _, err := readCredential(path); err == nil {
-		t.Fatal("a credential naming no user must be refused")
-	}
-	os.WriteFile(path, []byte(`not json`), 0o600)
-	if _, err := readCredential(path); err == nil {
-		t.Fatal("a file that is not a credential must be refused")
-	}
-}
-
 func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
 	t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) {
 		refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x")

modules/mesh-delivery-watchman/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-watchman
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-watchman/go.sum

@@ -1,7 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=