Stop committing mesh-sdk's Go dependencies: resolve them by go.sum (issue 1304)

ADR 0433 §2, §3: the watchman writes what stopped moving and the controller's
self-check raises it, so the key and value live in the SDK (ADR 0429 §2).

.gitignore

@@ -1,2 +1,3 @@
 node_modules/
 dist/
+vendor/

go/asks/condition.go

@@ -0,0 +1,149 @@
+package asks
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+)
+
+// ConditionSubject is where a module says a condition to the operator: a by-caller accept of the operator-channel
+// seat, `mesh.seat.operator-channel.accept.condition.<publisher>` (novox/hq ADR 0437 §9). The router reads the
+// verb at the fifth part and the caller at the sixth, so the publisher is one token, the module's own name; it is
+// refused when empty or when it holds a dot, a wildcard or whitespace.
+func ConditionSubject(publisher string) (string, error) {
+	if publisher == "" {
+		return "", errors.New("asks: the condition's publisher is empty")
+	}
+	if strings.ContainsAny(publisher, ".*> \t\r\n") || strings.ContainsFunc(publisher, func(r rune) bool { return r <= ' ' }) {
+		return "", fmt.Errorf("asks: the condition's publisher %q is not one token: it holds a dot, a wildcard or whitespace", publisher)
+	}
+	return "mesh.seat." + Seat + ".accept.condition." + publisher, nil
+}
+
+// ConditionVersion is the version of the condition value this package writes and the only one it reads.
+const ConditionVersion = 1
+
+// ErrUnknownConditionVersion is what DecodeCondition returns for a value whose version is missing or not
+// ConditionVersion. Test it with errors.Is.
+var ErrUnknownConditionVersion = errors.New("asks: unknown condition version")
+
+// The words a condition's explanation opens with (novox/hq ADR 0253 §1).
+const (
+	NothingToDo = "Nothing for you to do."
+	NeedsYou    = "Needs you: "
+)
+
+// The severities and states of a condition.
+const (
+	SeverityUrgent  = "urgent"
+	SeverityWarning = "warning"
+	StateRaised     = "raised"
+	StateCleared    = "cleared"
+)
+
+// Scope is what a condition is about, under the controller's names.
+type Scope struct {
+	Scope   string `json:"scope"`
+	ID      string `json:"id"`
+	Machine string `json:"machine,omitempty"`
+}
+
+// Action is one answer a notification offers, under the controller's names.
+type Action struct {
+	Label     string            `json:"label"`
+	Verb      string            `json:"verb"`
+	Machine   string            `json:"machine,omitempty"`
+	Arguments map[string]string `json:"arguments,omitempty"`
+	Level     Level             `json:"level,omitempty"`
+}
+
+// Condition is a condition a module says to the operator, under the names of the controller's conditions: its key,
+// severity, subject, state (raised or cleared), summary and the plain words of ADR 0253. A cleared condition needs
+// only its key, subject and state. The router alone chooses the channels.
+type Condition struct {
+	V           int      `json:"v"`
+	Key         string   `json:"key"`
+	Severity    string   `json:"severity,omitempty"`
+	Subject     Scope    `json:"subject"`
+	State       string   `json:"state"`
+	Summary     string   `json:"summary,omitempty"`
+	Headline    string   `json:"headline,omitempty"`
+	Needs       string   `json:"needs,omitempty"`
+	Explanation string   `json:"explanation,omitempty"`
+	Resolved    string   `json:"resolved,omitempty"`
+	Actions     []Action `json:"actions,omitempty"`
+}
+
+func (c Condition) check() error {
+	if c.Key == "" {
+		return errors.New("asks: a condition has no key")
+	}
+	if c.Subject.Scope == "" || c.Subject.ID == "" {
+		return errors.New("asks: a condition's subject needs its scope and id")
+	}
+	switch c.State {
+	case StateCleared:
+		return nil
+	case StateRaised:
+	default:
+		return fmt.Errorf("asks: the condition's state %q is not raised or cleared", c.State)
+	}
+	switch c.Severity {
+	case SeverityUrgent, SeverityWarning:
+	default:
+		return fmt.Errorf("asks: the condition's severity %q is not urgent or warning", c.Severity)
+	}
+	if c.Headline == "" {
+		return errors.New("asks: a raised condition has no headline")
+	}
+	if n := len([]rune(c.Headline)); n > HeadlineLength {
+		return fmt.Errorf("asks: the condition's headline is %d characters, more than %d", n, HeadlineLength)
+	}
+	verdict := NothingToDo
+	if c.Needs != "" {
+		verdict = NeedsYou + c.Needs
+	}
+	if !strings.HasPrefix(c.Explanation, verdict) {
+		return fmt.Errorf("asks: the condition's explanation must open with %q", verdict)
+	}
+	return nil
+}
+
+// EncodeCondition is c as JSON at ConditionVersion; it refuses what ADR 0253 and ADR 0437 §9 refuse.
+func EncodeCondition(c Condition) ([]byte, error) {
+	if err := c.check(); err != nil {
+		return nil, err
+	}
+	c.V = ConditionVersion
+	return json.Marshal(c)
+}
+
+// DecodeCondition reads a condition; it refuses an unknown version (ErrUnknownConditionVersion) and what
+// EncodeCondition refuses.
+func DecodeCondition(data []byte) (Condition, error) {
+	var probe struct {
+		V *int `json:"v"`
+	}
+	if err := json.Unmarshal(data, &probe); err != nil {
+		return Condition{}, fmt.Errorf("asks: the condition is not JSON: %w", err)
+	}
+	if probe.V == nil {
+		return Condition{}, fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownConditionVersion, ConditionVersion)
+	}
+	if *probe.V != ConditionVersion {
+		return Condition{}, fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownConditionVersion, *probe.V, ConditionVersion)
+	}
+	var c Condition
+	if err := json.Unmarshal(data, &c); err != nil {
+		return Condition{}, fmt.Errorf("asks: the condition does not fit its shape: %w", err)
+	}
+	if err := c.check(); err != nil {
+		return Condition{}, err
+	}
+	return c, nil
+}
+
+// SameCondition says whether two conditions are the same value.
+func SameCondition(a, b Condition) bool { return reflect.DeepEqual(a, b) }

go/asks/condition_test.go

@@ -0,0 +1,143 @@
+package asks
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestConditionSubject(t *testing.T) { // 1202/AC1
+	s, err := ConditionSubject("mesh-delivery-teller")
+	if err != nil || s != "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller" {
+		t.Fatalf("subject = %q, %v", s, err)
+	}
+	// The router splits an accept subject in six parts and reads the verb at 4 and the caller at 5.
+	parts := strings.Split(s, ".")
+	if len(parts) != 6 || parts[4] != "condition" || parts[5] != "mesh-delivery-teller" {
+		t.Errorf("subject %q does not keep the router's shape: %v", s, parts)
+	}
+	for _, bad := range []string{"", "a.b", "a*", "a>", "a b", "a\tb", "a\nb"} {
+		if _, err := ConditionSubject(bad); err == nil {
+			t.Errorf("publisher %q accepted", bad)
+		}
+	}
+}
+
+func raisedCondition() Condition {
+	return Condition{
+		Key: "delivery.stuck.mesh-mail", Severity: "warning",
+		Subject: Scope{Scope: "module", ID: "mesh-mail", Machine: "ace"}, State: "raised",
+		Summary: "plan 12 stopped", Headline: "Delivery of mesh-mail is stuck",
+		Needs: "approve the merge.", Explanation: "Needs you: approve the merge. The plan waits.",
+		Resolved: "The delivery of mesh-mail moved on.",
+		Actions: []Action{{Label: "Approve", Verb: "mesh-controller.assign", Machine: "ace",
+			Arguments: map[string]string{"why": ""}, Level: Approve}},
+	}
+}
+
+func clearedCondition() Condition {
+	return Condition{Key: "delivery.stuck.mesh-mail", Subject: Scope{Scope: "module", ID: "mesh-mail"}, State: "cleared"}
+}
+
+func TestConditionRoundTrip(t *testing.T) { // 1202/AC2
+	for _, c := range []Condition{raisedCondition(), clearedCondition()} {
+		data, err := EncodeCondition(c)
+		if err != nil {
+			t.Fatalf("%s: %v", c.State, err)
+		}
+		var probe map[string]any
+		if err := json.Unmarshal(data, &probe); err != nil || probe["v"] != float64(1) {
+			t.Fatalf("%s: %s does not carry v:1", c.State, data)
+		}
+		c.V = ConditionVersion
+		got, err := DecodeCondition(data)
+		if err != nil || !SameCondition(got, c) {
+			t.Errorf("%s round trip: %+v %v", c.State, got, err)
+		}
+	}
+	data, _ := EncodeCondition(raisedCondition())
+	var probe map[string]any
+	_ = json.Unmarshal(data, &probe)
+	for _, name := range []string{"key", "severity", "subject", "summary", "headline", "needs", "explanation", "resolved", "actions", "state"} {
+		if _, ok := probe[name]; !ok {
+			t.Errorf("the JSON has no %q: %s", name, data)
+		}
+	}
+	subject := probe["subject"].(map[string]any)
+	for _, name := range []string{"scope", "id", "machine"} {
+		if _, ok := subject[name]; !ok {
+			t.Errorf("the subject has no %q: %s", name, data)
+		}
+	}
+	action := probe["actions"].([]any)[0].(map[string]any)
+	for _, name := range []string{"label", "verb", "machine", "arguments", "level"} {
+		if _, ok := action[name]; !ok {
+			t.Errorf("the action has no %q: %s", name, data)
+		}
+	}
+	for _, in := range []string{`{"key":"k"}`, `{"v":2,"key":"k"}`} {
+		if _, err := DecodeCondition([]byte(in)); !errors.Is(err, ErrUnknownConditionVersion) {
+			t.Errorf("decode of %s = %v, want ErrUnknownConditionVersion", in, err)
+		}
+	}
+}
+
+func TestConditionVerdict(t *testing.T) { // 1202/AC3
+	c := raisedCondition()
+	c.Needs, c.Explanation = "", "The plan waits."
+	_, err := EncodeCondition(c)
+	if err == nil || !strings.Contains(err.Error(), "Nothing for you to do.") {
+		t.Errorf("empty needs without its verdict: %v", err)
+	}
+	c.Explanation = "Nothing for you to do. The plan waits."
+	if _, err := EncodeCondition(c); err != nil {
+		t.Errorf("empty needs with its verdict refused: %v", err)
+	}
+	c = raisedCondition()
+	c.Explanation = "Nothing for you to do. The plan waits."
+	_, err = EncodeCondition(c)
+	if err == nil || !strings.Contains(err.Error(), "Needs you: approve the merge.") {
+		t.Errorf("needs set with the wrong verdict: %v", err)
+	}
+	c.Explanation = "Needs you: something else. The plan waits."
+	if _, err = EncodeCondition(c); err == nil {
+		t.Error("needs set with other needs in the verdict accepted")
+	}
+}
+
+func TestConditionRefused(t *testing.T) { // 1202/AC4
+	mutate := func(f func(*Condition)) error {
+		c := raisedCondition()
+		f(&c)
+		_, err := EncodeCondition(c)
+		return err
+	}
+	for name, f := range map[string]func(*Condition){
+		"severity":      func(c *Condition) { c.Severity = "critical" },
+		"state":         func(c *Condition) { c.State = "open" },
+		"key":           func(c *Condition) { c.Key = "" },
+		"headline":      func(c *Condition) { c.Headline = "" },
+		"long headline": func(c *Condition) { c.Headline = strings.Repeat("h", HeadlineLength+1) },
+	} {
+		if err := mutate(f); err == nil {
+			t.Errorf("a condition with a bad %s accepted", name)
+		}
+	}
+	if err := mutate(func(c *Condition) { c.Headline = strings.Repeat("h", HeadlineLength) }); err != nil {
+		t.Errorf("a headline of the greatest length refused: %v", err)
+	}
+	if _, err := EncodeCondition(clearedCondition()); err != nil {
+		t.Errorf("a cleared condition with only key, subject and state refused: %v", err)
+	}
+	c := clearedCondition()
+	c.Key = ""
+	if _, err := EncodeCondition(c); err == nil {
+		t.Error("a cleared condition without its key accepted")
+	}
+	c = clearedCondition()
+	c.State = "gone"
+	if _, err := EncodeCondition(c); err == nil {
+		t.Error("a condition with an unknown state accepted")
+	}
+}

go/delivery/buildrecord.go

@@ -0,0 +1,41 @@
+package delivery
+
+import "time"
+
+// BuildRecord is the layout mesh-controller internal/chain/records.go Build writes (line 91), the fields a reader outside
+// the controller reads (novox/hq ADR 0437 §4 §5, issue 1223). The controller's record carries more; a decode ignores
+// what is not here. Path is the module's directory in its repository, as the controller writes it (Build.Path, records.go:100,
+// set from the build request trimmed of slashes at internal/inventory/chain_requests.go:19); empty for a module whose
+// manifest is the repository's root. SourceInputs has no json keys of its own: it is written under its Go field names, as the controller
+// writes it.
+type BuildRecord struct {
+	ID                string        `json:"id"`
+	State             string        `json:"state,omitempty"`
+	Repository        string        `json:"repository"`
+	Ref               string        `json:"ref,omitempty"`
+	Module            string        `json:"module,omitempty"`
+	Path              string        `json:"path,omitempty"`
+	Commit            string        `json:"commit,omitempty"`
+	At                time.Time     `json:"at,omitzero"`
+	SourceFingerprint string        `json:"source-fingerprint,omitempty"`
+	SourceInputs      *SourceInputs `json:"source-inputs,omitempty"`
+}
+
+// BuildsBucket is the bucket the builds are kept in: mesh-controller internal/broker/controller_buckets.go:101,
+// BucketName(ControllerSeat, ChainBuilds), with ControllerSeat "mesh-controller" (internal/broker/streams.go:359) and
+// ChainBuilds "builds" (controller_buckets.go:113).
+const BuildsBucket = "mesh-controller_builds"
+
+// BuildKey is a build's record under its module, a build that knew none under the request's key: copied from
+// mesh-controller internal/chain/keys.go:35-40 (BuildKey, with RequestKey at line 42).
+func BuildKey(module, id string) string {
+	if module == "" {
+		return "req." + id
+	}
+	return module + ".build." + id
+}
+
+// NewestKey is the key of a module's newest build, of any state: copied from mesh-controller internal/chain/keys.go:44.
+// The key holds no record: its value is the newest build's id as raw bytes (internal/inventory/chain_builds.go:210 and
+// :546), and the record lives at BuildKey(module, id). The pointer is a hint, checked against that record.
+func NewestKey(module string) string { return module + ".newest" }

go/delivery/buildrecord_test.go

@@ -0,0 +1,123 @@
+package delivery
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+	"time"
+)
+
+// The fixture's path is what the controller writes for a build request: strings.Trim(path, "/") at mesh-controller
+// internal/inventory/chain_requests.go:19, the module's directory in its repository (Build.Path, internal/chain/records.go:100).
+// The fixture is a record the controller's own encoder wrote (json.Marshal of a chain.Build, mesh-controller
+// internal/chain, a throwaway test run once and deleted) with build-agent's build-1792395447088936520 values, plus a
+// manifest-digest and one made artifact, which a BuildRecord does not carry.
+func loadBuildFixture(t *testing.T) []byte {
+	t.Helper()
+	raw, err := os.ReadFile("testdata/build-record-build-agent.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+func TestBuildRecordDecodes(t *testing.T) { // 1223/AC1
+	var r BuildRecord
+	if err := json.Unmarshal(loadBuildFixture(t), &r); err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range []struct{ name, got, want string }{
+		{"id", r.ID, "build-1792395447088936520"},
+		{"module", r.Module, "build-agent"},
+		{"repository", r.Repository, "novox/mesh-controller"},
+		{"ref", r.Ref, "main"},
+		{"commit", r.Commit, "7a74cee83f88e6585a5511c3cc53bdf0f5303d20"},
+		{"state", r.State, "built"},
+		{"source-fingerprint", r.SourceFingerprint, "src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7"},
+	} {
+		if c.got != c.want {
+			t.Errorf("%s = %q, want %q", c.name, c.got, c.want)
+		}
+	}
+	if want := time.Date(2026, 10, 19, 9, 38, 0, 0, time.UTC); !r.At.Equal(want) {
+		t.Errorf("at = %v, want %v", r.At, want)
+	}
+	if r.SourceInputs == nil {
+		t.Fatal("source-inputs missing")
+	}
+	fp, none := SourceFingerprint(*r.SourceInputs)
+	if none != "" || fp != r.SourceFingerprint {
+		t.Errorf("SourceFingerprint of the decoded inputs = %q (none %q), the record says %q", fp, none, r.SourceFingerprint)
+	}
+}
+
+func TestBuildRecordPath(t *testing.T) { // 1233/AC1
+	var r BuildRecord
+	if err := json.Unmarshal(loadBuildFixture(t), &r); err != nil {
+		t.Fatal(err)
+	}
+	if r.Path != "modules/build-agent" {
+		t.Errorf("path = %q, want modules/build-agent", r.Path)
+	}
+	// mesh-controller's manifest is its repository's root: the controller writes no path (omitempty), so the path is "".
+	var m map[string]json.RawMessage
+	if err := json.Unmarshal(loadBuildFixture(t), &m); err != nil {
+		t.Fatal(err)
+	}
+	delete(m, "path")
+	raw, _ := json.Marshal(m)
+	var root BuildRecord
+	if err := json.Unmarshal(raw, &root); err != nil {
+		t.Fatal(err)
+	}
+	if root.Path != "" {
+		t.Errorf("path of a root module = %q, want empty", root.Path)
+	}
+	// Every field 1223/AC1 checks is still equal in both.
+	if root.ID != r.ID || root.Module != r.Module || root.Repository != r.Repository || root.Ref != r.Ref ||
+		root.Commit != r.Commit || root.State != r.State || root.SourceFingerprint != r.SourceFingerprint || !root.At.Equal(r.At) {
+		t.Errorf("fields differ: %+v vs %+v", root, r)
+	}
+	// And the path survives a round trip through the SDK's own type.
+	back, _ := json.Marshal(r)
+	var again BuildRecord
+	if err := json.Unmarshal(back, &again); err != nil || again.Path != r.Path {
+		t.Errorf("round trip path = %q (%v)", again.Path, err)
+	}
+}
+
+func TestBuildRecordWithoutInputs(t *testing.T) { // 1223/AC2
+	var m map[string]json.RawMessage
+	if err := json.Unmarshal(loadBuildFixture(t), &m); err != nil {
+		t.Fatal(err)
+	}
+	delete(m, "source-inputs")
+	raw, _ := json.Marshal(m)
+	var r BuildRecord
+	if err := json.Unmarshal(raw, &r); err != nil {
+		t.Fatal(err)
+	}
+	if r.SourceInputs != nil {
+		t.Errorf("source-inputs = %+v, want nil", r.SourceInputs)
+	}
+	if r.Module != "build-agent" {
+		t.Errorf("module = %q, want build-agent", r.Module)
+	}
+}
+
+func TestBuildKeys(t *testing.T) { // 1223/AC3
+	// Literal values of mesh-controller internal/broker/controller_buckets.go:101 (BucketName(ControllerSeat, ChainBuilds)
+	// is "mesh-controller_builds") and internal/chain/keys.go:35-44.
+	if BuildsBucket != "mesh-controller_builds" {
+		t.Errorf("BuildsBucket = %q", BuildsBucket)
+	}
+	if got := BuildKey("build-agent", "build-1"); got != "build-agent.build.build-1" {
+		t.Errorf("BuildKey = %q", got)
+	}
+	if got := BuildKey("", "build-1"); got != "req.build-1" {
+		t.Errorf("BuildKey without a module = %q", got)
+	}
+	if got := NewestKey("build-agent"); got != "build-agent.newest" {
+		t.Errorf("NewestKey = %q", got)
+	}
+}

go/delivery/delivery.go

@@ -0,0 +1,390 @@
+// Package delivery is the one home of the keys and values the parts of delivery read and write on the bus
+// (novox/hq ADR 0429 §2: the key layouts another repository reads live in mesh-sdk; research 063 section 01). The
+// node-engine writes the have keys, the watchman of the repository mesh-delivery writes presence and pulse, the
+// view reads all of them, and the controller grants them; every one imports this package, and nothing else defines
+// these names.
+//
+// **Buckets** are named `<owner module>_<name>` as the broker names a module's state (provider.BucketOf): the
+// node-engine's `have`, and the watchman's `presence` and `pulse`.
+//
+// **Keys.** A have key is `<node>.<module>`, and a node's own facts are `<node>.node`, so a module may not be named
+// `node`. A presence key is `<node>` for a node and `part.<part>.<instance>` for a part's instance, so a node may
+// not be named `part`. A pulse key is `<bucket>.<key>`: the full name of the bucket watched, then that bucket's key,
+// which may hold dots, so it splits at the first dot. A name is refused when it is empty or holds a dot, a wildcard,
+// whitespace or a character a bus key refuses, and the error names it.
+//
+// **Values** are JSON of under about 500 bytes, each carrying its version in `v`. A reader refuses a value whose
+// version is missing or one it does not know, with an error that is ErrUnknownVersion (research 063 section 05
+// question 9 item 7): a writer writes a new version only once every reader's have shows a version that reads it. A
+// writer writes only when the new value differs from the one before (the Same functions).
+//
+// **Source, candidates, modules and the forge** (issue 1200; ADR 0437 §5, release 3): the gate's bucket `source` with
+// its one key `main` (every repository's merged head and the group that moved it), the indexer's `candidates.<group>`
+// and `modules.<module>` (a module's source fingerprint, or none and why), and the forge bridge's messages named
+// `forge.<owner>.<name>.<kind>` (source.go).
+//
+// **Merges, asks, owed and answers** (issue 1201; ADR 0437 §6, §8): the gate's `merges.<group>`, its asks
+// `merge.<group>.<heads hash>` (the router's ask whole), its owed keys `<repository>.<number>.<head>.<act>` that the
+// forge bridge performs, and the teller's answer under the ask's own key (gate.go).
+//
+// **The source fingerprint** (issue 1203; ADR 0437 §5): one function of named inputs that the controller's builder and
+// the indexer both call, so they agree byte for byte (fingerprint.go).
+//
+// Standard library only, and this module's own asks package.
+package delivery
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+)
+
+// The buckets of delivery's parts. Each is written by one owner: every node's node-engine writes the have keys of
+// its own node only, and the watchman writes presence and pulse.
+const (
+	HaveBucket     = "node-engine_have"
+	PresenceBucket = WatchmanModule + "_presence"
+	PulseBucket    = WatchmanModule + "_pulse"
+)
+
+// HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063
+// section 01).
+const (
+	HaveHistory     = 20
+	PresenceHistory = 20
+)
+
+// PulseMoveBound is how long a move may take: a pulse key lives for this time minus the move's age, and the bucket
+// expires it on its own (the server writes a delete marker), so a move that overruns it is seen to stall.
+const PulseMoveBound = 120 * time.Second
+
+// Version is the version of every value this package writes and the only one it reads.
+const Version = 1
+
+// ErrUnknownVersion is what a decode returns for a value whose version is missing or not Version. Test it with
+// errors.Is; the message names the version seen and the one read.
+var ErrUnknownVersion = errors.New("delivery: unknown value version")
+
+// Phase is how far a module's change has come on a node.
+type Phase string
+
+// The phases of a module's change on a node.
+const (
+	PhaseBackingUp Phase = "backing-up"
+	PhaseApplying  Phase = "applying"
+	PhaseChecking  Phase = "checking"
+	PhaseRunning   Phase = "running"
+	PhaseFailed    Phase = "failed"
+)
+
+// Health is what the module's own check says.
+type Health string
+
+// The healths a have carries.
+const (
+	HealthOK      Health = "ok"
+	HealthFailed  Health = "failed"
+	HealthWaits   Health = "waits"
+	HealthUnknown Health = "unknown"
+)
+
+var nameOK = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+$`)
+
+// checkName refuses a name that is empty or holds a dot, a wildcard, whitespace or a character a bus key refuses.
+func checkName(kind, name string) error {
+	if name == "" {
+		return fmt.Errorf("delivery: the %s name is empty", kind)
+	}
+	if !nameOK.MatchString(name) {
+		return fmt.Errorf("delivery: the %s name %q holds a dot, a wildcard, whitespace or a character a key refuses", kind, name)
+	}
+	return nil
+}
+
+// HaveKey is the key, in HaveBucket, of a module's have on a node: `<node>.<module>`. A module named `node` is
+// refused, since `<node>.node` is the node's own facts.
+func HaveKey(node, module string) (string, error) {
+	if err := checkName("node", node); err != nil {
+		return "", err
+	}
+	if err := checkName("module", module); err != nil {
+		return "", err
+	}
+	if module == "node" {
+		return "", fmt.Errorf(`delivery: the module name "node" is refused: %s.node is the node's own key`, node)
+	}
+	return node + "." + module, nil
+}
+
+// HaveNodeKey is the key, in HaveBucket, of a node's own facts: `<node>.node`.
+func HaveNodeKey(node string) (string, error) {
+	if err := checkName("node", node); err != nil {
+		return "", err
+	}
+	return node + ".node", nil
+}
+
+// ParseHaveKey is the inverse of HaveKey and HaveNodeKey: isNode is true for `<node>.node`, with module empty.
+func ParseHaveKey(key string) (node, module string, isNode bool, err error) {
+	node, module, ok := strings.Cut(key, ".")
+	if !ok {
+		return "", "", false, fmt.Errorf("delivery: %q is not a have key, <node>.<module>", key)
+	}
+	if err := checkName("node", node); err != nil {
+		return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
+	}
+	if err := checkName("module", module); err != nil {
+		return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
+	}
+	if module == "node" {
+		return node, "", true, nil
+	}
+	return node, module, false, nil
+}
+
+// PresenceNodeKey is the key, in PresenceBucket, of a node: `<node>`. A node named `part` is refused.
+func PresenceNodeKey(node string) (string, error) {
+	if err := checkName("node", node); err != nil {
+		return "", err
+	}
+	if node == "part" {
+		return "", errors.New(`delivery: the node name "part" is refused: part.<part>.<instance> is a part's key`)
+	}
+	return node, nil
+}
+
+// PresencePartKey is the key, in PresenceBucket, of a part's instance: `part.<part>.<instance>`.
+func PresencePartKey(part, instance string) (string, error) {
+	if err := checkName("part", part); err != nil {
+		return "", err
+	}
+	if err := checkName("instance", instance); err != nil {
+		return "", err
+	}
+	return "part." + part + "." + instance, nil
+}
+
+// ParsePresenceKey is the inverse of PresenceNodeKey and PresencePartKey: for a node's key part and instance are
+// empty, for a part's key node is.
+func ParsePresenceKey(key string) (node, part, instance string, err error) {
+	if !strings.Contains(key, ".") {
+		node, err = PresenceNodeKey(key)
+		if err != nil {
+			return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
+		}
+		return node, "", "", nil
+	}
+	fields := strings.Split(key, ".")
+	if len(fields) != 3 || fields[0] != "part" {
+		return "", "", "", fmt.Errorf("delivery: %q is not a presence key, <node> or part.<part>.<instance>", key)
+	}
+	if _, err := PresencePartKey(fields[1], fields[2]); err != nil {
+		return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
+	}
+	return "", fields[1], fields[2], nil
+}
+
+// PulseKey is the key, in PulseBucket, of the move watched at key of bucket: `<bucket>.<key>`. The bucket is the
+// full bucket name and holds no dot; the key may.
+func PulseKey(bucket, key string) (string, error) {
+	if err := checkName("bucket", bucket); err != nil {
+		return "", err
+	}
+	if key == "" || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") || strings.Contains(key, "..") {
+		return "", fmt.Errorf("delivery: the pulse key %q is empty or has an empty part", key)
+	}
+	for _, part := range strings.Split(key, ".") {
+		if err := checkName("key", part); err != nil {
+			return "", err
+		}
+	}
+	return bucket + "." + key, nil
+}
+
+// ParsePulseKey is the inverse of PulseKey, splitting at the first dot.
+func ParsePulseKey(pulseKey string) (bucket, key string, err error) {
+	bucket, key, ok := strings.Cut(pulseKey, ".")
+	if !ok {
+		return "", "", fmt.Errorf("delivery: %q is not a pulse key, <bucket>.<key>", pulseKey)
+	}
+	if _, err := PulseKey(bucket, key); err != nil {
+		return "", "", fmt.Errorf("delivery: %q is not a pulse key: %w", pulseKey, err)
+	}
+	return bucket, key, nil
+}
+
+// PulseTTL is the time to live of a pulse key for a move of the given age: PulseMoveBound minus the age. At or past
+// the bound the move is stalled, with no time left and nothing to write: the watchman raises it at once.
+func PulseTTL(moveAge time.Duration) (ttl time.Duration, stalled bool) {
+	if moveAge >= PulseMoveBound {
+		return 0, true
+	}
+	return PulseMoveBound - moveAge, false
+}
+
+// Have is a module's state on a node, written by that node's node-engine under HaveKey. FP is the module's
+// fingerprint once the indexer exists (design 47 release 3); until then Build is the build the node was sent, and
+// the view compares by build.
+type Have struct {
+	V      int    `json:"v"`
+	FP     string `json:"fp,omitempty"`
+	Build  string `json:"build,omitempty"`
+	Phase  Phase  `json:"phase"`
+	Health Health `json:"health"`
+	Detail string `json:"detail,omitempty"`
+}
+
+// HaveNode is a node's own facts, written under HaveNodeKey. Each field is the node-engine's own JSON, carried whole:
+// this package names the facts and does not type them.
+type HaveNode struct {
+	V       int             `json:"v"`
+	Windows json.RawMessage `json:"windows,omitempty"`
+	Busy    json.RawMessage `json:"busy,omitempty"`
+	Strays  json.RawMessage `json:"strays,omitempty"`
+	Filters json.RawMessage `json:"filters,omitempty"`
+	Profile json.RawMessage `json:"profile,omitempty"`
+}
+
+// Presence is whether a node or a part's instance is online; the entry's own time says since when. The watchman
+// writes it.
+type Presence struct {
+	V      int  `json:"v"`
+	Online bool `json:"online"`
+}
+
+// Pulse is the value of a pulse key: the watched key mirrored, its revision, and who owes the next move (a node or
+// a part instance). The watchman writes it with the time to live PulseTTL gives.
+type Pulse struct {
+	V        int    `json:"v"`
+	Bucket   string `json:"bucket"`
+	Key      string `json:"key"`
+	Revision uint64 `json:"revision"`
+	Owes     string `json:"owes,omitempty"`
+}
+
+func checkVersion(data []byte) error {
+	var probe struct {
+		V *int `json:"v"`
+	}
+	if err := json.Unmarshal(data, &probe); err != nil {
+		return fmt.Errorf("delivery: value is not JSON: %w", err)
+	}
+	if probe.V == nil {
+		return fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownVersion, Version)
+	}
+	if *probe.V != Version {
+		return fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownVersion, *probe.V, Version)
+	}
+	return nil
+}
+
+func decode(data []byte, into any) error {
+	if err := checkVersion(data); err != nil {
+		return err
+	}
+	if err := json.Unmarshal(data, into); err != nil {
+		return fmt.Errorf("delivery: value does not fit its shape: %w", err)
+	}
+	return nil
+}
+
+func (h Have) check() error {
+	switch h.Phase {
+	case PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed:
+	default:
+		return fmt.Errorf("delivery: phase %q is not backing-up, applying, checking, running or failed", h.Phase)
+	}
+	switch h.Health {
+	case HealthOK, HealthFailed, HealthWaits, HealthUnknown:
+	default:
+		return fmt.Errorf("delivery: health %q is not ok, failed, waits or unknown", h.Health)
+	}
+	return nil
+}
+
+// EncodeHave is h as JSON at Version; it refuses a phase or health outside the lists.
+func EncodeHave(h Have) ([]byte, error) {
+	if err := h.check(); err != nil {
+		return nil, err
+	}
+	h.V = Version
+	return json.Marshal(h)
+}
+
+// DecodeHave reads a have; it refuses an unknown version (ErrUnknownVersion) and a phase or health outside the lists.
+func DecodeHave(data []byte) (Have, error) {
+	var h Have
+	if err := decode(data, &h); err != nil {
+		return Have{}, err
+	}
+	if err := h.check(); err != nil {
+		return Have{}, err
+	}
+	return h, nil
+}
+
+// EncodeHaveNode is n as JSON at Version.
+func EncodeHaveNode(n HaveNode) ([]byte, error) {
+	n.V = Version
+	return json.Marshal(n)
+}
+
+// DecodeHaveNode reads a node's facts; it refuses an unknown version (ErrUnknownVersion).
+func DecodeHaveNode(data []byte) (HaveNode, error) {
+	var n HaveNode
+	err := decode(data, &n)
+	return n, err
+}
+
+// EncodePresence is p as JSON at Version.
+func EncodePresence(p Presence) ([]byte, error) {
+	p.V = Version
+	return json.Marshal(p)
+}
+
+// DecodePresence reads a presence; it refuses an unknown version (ErrUnknownVersion).
+func DecodePresence(data []byte) (Presence, error) {
+	var p Presence
+	err := decode(data, &p)
+	return p, err
+}
+
+// EncodePulse is p as JSON at Version.
+func EncodePulse(p Pulse) ([]byte, error) {
+	p.V = Version
+	return json.Marshal(p)
+}
+
+// DecodePulse reads a pulse; it refuses an unknown version (ErrUnknownVersion).
+func DecodePulse(data []byte) (Pulse, error) {
+	var p Pulse
+	err := decode(data, &p)
+	return p, err
+}
+
+// SameHave says whether two haves are the same value, so a writer writes only when they differ.
+func SameHave(a, b Have) bool { return a == b }
+
+// SameHaveNode says whether two node facts are the same value, comparing each carried JSON without its spacing.
+func SameHaveNode(a, b HaveNode) bool {
+	return a.V == b.V && sameJSON(a.Windows, b.Windows) && sameJSON(a.Busy, b.Busy) &&
+		sameJSON(a.Strays, b.Strays) && sameJSON(a.Filters, b.Filters) && sameJSON(a.Profile, b.Profile)
+}
+
+// SamePresence says whether two presences are the same value.
+func SamePresence(a, b Presence) bool { return a == b }
+
+// SamePulse says whether two pulses are the same value.
+func SamePulse(a, b Pulse) bool { return a == b }
+
+func sameJSON(a, b json.RawMessage) bool {
+	var ca, cb bytes.Buffer
+	if json.Compact(&ca, a) != nil || json.Compact(&cb, b) != nil {
+		return bytes.Equal(a, b)
+	}
+	return bytes.Equal(ca.Bytes(), cb.Bytes())
+}

go/delivery/delivery_test.go

@@ -0,0 +1,294 @@
+package delivery
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+// 1131/AC1: the have key forms and parses back; bad names are refused, naming them.
+func TestHaveKey(t *testing.T) {
+	if HaveBucket != "node-engine_have" {
+		t.Fatalf("HaveBucket = %q", HaveBucket)
+	}
+	key, err := HaveKey("ace", "plex")
+	if err != nil || key != "ace.plex" {
+		t.Fatalf("HaveKey = %q, %v", key, err)
+	}
+	node, module, isNode, err := ParseHaveKey(key)
+	if err != nil || node != "ace" || module != "plex" || isNode {
+		t.Fatalf("ParseHaveKey = %q %q %v %v", node, module, isNode, err)
+	}
+	if got, err := HaveNodeKey("ace"); err != nil || got != "ace.node" {
+		t.Fatalf("HaveNodeKey = %q, %v", got, err)
+	}
+	if _, err := HaveNodeKey("a b"); err == nil {
+		t.Error("HaveNodeKey with a space accepted")
+	}
+	node, module, isNode, err = ParseHaveKey("ace.node")
+	if err != nil || node != "ace" || module != "" || !isNode {
+		t.Fatalf("ParseHaveKey node = %q %q %v %v", node, module, isNode, err)
+	}
+	for _, c := range []struct{ node, module, named string }{
+		{"ace", "node", "node"},
+		{"ace", "", "module"},
+		{"", "plex", "node"},
+		{"a.b", "plex", "a.b"},
+		{"ace", "p*", "p*"},
+		{"ace", "p>", "p>"},
+		{"ace", "p x", "p x"},
+		{"ace", "p\tx", "module"},
+	} {
+		_, err := HaveKey(c.node, c.module)
+		if err == nil || !strings.Contains(err.Error(), c.named) {
+			t.Errorf("HaveKey(%q, %q) = %v, want an error naming %q", c.node, c.module, err, c.named)
+		}
+	}
+	for _, bad := range []string{"", "ace", "ace.", ".plex", "ace.plex.x", "ace.p x"} {
+		if _, _, _, err := ParseHaveKey(bad); err == nil {
+			t.Errorf("ParseHaveKey(%q) accepted", bad)
+		}
+	}
+}
+
+// 1131/AC2: the presence keys form and parse back; a node named part is refused.
+func TestPresenceKey(t *testing.T) {
+	if PresenceBucket != "mesh-delivery-watchman_presence" {
+		t.Fatalf("PresenceBucket = %q", PresenceBucket)
+	}
+	key, err := PresenceNodeKey("ace")
+	if err != nil || key != "ace" {
+		t.Fatalf("PresenceNodeKey = %q, %v", key, err)
+	}
+	node, part, instance, err := ParsePresenceKey(key)
+	if err != nil || node != "ace" || part != "" || instance != "" {
+		t.Fatalf("ParsePresenceKey node = %q %q %q %v", node, part, instance, err)
+	}
+	key, err = PresencePartKey("view", "novox")
+	if err != nil || key != "part.view.novox" {
+		t.Fatalf("PresencePartKey = %q, %v", key, err)
+	}
+	node, part, instance, err = ParsePresenceKey(key)
+	if err != nil || node != "" || part != "view" || instance != "novox" {
+		t.Fatalf("ParsePresenceKey part = %q %q %q %v", node, part, instance, err)
+	}
+	if _, err := PresenceNodeKey("part"); err == nil || !strings.Contains(err.Error(), "part") {
+		t.Errorf("PresenceNodeKey(part) = %v", err)
+	}
+	if _, err := PresenceNodeKey("a.b"); err == nil {
+		t.Error("PresenceNodeKey(a.b) accepted")
+	}
+	if _, err := PresencePartKey("", "x"); err == nil {
+		t.Error("PresencePartKey with empty part accepted")
+	}
+	if _, err := PresencePartKey("view", "x y"); err == nil {
+		t.Error("PresencePartKey with a space accepted")
+	}
+	for _, bad := range []string{"", "part", "part.view", "part.view.a.b", "a.b", "part..x"} {
+		if _, _, _, err := ParsePresenceKey(bad); err == nil {
+			t.Errorf("ParsePresenceKey(%q) accepted", bad)
+		}
+	}
+}
+
+// 1131/AC3: the pulse key splits at the first dot, the key holding dots.
+func TestPulseKey(t *testing.T) {
+	if PulseBucket != "mesh-delivery-watchman_pulse" {
+		t.Fatalf("PulseBucket = %q", PulseBucket)
+	}
+	key, err := PulseKey("node-engine_have", "ace.plex")
+	if err != nil || key != "node-engine_have.ace.plex" {
+		t.Fatalf("PulseKey = %q, %v", key, err)
+	}
+	bucket, rest, err := ParsePulseKey(key)
+	if err != nil || bucket != "node-engine_have" || rest != "ace.plex" {
+		t.Fatalf("ParsePulseKey = %q %q %v", bucket, rest, err)
+	}
+	for _, c := range [][2]string{{"", "k"}, {"a.b", "k"}, {"b", ""}, {"b", "k x"}, {"b", ".k"}, {"b", "k."}, {"b", "k*"}} {
+		if _, err := PulseKey(c[0], c[1]); err == nil {
+			t.Errorf("PulseKey(%q, %q) accepted", c[0], c[1])
+		}
+	}
+	for _, bad := range []string{"", "nodot", ".k", "b."} {
+		if _, _, err := ParsePulseKey(bad); err == nil {
+			t.Errorf("ParsePulseKey(%q) accepted", bad)
+		}
+	}
+}
+
+// 1131/AC4: each value encodes, decodes equal, and carries "v":1.
+func TestRoundTrip(t *testing.T) {
+	hasV1 := func(t *testing.T, b []byte) {
+		t.Helper()
+		var m map[string]any
+		if err := json.Unmarshal(b, &m); err != nil {
+			t.Fatal(err)
+		}
+		if m["v"] != float64(1) {
+			t.Fatalf("v = %v in %s", m["v"], b)
+		}
+		if len(b) > 500 {
+			t.Fatalf("value is %d bytes", len(b))
+		}
+	}
+	t.Run("have", func(t *testing.T) {
+		in := Have{Build: "b1", FP: "f1", Phase: PhaseRunning, Health: HealthOK, Detail: "up"}
+		b, err := EncodeHave(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		hasV1(t, b)
+		out, err := DecodeHave(b)
+		if err != nil {
+			t.Fatal(err)
+		}
+		in.V = Version
+		if out != in {
+			t.Fatalf("got %+v want %+v", out, in)
+		}
+	})
+	t.Run("have node", func(t *testing.T) {
+		in := HaveNode{Windows: json.RawMessage(`{"a":1}`), Busy: json.RawMessage(`true`)}
+		b, err := EncodeHaveNode(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		hasV1(t, b)
+		out, err := DecodeHaveNode(b)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !SameHaveNode(out, HaveNode{V: Version, Windows: in.Windows, Busy: in.Busy}) {
+			t.Fatalf("got %+v", out)
+		}
+	})
+	t.Run("presence", func(t *testing.T) {
+		b, err := EncodePresence(Presence{Online: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+		hasV1(t, b)
+		out, err := DecodePresence(b)
+		if err != nil || out != (Presence{V: Version, Online: true}) {
+			t.Fatalf("got %+v %v", out, err)
+		}
+	})
+	t.Run("pulse", func(t *testing.T) {
+		in := Pulse{Bucket: "node-engine_have", Key: "ace.plex", Revision: 7, Owes: "ace"}
+		b, err := EncodePulse(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		hasV1(t, b)
+		out, err := DecodePulse(b)
+		in.V = Version
+		if err != nil || out != in {
+			t.Fatalf("got %+v %v", out, err)
+		}
+	})
+}
+
+// 1131/AC5: a missing or unknown version is refused as ErrUnknownVersion, naming both versions.
+func TestUnknownVersion(t *testing.T) {
+	decoders := map[string]func([]byte) error{
+		"have":      func(b []byte) error { _, err := DecodeHave(b); return err },
+		"have node": func(b []byte) error { _, err := DecodeHaveNode(b); return err },
+		"presence":  func(b []byte) error { _, err := DecodePresence(b); return err },
+		"pulse":     func(b []byte) error { _, err := DecodePulse(b); return err },
+	}
+	for name, dec := range decoders {
+		t.Run(name, func(t *testing.T) {
+			err := dec([]byte(`{"v":2,"phase":"running","health":"ok"}`))
+			if !errors.Is(err, ErrUnknownVersion) || !strings.Contains(err.Error(), "2") || !strings.Contains(err.Error(), "1") {
+				t.Errorf("v 2: %v", err)
+			}
+			err = dec([]byte(`{"phase":"running","health":"ok"}`))
+			if !errors.Is(err, ErrUnknownVersion) {
+				t.Errorf("missing v: %v", err)
+			}
+			if err := dec([]byte(`not json`)); err == nil || errors.Is(err, ErrUnknownVersion) {
+				t.Errorf("garbage: %v", err)
+			}
+		})
+	}
+}
+
+// 1131/AC6: a phase or health outside the lists is refused on encode and decode.
+func TestPhaseHealthRefused(t *testing.T) {
+	for _, p := range []Phase{PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed} {
+		for _, h := range []Health{HealthOK, HealthFailed, HealthWaits, HealthUnknown} {
+			if _, err := EncodeHave(Have{Phase: p, Health: h}); err != nil {
+				t.Errorf("%s/%s refused: %v", p, h, err)
+			}
+		}
+	}
+	if _, err := EncodeHave(Have{Phase: "sleeping", Health: HealthOK}); err == nil || !strings.Contains(err.Error(), "sleeping") {
+		t.Errorf("bad phase encoded: %v", err)
+	}
+	if _, err := EncodeHave(Have{Phase: PhaseRunning, Health: "fine"}); err == nil || !strings.Contains(err.Error(), "fine") {
+		t.Errorf("bad health encoded: %v", err)
+	}
+	if _, err := EncodeHave(Have{Phase: PhaseRunning}); err == nil {
+		t.Error("empty health encoded")
+	}
+	if _, err := DecodeHave([]byte(`{"v":1,"phase":"sleeping","health":"ok"}`)); err == nil {
+		t.Error("bad phase decoded")
+	}
+	if _, err := DecodeHave([]byte(`{"v":1,"phase":"running","health":"fine"}`)); err == nil {
+		t.Error("bad health decoded")
+	}
+}
+
+// 1131/AC7: the pulse's time to live is 120 seconds minus the move's age; at the bound the move is stalled.
+func TestPulseTTL(t *testing.T) {
+	for _, c := range []struct {
+		age     time.Duration
+		ttl     time.Duration
+		stalled bool
+	}{
+		{0, 120 * time.Second, false},
+		{30 * time.Second, 90 * time.Second, false},
+		{119 * time.Second, time.Second, false},
+		{120 * time.Second, 0, true},
+		{200 * time.Second, 0, true},
+	} {
+		ttl, stalled := PulseTTL(c.age)
+		if ttl != c.ttl || stalled != c.stalled {
+			t.Errorf("PulseTTL(%v) = %v, %v; want %v, %v", c.age, ttl, stalled, c.ttl, c.stalled)
+		}
+	}
+	if PulseMoveBound != 120*time.Second || HaveHistory != 20 || PresenceHistory != 20 {
+		t.Error("constants")
+	}
+}
+
+// 1131/AC8: equal values say same, different ones do not.
+func TestSame(t *testing.T) {
+	a := Have{V: 1, Build: "b", Phase: PhaseRunning, Health: HealthOK}
+	b := a
+	if !SameHave(a, b) {
+		t.Error("equal haves differ")
+	}
+	b.Detail = "x"
+	if SameHave(a, b) {
+		t.Error("different haves same")
+	}
+	n := HaveNode{V: 1, Busy: json.RawMessage(`true`)}
+	if !SameHaveNode(n, HaveNode{V: 1, Busy: json.RawMessage(`true`)}) || SameHaveNode(n, HaveNode{V: 1}) {
+		t.Error("have node")
+	}
+	if !SamePresence(Presence{V: 1, Online: true}, Presence{V: 1, Online: true}) || SamePresence(Presence{V: 1}, Presence{V: 1, Online: true}) {
+		t.Error("presence")
+	}
+	p := Pulse{V: 1, Bucket: "b", Key: "k", Revision: 1, Owes: "ace"}
+	q := p
+	if !SamePulse(p, q) {
+		t.Error("equal pulses differ")
+	}
+	q.Revision = 2
+	if SamePulse(p, q) {
+		t.Error("different pulses same")
+	}
+}

go/delivery/fingerprint.go

@@ -0,0 +1,63 @@
+package delivery
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sort"
+	"strings"
+)
+
+// SourceInputs are what a module's source fingerprint is of (novox/hq ADR 0437 §5): the module, the git tree of
+// its directory, its bases (references, of which only the digest counts), the root tree of each repository its
+// artifacts read (by artifact), each artifact's toolchain line, and the reason the source does not pin the build
+// when it does not (a build that resolves packages when built). A context is the value the builder records for it:
+// a repository's root tree, or for an image that compiles a Go program the `narrow:` hash of the build source it was
+// handed (mesh-controller internal/builder/builder.go:1146-1156).
+type SourceInputs struct {
+	Module     string
+	Tree       string
+	Bases      []string
+	Contexts   map[string]string
+	Toolchains map[string]string
+	Unpinned   string
+}
+
+// SourceFingerprint is the module's source fingerprint, `src1:` and the hex sha256 of its inputs' lines, or none
+// and the reason when the source does not pin the build: Unpinned when set, "no tree" when Tree is empty. It is the
+// one function the controller's builder and the indexer both call, so they agree byte for byte. The lines are
+// "module <m>" and "tree <t>", then "base <digest>" (once each), "context <artifact> <tree>" and "toolchain
+// <artifact> <line>" sorted, all joined by a newline.
+func SourceFingerprint(in SourceInputs) (fp, none string) {
+	if in.Unpinned != "" {
+		return "", in.Unpinned
+	}
+	if in.Tree == "" {
+		return "", "no tree"
+	}
+	lines := []string{"module " + in.Module, "tree " + in.Tree}
+	bases := map[string]bool{}
+	for _, b := range in.Bases {
+		bases[ReferenceDigest(b)] = true
+	}
+	for b := range bases {
+		lines = append(lines, "base "+b)
+	}
+	for a, t := range in.Contexts {
+		lines = append(lines, "context "+a+" "+t)
+	}
+	for a, t := range in.Toolchains {
+		lines = append(lines, "toolchain "+a+" "+t)
+	}
+	sort.Strings(lines[2:])
+	sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
+	return SourcePrefix + hex.EncodeToString(sum[:]), ""
+}
+
+// ReferenceDigest is a reference's digest, `sha256:...`, so the registry address a base was copied into does not
+// enter the fingerprint; the reference itself when it carries none.
+func ReferenceDigest(reference string) string {
+	if _, digest, pinned := strings.Cut(reference, "@"); pinned && digest != "" {
+		return digest
+	}
+	return reference
+}

go/delivery/fingerprint_test.go

@@ -0,0 +1,178 @@
+package delivery
+
+import (
+	"strings"
+	"testing"
+)
+
+func inputs() SourceInputs {
+	return SourceInputs{
+		Module: "mesh-mail", Tree: "t-module",
+		Bases:      []string{"registry.example/base@sha256:aaa", "registry.example/other@sha256:bbb"},
+		Contexts:   map[string]string{"mail": "t-root", "tool": "t-root2"},
+		Toolchains: map[string]string{"mail": "go sha256:ccc 0123456789abcdef", "tool": "go sha256:ccc fedcba9876543210"},
+	}
+}
+
+func mustFP(t *testing.T, in SourceInputs) string {
+	t.Helper()
+	fp, none := SourceFingerprint(in)
+	if fp == "" || none != "" {
+		t.Fatalf("fingerprint = %q, none = %q", fp, none)
+	}
+	return fp
+}
+
+func TestFingerprintForm(t *testing.T) {
+	fp := mustFP(t, inputs())
+	hex := strings.TrimPrefix(fp, SourcePrefix)
+	if SourcePrefix != "src1:" || hex == fp || len(hex) != 64 || strings.Trim(hex, "0123456789abcdef") != "" {
+		t.Errorf("fingerprint %q is not src1: and 64 hex digits", fp)
+	}
+}
+
+func TestFingerprintOrderAndDuplicates(t *testing.T) { // 1203/AC2
+	want := mustFP(t, inputs())
+	in := inputs()
+	in.Bases = []string{in.Bases[1], in.Bases[0]}
+	if got := mustFP(t, in); got != want {
+		t.Errorf("the order of bases changes it: %s", got)
+	}
+	in = inputs()
+	in.Bases = append(in.Bases, in.Bases[0])
+	if got := mustFP(t, in); got != want {
+		t.Errorf("a base given twice changes it: %s", got)
+	}
+	in = inputs()
+	in.Bases = append(in.Bases, "other.registry.example/base@sha256:aaa")
+	if got := mustFP(t, in); got != want {
+		t.Errorf("two references to one digest change it: %s", got)
+	}
+	// Maps have no order, but build them in many insertion orders anyway.
+	for i := 0; i < 20; i++ {
+		in = inputs()
+		in.Contexts = map[string]string{}
+		in.Toolchains = map[string]string{}
+		for _, k := range []string{"tool", "mail"} {
+			in.Contexts[k] = inputs().Contexts[k]
+			in.Toolchains[k] = inputs().Toolchains[k]
+		}
+		if got := mustFP(t, in); got != want {
+			t.Fatalf("map order changes it: %s", got)
+		}
+	}
+}
+
+func TestFingerprintChanges(t *testing.T) { // 1203/AC3
+	want := mustFP(t, inputs())
+	for name, f := range map[string]func(*SourceInputs){
+		"module":           func(in *SourceInputs) { in.Module = "other" },
+		"tree":             func(in *SourceInputs) { in.Tree = "t-other" },
+		"base":             func(in *SourceInputs) { in.Bases[0] = "registry.example/base@sha256:zzz" },
+		"a base added":     func(in *SourceInputs) { in.Bases = append(in.Bases, "x@sha256:new") },
+		"context":          func(in *SourceInputs) { in.Contexts["mail"] = "t-changed" },
+		"context artifact": func(in *SourceInputs) { in.Contexts["extra"] = "t-root" },
+		"toolchain":        func(in *SourceInputs) { in.Toolchains["tool"] = "go sha256:ccc changed" },
+	} {
+		in := inputs()
+		f(&in)
+		if got := mustFP(t, in); got == want {
+			t.Errorf("a changed %s keeps the fingerprint", name)
+		}
+	}
+}
+
+func TestFingerprintNone(t *testing.T) { // 1203/AC4
+	in := inputs()
+	in.Unpinned = "the build resolves packages when built"
+	fp, none := SourceFingerprint(in)
+	if fp != "" || none != in.Unpinned {
+		t.Errorf("unpinned: fp = %q, none = %q", fp, none)
+	}
+	in = inputs()
+	in.Tree = ""
+	fp, none = SourceFingerprint(in)
+	if fp != "" || none != "no tree" {
+		t.Errorf("no tree: fp = %q, none = %q", fp, none)
+	}
+	in = inputs()
+	in.Tree, in.Unpinned = "", "the build resolves packages when built"
+	if _, none = SourceFingerprint(in); none != in.Unpinned {
+		t.Errorf("the reason for unpinned stands over no tree: %q", none)
+	}
+}
+
+func TestReferenceDigest(t *testing.T) {
+	for in, want := range map[string]string{
+		"registry.example/base@sha256:aaa": "sha256:aaa",
+		"registry.example/base":            "registry.example/base",
+		"registry.example/base@":           "registry.example/base@",
+		"":                                 "",
+	} {
+		if got := ReferenceDigest(in); got != want {
+			t.Errorf("ReferenceDigest(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+// 1203/AC1: three builds the live builder recorded a source fingerprint for. Each value is captured, never written
+// by hand: the recorded fingerprint was read with mesh-controller.builds (after issues 1204 and 1208 were
+// delivered); each tree is `git rev-parse <commit>:<dir>` in the repository's clone; the bases are from the
+// build's `[bases]` log lines; the toolchain lines are mesh-controller's toolchainOf over its toolchains values.
+func TestFingerprintLiveBuilds(t *testing.T) {
+	const ts = "typescript sha256:4b81682df5665c1f461d3f720a3a0fd7593f4aa71e4fc272dbfdc4bbd235b489 20125bd590041690"
+	for _, c := range []struct {
+		name     string
+		in       SourceInputs
+		recorded string
+	}{
+		{
+			// mailu, build-1792391134620131370: mesh-catalog at b4527d9091e972f28cb365623d9280045e7741a3, dir modules/mailu.
+			name: "mailu",
+			in: SourceInputs{
+				Module:     "mailu",
+				Tree:       "cfc6fa8ff5fc1e220132e51ade653b9d3f387b3e",
+				Bases:      []string{"sha256:25f3cfeaceca14921366af4d1240b56457ef46273bdb508c7b0e8f469f6fd228"},
+				Toolchains: map[string]string{"code": ts},
+			},
+			recorded: "src1:539f2e64000d436c572ef53b40e524012455945f8c5a1c6bb173ee0707e88ef4",
+		},
+		{
+			// radarr, build-1792391147149753355: mesh-media-catalog at 197abb061d9b77d67c7f11e9e567528ea7a85436,
+			// dir modules/radarr; no bases.
+			name: "radarr",
+			in: SourceInputs{
+				Module: "radarr",
+				Tree:   "42b9c918b5af75cba2bbc75845012e7f64e16df9",
+				Toolchains: map[string]string{
+					"code":     ts,
+					"tools-go": "go sha256:2054591ca034b68323d87ec887ba34f56819cb3364034360c0777da4b314dd54 0995d85edc9a6a23",
+				},
+			},
+			recorded: "src1:b3c150d39c395b99dd68974a48490e9a42467aa5d486d0c604d6ef1e8e9c910a",
+		},
+		{
+			// build-agent, build-1792395447088936520: mesh-controller at 7a74cee83f88e6585a5511c3cc53bdf0f5303d20,
+			// dir modules/build-agent; the context is narrowTree's hash of the 185 paths GoBuildSource gave for
+			// cmd/mesh-builder at 7a74cee8.
+			name: "build-agent",
+			in: SourceInputs{
+				Module: "build-agent",
+				Tree:   "11c0b4f90cae0d6e4ef9cc48cbfdf9f6dbe52bca",
+				Bases: []string{
+					"sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc",
+					"sha256:8ac98ca534ac3f51e1f420a1dd2c15e74c75cfa0f23f3ad27eb5d7236c349a0c",
+				},
+				Contexts: map[string]string{"server": "narrow:d51ca744fd11af59d616c858f76c27b3e58e16e11fbdc0c720eb9d7d19d55fac"},
+			},
+			recorded: "src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7",
+		},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			fp, none := SourceFingerprint(c.in)
+			if none != "" || fp != c.recorded {
+				t.Errorf("fingerprint = %q (none %q), the live builder recorded %q", fp, none, c.recorded)
+			}
+		})
+	}
+}

go/delivery/gate.go

@@ -0,0 +1,404 @@
+package delivery
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"regexp"
+	"strconv"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+)
+
+// The gate's buckets and the teller's (ADR 0437 §6, §8). The gate writes merges (each pull request's expected
+// head, its state and each merge commit), its asks (`merge.<group>.<heads hash>`, the router's ask whole) and owed
+// (one key per act it owes the forge, which the forge bridge performs once per revision). The teller writes its
+// answer under the ask's own key in its bucket answers.
+const (
+	MergesBucket   = GateModule + "_merges"
+	AsksBucket     = GateModule + "_asks"
+	OwedBucket     = GateModule + "_owed"
+	AnswersBucket  = TellerModule + "_answers"
+	MergesHistory  = 10
+	AsksHistory    = 5
+	OwedHistory    = 5
+	AnswersHistory = 5
+)
+
+// AskKindMerge is the kind of the gate's ask for a group's merge, the first part of its key.
+const AskKindMerge = "merge"
+
+// The acts the gate owes the forge, the last part of an owed key.
+const (
+	OwedStatus  = "status"
+	OwedView    = "view"
+	OwedNote    = "note"
+	OwedComment = "comment"
+)
+
+// The states of a pull request in a merge.
+type MergeState string
+
+// The states a merge's pull request is in.
+const (
+	MergeMerging MergeState = "merging"
+	MergeMerged  MergeState = "merged"
+	MergeRefused MergeState = "refused"
+)
+
+// The states a status the gate owes the forge may carry.
+var statusStates = []string{"pending", "success", "failure", "error", "warning"}
+
+var hashOK = regexp.MustCompile(`^[0-9a-f]{12}$`)
+
+// MergesKey is the key, in MergesBucket, of a group: `<group>`.
+func MergesKey(group string) (string, error) {
+	if err := checkName("group", group); err != nil {
+		return "", err
+	}
+	return group, nil
+}
+
+// MergeAskKey is the key, in AsksBucket, of the ask for merging a group at these heads: `merge.<group>.<heads
+// hash>`. The teller's answer has the same key in AnswersBucket (AnswerKey).
+func MergeAskKey(group string, heads []Head) (string, error) {
+	if err := checkName("group", group); err != nil {
+		return "", err
+	}
+	hash, err := HeadsHash(heads)
+	if err != nil {
+		return "", err
+	}
+	return AskKindMerge + "." + group + "." + hash, nil
+}
+
+// ParseAskKey is the inverse of MergeAskKey: the kind, the group and the heads hash.
+func ParseAskKey(key string) (kind, group, hash string, err error) {
+	parts := strings.Split(key, ".")
+	if len(parts) != 3 || parts[0] != AskKindMerge {
+		return "", "", "", fmt.Errorf("delivery: %q is not an ask key, merge.<group>.<heads hash>", key)
+	}
+	if err := checkName("group", parts[1]); err != nil {
+		return "", "", "", fmt.Errorf("delivery: %q is not an ask key: %w", key, err)
+	}
+	if !hashOK.MatchString(parts[2]) {
+		return "", "", "", fmt.Errorf("delivery: %q is not an ask key: %q is not 12 hex digits", key, parts[2])
+	}
+	return parts[0], parts[1], parts[2], nil
+}
+
+// AnswerKey is the key, in AnswersBucket, of the teller's answer to the ask at askKey: the same string.
+func AnswerKey(askKey string) string { return askKey }
+
+// AskID is the id the teller gives the router for the ask at askKey. The router's ids refuse dots (asks.UsableID),
+// so in the readable part a dot becomes `-` and any other byte outside letters, digits, `_` and `-` becomes `_`; that part is cut to fit, and every id ends in `-` and
+// the first 12 hex digits of the sha256 of the whole key, so two keys never give one id. It cannot be read back:
+// the teller keeps the key beside it.
+func AskID(askKey string) string {
+	const most = 64 // the router's bound on an id
+	const tail = 13 // "-" and 12 hex digits
+	readable := []byte(askKey)
+	for i, c := range readable {
+		switch {
+		case c >= 'A' && c <= 'Z', c >= 'a' && c <= 'z', c >= '0' && c <= '9':
+		case c == '.':
+			readable[i] = '-'
+		default: // a slash, an equals sign or any other byte the router's id refuses
+			readable[i] = '_'
+		}
+	}
+	if len(readable) > 0 && (readable[0] == '-' || readable[0] == '_') {
+		readable[0] = 'k' // an id starts with a letter or digit
+	}
+	if len(readable) > most-tail {
+		readable = readable[:most-tail]
+	}
+	sum := sha256.Sum256([]byte(askKey))
+	return string(readable) + "-" + hex.EncodeToString(sum[:])[:12]
+}
+
+func checkOwedAct(act string) (kind, context string, err error) {
+	kind, context, hasContext := strings.Cut(act, "=")
+	switch kind {
+	case OwedStatus:
+		if !hasContext {
+			return "", "", fmt.Errorf("delivery: the owed act %q is a status without its context, status=<context>", act)
+		}
+		if err := checkName("status context", context); err != nil {
+			return "", "", err
+		}
+		return kind, context, nil
+	case OwedView, OwedNote, OwedComment:
+		if hasContext {
+			return "", "", fmt.Errorf("delivery: the owed act %q takes no context", act)
+		}
+		return kind, "", nil
+	}
+	return "", "", fmt.Errorf("delivery: the owed act %q is not status=<context>, view, note or comment", act)
+}
+
+// ParseOwedAct splits an owed act into its kind and, for a status, its context.
+func ParseOwedAct(act string) (kind, context string, err error) { return checkOwedAct(act) }
+
+// OwedKey is the key, in OwedBucket, of an act the gate owes the forge on a pull request at a head:
+// `<repository>.<number>.<head>.<act>`, the act being `status=<context>` (mesh/merge-gate), `view`, `note` or
+// `comment`.
+func OwedKey(repository string, number int, head, act string) (string, error) {
+	if err := checkRepository(repository); err != nil {
+		return "", err
+	}
+	if number <= 0 {
+		return "", fmt.Errorf("delivery: an owed key needs the pull request number, not %d", number)
+	}
+	if err := checkName("head", head); err != nil {
+		return "", err
+	}
+	if _, _, err := checkOwedAct(act); err != nil {
+		return "", err
+	}
+	return repository + "." + strconv.Itoa(number) + "." + head + "." + act, nil
+}
+
+// ParseOwedKey is the inverse of OwedKey; act is whole, as `status=<context>` for a status.
+func ParseOwedKey(key string) (repository string, number int, head, act string, err error) {
+	parts := strings.Split(key, ".")
+	if len(parts) != 4 {
+		return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key, <repository>.<number>.<head>.<act>", key)
+	}
+	number, convErr := strconv.Atoi(parts[1])
+	if convErr != nil {
+		return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key: %q is not a pull request number", key, parts[1])
+	}
+	if _, err := OwedKey(parts[0], number, parts[2], parts[3]); err != nil {
+		return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key: %w", key, err)
+	}
+	return parts[0], number, parts[2], parts[3], nil
+}
+
+// Merge is the value of a merges key: the ask approved and, for each pull request, the head the approval expects
+// and how far its merge has come.
+type Merge struct {
+	V     int         `json:"v"`
+	Group string      `json:"group"`
+	Ask   string      `json:"ask"`
+	Pulls []MergePull `json:"pulls"`
+}
+
+// MergePull is one pull request of a merge: its expected head, its state, the merge commit once merged, the
+// forge's words when refused.
+type MergePull struct {
+	Head
+	State  MergeState `json:"state"`
+	Commit string     `json:"commit,omitempty"`
+	Words  string     `json:"words,omitempty"`
+}
+
+func (m Merge) check() error {
+	if err := checkName("group", m.Group); err != nil {
+		return err
+	}
+	if m.Ask == "" {
+		return fmt.Errorf("delivery: a merge names no ask")
+	}
+	if len(m.Pulls) == 0 {
+		return fmt.Errorf("delivery: a merge names no pull request")
+	}
+	for _, p := range m.Pulls {
+		who := fmt.Sprintf("%s#%d", p.Repository, p.Number)
+		switch p.State {
+		case MergeMerging:
+		case MergeMerged:
+			if p.Commit == "" {
+				return fmt.Errorf("delivery: the merged pull request %s has no merge commit", who)
+			}
+		case MergeRefused:
+			if p.Words == "" {
+				return fmt.Errorf("delivery: the refused pull request %s has no words of the forge", who)
+			}
+		default:
+			return fmt.Errorf("delivery: the state %q of %s is not merging, merged or refused", p.State, who)
+		}
+	}
+	return nil
+}
+
+// Ask is the value of an asks key: the router's ask whole as the gate composes it, so the teller publishes it
+// unchanged, and the group and heads it was composed for.
+type Ask struct {
+	V     int      `json:"v"`
+	Group string   `json:"group"`
+	Heads []Head   `json:"heads"`
+	Ask   asks.Ask `json:"ask"`
+}
+
+func (a Ask) check() error {
+	key, err := MergeAskKey(a.Group, a.Heads)
+	if err != nil {
+		return fmt.Errorf("delivery: an ask needs its group and at least one pull request's head: %w", err)
+	}
+	if want := AskID(key); a.Ask.ID != want {
+		return fmt.Errorf("delivery: the ask's id %q is not %q, the id its group and heads give", a.Ask.ID, want)
+	}
+	if len(a.Ask.Options) == 0 {
+		return fmt.Errorf("delivery: the ask has no option")
+	}
+	return nil
+}
+
+// Answer is the teller's answer to an ask: the option chosen, the warrant's id and its words.
+type Answer struct {
+	V       int    `json:"v"`
+	Option  string `json:"option"`
+	Warrant string `json:"warrant"`
+	Words   string `json:"words"`
+}
+
+func (a Answer) check() error {
+	for _, f := range []struct{ name, value string }{{"option", a.Option}, {"warrant", a.Warrant}, {"words", a.Words}} {
+		if f.value == "" {
+			return fmt.Errorf("delivery: an answer has no %s", f.name)
+		}
+	}
+	return nil
+}
+
+// Owed is the value of an owed key: what the forge bridge performs. A status needs Context and State, a view and a
+// comment need Text, a note needs Text and Ref (the notes ref, mesh-plan).
+type Owed struct {
+	V       int    `json:"v"`
+	Act     string `json:"act"`
+	Context string `json:"context,omitempty"`
+	State   string `json:"state,omitempty"`
+	Text    string `json:"text,omitempty"`
+	Ref     string `json:"ref,omitempty"`
+}
+
+func (o Owed) check() error {
+	switch o.Act {
+	case OwedStatus:
+		if o.Context == "" {
+			return fmt.Errorf("delivery: an owed status has no context")
+		}
+		for _, s := range statusStates {
+			if o.State == s {
+				return nil
+			}
+		}
+		return fmt.Errorf("delivery: the owed status state %q is not %s", o.State, strings.Join(statusStates, ", "))
+	case OwedView, OwedComment:
+		if o.Text == "" {
+			return fmt.Errorf("delivery: an owed %s has no text", o.Act)
+		}
+	case OwedNote:
+		if o.Text == "" {
+			return fmt.Errorf("delivery: an owed note has no text")
+		}
+		if o.Ref == "" {
+			return fmt.Errorf("delivery: an owed note has no ref")
+		}
+	default:
+		return fmt.Errorf("delivery: the owed act %q is not status, view, note or comment", o.Act)
+	}
+	return nil
+}
+
+// EncodeMerge is m as JSON at Version; it refuses a merged pull request without its merge commit, a refused one
+// without the forge's words and a state outside the list.
+func EncodeMerge(m Merge) ([]byte, error) {
+	if err := m.check(); err != nil {
+		return nil, err
+	}
+	m.V = Version
+	return json.Marshal(m)
+}
+
+// DecodeMerge reads a merge; it refuses an unknown version (ErrUnknownVersion) and what EncodeMerge refuses.
+func DecodeMerge(data []byte) (Merge, error) {
+	var m Merge
+	if err := decode(data, &m); err != nil {
+		return Merge{}, err
+	}
+	if err := m.check(); err != nil {
+		return Merge{}, err
+	}
+	return m, nil
+}
+
+// EncodeAsk is a as JSON at Version; it refuses an ask whose id is not the one its group and heads give, or with
+// no option.
+func EncodeAsk(a Ask) ([]byte, error) {
+	if err := a.check(); err != nil {
+		return nil, err
+	}
+	a.V = Version
+	return json.Marshal(a)
+}
+
+// DecodeAsk reads an ask; it refuses an unknown version (ErrUnknownVersion) and what EncodeAsk refuses.
+func DecodeAsk(data []byte) (Ask, error) {
+	var a Ask
+	if err := decode(data, &a); err != nil {
+		return Ask{}, err
+	}
+	if err := a.check(); err != nil {
+		return Ask{}, err
+	}
+	return a, nil
+}
+
+// EncodeAnswer is a as JSON at Version; it refuses an answer missing its option, warrant or words.
+func EncodeAnswer(a Answer) ([]byte, error) {
+	if err := a.check(); err != nil {
+		return nil, err
+	}
+	a.V = Version
+	return json.Marshal(a)
+}
+
+// DecodeAnswer reads an answer; it refuses an unknown version (ErrUnknownVersion) and what EncodeAnswer refuses.
+func DecodeAnswer(data []byte) (Answer, error) {
+	var a Answer
+	if err := decode(data, &a); err != nil {
+		return Answer{}, err
+	}
+	if err := a.check(); err != nil {
+		return Answer{}, err
+	}
+	return a, nil
+}
+
+// EncodeOwed is o as JSON at Version; it refuses an act missing what it needs.
+func EncodeOwed(o Owed) ([]byte, error) {
+	if err := o.check(); err != nil {
+		return nil, err
+	}
+	o.V = Version
+	return json.Marshal(o)
+}
+
+// DecodeOwed reads an owed act; it refuses an unknown version (ErrUnknownVersion) and what EncodeOwed refuses.
+func DecodeOwed(data []byte) (Owed, error) {
+	var o Owed
+	if err := decode(data, &o); err != nil {
+		return Owed{}, err
+	}
+	if err := o.check(); err != nil {
+		return Owed{}, err
+	}
+	return o, nil
+}
+
+// SameMerge says whether two merges are the same value, so a writer writes only when they differ.
+func SameMerge(a, b Merge) bool { return reflect.DeepEqual(a, b) }
+
+// SameAsk says whether two asks are the same value.
+func SameAsk(a, b Ask) bool { return reflect.DeepEqual(a, b) }
+
+// SameAnswer says whether two answers are the same value.
+func SameAnswer(a, b Answer) bool { return a == b }
+
+// SameOwed says whether two owed acts are the same value.
+func SameOwed(a, b Owed) bool { return a == b }

go/delivery/gate_test.go

@@ -0,0 +1,301 @@
+package delivery
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+)
+
+func gateHeads() []Head {
+	return []Head{
+		{Repository: "novox/mesh-sdk", Number: 36, Head: "aaaa"},
+		{Repository: "novox/mesh-control", Number: 7, Head: "bbbb"},
+	}
+}
+
+func TestMergeKeys(t *testing.T) { // 1201/AC1
+	if k, err := MergesKey("feat/x"); err != nil || k != "feat/x" {
+		t.Fatalf("merges key = %q, %v", k, err)
+	}
+	if _, err := MergesKey("a.b"); err == nil {
+		t.Error("a group with a dot accepted")
+	}
+	hh, _ := HeadsHash(gateHeads())
+	key, err := MergeAskKey("feat/x", gateHeads())
+	if err != nil || key != "merge.feat/x."+hh {
+		t.Fatalf("ask key = %q, %v", key, err)
+	}
+	kind, group, hash, err := ParseAskKey(key)
+	if err != nil || kind != AskKindMerge || group != "feat/x" || hash != hh {
+		t.Fatalf("parse ask key = %q %q %q %v", kind, group, hash, err)
+	}
+	if AnswerKey(key) != key {
+		t.Error("the answer's key is not the ask's key")
+	}
+	if _, err := MergeAskKey("feat/x", nil); err == nil {
+		t.Error("an ask key without heads accepted")
+	}
+	if _, err := MergeAskKey("", gateHeads()); err == nil {
+		t.Error("an ask key without a group accepted")
+	}
+	for _, bad := range []string{"", "merge", "merge.g", "merge.g.h.x", "other.g.0123456789ab", "merge..0123456789ab", "merge.g.nothex"} {
+		if _, _, _, err := ParseAskKey(bad); err == nil {
+			t.Errorf("parse ask key %q accepted", bad)
+		}
+	}
+}
+
+func TestAskID(t *testing.T) { // 1201/AC2
+	k1, _ := MergeAskKey("feat/x", gateHeads())
+	k2, _ := MergeAskKey("feat/y", gateHeads())
+	k3, _ := MergeAskKey("feat/x", gateHeads()[:1])
+	long, _ := MergeAskKey("feat/"+strings.Repeat("long-", 20), gateHeads())
+	long2, _ := MergeAskKey("feat/"+strings.Repeat("long-", 20)+"2", gateHeads())
+	slash, _ := MergeAskKey("a/b", gateHeads())
+	under, _ := MergeAskKey("a_b", gateHeads())
+	eq, _ := MergeAskKey("feat=x", gateHeads())
+	eqs, _ := MergeAskKey("feat_x", gateHeads())
+	seen := map[string]string{}
+	for _, k := range []string{k1, k2, k3, long, long2, slash, under, eq, eqs, "=odd.key"} {
+		id := AskID(k)
+		if strings.Contains(id, ".") {
+			t.Errorf("id %q holds a dot", id)
+		}
+		if !asks.UsableID(id) {
+			t.Errorf("id %q is not usable by the router", id)
+		}
+		if other, dup := seen[id]; dup {
+			t.Errorf("keys %q and %q give the id %q", k, other, id)
+		}
+		seen[id] = k
+	}
+	if AskID(k1) != AskID(k1) {
+		t.Error("the id is not stable")
+	}
+}
+
+func TestOwedKey(t *testing.T) { // 1201/AC3
+	for _, act := range []string{"status=mesh/merge-gate", "view", "note", "comment"} {
+		key, err := OwedKey("novox/mesh-sdk", 36, "aaaa", act)
+		if err != nil || key != "novox/mesh-sdk.36.aaaa."+act {
+			t.Fatalf("owed key for %s = %q, %v", act, key, err)
+		}
+		repo, number, head, gotAct, err := ParseOwedKey(key)
+		if err != nil || repo != "novox/mesh-sdk" || number != 36 || head != "aaaa" || gotAct != act {
+			t.Fatalf("parse %q = %q %d %q %q %v", key, repo, number, head, gotAct, err)
+		}
+	}
+	kind, context, err := ParseOwedAct("status=mesh/merge-gate")
+	if err != nil || kind != OwedStatus || context != "mesh/merge-gate" {
+		t.Errorf("owed act = %q %q %v", kind, context, err)
+	}
+	for _, bad := range []string{"", "merge", "status", "status=", "view=x", "status=a.b", "statuses=x"} {
+		if _, err := OwedKey("novox/mesh-sdk", 36, "aaaa", bad); err == nil {
+			t.Errorf("owed act %q accepted", bad)
+		}
+	}
+	if _, err := OwedKey("nope", 36, "aaaa", "view"); err == nil {
+		t.Error("a bad repository accepted")
+	}
+	if _, err := OwedKey("novox/mesh-sdk", 0, "aaaa", "view"); err == nil {
+		t.Error("pull request number 0 accepted")
+	}
+	if _, err := OwedKey("novox/mesh-sdk", 36, "", "view"); err == nil {
+		t.Error("an empty head accepted")
+	}
+	for _, bad := range []string{"", "novox/mesh-sdk.36.aaaa", "novox/mesh-sdk.x.aaaa.view", "novox/mesh-sdk.36.aaaa.merge", "a.b.36.aaaa.view"} {
+		if _, _, _, _, err := ParseOwedKey(bad); err == nil {
+			t.Errorf("parse owed key %q accepted", bad)
+		}
+	}
+}
+
+func sampleMerge() Merge {
+	return Merge{Group: "feat/x", Ask: "merge.feat/x.0123456789ab", Pulls: []MergePull{
+		{Head: gateHeads()[0], State: MergeMerged, Commit: "cccc"},
+		{Head: gateHeads()[1], State: MergeRefused, Words: "the forge says no"},
+		{Head: Head{Repository: "novox/other", Number: 1, Head: "dddd"}, State: MergeMerging},
+	}}
+}
+
+func sampleAsk() Ask {
+	key, _ := MergeAskKey("feat/x", gateHeads())
+	return Ask{Group: "feat/x", Heads: gateHeads(), Ask: asks.Ask{
+		ID: AskID(key), Headline: "Merge into main: group feat/x", Explanation: "The group passed its checks.",
+		Who: asks.Operator, Expires: time.Date(2026, 10, 20, 8, 0, 0, 0, time.UTC), OnExpiry: "nothing is merged",
+		Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: "sha256:00"}},
+	}}
+}
+
+func sampleOwed() []Owed {
+	return []Owed{
+		{Act: OwedStatus, Context: "mesh/merge-gate", State: "success"},
+		{Act: OwedView, Text: "the delivery reaches ace"},
+		{Act: OwedNote, Text: "plan: mesh-mail to ace", Ref: "mesh-plan"},
+		{Act: OwedComment, Text: "merged"},
+	}
+}
+
+func TestGateValuesRoundTrip(t *testing.T) { // 1201/AC4
+	hasV := func(name string, data []byte, err error) {
+		t.Helper()
+		if err != nil {
+			t.Fatalf("%s: %v", name, err)
+		}
+		var probe map[string]any
+		if json.Unmarshal(data, &probe) != nil || probe["v"] != float64(1) {
+			t.Fatalf("%s: %s does not carry v:1", name, data)
+		}
+	}
+	m := sampleMerge()
+	data, err := EncodeMerge(m)
+	hasV("merge", data, err)
+	m.V = Version
+	if got, err := DecodeMerge(data); err != nil || !SameMerge(got, m) {
+		t.Errorf("merge round trip: %+v %v", got, err)
+	}
+	a := sampleAsk()
+	data, err = EncodeAsk(a)
+	hasV("ask", data, err)
+	a.V = Version
+	if got, err := DecodeAsk(data); err != nil || !SameAsk(got, a) {
+		t.Errorf("ask round trip: %+v %v", got, err)
+	}
+	an := Answer{Option: "approve", Warrant: "w-1", Words: "approved on telegram"}
+	data, err = EncodeAnswer(an)
+	hasV("answer", data, err)
+	an.V = Version
+	if got, err := DecodeAnswer(data); err != nil || !SameAnswer(got, an) {
+		t.Errorf("answer round trip: %+v %v", got, err)
+	}
+	for _, o := range sampleOwed() {
+		data, err = EncodeOwed(o)
+		hasV("owed "+o.Act, data, err)
+		o.V = Version
+		if got, err := DecodeOwed(data); err != nil || !SameOwed(got, o) {
+			t.Errorf("owed %s round trip: %+v %v", o.Act, got, err)
+		}
+	}
+	for name, dec := range map[string]func([]byte) error{
+		"merge":  func(b []byte) error { _, err := DecodeMerge(b); return err },
+		"ask":    func(b []byte) error { _, err := DecodeAsk(b); return err },
+		"answer": func(b []byte) error { _, err := DecodeAnswer(b); return err },
+		"owed":   func(b []byte) error { _, err := DecodeOwed(b); return err },
+	} {
+		for _, in := range []string{`{"group":"g"}`, `{"v":2}`} {
+			if err := dec([]byte(in)); !errors.Is(err, ErrUnknownVersion) {
+				t.Errorf("%s decode of %s = %v, want ErrUnknownVersion", name, in, err)
+			}
+		}
+	}
+}
+
+func TestGateValuesRefused(t *testing.T) { // 1201/AC5
+	wantErr := func(name string, err error, words ...string) {
+		t.Helper()
+		if err == nil {
+			t.Errorf("%s accepted", name)
+			return
+		}
+		for _, w := range words {
+			if !strings.Contains(err.Error(), w) {
+				t.Errorf("%s: error %q does not name %q", name, err, w)
+			}
+		}
+	}
+	m := sampleMerge()
+	m.Pulls[0].Commit = ""
+	_, err := EncodeMerge(m)
+	wantErr("merged without a merge commit", err, "commit")
+	m = sampleMerge()
+	m.Pulls[1].Words = ""
+	_, err = EncodeMerge(m)
+	wantErr("refused without words", err, "words")
+	m = sampleMerge()
+	m.Pulls[2].State = "done"
+	_, err = EncodeMerge(m)
+	wantErr("unknown state", err, "done")
+	m = sampleMerge()
+	m.Pulls = nil
+	_, err = EncodeMerge(m)
+	wantErr("merge without pulls", err, "pull")
+	m = sampleMerge()
+	m.Ask = ""
+	_, err = EncodeMerge(m)
+	wantErr("merge without its ask", err, "ask")
+
+	a := sampleAsk()
+	a.Ask.ID = "something-else"
+	_, err = EncodeAsk(a)
+	wantErr("ask with another id", err, "something-else", AskID("merge.feat/x."+mustHash(gateHeads())))
+	a = sampleAsk()
+	a.Ask.Options = nil
+	_, err = EncodeAsk(a)
+	wantErr("ask without options", err, "option")
+	a = sampleAsk()
+	a.Heads = nil
+	_, err = EncodeAsk(a)
+	wantErr("ask without heads", err, "pull request")
+
+	for name, an := range map[string]Answer{
+		"option":  {Warrant: "w", Words: "x"},
+		"warrant": {Option: "approve", Words: "x"},
+		"words":   {Option: "approve", Warrant: "w"},
+	} {
+		_, err = EncodeAnswer(an)
+		wantErr("answer without "+name, err, name)
+	}
+
+	for name, o := range map[string]Owed{
+		"context": {Act: OwedStatus, State: "success"},
+		"state":   {Act: OwedStatus, Context: "mesh/merge-gate", State: "green"},
+		"text":    {Act: OwedView},
+		"ref":     {Act: OwedNote, Text: "line"},
+		"comment": {Act: OwedComment},
+		"act":     {Act: "merge"},
+	} {
+		_, err = EncodeOwed(o)
+		wantErr("owed missing "+name, err)
+	}
+	_, err = EncodeOwed(Owed{Act: OwedNote, Text: "line"})
+	wantErr("note without ref", err, "ref")
+	_, err = EncodeOwed(Owed{Act: OwedStatus, State: "success"})
+	wantErr("status without context", err, "context")
+}
+
+func mustHash(h []Head) string { s, _ := HeadsHash(h); return s }
+
+func TestGateSame(t *testing.T) { // 1201/AC6
+	if !SameMerge(sampleMerge(), sampleMerge()) || !SameAsk(sampleAsk(), sampleAsk()) {
+		t.Error("equal values differ")
+	}
+	m := sampleMerge()
+	m.Pulls[0].Commit = "other"
+	if SameMerge(m, sampleMerge()) {
+		t.Error("different merges are the same")
+	}
+	a := sampleAsk()
+	a.Ask.Headline = "other"
+	if SameAsk(a, sampleAsk()) {
+		t.Error("different asks are the same")
+	}
+	if !SameAnswer(Answer{Option: "a", Warrant: "w", Words: "x"}, Answer{Option: "a", Warrant: "w", Words: "x"}) ||
+		SameAnswer(Answer{Option: "a", Warrant: "w", Words: "x"}, Answer{Option: "b", Warrant: "w", Words: "x"}) {
+		t.Error("SameAnswer wrong")
+	}
+	if !SameOwed(sampleOwed()[0], sampleOwed()[0]) || SameOwed(sampleOwed()[0], sampleOwed()[1]) {
+		t.Error("SameOwed wrong")
+	}
+}
+
+func TestGateBuckets(t *testing.T) {
+	if MergesBucket != "mesh-delivery-gate_merges" || AsksBucket != "mesh-delivery-gate_asks" ||
+		OwedBucket != "mesh-delivery-gate_owed" || AnswersBucket != "mesh-delivery-teller_answers" ||
+		MergesHistory != 10 || AsksHistory != 5 || OwedHistory != 5 || AnswersHistory != 5 {
+		t.Error("a bucket or history differs from the decided layout")
+	}
+}

go/delivery/source.go

@@ -0,0 +1,420 @@
+package delivery
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+)
+
+// The modules of the parts release 3 adds (novox/hq ADR 0437): their buckets are their own state, `<module>_<name>`.
+const (
+	GateModule        = "mesh-delivery-gate"
+	IndexerModule     = "mesh-delivery-indexer"
+	ForgeBridgeModule = "mesh-delivery-forge-bridge"
+	TellerModule      = "mesh-delivery-teller"
+)
+
+// The gate's source bucket and the indexer's candidates and modules buckets (ADR 0437 §5). The source bucket has
+// one key, SourceMainKey: every repository's merged head and the group that moved it, written by the gate alone by
+// compare-and-set; its revision is the mesh's source revision. The indexer writes the other two.
+const (
+	SourceBucket     = GateModule + "_source"
+	SourceMainKey    = "main"
+	SourceHistory    = 64
+	CandidatesBucket = IndexerModule + "_candidates"
+	// CandidatesHistory is the revisions the candidates bucket keeps per key.
+	CandidatesHistory = 10
+	ModulesBucket     = IndexerModule + "_modules"
+	// ModulesHistory is the revisions the modules bucket keeps per key.
+	ModulesHistory = 20
+)
+
+// ForgeStream is the forge bridge's stream as its manifest declares it; the broker derives the real names.
+const ForgeStream = "forge"
+
+// The kinds of forge message, the last part of a forge event's name.
+const (
+	ForgePull    = "pull"
+	ForgeMerged  = "merged"
+	ForgeRefused = "refused"
+)
+
+// The actions of a pull message.
+const (
+	PullOpened  = "opened"
+	PullUpdated = "updated"
+	PullClosed  = "closed"
+)
+
+// SourcePrefix opens every source fingerprint.
+const SourcePrefix = "src1:"
+
+// CandidatesKey is the key, in CandidatesBucket, of a group: `<group>`, the branch name the group's pull requests
+// share (a slash is allowed); a lone pull request is a group of one under its branch name.
+func CandidatesKey(group string) (string, error) {
+	if err := checkName("group", group); err != nil {
+		return "", err
+	}
+	return group, nil
+}
+
+// ParseCandidatesKey is the inverse of CandidatesKey.
+func ParseCandidatesKey(key string) (group string, err error) {
+	if group, err = CandidatesKey(key); err != nil {
+		return "", fmt.Errorf("delivery: %q is not a candidates key: %w", key, err)
+	}
+	return group, nil
+}
+
+// ModulesKey is the key, in ModulesBucket, of a module: `<module>`.
+func ModulesKey(module string) (string, error) {
+	if err := checkName("module", module); err != nil {
+		return "", err
+	}
+	return module, nil
+}
+
+// ParseModulesKey is the inverse of ModulesKey.
+func ParseModulesKey(key string) (module string, err error) {
+	if module, err = ModulesKey(key); err != nil {
+		return "", fmt.Errorf("delivery: %q is not a modules key: %w", key, err)
+	}
+	return module, nil
+}
+
+// checkRepository refuses a repository that is not `<owner>/<name>`: exactly one slash, each part a valid name.
+func checkRepository(repository string) error {
+	owner, name, ok := strings.Cut(repository, "/")
+	if !ok || strings.Contains(name, "/") {
+		return fmt.Errorf("delivery: the repository %q is not <owner>/<name>, with exactly one slash", repository)
+	}
+	if err := checkName("repository owner", owner); err != nil {
+		return fmt.Errorf("delivery: the repository %q: %w", repository, err)
+	}
+	if err := checkName("repository name", name); err != nil {
+		return fmt.Errorf("delivery: the repository %q: %w", repository, err)
+	}
+	if strings.Contains(owner, "/") {
+		return fmt.Errorf("delivery: the repository %q is not <owner>/<name>, with exactly one slash", repository)
+	}
+	return nil
+}
+
+func checkForgeKind(kind string) error {
+	switch kind {
+	case ForgePull, ForgeMerged, ForgeRefused:
+		return nil
+	}
+	return fmt.Errorf("delivery: the forge kind %q is not pull, merged or refused", kind)
+}
+
+// ForgeEvent is the name of a forge message of the given kind: `forge.<owner>.<name>.<kind>`.
+func ForgeEvent(repository, kind string) (string, error) {
+	if err := checkRepository(repository); err != nil {
+		return "", err
+	}
+	if err := checkForgeKind(kind); err != nil {
+		return "", err
+	}
+	owner, name, _ := strings.Cut(repository, "/")
+	return "forge." + owner + "." + name + "." + kind, nil
+}
+
+// ParseForgeEvent is the inverse of ForgeEvent.
+func ParseForgeEvent(event string) (repository, kind string, err error) {
+	parts := strings.Split(event, ".")
+	if len(parts) != 4 || parts[0] != "forge" {
+		return "", "", fmt.Errorf("delivery: %q is not a forge event, forge.<owner>.<name>.<kind>", event)
+	}
+	repository = parts[1] + "/" + parts[2]
+	if _, err := ForgeEvent(repository, parts[3]); err != nil {
+		return "", "", fmt.Errorf("delivery: %q is not a forge event: %w", event, err)
+	}
+	return repository, parts[3], nil
+}
+
+// Head is one pull request at one commit: the repository as `<owner>/<name>` and the full sha.
+type Head struct {
+	Repository string `json:"repository"`
+	Number     int    `json:"number"`
+	Head       string `json:"head"`
+}
+
+// HeadsHash is the first 12 hex digits of the sha256 over the lines `<repository>#<number>=<head>`, sorted and
+// joined by a newline: the same for every order of heads, different when any head changes. An empty list is refused.
+func HeadsHash(heads []Head) (string, error) {
+	if len(heads) == 0 {
+		return "", fmt.Errorf("delivery: a heads hash needs at least one pull request")
+	}
+	lines := make([]string, len(heads))
+	for i, h := range heads {
+		lines[i] = fmt.Sprintf("%s#%d=%s", h.Repository, h.Number, h.Head)
+	}
+	sort.Strings(lines)
+	sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
+	return hex.EncodeToString(sum[:])[:12], nil
+}
+
+// Source is the value of SourceMainKey: every repository's merged head (repository to full sha, not only the
+// group's) and the group that moved it.
+type Source struct {
+	V     int               `json:"v"`
+	Group string            `json:"group"`
+	Heads map[string]string `json:"heads"`
+}
+
+func (s Source) check() error {
+	if len(s.Heads) == 0 {
+		return fmt.Errorf("delivery: a source names no repository")
+	}
+	for repository, head := range s.Heads {
+		if repository == "" {
+			return fmt.Errorf("delivery: a source names a repository with no name")
+		}
+		if head == "" {
+			return fmt.Errorf("delivery: the source's head of %s is empty", repository)
+		}
+	}
+	return nil
+}
+
+// Candidate is the value of a candidates key: the group's pull requests and heads, the heads hash, the base (the
+// source revision it was previewed onto) and the modules each pull request touches. It carries no fingerprint
+// until release 4 (ADR 0437 §5).
+type Candidate struct {
+	V         int             `json:"v"`
+	Group     string          `json:"group"`
+	Base      uint64          `json:"base"`
+	HeadsHash string          `json:"heads"`
+	Pulls     []CandidatePull `json:"pulls"`
+}
+
+// CandidatePull is one pull request of a candidate and the modules it touches.
+type CandidatePull struct {
+	Head
+	Modules []string `json:"modules"`
+}
+
+func (c Candidate) check() error {
+	if len(c.Pulls) == 0 {
+		return fmt.Errorf("delivery: a candidate names no pull request")
+	}
+	heads := make([]Head, len(c.Pulls))
+	for i, p := range c.Pulls {
+		heads[i] = p.Head
+	}
+	want, _ := HeadsHash(heads)
+	if c.HeadsHash != want {
+		return fmt.Errorf("delivery: the candidate's heads hash %q is not the hash %q of its pull requests' heads", c.HeadsHash, want)
+	}
+	return nil
+}
+
+// Module is the value of a modules key: the module's source fingerprint, or none and why (a build that resolves
+// packages when built gets none, said here), and where it was computed: Repository and Commit are the trunk merge,
+// Src the source revision then (0 before the gate writes source.main). Exactly one of FP and None is set.
+type Module struct {
+	V          int    `json:"v"`
+	FP         string `json:"fp,omitempty"`
+	None       string `json:"none,omitempty"`
+	Src        uint64 `json:"src,omitempty"`
+	Repository string `json:"repository"`
+	Commit     string `json:"commit"`
+}
+
+func (m Module) check() error {
+	switch {
+	case m.FP != "" && m.None != "":
+		return fmt.Errorf("delivery: a module has both a fingerprint and a reason for none")
+	case m.FP == "" && m.None == "":
+		return fmt.Errorf("delivery: a module has neither a fingerprint nor a reason for none")
+	case m.FP != "" && !strings.HasPrefix(m.FP, SourcePrefix):
+		return fmt.Errorf("delivery: the module's fingerprint %q does not start with %s", m.FP, SourcePrefix)
+	}
+	return nil
+}
+
+// Forge is a message of the forge bridge's stream, trimmed to repository, branch, pull request and head. A pull
+// message says Action; a merged message carries the merge commit, the tree of every directory holding a
+// module.json and the repository's root tree; a refused message names the owed act it answers and the forge's own
+// words. A pull message may also carry the module directories its head changes (Dirs, in the order the forge said) and
+// DirsKnown, true when the forge could say them: known and empty is DirsKnown with no Dirs, not known is neither
+// (the forge's module_dirs and module_dirs_said, mesh-catalog modules/gitea/trunk.ts). Other messages carry neither.
+type Forge struct {
+	V          int               `json:"v"`
+	Kind       string            `json:"kind"`
+	Repository string            `json:"repository"`
+	Branch     string            `json:"branch,omitempty"`
+	Number     int               `json:"number,omitempty"`
+	Head       string            `json:"head,omitempty"`
+	Action     string            `json:"action,omitempty"`
+	Dirs       []string          `json:"dirs,omitempty"`
+	DirsKnown  bool              `json:"dirs-known,omitempty"`
+	Merge      string            `json:"merge,omitempty"`
+	Trees      map[string]string `json:"trees,omitempty"`
+	Root       string            `json:"root,omitempty"`
+	Act        string            `json:"act,omitempty"`
+	Words      string            `json:"words,omitempty"`
+}
+
+func (f Forge) check() error {
+	if err := checkRepository(f.Repository); err != nil {
+		return err
+	}
+	if err := checkForgeKind(f.Kind); err != nil {
+		return err
+	}
+	switch f.Kind {
+	case ForgePull:
+		if f.Number <= 0 {
+			return fmt.Errorf("delivery: a pull message needs its pull request number")
+		}
+		if f.Head == "" {
+			return fmt.Errorf("delivery: a pull message needs its head")
+		}
+		switch f.Action {
+		case PullOpened, PullUpdated, PullClosed:
+		default:
+			return fmt.Errorf("delivery: the pull action %q is not opened, updated or closed", f.Action)
+		}
+		if len(f.Dirs) > 0 && !f.DirsKnown {
+			return fmt.Errorf("delivery: a pull message with module directories must say it knows them (dirs-known)")
+		}
+	case ForgeMerged:
+		if len(f.Dirs) > 0 || f.DirsKnown {
+			return fmt.Errorf("delivery: a merged message carries no module directories")
+		}
+		if f.Number <= 0 {
+			return fmt.Errorf("delivery: a merged message needs its pull request number")
+		}
+		if f.Head == "" {
+			return fmt.Errorf("delivery: a merged message needs its head")
+		}
+		if f.Merge == "" {
+			return fmt.Errorf("delivery: a merged message needs its merge commit")
+		}
+		if f.Root == "" {
+			return fmt.Errorf("delivery: a merged message needs the repository's root tree")
+		}
+		if len(f.Trees) == 0 {
+			return fmt.Errorf("delivery: a merged message needs the tree of at least one directory holding a module.json")
+		}
+		for dir, tree := range f.Trees {
+			if tree == "" {
+				return fmt.Errorf("delivery: a merged message's tree of the directory %q is empty", dir)
+			}
+		}
+	case ForgeRefused:
+		if len(f.Dirs) > 0 || f.DirsKnown {
+			return fmt.Errorf("delivery: a refused message carries no module directories")
+		}
+		if f.Act == "" {
+			return fmt.Errorf("delivery: a refused message needs the act it answers")
+		}
+		if f.Words == "" {
+			return fmt.Errorf("delivery: a refused message needs the forge's words")
+		}
+	}
+	return nil
+}
+
+// EncodeSource is s as JSON at Version; it refuses a source with no repository or an empty head.
+func EncodeSource(s Source) ([]byte, error) {
+	if err := s.check(); err != nil {
+		return nil, err
+	}
+	s.V = Version
+	return json.Marshal(s)
+}
+
+// DecodeSource reads a source; it refuses an unknown version (ErrUnknownVersion) and what EncodeSource refuses.
+func DecodeSource(data []byte) (Source, error) {
+	var s Source
+	if err := decode(data, &s); err != nil {
+		return Source{}, err
+	}
+	if err := s.check(); err != nil {
+		return Source{}, err
+	}
+	return s, nil
+}
+
+// EncodeCandidate is c as JSON at Version; it refuses a heads hash that is not the hash of its pull requests' heads.
+func EncodeCandidate(c Candidate) ([]byte, error) {
+	if err := c.check(); err != nil {
+		return nil, err
+	}
+	c.V = Version
+	return json.Marshal(c)
+}
+
+// DecodeCandidate reads a candidate; it refuses an unknown version (ErrUnknownVersion) and what EncodeCandidate refuses.
+func DecodeCandidate(data []byte) (Candidate, error) {
+	var c Candidate
+	if err := decode(data, &c); err != nil {
+		return Candidate{}, err
+	}
+	if err := c.check(); err != nil {
+		return Candidate{}, err
+	}
+	return c, nil
+}
+
+// EncodeModule is m as JSON at Version; it refuses both or neither of a fingerprint and a reason, and a
+// fingerprint not of the form src1:.
+func EncodeModule(m Module) ([]byte, error) {
+	if err := m.check(); err != nil {
+		return nil, err
+	}
+	m.V = Version
+	return json.Marshal(m)
+}
+
+// DecodeModule reads a module; it refuses an unknown version (ErrUnknownVersion) and what EncodeModule refuses.
+func DecodeModule(data []byte) (Module, error) {
+	var m Module
+	if err := decode(data, &m); err != nil {
+		return Module{}, err
+	}
+	if err := m.check(); err != nil {
+		return Module{}, err
+	}
+	return m, nil
+}
+
+// EncodeForge is f as JSON at Version; it refuses a message missing what its kind needs.
+func EncodeForge(f Forge) ([]byte, error) {
+	if err := f.check(); err != nil {
+		return nil, err
+	}
+	f.V = Version
+	return json.Marshal(f)
+}
+
+// DecodeForge reads a forge message; it refuses an unknown version (ErrUnknownVersion) and what EncodeForge refuses.
+func DecodeForge(data []byte) (Forge, error) {
+	var f Forge
+	if err := decode(data, &f); err != nil {
+		return Forge{}, err
+	}
+	if err := f.check(); err != nil {
+		return Forge{}, err
+	}
+	return f, nil
+}
+
+// SameSource says whether two sources are the same value, so a writer writes only when they differ.
+func SameSource(a, b Source) bool { return reflect.DeepEqual(a, b) }
+
+// SameCandidate says whether two candidates are the same value.
+func SameCandidate(a, b Candidate) bool { return reflect.DeepEqual(a, b) }
+
+// SameModule says whether two modules are the same value.
+func SameModule(a, b Module) bool { return a == b }
+
+// SameForge says whether two forge messages are the same value.
+func SameForge(a, b Forge) bool { return reflect.DeepEqual(a, b) }

go/delivery/source_test.go

@@ -0,0 +1,340 @@
+package delivery
+
+import (
+	"encoding/json"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestSourceKeys(t *testing.T) { // 1200/AC1
+	if k, err := CandidatesKey("feat/delivery-source-keys"); err != nil || k != "feat/delivery-source-keys" {
+		t.Fatalf("candidates key = %q, %v", k, err)
+	}
+	if g, err := ParseCandidatesKey("feat/x"); err != nil || g != "feat/x" {
+		t.Fatalf("parse candidates = %q, %v", g, err)
+	}
+	if k, err := ModulesKey("mesh-mail"); err != nil || k != "mesh-mail" {
+		t.Fatalf("modules key = %q, %v", k, err)
+	}
+	if m, err := ParseModulesKey("mesh-mail"); err != nil || m != "mesh-mail" {
+		t.Fatalf("parse modules = %q, %v", m, err)
+	}
+	for _, bad := range []string{"", "a.b", "a*b", "a>b", "a b", "a\tb"} {
+		if _, err := CandidatesKey(bad); err == nil {
+			t.Errorf("candidates key %q accepted", bad)
+		}
+		if _, err := ModulesKey(bad); err == nil {
+			t.Errorf("modules key %q accepted", bad)
+		}
+		if _, err := ParseCandidatesKey(bad); err == nil {
+			t.Errorf("parse candidates %q accepted", bad)
+		}
+		if _, err := ParseModulesKey(bad); err == nil {
+			t.Errorf("parse modules %q accepted", bad)
+		}
+	}
+	ev, err := ForgeEvent("novox/mesh-sdk", ForgeMerged)
+	if err != nil || ev != "forge.novox.mesh-sdk.merged" {
+		t.Fatalf("forge event = %q, %v", ev, err)
+	}
+	repo, kind, err := ParseForgeEvent(ev)
+	if err != nil || repo != "novox/mesh-sdk" || kind != ForgeMerged {
+		t.Fatalf("parse forge event = %q %q %v", repo, kind, err)
+	}
+	for _, k := range []string{ForgePull, ForgeMerged, ForgeRefused} {
+		if _, err := ForgeEvent("a/b", k); err != nil {
+			t.Errorf("kind %s refused: %v", k, err)
+		}
+	}
+	for _, bad := range []struct{ repo, kind string }{
+		{"", "pull"}, {"novox", "pull"}, {"a/b/c", "pull"}, {"/b", "pull"}, {"a/", "pull"},
+		{"a.b/c", "pull"}, {"a/b c", "pull"}, {"a/b", "closed"}, {"a/b", ""},
+	} {
+		if _, err := ForgeEvent(bad.repo, bad.kind); err == nil {
+			t.Errorf("forge event %q %q accepted", bad.repo, bad.kind)
+		} else if bad.kind == "closed" && !strings.Contains(err.Error(), "closed") {
+			t.Errorf("error does not name the kind: %v", err)
+		}
+	}
+	for _, bad := range []string{"", "forge.a.b", "forge.a.b.closed", "x.a.b.pull", "forge.a.b.pull.x", "forge..b.pull"} {
+		if _, _, err := ParseForgeEvent(bad); err == nil {
+			t.Errorf("parse forge event %q accepted", bad)
+		}
+	}
+}
+
+func TestHeadsHash(t *testing.T) { // 1200/AC2
+	a := Head{Repository: "novox/mesh-sdk", Number: 36, Head: "aaaa"}
+	b := Head{Repository: "novox/mesh-control", Number: 7, Head: "bbbb"}
+	h1, err := HeadsHash([]Head{a, b})
+	if err != nil {
+		t.Fatal(err)
+	}
+	h2, _ := HeadsHash([]Head{b, a})
+	if h1 != h2 {
+		t.Errorf("order changes the hash: %s %s", h1, h2)
+	}
+	if len(h1) != 12 || strings.Trim(h1, "0123456789abcdef") != "" {
+		t.Errorf("hash %q is not 12 hex digits", h1)
+	}
+	for _, changed := range []Head{
+		{Repository: "novox/mesh-sdk", Number: 36, Head: "aaab"},
+		{Repository: "novox/mesh-sdk", Number: 37, Head: "aaaa"},
+		{Repository: "novox/other", Number: 36, Head: "aaaa"},
+	} {
+		h3, _ := HeadsHash([]Head{changed, b})
+		if h3 == h1 {
+			t.Errorf("changing %v keeps the hash", changed)
+		}
+	}
+	if _, err := HeadsHash(nil); err == nil {
+		t.Error("an empty list is accepted")
+	}
+}
+
+func sampleSource() Source {
+	return Source{Group: "feat/x", Heads: map[string]string{"novox/mesh-sdk": "abc123", "novox/mesh-control": "def456"}}
+}
+
+func sampleCandidate() Candidate {
+	pulls := []CandidatePull{
+		{Head: Head{Repository: "novox/mesh-sdk", Number: 36, Head: "aaaa"}, Modules: []string{"mesh-mail"}},
+		{Head: Head{Repository: "novox/mesh-control", Number: 7, Head: "bbbb"}, Modules: []string{"mesh-controller"}},
+	}
+	hh, _ := HeadsHash([]Head{pulls[0].Head, pulls[1].Head})
+	return Candidate{Group: "feat/x", Base: 12, HeadsHash: hh, Pulls: pulls}
+}
+
+func sampleForges() []Forge {
+	return []Forge{
+		{Kind: ForgePull, Repository: "novox/mesh-sdk", Branch: "feat/x", Number: 36, Head: "aaaa", Action: "opened"},
+		{Kind: ForgeMerged, Repository: "novox/mesh-sdk", Number: 36, Head: "aaaa", Merge: "cccc",
+			Trees: map[string]string{"modules/a": "t1"}, Root: "t0"},
+		{Kind: ForgeRefused, Repository: "novox/mesh-sdk", Act: "novox/mesh-sdk.36.aaaa.view", Words: "403 forbidden"},
+	}
+}
+
+func TestSourceValuesRoundTrip(t *testing.T) { // 1200/AC3
+	check := func(name string, data []byte, err error) {
+		t.Helper()
+		if err != nil {
+			t.Fatalf("%s: %v", name, err)
+		}
+		var probe map[string]any
+		if err := json.Unmarshal(data, &probe); err != nil || probe["v"] != float64(1) {
+			t.Fatalf("%s: json %s does not carry v:1", name, data)
+		}
+	}
+	s := sampleSource()
+	data, err := EncodeSource(s)
+	check("source", data, err)
+	s.V = Version
+	if got, err := DecodeSource(data); err != nil || !SameSource(got, s) {
+		t.Errorf("source round trip: %+v %v", got, err)
+	}
+	c := sampleCandidate()
+	data, err = EncodeCandidate(c)
+	check("candidate", data, err)
+	c.V = Version
+	if got, err := DecodeCandidate(data); err != nil || !SameCandidate(got, c) {
+		t.Errorf("candidate round trip: %+v %v", got, err)
+	}
+	for _, m := range []Module{
+		{FP: "src1:abc", Src: 5, Repository: "novox/mesh-sdk", Commit: "cccc"},
+		{None: "a build that resolves packages when built", Repository: "novox/mesh-sdk", Commit: "cccc"},
+	} {
+		data, err = EncodeModule(m)
+		check("module", data, err)
+		m.V = Version
+		if got, err := DecodeModule(data); err != nil || !SameModule(got, m) {
+			t.Errorf("module round trip: %+v %v", got, err)
+		}
+	}
+	for _, f := range sampleForges() {
+		data, err = EncodeForge(f)
+		check("forge "+f.Kind, data, err)
+		f.V = Version
+		if got, err := DecodeForge(data); err != nil || !SameForge(got, f) {
+			t.Errorf("forge %s round trip: %+v %v", f.Kind, got, err)
+		}
+	}
+	for name, dec := range map[string]func([]byte) error{
+		"source":    func(b []byte) error { _, err := DecodeSource(b); return err },
+		"candidate": func(b []byte) error { _, err := DecodeCandidate(b); return err },
+		"module":    func(b []byte) error { _, err := DecodeModule(b); return err },
+		"forge":     func(b []byte) error { _, err := DecodeForge(b); return err },
+	} {
+		for _, in := range []string{`{"group":"g"}`, `{"v":2,"group":"g"}`} {
+			if err := dec([]byte(in)); !errors.Is(err, ErrUnknownVersion) {
+				t.Errorf("%s decode of %s = %v, want ErrUnknownVersion", name, in, err)
+			}
+		}
+	}
+}
+
+func TestSourceValuesRefused(t *testing.T) { // 1200/AC4
+	wantErr := func(name string, err error, words ...string) {
+		t.Helper()
+		if err == nil {
+			t.Errorf("%s accepted", name)
+			return
+		}
+		for _, w := range words {
+			if !strings.Contains(err.Error(), w) {
+				t.Errorf("%s: error %q does not name %q", name, err, w)
+			}
+		}
+	}
+	_, err := EncodeSource(Source{Group: "g"})
+	wantErr("source without repositories", err, "repositor")
+	_, err = EncodeSource(Source{Group: "g", Heads: map[string]string{"a/b": ""}})
+	wantErr("source with an empty head", err, "a/b")
+	_, err = EncodeSource(Source{Group: "g", Heads: map[string]string{"": "x"}})
+	wantErr("source with an empty repository", err, "repositor")
+
+	c := sampleCandidate()
+	c.HeadsHash = "000000000000"
+	_, err = EncodeCandidate(c)
+	wantErr("candidate with a stale hash", err, "heads hash")
+	c = sampleCandidate()
+	c.Pulls = nil
+	_, err = EncodeCandidate(c)
+	wantErr("candidate without pulls", err, "pull")
+
+	_, err = EncodeModule(Module{FP: "src1:x", None: "why", Repository: "a/b", Commit: "c"})
+	wantErr("module with both", err, "both")
+	_, err = EncodeModule(Module{Repository: "a/b", Commit: "c"})
+	wantErr("module with neither", err, "fingerprint", "none")
+	_, err = EncodeModule(Module{FP: "abc", Repository: "a/b", Commit: "c"})
+	wantErr("module with a bad fingerprint", err, "src1:")
+
+	f := sampleForges()[1]
+	f.Merge = ""
+	_, err = EncodeForge(f)
+	wantErr("merged without merge commit", err, "merge")
+	f = sampleForges()[1]
+	f.Root = ""
+	_, err = EncodeForge(f)
+	wantErr("merged without root", err, "root")
+	f = sampleForges()[1]
+	f.Trees = nil
+	_, err = EncodeForge(f)
+	wantErr("merged without trees", err, "tree")
+	f = sampleForges()[1]
+	f.Trees = map[string]string{"modules/a": ""}
+	_, err = EncodeForge(f)
+	wantErr("merged with an empty tree", err, "modules/a")
+	f = sampleForges()[0]
+	f.Action = "reopened"
+	_, err = EncodeForge(f)
+	wantErr("pull with an unknown action", err, "reopened")
+	f = sampleForges()[0]
+	f.Head = ""
+	_, err = EncodeForge(f)
+	wantErr("pull without head", err, "head")
+	f = sampleForges()[2]
+	f.Words = ""
+	_, err = EncodeForge(f)
+	wantErr("refused without words", err, "words")
+	f = sampleForges()[2]
+	f.Act = ""
+	_, err = EncodeForge(f)
+	wantErr("refused without act", err, "act")
+	_, err = EncodeForge(Forge{Kind: "closed", Repository: "a/b"})
+	wantErr("unknown kind", err, "closed")
+	_, err = EncodeForge(Forge{Kind: ForgePull, Repository: "nope", Number: 1, Head: "h", Action: "opened"})
+	wantErr("bad repository", err, "nope")
+}
+
+func TestSourceSame(t *testing.T) { // 1200/AC5
+	if !SameSource(sampleSource(), sampleSource()) {
+		t.Error("equal sources differ")
+	}
+	s := sampleSource()
+	s.Heads["novox/mesh-sdk"] = "zzz"
+	if SameSource(s, sampleSource()) {
+		t.Error("different sources are the same")
+	}
+	if !SameCandidate(sampleCandidate(), sampleCandidate()) {
+		t.Error("equal candidates differ")
+	}
+	c := sampleCandidate()
+	c.Pulls[0].Modules = []string{"other"}
+	if SameCandidate(c, sampleCandidate()) {
+		t.Error("different candidates are the same")
+	}
+	m := Module{FP: "src1:a", Repository: "a/b", Commit: "c"}
+	if !SameModule(m, m) || SameModule(m, Module{FP: "src1:b", Repository: "a/b", Commit: "c"}) {
+		t.Error("SameModule wrong")
+	}
+	f := sampleForges()[1]
+	g := sampleForges()[1]
+	if !SameForge(f, g) {
+		t.Error("equal forge messages differ")
+	}
+	g.Trees["modules/a"] = "other"
+	if SameForge(f, g) {
+		t.Error("different forge messages are the same")
+	}
+}
+
+func TestSourceBuckets(t *testing.T) {
+	if SourceBucket != "mesh-delivery-gate_source" || CandidatesBucket != "mesh-delivery-indexer_candidates" ||
+		ModulesBucket != "mesh-delivery-indexer_modules" || SourceMainKey != "main" ||
+		SourceHistory != 64 || CandidatesHistory != 10 || ModulesHistory != 20 {
+		t.Error("a bucket or history differs from the decided layout")
+	}
+}
+
+func TestForgeDirs(t *testing.T) {
+	pull := func(dirs []string, known bool) Forge {
+		return Forge{Kind: ForgePull, Repository: "novox/mesh-sdk", Number: 36, Head: "aaaa", Action: "updated", Dirs: dirs, DirsKnown: known}
+	}
+	round := func(f Forge) Forge {
+		t.Helper()
+		data, err := EncodeForge(f)
+		if err != nil {
+			t.Fatal(err)
+		}
+		got, err := DecodeForge(data)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return got
+	}
+	// 1233/AC2: the directories come back equal and in order.
+	got := round(pull([]string{"modules/a", "modules/b"}, true))
+	if !reflect.DeepEqual(got.Dirs, []string{"modules/a", "modules/b"}) || !got.DirsKnown {
+		t.Errorf("dirs = %v known %v", got.Dirs, got.DirsKnown)
+	}
+	// 1233/AC3: a head changing no module directory is told apart from one whose directories could not be said.
+	none, unknown := round(pull(nil, true)), round(pull(nil, false))
+	if !none.DirsKnown || len(none.Dirs) != 0 {
+		t.Errorf("known and empty: dirs %v known %v", none.Dirs, none.DirsKnown)
+	}
+	if unknown.DirsKnown || len(unknown.Dirs) != 0 {
+		t.Errorf("not known: dirs %v known %v", unknown.Dirs, unknown.DirsKnown)
+	}
+	// 1233/AC4: a merged or a refused message carrying directories is refused.
+	m := sampleForges()[1]
+	m.Dirs, m.DirsKnown = []string{"modules/a"}, true
+	if _, err := EncodeForge(m); err == nil {
+		t.Error("merged message with dirs accepted")
+	}
+	r := sampleForges()[2]
+	r.Dirs, r.DirsKnown = []string{"modules/a"}, true
+	if _, err := EncodeForge(r); err == nil {
+		t.Error("refused message with dirs accepted")
+	}
+	r = sampleForges()[2]
+	r.DirsKnown = true
+	if _, err := EncodeForge(r); err == nil {
+		t.Error("refused message with dirs-known accepted")
+	}
+	// directories on a pull message that does not say it knows them are refused too.
+	if _, err := EncodeForge(pull([]string{"modules/a"}, false)); err == nil {
+		t.Error("dirs without dirs-known accepted")
+	}
+}

go/delivery/stuck.go

@@ -0,0 +1,65 @@
+package delivery
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// WatchmanModule is the watchman's module: its buckets are its own state, `<module>_<name>`, and the controller's
+// broker grants it alone the bus's connect and disconnect events (novox/hq ADR 0431).
+const WatchmanModule = "mesh-delivery-watchman"
+
+// StuckBucket holds what stopped moving (novox/hq ADR 0433 §2): one key per key whose pulse expired, written by the
+// watchman, read by the controller's self-check, which raises each as a condition until it says moved. No key of it
+// is removed: a key that moves again is written as moved.
+const StuckBucket = WatchmanModule + "_stuck"
+
+// StuckKey is the key, in StuckBucket, of the key that stopped moving: `<bucket>.<key>`, formed as PulseKey forms a
+// pulse's, so a node or part gone offline is `mesh-delivery-watchman_presence.<presence key>`.
+func StuckKey(bucket, key string) (string, error) {
+	stuck, err := PulseKey(bucket, key)
+	if err != nil {
+		return "", fmt.Errorf("delivery: no stuck key: %w", err)
+	}
+	return stuck, nil
+}
+
+// ParseStuckKey is the inverse of StuckKey, splitting at the first dot.
+func ParseStuckKey(stuckKey string) (bucket, key string, err error) {
+	bucket, key, err = ParsePulseKey(stuckKey)
+	if err != nil {
+		return "", "", fmt.Errorf("delivery: %q is not a stuck key: %w", stuckKey, err)
+	}
+	return bucket, key, nil
+}
+
+// Stuck is the value of a stuck key: the key that stopped moving, what it waited for in words, since when, and
+// whether it has moved again since. The watchman writes it; the controller reads it.
+type Stuck struct {
+	V      int       `json:"v"`
+	Bucket string    `json:"bucket"`
+	Key    string    `json:"key"`
+	Waited string    `json:"waited,omitempty"`
+	Since  time.Time `json:"since"`
+	Moved  bool      `json:"moved,omitempty"`
+}
+
+// EncodeStuck is s as JSON at Version.
+func EncodeStuck(s Stuck) ([]byte, error) {
+	s.V = Version
+	return json.Marshal(s)
+}
+
+// DecodeStuck reads a stuck value; it refuses an unknown version (ErrUnknownVersion).
+func DecodeStuck(data []byte) (Stuck, error) {
+	var s Stuck
+	err := decode(data, &s)
+	return s, err
+}
+
+// SameStuck says whether two stuck values are the same, so a writer writes only when they differ.
+func SameStuck(a, b Stuck) bool {
+	return a.V == b.V && a.Bucket == b.Bucket && a.Key == b.Key && a.Waited == b.Waited && a.Since.Equal(b.Since) &&
+		a.Moved == b.Moved
+}

go/delivery/stuck_test.go

@@ -0,0 +1,88 @@
+package delivery
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+// 1153/AC4: the watchman's module and its three buckets, each that module's state.
+func TestWatchmanBuckets(t *testing.T) {
+	if WatchmanModule != "mesh-delivery-watchman" {
+		t.Fatalf("WatchmanModule = %q", WatchmanModule)
+	}
+	for name, bucket := range map[string]string{"presence": PresenceBucket, "pulse": PulseBucket, "stuck": StuckBucket} {
+		if bucket != WatchmanModule+"_"+name {
+			t.Errorf("the %s bucket is %q, not %s_%s", name, bucket, WatchmanModule, name)
+		}
+	}
+}
+
+// 1153/AC1: the stuck key is the stopped key's bucket and key, split at the first dot, refused as a pulse key is.
+func TestStuckKey(t *testing.T) {
+	key, err := StuckKey(PresenceBucket, "ace")
+	if err != nil || key != "mesh-delivery-watchman_presence.ace" {
+		t.Fatalf("StuckKey = %q, %v", key, err)
+	}
+	key, err = StuckKey("mesh-controller_builds", "zsh.build.b-1")
+	if err != nil || key != "mesh-controller_builds.zsh.build.b-1" {
+		t.Fatalf("StuckKey = %q, %v", key, err)
+	}
+	bucket, rest, err := ParseStuckKey(key)
+	if err != nil || bucket != "mesh-controller_builds" || rest != "zsh.build.b-1" {
+		t.Fatalf("ParseStuckKey = %q %q %v", bucket, rest, err)
+	}
+	for _, c := range [][2]string{{"", "k"}, {"a.b", "k"}, {"b", ""}, {"b", "k x"}, {"b", ".k"}, {"b", "k."}, {"b", "k>"}} {
+		if _, err := StuckKey(c[0], c[1]); err == nil {
+			t.Errorf("StuckKey(%q, %q) accepted", c[0], c[1])
+		}
+	}
+	for _, bad := range []string{"", "nodot", ".k", "b."} {
+		if _, _, err := ParseStuckKey(bad); err == nil {
+			t.Errorf("ParseStuckKey(%q) accepted", bad)
+		}
+	}
+}
+
+// 1153/AC2: a stuck value round-trips with "v":1, and an unknown or missing version is refused loudly.
+func TestStuckValue(t *testing.T) {
+	s := Stuck{Bucket: PresenceBucket, Key: "ace", Waited: "ace to come back online",
+		Since: time.Date(2026, 10, 18, 20, 0, 0, 0, time.UTC)}
+	raw, err := EncodeStuck(s)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var probe map[string]any
+	if err := json.Unmarshal(raw, &probe); err != nil || probe["v"] != float64(1) {
+		t.Fatalf("encoded %s: v = %v (%v)", raw, probe["v"], err)
+	}
+	back, err := DecodeStuck(raw)
+	s.V = Version
+	if err != nil || !SameStuck(back, s) {
+		t.Fatalf("decoded %+v, %v; want %+v", back, err, s)
+	}
+	for _, bad := range []string{`{"v":2,"bucket":"b","key":"k"}`, `{"bucket":"b","key":"k"}`} {
+		_, err := DecodeStuck([]byte(bad))
+		if !errors.Is(err, ErrUnknownVersion) {
+			t.Errorf("DecodeStuck(%s) = %v, want ErrUnknownVersion", bad, err)
+		}
+		if strings.Contains(bad, `"v":2`) && (err == nil || !strings.Contains(err.Error(), "2") || !strings.Contains(err.Error(), "1")) {
+			t.Errorf("DecodeStuck(%s) = %v: does not name the version seen and the one read", bad, err)
+		}
+	}
+}
+
+// 1153/AC3: equal stuck values are the same; one moved is not.
+func TestSameStuck(t *testing.T) {
+	a := Stuck{V: Version, Bucket: "b", Key: "k", Waited: "w", Since: time.Unix(100, 0).UTC()}
+	b := a
+	if !SameStuck(a, b) {
+		t.Fatal("equal values are not the same")
+	}
+	b.Moved = true
+	if SameStuck(a, b) {
+		t.Fatal("a moved value is the same as a stuck one")
+	}
+}

go/delivery/testdata/build-record-build-agent.json

@@ -0,0 +1 @@
+{"id":"build-1792395447088936520","state":"built","repository":"novox/mesh-controller","ref":"main","module":"build-agent","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/build-agent","manifest-digest":"sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef","source-fingerprint":"src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7","source-inputs":{"Module":"build-agent","Tree":"11c0b4f90cae0d6e4ef9cc48cbfdf9f6dbe52bca","Bases":["sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc","sha256:8ac98ca534ac3f51e1f420a1dd2c15e74c75cfa0f23f3ad27eb5d7236c349a0c"],"Contexts":{"server":"narrow:d51ca744fd11af59d616c858f76c27b3e58e16e11fbdc0c720eb9d7d19d55fac"},"Toolchains":null,"Unpinned":""},"made":[{"name":"server","kind":"image","reference":"registry.example/build-agent@sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc"}],"at":"2026-10-19T09:38:00Z"}

go/emit_message_id_test.go

@@ -0,0 +1,110 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+)
+
+// emitRuntime drives a bundle over pipes as the runtime does, reading every line it says.
+type emitRuntime struct {
+	t     *testing.T
+	lines *bufio.Scanner
+	in    io.Writer
+}
+
+func (r *emitRuntime) next() map[string]any {
+	done := make(chan map[string]any, 1)
+	go func() {
+		if r.lines.Scan() {
+			var m map[string]any
+			_ = json.Unmarshal(r.lines.Bytes(), &m)
+			done <- m
+		}
+	}()
+	select {
+	case m := <-done:
+		return m
+	case <-time.After(3 * time.Second):
+		r.t.Fatal("the bundle said nothing")
+		return nil
+	}
+}
+
+func serveEmit(t *testing.T, tools []Tool) *emitRuntime {
+	t.Helper()
+	t.Setenv(ServedAs, "chime")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	go func() { _ = ServeOn("", tools, inR, outW) }()
+	t.Cleanup(func() { _ = inW.Close() })
+	return &emitRuntime{t: t, lines: bufio.NewScanner(outR), in: inW}
+}
+
+func (r *emitRuntime) call(id int, tool string) {
+	b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call",
+		"params": map[string]any{"name": tool, "arguments": map[string]any{}}})
+	_, _ = r.in.Write(append(b, '\n'))
+}
+
+func (r *emitRuntime) answer(id any) {
+	b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "result": map[string]any{}})
+	_, _ = r.in.Write(append(b, '\n'))
+}
+
+// eventIDOf is a "mesh/publish" ask's x-event-id header.
+func eventIDOf(asked map[string]any) string {
+	return asked["params"].(map[string]any)["headers"].(map[string]any)["x-event-id"].(string)
+}
+
+// 1187/AC1: given EmitOptions with a message id, when a bundle emits, then the event's x-event-id header is
+// that id; and a random one when none is given, as on main.
+func TestEmitSendsTheChosenMessageIDAsXEventID(t *testing.T) {
+	rt := serveEmit(t, []Tool{
+		{Name: "chosen", Run: func(map[string]any) (any, error) {
+			return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-42"})
+		}},
+		{Name: "unchosen", Run: func(map[string]any) (any, error) {
+			return nil, Emit("sent", map[string]any{})
+		}},
+	})
+	rt.call(1, "chosen")
+	asked := rt.next()
+	if got := eventIDOf(asked); got != "send-42" {
+		t.Fatalf("x-event-id: %q, want the chosen id %q", got, "send-42")
+	}
+	rt.answer(asked["id"])
+	rt.next() // the tool's own answer
+
+	rt.call(2, "unchosen")
+	asked = rt.next()
+	if got := eventIDOf(asked); len(got) != 36 {
+		t.Fatalf("x-event-id with no message id chosen: %q, want a random UUID as on main", got)
+	}
+	rt.answer(asked["id"])
+	rt.next()
+}
+
+// 1187/AC2: given two emits with the same chosen id, when sent through the runtime's protocol stand-in,
+// then both carry the same x-event-id, which the tool runner gives the bus as Nats-Msg-Id (mesh-tools
+// node-tools/internal/bus/bus.go ~748-773, :770): a repeated publish is kept to one by the bus's dedup.
+func TestEmitRepeatedWithTheSameChosenIDCarriesTheSameXEventIDTwice(t *testing.T) {
+	rt := serveEmit(t, []Tool{
+		{Name: "chosen", Run: func(map[string]any) (any, error) {
+			return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-77"})
+		}},
+	})
+	var ids []string
+	for i := 1; i <= 2; i++ {
+		rt.call(i, "chosen")
+		asked := rt.next()
+		ids = append(ids, eventIDOf(asked))
+		rt.answer(asked["id"])
+		rt.next()
+	}
+	if ids[0] != "send-77" || ids[1] != "send-77" {
+		t.Fatalf("the two emits' x-event-id: %v, want %q both times", ids, "send-77")
+	}
+}

go/go.mod

@@ -1,3 +1,21 @@
 module git.novox.be/novox/mesh-sdk/go
 
-go 1.22
+go 1.26.0
+
+require (
+	github.com/nats-io/nats-server/v2 v2.11.17
+	github.com/nats-io/nats.go v1.54.0
+)
+
+require (
+	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
+	github.com/google/go-tpm v0.9.8 // indirect
+	github.com/klauspost/compress v1.20.0 // indirect
+	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/nats-io/jwt/v2 v2.8.1 // indirect
+	github.com/nats-io/nkeys v0.4.16 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	golang.org/x/crypto v0.57.0 // indirect
+	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
+)

go/go.sum

@@ -0,0 +1,25 @@
+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=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
+github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
+github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
+github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
+github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
+github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
+golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=

go/internal/testbus/testbus.go

@@ -0,0 +1,74 @@
+// Package testbus is the bus an integration test of the SDK's Go binding runs against: a server of
+// its own, of the release go.mod pins, started for the one test and gone after it.
+//
+// **Why one per test.** A test that reaches a shared bus asserts, reads and removes objects by fixed
+// name — a stream, a consumer — and two tests at once would step on each other's: one deleting what
+// the other is reading, one leaving a stream behind for the next to trip on. A server per test makes
+// every test start from nothing and leave nothing, so the suite is the same run in any order, at any
+// parallelism, under the race detector, on a workstation or on the build seat.
+//
+// **No catalogue pin check.** mesh-controller's testbus also checks the server's release against the
+// catalogue's bus image and the mesh's facts snapshot (novox/hq ADR 0227 rule 9) — the SDK has no
+// catalogue of its own to check against, so that test is left out here. Only the version this
+// module's go.mod pins is what these tests run.
+//
+// A person may still point a run at a bus of their own — another release, a lab's — with
+// MESH_TEST_NATS_EXTERNAL=1 and MESH_TEST_NATS; then the run is theirs to serialise.
+package testbus
+
+import (
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+)
+
+// Version is the release of the server the tests run.
+const Version = server.VERSION
+
+// URL is the bus of this test alone: started at its first ask, the same one at every ask after — a
+// test that dials twice, or hands the address to the code it tests, reaches one bus — and shut down
+// when the test ends.
+func URL(t testing.TB) string {
+	t.Helper()
+	if os.Getenv("MESH_TEST_NATS_EXTERNAL") == "1" {
+		if url := os.Getenv("MESH_TEST_NATS"); url != "" {
+			return url
+		}
+		t.Fatal("MESH_TEST_NATS_EXTERNAL=1 and MESH_TEST_NATS names no bus")
+	}
+	if url, ok := buses.Load(t); ok {
+		return url.(string)
+	}
+	url := Start(t).ClientURL()
+	buses.Store(t, url)
+	t.Cleanup(func() { buses.Delete(t) })
+	return url
+}
+
+// buses are the running tests' buses, by test.
+var buses sync.Map
+
+// Start is a server with JetStream, for this test alone: on loopback, on a port the system chooses, its
+// store in the test's own directory.
+func Start(t testing.TB) *server.Server {
+	t.Helper()
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
+		NoLog: true, NoSigs: true}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatalf("a bus for this test could not be made: %v", err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(30 * time.Second) {
+		s.Shutdown()
+		t.Fatal("a bus for this test did not come up within 30s")
+	}
+	t.Cleanup(func() {
+		s.Shutdown()
+		s.WaitForShutdown()
+	})
+	return s
+}

go/internal/testbus/testbus_integration_test.go

@@ -0,0 +1,87 @@
+//go:build integration
+
+package testbus
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// 1175/AC3, 1171/AC1: a JetStream bus of this test alone answers a publish and a key-value create.
+func TestBusAnswersPublishAndKeyValueCreate(t *testing.T) {
+	conn, err := nats.Connect(URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	js, err := jetstream.New(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.>"}}); err != nil {
+		t.Fatalf("the bus did not answer a stream create: %v", err)
+	}
+	ack, err := js.Publish(ctx, "mesh.test", []byte("hello"))
+	if err != nil {
+		t.Fatalf("the bus did not answer a publish: %v", err)
+	}
+	if ack.Stream != "EVENTS" {
+		t.Errorf("published into %q, want EVENTS", ack.Stream)
+	}
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TESTKV"})
+	if err != nil {
+		t.Fatalf("the bus did not answer a key-value create: %v", err)
+	}
+	if _, err := kv.Put(ctx, "k", []byte("v")); err != nil {
+		t.Fatal(err)
+	}
+	entry, err := kv.Get(ctx, "k")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(entry.Value()) != "v" {
+		t.Errorf("read back %q, want %q", entry.Value(), "v")
+	}
+}
+
+// 1175/AC3, 1171/AC1: the bus is gone when the test that started it ends.
+func TestBusGoneWhenItsTestEnds(t *testing.T) {
+	var url string
+	t.Run("one", func(t *testing.T) {
+		url = URL(t)
+		conn, err := nats.Connect(url)
+		if err != nil {
+			t.Fatal(err)
+		}
+		conn.Close()
+	})
+	if _, err := nats.Connect(url, nats.Timeout(time.Second)); err == nil {
+		t.Fatal("the subtest's bus still answers now that the subtest has ended")
+	}
+}
+
+// 1175/AC3: two tests never share a bus.
+func TestEveryTestHasABusOfItsOwn(t *testing.T) {
+	seen := map[string]bool{}
+	for i := 0; i < 2; i++ {
+		t.Run("one", func(t *testing.T) {
+			url := URL(t)
+			if seen[url] {
+				t.Fatalf("two tests were handed %s", url)
+			}
+			seen[url] = true
+		})
+	}
+	// Asked twice, one test reaches one bus.
+	if a, b := URL(t), URL(t); a != b {
+		t.Errorf("one test was handed two buses: %s and %s", a, b)
+	}
+}

go/live_before_serve_test.go

@@ -0,0 +1,36 @@
+package stdio
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+)
+
+// 1171/AC4: before Serve, a bundle's bus functions still refuse exactly as on main (stdio.go:391-398,
+// unchanged for bundles). Run in a fresh process: once any other test in this package has called Serve,
+// current stays set for the rest of the test binary, so "before Serve" can only be checked honestly in a
+// process of its own (the self-exec pattern Go's own testing package uses for os.Exit).
+func TestBusFunctionsRefuseBeforeServeInAFreshProcess(t *testing.T) {
+	const marker = "MESH_SDK_TEST_FRESH_PROCESS"
+	if os.Getenv(marker) == "1" {
+		if err := Emit("x", map[string]any{}); err == nil ||
+			err.Error() != "emit before Serve: a bundle emits through the runtime that launched it" {
+			t.Fatalf("Emit before Serve: %v", err)
+		}
+		if _, err := Ask("m.t", nil); err == nil ||
+			err.Error() != "before Serve: a bundle reaches the bus through the runtime that launched it" {
+			t.Fatalf("Ask before Serve: %v", err)
+		}
+		if err := Subscribe("x", func(Envelope) error { return nil }); err == nil ||
+			err.Error() != "before Serve: a bundle reaches the bus through the runtime that launched it" {
+			t.Fatalf("Subscribe before Serve: %v", err)
+		}
+		return
+	}
+	cmd := exec.Command(os.Args[0], "-test.run=^TestBusFunctionsRefuseBeforeServeInAFreshProcess$", "-test.v")
+	cmd.Env = append(os.Environ(), marker+"=1")
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("a fresh process did not refuse every bus function before Serve: %v\n%s", err, out)
+	}
+}

go/merge-check.sh

@@ -5,16 +5,127 @@
 # seat in the mesh's Go toolchain, its own container, because the SDK's merge-check.sh declares it
 # (`# mesh-check-also: go go/merge-check.sh`) — and by hand, from anywhere: `sh go/merge-check.sh`.
 #
-# Formatted, vetted, and its whole suite run: the stdio loop and the local protocol mesh-cli speaks to the
-# node-engine. Standard library only, so nothing is fetched.
+# Formatted, vetted (with and without the build tag `integration`), and run at two levels (novox/hq ADR 0391):
+#
+#   1. the unit level: `go test ./...` untagged, no bus or other process, timed and held to a bound
+#      (unit_bound below); over it the check fails and names the slowest tests (issue 895);
+#   2. the integration level: the whole suite with `-tags integration`, each test against a bus of its
+#      own (internal/testbus, which issue 1171 brings with the first test that needs a bus) so nothing is
+#      serialised. With no tagged test yet, the level runs the unit tests again and says their count.
+#
+# **A test that starts a bus or another process carries `//go:build integration`** as the first line of
+# its file; a file that held both kinds is two files (`<name>_integration_test.go` beside `<name>_test.go`).
+# Each level says its test count and its time.
+#
+# No dependency is committed (no vendor directory: .gitignore; the operator, 2026-10-19; issue 1304). Every dependency
+# is resolved by the version and checksum go.sum names (novox/hq ADR 0405 §5), from its public source: this binding
+# requires nothing of the mesh's own forge. `go mod verify` fails by module name when a downloaded dependency does not
+# match go.sum, and -mod=readonly fails a build whose go.sum lacks an entry instead of writing it. gofmt reads only
+# the binding's own files.
+#
+# Exit 0 is a pass; anything else fails `mesh/repo-check` with the last line printed.
 set -eu
 cd "$(dirname "$0")"
+# The toolchain the check runs in sets -mod=vendor; there is no vendor directory, so it is replaced.
+GOFLAGS="$(printf '%s' "${GOFLAGS:-}" | sed 's/-mod=[a-z]*//g') -mod=readonly"
+export GOFLAGS
+export GOPRIVATE=git.novox.be
 
-unformatted=$(gofmt -l .)
+# **The unit level's bound, in seconds** (issue 895): the slowest package's own test time of the untagged run,
+# the compile not counted. A run over it fails the check and names the slowest tests. The SDK's unit suite is
+# small (its slowest package measured well under a second on a quiet machine), so 2 s leaves room for a busier
+# build node without hiding a test that should carry `integration` instead.
+unit_bound=2
+
+# judge_unit_run says the unit level as `go test -v` reported it and fails when it was over the bound: $1 the
+# file holding the output of the untagged `go test -v` of every package (packages one after another, each
+# ending in its `ok <package> <time>` line), $2 the bound in seconds. Packages run side by side, so the
+# level's time is that of its slowest package; the five slowest tests of the run are named when it is over,
+# so that the bound is kept by making them cheap or by tagging them `integration`, and never by raising it
+# (issue 895).
+judge_unit_run() {
+	out=$1 bound=$2
+	awk '
+		/^--- (PASS|FAIL|SKIP): / {
+			name = $3; took = $4; gsub(/[()s]/, "", took)
+			pending[++n] = took "\t" name
+			if ($2 == "SKIP:") skipped++
+			next
+		}
+		/^(ok|FAIL)[ \t]/ && NF >= 3 && $3 ~ /^[0-9.]+s$/ {
+			t = $3; sub(/s$/, "", t)
+			for (i = 1; i <= n; i++) print "T\t" pending[i] "\t" $2
+			n = 0
+			print "P\t" t "\t" $2
+		}
+		END { print "S\t" skipped + 0 }
+	' "$out" >"$out.judged"
+	tests=$(grep -c '^T' "$out.judged" || true)
+	skipped=$(awk -F'\t' '$1 == "S" {print $2}' "$out.judged")
+	level=$(awk -F'\t' '$1 == "P" && $2 + 0 > m {m = $2 + 0} END {printf "%.2f", m}' "$out.judged")
+	echo "unit level: $tests tests ($skipped skipped), $level s (bound $bound s)"
+	if awk -v t="$level" -v b="$bound" 'BEGIN {exit !(t + 0 > b + 0)}'; then
+		echo "the unit level took $level s, over its bound of $bound s; the slowest tests:"
+		awk -F'\t' '$1 == "T" {print $2 "\t" $4 "\t" $3}' "$out.judged" | sort -t"$(printf '\t')" -k1,1 -nr | head -n 5 |
+			awk -F'\t' '{printf "  %s s  %s  %s\n", $1, $3, $2}'
+		echo "make them cheap (no wait, no process), or give their files //go:build integration if they start a bus or another process"
+		return 1
+	fi
+}
+# `sh go/merge-check.sh judge-unit-run <output> <bound>` runs only the judging of the unit level, on a
+# recorded output.
+if [ "${1:-}" = judge-unit-run ]; then
+	shift
+	judge_unit_run "$@"
+	exit
+fi
+
+if ! fetched=$(go mod download 2>&1); then
+	echo "FAILED: the dependencies cannot be fetched: $(printf '%s\n' "$fetched" | tail -n 1)"
+	exit 1
+fi
+if ! verified=$(go mod verify 2>&1); then
+	echo "FAILED: a dependency does not match go.sum: $verified"
+	exit 1
+fi
+
+unformatted=$(find . -name '*.go' -not -path './vendor/*' | xargs gofmt -l)
 if [ -n "$unformatted" ]; then
 	echo "not gofmt'd:"
 	echo "$unformatted"
 	exit 1
 fi
 go vet ./...
-go test -count=1 ./...
+# The files tagged `integration` are vetted too, or nothing would read them.
+go vet -tags integration ./...
+
+# NOT RACE-CHECKED here: the SDK's suite is small and carries no store or other shared state between
+# packages, so the race detector is not required of this check (novox/hq issue 1175); a module that needs
+# it asks for it in its own merge-check.sh.
+echo "NOT RACE-CHECKED: the Go binding's check runs without the race detector"
+
+work=$(mktemp -d)
+trap 'rm -rf "$work"' EXIT
+
+# The unit level first and alone: every package untagged, no bus or other process (judged above).
+status=0
+go test -count=1 -v ./... >"$work/unit.out" 2>&1 || status=$?
+if [ "$status" -ne 0 ]; then
+	grep -vE '^(=== |[[:space:]]*--- PASS)' "$work/unit.out"
+	echo "the unit level failed (go test exited $status)"
+	exit 1
+fi
+judge_unit_run "$work/unit.out" "$unit_bound"
+
+# The integration level: the whole suite with the tag, each test against its own bus (internal/testbus).
+integration_start=$(date +%s)
+status=0
+go test -tags integration -count=1 -v ./... >"$work/integration.out" 2>&1 || status=$?
+grep -vE '^(=== |[[:space:]]*--- PASS)' "$work/integration.out" || true
+tests=$(grep -cE '^--- (PASS|FAIL|SKIP): ' "$work/integration.out" || true)
+skipped=$(grep -cE '^--- SKIP: ' "$work/integration.out" || true)
+echo "integration level: $tests tests ($skipped skipped), $(($(date +%s) - integration_start)) s"
+if [ "$status" -ne 0 ]; then
+	echo "the integration level failed (go test exited $status)"
+	exit 1
+fi

go/process/account.go

@@ -0,0 +1,54 @@
+// Package process is a module's own process's connection to the bus (novox/hq ADR 0436 clause 4): it reads
+// the account the controller delivers, connects as it with its pinned fingerprint and its inbox prefix, and
+// offers emitting an event, the module's states with compare-and-set, consuming a work queue, reading the
+// router's records of the module's own asks, and calling a tool or verb — each only as the account's grants
+// allow. It offers no Serve and no way to serve a tool: a process never answers a call. A sibling of stdio,
+// which it imports only for stdio's pure types (ErrChanged, CodeChanged, StateEntry, EmitOptions); it never
+// touches stdio's current, channel or live(), which stay the bundle's own way to the bus.
+package process
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+)
+
+// Account is a process's own bus account, exactly as the controller delivers it as the module's own secret
+// broker-<process id> (mesh-controller cmd/mesh-controller/modules.go issueProcess, heldProcessAccount, and
+// issueWith for a module's own broker account, which also carries claims). A process account carries no
+// claims at all; this type reads either file without error.
+type Account struct {
+	URL         string `json:"url"`
+	Fingerprint string `json:"fingerprint,omitempty"`
+	Node        string `json:"node"`
+	Module      string `json:"module"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+	// Claims rides on a module's own broker account only; a process holds no seat, so its own account
+	// carries none (novox/hq ADR 0436 §2, §3). Kept raw: this package never reads it.
+	Claims json.RawMessage `json:"claims,omitempty"`
+}
+
+// InboxPrefix is this account's own inbox, and nothing wider: `_INBOX.<user>`.
+func (a Account) InboxPrefix() string { return "_INBOX." + a.User }
+
+// ReadAccount reads and validates the account file the controller delivers: it refuses a file missing url,
+// user or password, naming the field — never the password, which does not appear in an error.
+func ReadAccount(path string) (Account, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return Account{}, fmt.Errorf("reading the process account at %s: %w", path, err)
+	}
+	var a Account
+	if err := json.Unmarshal(data, &a); err != nil {
+		return Account{}, fmt.Errorf("the process account at %s is not JSON: %w", path, err)
+	}
+	for _, f := range []struct{ name, value string }{
+		{"url", a.URL}, {"user", a.User}, {"password", a.Password},
+	} {
+		if f.value == "" {
+			return Account{}, fmt.Errorf("the process account at %s has no %s", path, f.name)
+		}
+	}
+	return a, nil
+}

go/process/account_test.go

@@ -0,0 +1,84 @@
+package process
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
+// module, user, password, claims}, when the process package reads it, then it yields those fields and
+// tolerates claims.
+func TestReadAccountYieldsEveryFieldAndToleratesClaims(t *testing.T) {
+	path := writeAccount(t, map[string]any{
+		"url": "nats://novox:4222", "fingerprint": "sha256:ab", "node": "novox", "module": "mesh-mail",
+		"user": "novox.mesh-mail.send", "password": "secret-pw",
+		"claims": []map[string]any{{"seat": "operator-channel", "verbs": []string{"tell"}}},
+	})
+	a, err := ReadAccount(path)
+	if err != nil {
+		t.Fatalf("a well-formed account file was refused: %v", err)
+	}
+	if a.URL != "nats://novox:4222" || a.Fingerprint != "sha256:ab" || a.Node != "novox" ||
+		a.Module != "mesh-mail" || a.User != "novox.mesh-mail.send" || a.Password != "secret-pw" {
+		t.Fatalf("the account read back: %+v", a)
+	}
+}
+
+// 1171/AC2: a process account carries no claims at all (heldProcessAccount); reading it is not refused for
+// the field being absent.
+func TestReadAccountToleratesNoClaimsAtAll(t *testing.T) {
+	path := writeAccount(t, map[string]any{
+		"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail",
+		"user": "novox.mesh-mail.send", "password": "secret-pw",
+	})
+	a, err := ReadAccount(path)
+	if err != nil {
+		t.Fatalf("a process account with no claims was refused: %v", err)
+	}
+	if a.User != "novox.mesh-mail.send" {
+		t.Fatalf("the account read back: %+v", a)
+	}
+}
+
+// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
+// in the error even when it is the field present and another is missing.
+func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
+	for _, tc := range []struct {
+		field  string
+		values map[string]any
+	}{
+		{"url", map[string]any{"node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "secret-pw"}},
+		{"user", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "password": "secret-pw"}},
+		{"password", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send"}},
+	} {
+		t.Run(tc.field, func(t *testing.T) {
+			path := writeAccount(t, tc.values)
+			_, err := ReadAccount(path)
+			if err == nil {
+				t.Fatalf("an account file missing %s was accepted", tc.field)
+			}
+			if !strings.Contains(err.Error(), tc.field) {
+				t.Errorf("the error %q does not name the missing field %s", err.Error(), tc.field)
+			}
+			if strings.Contains(err.Error(), "secret-pw") {
+				t.Errorf("the password appears in the error: %q", err.Error())
+			}
+		})
+	}
+}
+
+func writeAccount(t *testing.T, fields map[string]any) string {
+	t.Helper()
+	b, err := json.Marshal(fields)
+	if err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), "broker-send")
+	if err := os.WriteFile(path, b, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}

go/process/asks.go

@@ -0,0 +1,68 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// Ask reads the router's record of one of this process's module's own asks, under its own module and no
+// other's (novox/hq ADR 0436 clause 4, ADR 0259 §3): by direct get on bucket, the same subject the tool
+// runner reads (mesh-tools node-tools/internal/bus/seat.go ~300-345):
+//
+//	$JS.API.DIRECT.GET.KV_<bucket>.$KV.<bucket>.<module>.<id>
+//
+// bucket is the router's records bucket as the caller's own manifest names it (operator-channel's is
+// messenger_asks in the controller's own test fixtures) — never hard-coded to one bucket here, since a
+// process may read records of more than one seat's router. A missing, deleted or purged key is nil, not
+// an error.
+func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) {
+	if err := checkAskID(id); err != nil {
+		return nil, err
+	}
+	subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	reply, err := c.nc.RequestWithContext(ctx, subject, nil)
+	if err != nil {
+		return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err)
+	}
+	if status := reply.Header.Get("Status"); status == "404" {
+		return nil, nil
+	} else if status != "" {
+		return nil, fmt.Errorf("the record %s.%s could not be read: %s %s", c.account.Module, id, status,
+			reply.Header.Get("Description"))
+	}
+	if op := reply.Header.Get("KV-Operation"); op == "DEL" || op == "PURGE" {
+		return nil, nil
+	}
+	var revision uint64
+	if _, err := fmt.Sscan(reply.Header.Get("Nats-Sequence"), &revision); err != nil {
+		return nil, fmt.Errorf("the record %s.%s carries no readable revision: %w", c.account.Module, id, err)
+	}
+	return &stdio.StateEntry{Key: c.account.Module + "." + id, Value: askValueOf(reply.Data), Revision: revision}, nil
+}
+
+// checkAskID refuses an id holding a dot or a wildcard before anything is sent: either would name a
+// different key under the direct-get subject's own dot-joined path, never this one (novox/hq issue 1189
+// AC2).
+func checkAskID(id string) error {
+	if id == "" || strings.ContainsAny(id, ".*>") {
+		return fmt.Errorf("%q is not an ask id: an ask id holds no dot and no wildcard, naming one key and "+
+			"no other", id)
+	}
+	return nil
+}
+
+// askValueOf is a record's raw bytes as JSON, quoting them when they are not already valid JSON
+// (mesh-tools node-tools/internal/bus/state.go:351-357, valueOf).
+func askValueOf(raw []byte) json.RawMessage {
+	if len(raw) == 0 || !json.Valid(raw) {
+		b, _ := json.Marshal(string(raw))
+		return b
+	}
+	return json.RawMessage(append([]byte(nil), raw...))
+}

go/process/asks_integration_test.go

@@ -0,0 +1,60 @@
+//go:build integration
+
+package process
+
+import (
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// 1189/AC1: given the router's records bucket on a test bus holding a record under <module>.<ask id>,
+// when the process connection reads its module's ask of that id, then it gets the record's bytes and
+// revision by direct get; a deleted or purged key is said as absent; a missing key as absent.
+func TestAskReadsTheRoutersRecordOfItsOwnAsk(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	kv, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"})
+	if err != nil {
+		t.Fatalf("the admin could not make the records bucket: %v", err)
+	}
+	rev, err := kv.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	e, err := conn.Ask("messenger_asks", "ask-1")
+	if err != nil {
+		t.Fatalf("reading an existing record: %v", err)
+	}
+	if e == nil || e.Revision != rev || string(e.Value) != `{"state":"decided"}` {
+		t.Fatalf("read back %+v, want revision %d and {\"state\":\"decided\"}", e, rev)
+	}
+
+	if err := kv.Delete(ctxFor(t), "mesh-mail.ask-1"); err != nil {
+		t.Fatal(err)
+	}
+	if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e != nil {
+		t.Fatalf("a deleted record: %+v, %v, want absent", e, err)
+	}
+
+	if _, err := kv.Create(ctxFor(t), "mesh-mail.ask-2", []byte(`{"state":"asked"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if err := kv.Purge(ctxFor(t), "mesh-mail.ask-2"); err != nil {
+		t.Fatal(err)
+	}
+	if e, err := conn.Ask("messenger_asks", "ask-2"); err != nil || e != nil {
+		t.Fatalf("a purged record: %+v, %v, want absent", e, err)
+	}
+
+	if e, err := conn.Ask("messenger_asks", "ask-missing"); err != nil || e != nil {
+		t.Fatalf("a missing record: %+v, %v, want absent", e, err)
+	}
+}

go/process/asks_test.go

@@ -0,0 +1,28 @@
+package process
+
+import "testing"
+
+// 1189/AC2: given an ask id holding a dot or a wildcard, when read, then it is refused before anything is
+// sent — proven on a Conn with no live connection at all: if Ask tried to send, it would panic on the nil
+// *nats.Conn instead of returning a clean error.
+func TestAskRefusesAnInvalidIDBeforeSendingAnything(t *testing.T) {
+	c := &Conn{account: Account{Module: "mesh-mail"}}
+	for _, id := range []string{"", "a.b", "a*", "a>", "ask.1", "ask.*"} {
+		if _, err := c.Ask("messenger_asks", id); err == nil {
+			t.Errorf("Ask(%q) was not refused", id)
+		}
+	}
+}
+
+func TestCheckAskID(t *testing.T) {
+	for _, id := range []string{"", "a.b", "a*b", "a>b", ".", "*"} {
+		if err := checkAskID(id); err == nil {
+			t.Errorf("checkAskID(%q) was accepted", id)
+		}
+	}
+	for _, id := range []string{"ask-1", "a1b2c3", "send_42"} {
+		if err := checkAskID(id); err != nil {
+			t.Errorf("checkAskID(%q) was refused: %v", id, err)
+		}
+	}
+}

go/process/call.go

@@ -0,0 +1,84 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35,
+// RequestTimeout): what a module already expects.
+var CallTimeout = 30 * time.Second
+
+// Call invokes a tool or a seat verb as this process's own account (novox/hq ADR 0436 clause 4), on the
+// caller-named subject, falling back to the subject that names no caller when nothing answers it —
+// exactly the provisioner client's own pattern (mesh-catalog modules/mesh-provisioner-client/cmd/
+// mesh-provisioner-client/secrets.go:59-61, 93-97, 119, 125-133). key is "seat:<seat>.<verb>" for a seat's
+// verb, or "<module>.<tool>" for a module's own tool, as a process's bus.invokes names it. The reply is
+// {result, error, node} (mesh-tools node-tools/internal/bus/bus.go answerOf); a non-empty error is the
+// call's failure. It offers no way to answer a call — a process reaches the bus only as itself, and
+// never serves anything.
+func (c *Conn) Call(key string, body any) (json.RawMessage, error) {
+	called, fallback, err := callSubjects(key, c.callerToken())
+	if err != nil {
+		return nil, err
+	}
+	data, err := json.Marshal(body)
+	if err != nil {
+		return nil, err
+	}
+	result, err := c.request(called, data)
+	if errors.Is(err, nats.ErrNoResponders) {
+		result, err = c.request(fallback, data)
+	}
+	return result, err
+}
+
+// callerToken is this account's user name as the last token of a call it publishes, each dot written ~
+// (mesh-tools node-tools/internal/bus/caller.go:43, CallerToken).
+func (c *Conn) callerToken() string { return strings.ReplaceAll(c.account.User, ".", "~") }
+
+// callSubjects is the caller-named subject key is called on, and its plain fallback (mesh-tools
+// node-tools/internal/bus/caller.go CalledSubject, toolShape, calledPattern).
+func callSubjects(key, token string) (called, fallback string, err error) {
+	if rest, ok := strings.CutPrefix(key, "seat:"); ok {
+		seat, verb, ok := strings.Cut(rest, ".")
+		if !ok || seat == "" || verb == "" {
+			return "", "", fmt.Errorf("%q names no seat and verb (seat:<seat>.<verb>)", key)
+		}
+		return "mesh.seat." + seat + ".call." + verb + "." + token, "mesh.seat." + seat + ".tool." + verb, nil
+	}
+	module, tool, ok := strings.Cut(key, ".")
+	if !ok || module == "" || tool == "" {
+		return "", "", fmt.Errorf("%q is not <module>.<tool>, and does not start seat:<seat>.<verb>", key)
+	}
+	return "mesh.mod." + module + ".call." + tool + "." + token, "mesh.mod." + module + ".tool." + tool, nil
+}
+
+// request asks one subject once and reads its reply ({result, error, node}); a reply naming an error is
+// the call's failure, not a transport one.
+func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), CallTimeout)
+	defer cancel()
+	msg, err := c.nc.RequestWithContext(ctx, subject, data)
+	if err != nil {
+		return nil, err
+	}
+	var r struct {
+		Result json.RawMessage `json:"result"`
+		Error  string          `json:"error"`
+		Node   string          `json:"node"`
+	}
+	if err := json.Unmarshal(msg.Data, &r); err != nil {
+		return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err)
+	}
+	if r.Error != "" {
+		return nil, errors.New(r.Error)
+	}
+	return r.Result, nil
+}

go/process/call_integration_test.go

@@ -0,0 +1,237 @@
+//go:build integration
+
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// 1191/AC1: given a stand-in holder answering seat operator-channel verb tell on a test bus, when the
+// process connection of novox.mesh-mail.send calls seat:operator-channel.tell with {headline, lines,
+// subject}, then the request goes to mesh.seat.operator-channel.call.tell.novox~mesh-mail~send with the
+// arguments as the body, and the holder's {result} comes back; a reply with error set comes back as an
+// error with those words.
+func TestCallReachesTheCallerNamedSubjectAndReadsTheReply(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+
+	heard := make(chan *nats.Msg, 1)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		heard <- m
+		_ = m.Respond([]byte(`{"result":{"sent":true}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	result, err := conn.Call("seat:operator-channel.tell",
+		map[string]any{"headline": "h", "lines": []string{"l1"}, "subject": "s"})
+	if err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+	if string(result) != `{"sent":true}` {
+		t.Fatalf("result: %s", result)
+	}
+
+	select {
+	case msg := <-heard:
+		var body map[string]any
+		if err := json.Unmarshal(msg.Data, &body); err != nil {
+			t.Fatal(err)
+		}
+		if body["headline"] != "h" || body["subject"] != "s" {
+			t.Errorf("the arguments were not the call's body: %v", body)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("the holder never heard the call")
+	}
+}
+
+// 1191/AC1: a reply with error set comes back as an error with those words.
+func TestCallReadsAnErrorReplyAsAnError(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"error":"refused: no such warrant"}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	_, err = conn.Call("seat:operator-channel.tell", map[string]any{})
+	if err == nil || err.Error() != "refused: no such warrant" {
+		t.Fatalf("Call with an error reply: %v, want the words of the error", err)
+	}
+}
+
+// 1191/AC2: given no responder on the caller-named subject, when the call is made, then it is asked
+// again on mesh.seat.operator-channel.tool.tell (the tool runner's fallback).
+func TestCallFallsBackToThePlainSubjectWhenNothingAnswersTheCallerNamedOne(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.tool.tell", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"result":{"via":"fallback"}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	result, err := conn.Call("seat:operator-channel.tell", map[string]any{})
+	if err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+	if string(result) != `{"via":"fallback"}` {
+		t.Fatalf("result: %s, want the fallback holder's answer", result)
+	}
+}
+
+// 1191/AC4, release 2's acceptance test: given the test bus with accounts composed as the controller
+// composes a process account (the subjects of mesh-controller internal/broker/process_grants.go), when
+// one process connection emits, updates a state, reads an own ask's record, consumes and acknowledges its
+// queue and calls tell, then each succeeds; and its attempt to subscribe a tool-call subject is refused
+// by the bus.
+func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
+		t.Fatalf("the admin could not make the state bucket: %v", err)
+	}
+	asksKV, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"})
+	if err != nil {
+		t.Fatalf("the admin could not make the records bucket: %v", err)
+	}
+	if _, err := asksKV.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
+		Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
+	}); err != nil {
+		t.Fatalf("the admin could not make the events stream: %v", err)
+	}
+	makeQueue(t, url, account, 30*time.Second)
+	publishApproved(t, url, `{"send":"s1"}`)
+
+	adminNC, _ := adminJS(t, url)
+	holderSub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"result":{}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = holderSub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	// 1. Emits.
+	if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "ev-1"}); err != nil {
+		t.Fatalf("Emit: %v", err)
+	}
+
+	// 2. Updates a state.
+	s := conn.State("sends")
+	rev, err := s.Create("k", map[string]any{"v": 1})
+	if err != nil {
+		t.Fatalf("State.Create: %v", err)
+	}
+	if _, err := s.Update("k", map[string]any{"v": 2}, rev); err != nil {
+		t.Fatalf("State.Update: %v", err)
+	}
+
+	// 3. Reads an own ask's record.
+	if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e == nil {
+		t.Fatalf("Ask: %+v, %v", e, err)
+	}
+
+	// 4. Consumes and acknowledges its queue.
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	consumed := make(chan error, 1)
+	go func() {
+		consumed <- conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
+			if err := m.Ack(); err != nil {
+				t.Errorf("Ack: %v", err)
+			}
+			cancel()
+		})
+	}()
+	select {
+	case <-consumed:
+	case <-time.After(8 * time.Second):
+		t.Fatal("the process did not consume its own queue")
+	}
+
+	// 5. Calls tell.
+	if _, err := conn.Call("seat:operator-channel.tell", map[string]any{}); err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+
+	// 6. Refused: subscribing a tool-call subject — a fresh connection, so the error handler catches
+	// exactly this one refusal and nothing earlier.
+	refusals := make(chan error, 1)
+	watched, err := Connect(account, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer watched.Close()
+	if _, err := watched.nc.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
+		t.Fatal(err)
+	}
+	if err := watched.nc.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-refusals:
+		if !errors.Is(err, nats.ErrPermissionViolation) {
+			t.Errorf("subscribing a tool-call subject was not refused as a permissions violation: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Error("subscribing a tool-call subject the process does not serve was not refused")
+	}
+}

go/process/call_test.go

@@ -0,0 +1,41 @@
+package process
+
+import "testing"
+
+// 1191/AC3: given a module tool key <module>.<tool>, when called, then the subject is
+// mesh.mod.<module>.call.<tool>.<token> with the same fallback (mesh.mod.<module>.tool.<tool>).
+func TestCallSubjectsForAModuleTool(t *testing.T) {
+	called, fallback, err := callSubjects("mesh-mail.send", "novox~mesh-mail~send")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if called != "mesh.mod.mesh-mail.call.send.novox~mesh-mail~send" {
+		t.Errorf("called: %q", called)
+	}
+	if fallback != "mesh.mod.mesh-mail.tool.send" {
+		t.Errorf("fallback: %q", fallback)
+	}
+}
+
+// 1191/AC1's subject (the seat form): seat:<seat>.<verb> calls mesh.seat.<seat>.call.<verb>.<token>,
+// falling back to mesh.seat.<seat>.tool.<verb>.
+func TestCallSubjectsForASeatVerb(t *testing.T) {
+	called, fallback, err := callSubjects("seat:operator-channel.tell", "novox~mesh-mail~send")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if called != "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send" {
+		t.Errorf("called: %q", called)
+	}
+	if fallback != "mesh.seat.operator-channel.tool.tell" {
+		t.Errorf("fallback: %q", fallback)
+	}
+}
+
+func TestCallSubjectsRefusesAKeyNamingNeither(t *testing.T) {
+	for _, key := range []string{"", "nodot", "seat:", "seat:noverb", "."} {
+		if _, _, err := callSubjects(key, "token"); err == nil {
+			t.Errorf("callSubjects(%q) was accepted", key)
+		}
+	}
+}

go/process/conn.go

@@ -0,0 +1,162 @@
+package process
+
+import (
+	"context"
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"regexp"
+	"sync"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Conn is one process's own connection to the bus (novox/hq ADR 0436 clause 4): dialled with the account
+// the controller delivered, pinned to its fingerprint, on its own inbox prefix. It offers no Serve and no
+// way to serve a tool — a process reaches the bus only as itself, and never answers a call.
+type Conn struct {
+	nc      *nats.Conn
+	account Account
+
+	mu      sync.Mutex
+	next    uint64
+	pending map[string]map[uint64]context.CancelCauseFunc
+}
+
+// errPermissionRefused marks a context cancelled because the bus refused this account's publish of the
+// subject a pending call was waiting an answer on: a denied publish is refused only as an async protocol
+// error with no reply (mesh-tools node-tools/internal/bus/caller.go publishRefused), so without this a
+// call would wait out its whole timeout instead of failing at once.
+var errPermissionRefused = errors.New("the bus refused this account's publish")
+
+// refusedPublish reads the subject out of the bus's "-ERR 'Permissions Violation for Publish to
+// "<subject>"'" (mesh-tools node-tools/internal/bus/caller.go's own regexp).
+var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`)
+
+// Connect dials the bus with account: pinned to its fingerprint when it carries one (refused with any
+// other certificate), authenticated as its user and password, on the inbox prefix _INBOX.<user> and no
+// other, reconnecting without limit as the tool runner does (novox/hq issue 1192: a process outlives a
+// bus restart, never giving up the way the client library's own default of 60 attempts would). Ported
+// from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned, PinnedToFingerprint) since this
+// package may not import the controller.
+//
+// An ErrorHandler among opts is chained behind this package's own, which watches every async error for a
+// permission violation refusing a publish this connection is waiting an answer to, and cancels that wait
+// at once.
+func Connect(account Account, opts ...nats.Option) (*Conn, error) {
+	c := &Conn{account: account, pending: map[string]map[uint64]context.CancelCauseFunc{}}
+	nc, err := nats.Connect(account.URL, c.options(opts)...)
+	if err != nil {
+		return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err)
+	}
+	c.nc = nc
+	return c, nil
+}
+
+// options is the full option list Connect dials with, apart from Connect so a unit test can check what
+// it carries without a live bus: this package's own (the inbox prefix, the credentials, the pin,
+// reconnecting without limit, and an ErrorHandler chaining any the caller passed), then opts, then this
+// package's own ErrorHandler again last, so it is the one AsyncErrorCB nats.Connect keeps however opts
+// ordered theirs.
+func (c *Conn) options(opts []nats.Option) []nats.Option {
+	// opts are functions over a *nats.Options; applied to a scratch value, this reads off a caller's own
+	// ErrorHandler (if any) to chain, without needing nats.Connect's own URL/Servers parsing here.
+	var scratch nats.Options
+	for _, opt := range opts {
+		if opt != nil {
+			_ = opt(&scratch)
+		}
+	}
+	chained := scratch.AsyncErrorCB
+
+	all := []nats.Option{
+		nats.CustomInboxPrefix(c.account.InboxPrefix()),
+		nats.UserInfo(c.account.User, c.account.Password),
+		nats.MaxReconnects(-1),
+	}
+	if c.account.Fingerprint != "" {
+		all = append(all, nats.Secure(pinnedToFingerprint(c.account.Fingerprint)))
+	}
+	all = append(all, opts...)
+	return append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
+		c.publishRefused(err)
+		if chained != nil {
+			chained(nc, sub, err)
+		}
+	}))
+}
+
+// Close ends the connection.
+func (c *Conn) Close() { c.nc.Close() }
+
+// withRefusal derives a context from ctx that is also cancelled, with errPermissionRefused as its cause,
+// the moment the bus refuses this account's publish of subject — before ctx's own deadline, which a
+// denied publish would otherwise run out, refused only as an async error with no reply. Every call that
+// requests on the bus registers the subject it is about to publish to here first.
+func (c *Conn) withRefusal(ctx context.Context, subject string) (context.Context, func()) {
+	derived, cancel := context.WithCancelCause(ctx)
+	c.mu.Lock()
+	c.next++
+	id := c.next
+	if c.pending[subject] == nil {
+		c.pending[subject] = map[uint64]context.CancelCauseFunc{}
+	}
+	c.pending[subject][id] = cancel
+	c.mu.Unlock()
+	return derived, func() {
+		c.mu.Lock()
+		if list := c.pending[subject]; list != nil {
+			delete(list, id)
+			if len(list) == 0 {
+				delete(c.pending, subject)
+			}
+		}
+		c.mu.Unlock()
+		cancel(nil)
+	}
+}
+
+// publishRefused ends every call waiting on the subject the bus's error names, with errPermissionRefused
+// as the cause (mesh-tools node-tools/internal/bus/caller.go publishRefused): true when there was one
+// waiting.
+func (c *Conn) publishRefused(err error) bool {
+	if !errors.Is(err, nats.ErrPermissionViolation) {
+		return false
+	}
+	m := refusedPublish.FindStringSubmatch(err.Error())
+	if len(m) < 2 {
+		return false
+	}
+	c.mu.Lock()
+	cancels := c.pending[m[1]]
+	delete(c.pending, m[1])
+	c.mu.Unlock()
+	for _, cancel := range cancels {
+		cancel(errPermissionRefused)
+	}
+	return len(cancels) > 0
+}
+
+// pinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its
+// name and chain deliberately not consulted, as mesh-controller's own pin does (jetstream.go:116-132).
+func pinnedToFingerprint(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])
+			got := "sha256:" + hex.EncodeToString(sum[:])
+			if got != want {
+				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s…), expected %s…",
+					got[:23], want[:23])
+			}
+			return nil
+		},
+	}
+}

go/process/conn_integration_test.go

@@ -0,0 +1,115 @@
+//go:build integration
+
+package process
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/hex"
+	"encoding/json"
+	"math/big"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+)
+
+// selfSignedServer is a nats-server with JetStream and TLS, self-signed, for this test alone (mirroring
+// go/internal/testbus.Start, with a certificate added since testbus has none): a mesh's bus presents its own
+// certificate in no public trust store, which is exactly what a process pins to (ADR 0436 clause 4).
+func selfSignedServer(t *testing.T) (url, fingerprint string) {
+	t.Helper()
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "mesh-sdk-test-bus"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sum := sha256.Sum256(der)
+	fingerprint = "sha256:" + hex.EncodeToString(sum[:])
+	cert := tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
+		NoLog: true, NoSigs: true, TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}},
+		Username: "novox.mesh-mail.send", Password: "the-pw"}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatalf("a TLS bus for this test could not be made: %v", err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(10 * time.Second) {
+		s.Shutdown()
+		t.Fatal("a TLS bus for this test did not come up within 10s")
+	}
+	t.Cleanup(func() {
+		s.Shutdown()
+		s.WaitForShutdown()
+	})
+	return s.ClientURL(), fingerprint
+}
+
+// mailSenderAccount is the account novox.mesh-mail.send would be delivered as, the module's own secret
+// broker-send: written to a file and read back through ReadAccount (1171/AC2), exactly as a process
+// reads it, rather than built by hand.
+func mailSenderAccount(t *testing.T, url, fingerprint string) Account {
+	t.Helper()
+	data, err := json.Marshal(map[string]any{
+		"url": url, "fingerprint": fingerprint, "node": "novox", "module": "mesh-mail",
+		"user": "novox.mesh-mail.send", "password": "the-pw",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), "broker-send")
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	account, err := ReadAccount(path)
+	if err != nil {
+		t.Fatalf("ReadAccount on a well-formed file: %v", err)
+	}
+	return account
+}
+
+// 1171/AC3: given a test bus with TLS and the account composed for novox.mesh-mail.send, when the package
+// connects with the file, then it connects with the inbox prefix _INBOX.novox.mesh-mail.send.
+func TestConnectPinsTheFingerprintAndSetsTheInboxPrefix(t *testing.T) {
+	url, fingerprint := selfSignedServer(t)
+	account := mailSenderAccount(t, url, fingerprint)
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatalf("connecting with the right fingerprint was refused: %v", err)
+	}
+	defer conn.Close()
+	if !conn.nc.IsConnected() {
+		t.Fatal("the connection did not come up")
+	}
+	if got := conn.nc.Opts.InboxPrefix; got != "_INBOX.novox.mesh-mail.send" {
+		t.Errorf("inbox prefix: %q, want %q", got, "_INBOX.novox.mesh-mail.send")
+	}
+}
+
+// 1171/AC3: connecting with a fingerprint that is not the bus's own leaf certificate is refused.
+func TestConnectRefusesTheWrongFingerprint(t *testing.T) {
+	url, _ := selfSignedServer(t)
+	account := mailSenderAccount(t, url, "sha256:0000000000000000000000000000000000000000000000000000000000000000")
+	if _, err := Connect(account); err == nil {
+		t.Fatal("connecting with the wrong fingerprint was not refused")
+	}
+}

go/process/emit.go

@@ -0,0 +1,69 @@
+package process
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// Emit publishes an event as this process's module (novox/hq ADR 0436 clause 4), under its module's own
+// namespace (mesh.mod.<module>.event.<type>, the same subject the module's own principal would publish it
+// on), with the SDK's own envelope and headers (stdio.go:498-509). A chosen message id
+// (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second
+// publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main.
+func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error {
+	data, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+	id := ""
+	for _, o := range opts {
+		if o.MessageID != "" {
+			id = o.MessageID
+		}
+	}
+	if id == "" {
+		id = eventID()
+	}
+	msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType)
+	msg.Header.Set("x-event-id", id)
+	msg.Header.Set("x-source", c.account.Module)
+	msg.Header.Set("x-node", c.account.Node)
+	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+	msg.Header.Set("content-type", "application/json")
+	for _, o := range opts {
+		if o.CausationID != "" {
+			msg.Header.Set("x-causation-id", o.CausationID)
+		}
+		if o.Schema != "" {
+			msg.Header.Set("x-schema", o.Schema)
+		}
+	}
+	msg.Data = data
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
+	return err
+}
+
+// eventID is a random UUID, version 4, as stdio's own eventID (go/stdio.go:515-522) — duplicated rather
+// than imported, since stdio's is unexported and this package touches none of stdio's bundle machinery.
+func eventID() string {
+	var b [16]byte
+	_, _ = rand.Read(b[:])
+	b[6] = (b[6] & 0x0f) | 0x40
+	b[8] = (b[8] & 0x3f) | 0x80
+	h := hex.EncodeToString(b[:])
+	return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
+}

go/process/emit_integration_test.go

@@ -0,0 +1,57 @@
+//go:build integration
+
+package process
+
+import (
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// 1188/AC3: given a stream accepting mail.send.sent, when the process emits it twice with the same
+// message id inside the duplicate window, then the stream holds one message; with the headers stdio's
+// Emit sets (stdio.go:498-509).
+func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	account.Node = "novox"
+	_, adminJetStream := adminJS(t, url)
+	stream, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
+		Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
+	})
+	if err != nil {
+		t.Fatalf("the admin could not make the events stream: %v", err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	for i := 0; i < 2; i++ {
+		if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "send-s1"}); err != nil {
+			t.Fatalf("emit %d: %v", i, err)
+		}
+	}
+
+	info, err := stream.Info(ctxFor(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.State.Msgs != 1 {
+		t.Fatalf("the stream holds %d messages, want exactly 1 (the second publish is a duplicate)", info.State.Msgs)
+	}
+
+	msg, err := stream.GetMsg(ctxFor(t), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg.Header.Get("x-event-id") != "send-s1" || msg.Header.Get("x-source") != "mesh-mail" ||
+		msg.Header.Get("x-node") != "novox" || msg.Header.Get("content-type") != "application/json" ||
+		msg.Header.Get("x-time") == "" {
+		t.Errorf("the stored message's headers are not the SDK's: %v", msg.Header)
+	}
+}

go/process/fixture_integration_test.go

@@ -0,0 +1,100 @@
+//go:build integration
+
+package process
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// sendUser and sendPassword are mesh-mail's process "send" on novox, as mesh-controller would mint it
+// (mesh-controller internal/broker/process_grants_test.go mailSenderProcess, Principal.Username).
+const (
+	sendUser     = "novox.mesh-mail.send"
+	sendPassword = "the-pw"
+	adminUser    = "admin"
+	adminPass    = "admin-pw"
+)
+
+// sendGrants is the process principal mailSenderProcess's whole publish set, copied from
+// mesh-controller internal/broker/process_grants_test.go (TestAProcessIsGrantedExactlyItsBusDeclaration,
+// "dumped from a passing run... 16 subjects, correct today"), since this package may not import the
+// controller: the exact grant the controller composes for novox.mesh-mail.send.
+var sendGrants = []string{
+	"$JS.ACK.MAIL_SENDS.novox_mesh-mail_send.>",
+	"$JS.API.CONSUMER.CREATE.KV_mesh-mail_sends.>",
+	"$JS.API.CONSUMER.DELETE.KV_mesh-mail_sends.>",
+	"$JS.API.CONSUMER.INFO.MAIL_SENDS.novox_mesh-mail_send",
+	"$JS.API.CONSUMER.MSG.NEXT.MAIL_SENDS.novox_mesh-mail_send",
+	"$JS.API.DIRECT.GET.KV_mesh-mail_sends.>",
+	"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-mail.>",
+	"$JS.API.STREAM.INFO.KV_mesh-mail_sends",
+	"$JS.FC.KV_mesh-mail_sends.>",
+	"$KV.mesh-mail_sends.>",
+	"mesh.mod.mesh-mail.event.mail.send.failed",
+	"mesh.mod.mesh-mail.event.mail.send.sent",
+	"mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send",
+	"mesh.seat.operator-channel.call.tell.novox~mesh-mail~send",
+	"mesh.seat.operator-channel.tool.tell",
+	"mesh.seat.operator-channel.tool.tell.*",
+}
+
+// grantedSendProcess is a test bus with JetStream, with two users: novox.mesh-mail.send, permissioned to
+// exactly sendGrants and its own inbox (nats-server server.Options Users with Permissions, as
+// mesh-controller's cmd/mesh-controller/process_live_test.go composes accounts for a real bus — copied as
+// literal grants here since this package may not import the controller); and an unrestricted admin for the
+// test's own setup (creating streams and buckets, standing in for the holder and the runtime).
+func grantedSendProcess(t *testing.T) (url string, account Account) {
+	t.Helper()
+	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
+		NoLog: true, NoSigs: true,
+		Users: []*server.User{
+			{Username: adminUser, Password: adminPass},
+			{Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{
+				Publish:   &server.SubjectPermission{Allow: sendGrants},
+				Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>"}},
+			}},
+		},
+	}
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatalf("a granted test bus could not be made: %v", err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(10 * time.Second) {
+		s.Shutdown()
+		t.Fatal("a granted test bus did not come up within 10s")
+	}
+	t.Cleanup(func() {
+		s.Shutdown()
+		s.WaitForShutdown()
+	})
+	url = s.ClientURL()
+	return url, Account{URL: url, Node: "novox", Module: "mesh-mail", User: sendUser, Password: sendPassword}
+}
+
+// adminJS is the test's own unrestricted connection to url, for setting up fixtures.
+func adminJS(t *testing.T, url string) (*nats.Conn, jetstream.JetStream) {
+	t.Helper()
+	nc, err := nats.Connect(url, nats.UserInfo(adminUser, adminPass), nats.CustomInboxPrefix("_INBOX."+adminUser))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return nc, js
+}
+
+func ctxFor(t *testing.T) context.Context {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}

go/process/no_serve_test.go

@@ -0,0 +1,37 @@
+package process
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+)
+
+// 1171/AC3: it offers no Serve and no way to serve a tool — a process reaches the bus only as itself, and
+// never answers a call. Checked by reflection over *Conn's whole method set and field list, so a later
+// addition cannot bring one back without this test naming it.
+func TestConnOffersNoServeAndExposesNoRawConnection(t *testing.T) {
+	ptrType := reflect.TypeOf(&Conn{})
+	for _, forbidden := range []string{"Serve", "ServeOn", "Handle", "Subscribe", "Respond", "Answer"} {
+		if _, ok := ptrType.MethodByName(forbidden); ok {
+			t.Errorf("*Conn has a method %s: a process offers no way to serve a tool", forbidden)
+		}
+	}
+
+	natsConnType := reflect.TypeOf(&nats.Conn{})
+	structType := reflect.TypeOf(Conn{})
+	for i := 0; i < structType.NumField(); i++ {
+		f := structType.Field(i)
+		if f.IsExported() && (f.Type == natsConnType || f.Type == natsConnType.Elem()) {
+			t.Errorf("Conn exposes its raw connection through the exported field %s", f.Name)
+		}
+	}
+	for i := 0; i < ptrType.NumMethod(); i++ {
+		m := ptrType.Method(i)
+		for o := 0; o < m.Type.NumOut(); o++ {
+			if m.Type.Out(o) == natsConnType || m.Type.Out(o) == natsConnType.Elem() {
+				t.Errorf("*Conn.%s returns the raw *nats.Conn", m.Name)
+			}
+		}
+	}
+}

go/process/queue.go

@@ -0,0 +1,127 @@
+package process
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Message is one piece of work this process's queue consumer delivered, with how many times the bus has
+// delivered it. Nothing acts on its behalf: the handler decides its fate exactly once, by calling Ack,
+// NakWithDelay or Term (mesh-tools node-tools/internal/bus/seat.go's seat work Consume, with the same
+// three outcomes, offered here explicitly instead of by a handler's return value, since the caller
+// chooses its own delay before a message comes back).
+type Message struct {
+	Subject string
+	Body    []byte
+	Headers map[string]string
+	// Delivered is how many times the bus has handed this message over, this time included.
+	Delivered uint64
+
+	ack  func() error
+	nak  func(time.Duration) error
+	term func() error
+}
+
+// Ack acknowledges the message: the bus never delivers it again.
+func (m *Message) Ack() error { return m.ack() }
+
+// NakWithDelay asks for the message to be delivered again no sooner than delay, its delivery count
+// raised by one.
+func (m *Message) NakWithDelay(delay time.Duration) error { return m.nak(delay) }
+
+// Term terminates the message: the bus never delivers it again, and does not count it a dead letter the
+// way running out of deliveries does.
+func (m *Message) Term() error { return m.term() }
+
+// ProcessID is this account's own process id, the last segment of its user name <node>.<module>.<id>
+// (mesh-controller internal/broker, Principal.Username for KindProcess).
+func (a Account) ProcessID() string {
+	prefix := a.Node + "." + a.Module + "."
+	if !strings.HasPrefix(a.User, prefix) {
+		return ""
+	}
+	return strings.TrimPrefix(a.User, prefix)
+}
+
+// QueueConsumerName is the one durable consumer a process's bus.queues names it on its work-queue stream
+// (novox/hq ADR 0436 clauses 1, 2, 6; mesh-controller internal/broker/modstreams.go QueueConsumer):
+// <node>_<module>_<process id>.
+func (a Account) QueueConsumerName() string { return a.Node + "_" + a.Module + "_" + a.ProcessID() }
+
+// errNoMessage is one pull that fetched nothing within its wait: not an error to the caller, just a
+// reason to pull again.
+var errNoMessage = errors.New("no message this pull")
+
+// Consume binds this process's one durable consumer on stream (QueueConsumerName) and pulls from it, one
+// message at a time, handing each to handle until ctx ends: it then stops pulling and returns ctx.Err(),
+// never acknowledging a message still in handle's hands — only handle decides a message's fate, by
+// calling Ack, NakWithDelay or Term on it.
+func (c *Conn) Consume(ctx context.Context, stream string, handle func(*Message)) error {
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return err
+	}
+	name := c.account.QueueConsumerName()
+	cons, err := js.Consumer(ctx, stream, name)
+	if err != nil {
+		return fmt.Errorf("binding %s's queue consumer %s on %s: %w", c.account.Module, name, stream, err)
+	}
+	return runQueue(ctx, fetchOne(cons), handle)
+}
+
+// fetchOne is one pull of at most one message, non-blocking past ctx: Fetch itself does not take a
+// context, so this selects on its message channel and ctx together, and returns promptly once ctx ends
+// even while a fetch is still outstanding.
+func fetchOne(cons jetstream.Consumer) func(ctx context.Context) (*Message, error) {
+	return func(ctx context.Context) (*Message, error) {
+		batch, err := cons.Fetch(1, jetstream.FetchMaxWait(5*time.Second))
+		if err != nil {
+			return nil, err
+		}
+		select {
+		case msg, ok := <-batch.Messages():
+			if !ok {
+				return nil, errNoMessage
+			}
+			headers := map[string]string{}
+			for k := range msg.Headers() {
+				headers[k] = msg.Headers().Get(k)
+			}
+			var delivered uint64 = 1
+			if meta, err := msg.Metadata(); err == nil && meta != nil {
+				delivered = meta.NumDelivered
+			}
+			return &Message{Subject: msg.Subject(), Body: append([]byte(nil), msg.Data()...), Headers: headers,
+				Delivered: delivered, ack: msg.Ack, nak: msg.NakWithDelay, term: msg.Term}, nil
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+}
+
+// runQueue is Consume's loop, apart from the bus so a unit test can drive it with a fake next
+// (novox/hq issue 1190 AC4): it checks ctx before every pull, and stops as soon as next or the check
+// says ctx has ended, without touching the message past handing it to handle.
+func runQueue(ctx context.Context, next func(context.Context) (*Message, error), handle func(*Message)) error {
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		msg, err := next(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			if errors.Is(err, errNoMessage) {
+				continue
+			}
+			return err
+		}
+		handle(msg)
+	}
+}

go/process/queue_integration_test.go

@@ -0,0 +1,223 @@
+//go:build integration
+
+package process
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// mailSendsApprovedSubject is where the runtime publishes a work-queue message on mesh-mail's own stream
+// (mesh-controller internal/broker/modstreams.go ModuleStreamSubject, mail-sends' one event).
+const mailSendsApprovedSubject = "mesh.stream.mesh-mail.event.mail.send.approved"
+
+// makeQueue sets up MAIL_SENDS as a work-queue stream and binds the one durable consumer a process's
+// bus.queues names it (QueueConsumerName), both as the admin, mirroring what the controller composes
+// (mesh-controller internal/broker/modstreams.go QueueConsumer), named <node>_<module>_<process id>.
+// ackWait is the controller's 30s in most tests, shortened in the Term test so waiting past it (to prove
+// no redelivery) does not make that test slow.
+func makeQueue(t *testing.T, url string, account Account, ackWait time.Duration) {
+	t.Helper()
+	_, adminJetStream := adminJS(t, url)
+	if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
+		Name: "MAIL_SENDS", Subjects: []string{mailSendsApprovedSubject}, Retention: jetstream.WorkQueuePolicy,
+		Discard: jetstream.DiscardNew,
+	}); err != nil {
+		t.Fatalf("the admin could not make the work-queue stream: %v", err)
+	}
+	if _, err := adminJetStream.CreateConsumer(ctxFor(t), "MAIL_SENDS", jetstream.ConsumerConfig{
+		Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait,
+		MaxDeliver: 5,
+	}); err != nil {
+		t.Fatalf("the admin could not make the queue consumer: %v", err)
+	}
+}
+
+func publishApproved(t *testing.T, url string, body string) {
+	t.Helper()
+	nc, _ := adminJS(t, url)
+	if err := nc.Publish(mailSendsApprovedSubject, []byte(body)); err != nil {
+		t.Fatal(err)
+	}
+	if err := nc.FlushTimeout(3 * time.Second); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// recv reads one value off ch, failing the test rather than hanging forever if nothing arrives within
+// timeout.
+func recv[T any](t *testing.T, ch <-chan T, timeout time.Duration, what string) T {
+	t.Helper()
+	select {
+	case v := <-ch:
+		return v
+	case <-time.After(timeout):
+		t.Fatalf("%s: nothing arrived within %s", what, timeout)
+		var zero T
+		return zero
+	}
+}
+
+// streamMsgs is how many messages MAIL_SENDS still holds.
+func streamMsgs(t *testing.T, url string) uint64 {
+	t.Helper()
+	_, adminJetStream := adminJS(t, url)
+	stream, err := adminJetStream.Stream(ctxFor(t), "MAIL_SENDS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	info, err := stream.Info(ctxFor(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return info.State.Msgs
+}
+
+// 1190/AC1: given a work-queue stream with one durable pull consumer and a process connection, when a
+// message is consumed and acknowledged, then it is gone from the stream.
+func TestConsumeAckRemovesTheMessageFromTheStream(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	makeQueue(t, url, account, 30*time.Second)
+	publishApproved(t, url, `{"send":"s1"}`)
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	got := make(chan *Message, 1)
+	go func() {
+		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
+			got <- m
+			cancel()
+		})
+	}()
+	msg := recv(t, got, 8*time.Second, "the message")
+	<-ctx.Done()
+	if err := msg.Ack(); err != nil {
+		t.Fatalf("Ack: %v", err)
+	}
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		if n := streamMsgs(t, url); n == 0 {
+			return
+		} else if time.Now().After(deadline) {
+			t.Fatalf("the stream still holds %d messages after the ack", n)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+}
+
+// 1190/AC2: given a consumed message, when the handler asks for it to be delivered again after a delay,
+// then it comes back no sooner than the delay with its delivery count raised by one.
+func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	makeQueue(t, url, account, 30*time.Second)
+	publishApproved(t, url, `{"send":"s2"}`)
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	const delay = 2 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+	defer cancel()
+	type delivery struct {
+		at        time.Time
+		delivered uint64
+	}
+	deliveries := make(chan delivery, 2)
+	go func() {
+		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
+			deliveries <- delivery{time.Now(), m.Delivered}
+			if m.Delivered == 1 {
+				_ = m.NakWithDelay(delay)
+			} else {
+				_ = m.Ack()
+				cancel()
+			}
+		})
+	}()
+
+	first := recv(t, deliveries, 8*time.Second, "the first delivery")
+	if first.delivered != 1 {
+		t.Fatalf("first delivery count: %d, want 1", first.delivered)
+	}
+	second := recv(t, deliveries, delay+8*time.Second, "the redelivery")
+	if second.delivered != 2 {
+		t.Fatalf("second delivery count: %d, want 2", second.delivered)
+	}
+	if elapsed := second.at.Sub(first.at); elapsed < delay {
+		t.Fatalf("redelivered after %s, want no sooner than %s", elapsed, delay)
+	}
+}
+
+// 1190/AC3: given a consumed message, when the handler terminates it, then it is never delivered again.
+// A short AckWait (well under the 4s this then waits past) is what would otherwise make this pass
+// whether or not Term did anything at all: without it, the message simply sits unredelivered until the
+// controller's ordinary 30s AckWait, and the test would see the same "nothing arrived" result either way.
+func TestConsumeTermNeverDeliversTheMessageAgain(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	const ackWait = time.Second
+	makeQueue(t, url, account, ackWait)
+	publishApproved(t, url, `{"send":"s3"}`)
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	deliveries := make(chan *Message, 4)
+	go func() {
+		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
+			deliveries <- m
+			_ = m.Term()
+		})
+	}()
+
+	first := recv(t, deliveries, 8*time.Second, "the message")
+	if first.Delivered != 1 {
+		t.Fatalf("first delivery count: %d, want 1", first.Delivered)
+	}
+
+	// Term removes it from the work-queue stream and leaves the consumer with nothing pending — not just
+	// "has not come back yet", which a message merely sitting past an un-acted AckWait would also show.
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		if n := streamMsgs(t, url); n == 0 {
+			break
+		} else if time.Now().After(deadline) {
+			t.Fatalf("the stream still holds %d messages after Term", n)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	_, adminJetStream := adminJS(t, url)
+	cons, err := adminJetStream.Consumer(ctxFor(t), "MAIL_SENDS", account.QueueConsumerName())
+	if err != nil {
+		t.Fatal(err)
+	}
+	info, err := cons.Info(ctxFor(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.NumAckPending != 0 {
+		t.Fatalf("the consumer still has %d acks pending after Term", info.NumAckPending)
+	}
+
+	select {
+	case again := <-deliveries:
+		t.Fatalf("a terminated message was delivered again: %+v", again)
+	case <-time.After(4 * ackWait):
+	}
+}

go/process/queue_test.go

@@ -0,0 +1,71 @@
+package process
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+// 1190/AC4: given the consume loop, when its context ends, then it stops pulling and returns without
+// acknowledging a message in hand.
+func TestRunQueueStopsPullingAndNeverActsOnAMessageWhenContextEnds(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	acked := false
+	msg := &Message{Subject: "test", ack: func() error { acked = true; return nil },
+		nak: func(time.Duration) error { return nil }, term: func() error { return nil }}
+
+	pulls := 0
+	next := func(ctx context.Context) (*Message, error) {
+		pulls++
+		return msg, nil
+	}
+
+	handled := 0
+	done := make(chan error, 1)
+	go func() {
+		done <- runQueue(ctx, next, func(m *Message) {
+			handled++
+			// The handler does not act on the message at all: a redelivery, a termination or an
+			// acknowledgement is its choice alone, never the loop's.
+			cancel()
+		})
+	}()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("runQueue returned %v, want context.Canceled", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("runQueue did not stop when its context ended")
+	}
+	if handled != 1 {
+		t.Fatalf("handle was called %d times, want 1", handled)
+	}
+	if acked {
+		t.Error("runQueue acknowledged the message itself; only the handler may")
+	}
+	if pulls != 1 {
+		t.Fatalf("next was called %d times, want exactly 1: once the handler ends the context, runQueue's "+
+			"own check stops it before pulling again", pulls)
+	}
+}
+
+// 1190/AC4: a context already ended before the loop starts pulls nothing at all.
+func TestRunQueueNeverPullsOnAnAlreadyEndedContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	calls := 0
+	err := runQueue(ctx, func(context.Context) (*Message, error) {
+		calls++
+		return nil, context.Canceled
+	}, func(*Message) { t.Fatal("handle was called on an already-ended context") })
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("runQueue returned %v, want context.Canceled", err)
+	}
+	if calls != 0 {
+		t.Fatalf("next was called %d times, want 0", calls)
+	}
+}

go/process/reconnect_test.go

@@ -0,0 +1,27 @@
+package process
+
+import (
+	"testing"
+
+	"github.com/nats-io/nats.go"
+)
+
+// novox/hq issue 1192 (taken into this round, release 2's reviewer): Connect reconnects without limit, as
+// the tool runner does, so a process outlives a bus restart instead of giving up after the client
+// library's default of 60 attempts. Checked without a live bus: the option list Connect dials with is
+// applied to a scratch nats.Options and MaxReconnect read back.
+func TestConnectOptionsReconnectWithoutLimit(t *testing.T) {
+	c := &Conn{account: Account{Node: "novox", Module: "mesh-mail", User: "novox.mesh-mail.send", Password: "pw"}}
+	var o nats.Options
+	for _, opt := range c.options(nil) {
+		if opt == nil {
+			continue
+		}
+		if err := opt(&o); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if o.MaxReconnect != -1 {
+		t.Errorf("MaxReconnect: %d, want -1 (no limit)", o.MaxReconnect)
+	}
+}

go/process/state.go

@@ -0,0 +1,130 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
+// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so
+// this stays the ordinary request bound and is not exposed for a test to shrink.
+const stateTimeout = 30 * time.Second
+
+// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
+// clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
+// account's grants allow. It satisfies go/asker's Store.
+type State struct {
+	Module string
+	Name   string
+	// kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a
+	// unit test, a fake that never reaches a bus.
+	kv func(ctx context.Context) (jetstream.KeyValue, error)
+}
+
+// bucket is this state's name on the bus.
+func (s State) bucket() string { return s.Module + "_" + s.Name }
+
+// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
+// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself
+// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
+// not granted is refused there, promptly, as nats.ErrPermissionViolation.
+func (c *Conn) State(name string) State {
+	bucket := c.account.Module + "_" + name
+	return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
+		js, err := jetstream.New(c.nc)
+		if err != nil {
+			return nil, err
+		}
+		subject := "$JS.API.STREAM.INFO.KV_" + bucket
+		ctx, stop := c.withRefusal(ctx, subject)
+		defer stop()
+		kv, err := js.KeyValue(ctx, bucket)
+		if err != nil {
+			if errors.Is(context.Cause(ctx), errPermissionRefused) {
+				return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject)
+			}
+			return nil, err
+		}
+		return kv, nil
+	}}
+}
+
+func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err)
+	}
+	return kv, nil
+}
+
+// Get is a key's current value, or nil when it has none.
+func (s State) Get(key string) (*stdio.StateEntry, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	kv, err := s.store(ctx)
+	if err != nil {
+		return nil, err
+	}
+	entry, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
+	}
+	return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil
+}
+
+// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
+// stdio.ErrChanged.
+func (s State) Create(key string, value any) (uint64, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	kv, err := s.store(ctx)
+	if err != nil {
+		return 0, err
+	}
+	data, err := json.Marshal(value)
+	if err != nil {
+		return 0, err
+	}
+	rev, err := kv.Create(ctx, key, data)
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
+	}
+	if err != nil {
+		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
+	}
+	return rev, nil
+}
+
+// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
+// is refused as stdio.ErrChanged.
+func (s State) Update(key string, value any, revision uint64) (uint64, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	kv, err := s.store(ctx)
+	if err != nil {
+		return 0, err
+	}
+	data, err := json.Marshal(value)
+	if err != nil {
+		return 0, err
+	}
+	rev, err := kv.Update(ctx, key, data, revision)
+	if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
+	}
+	if err != nil {
+		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
+	}
+	return rev, nil
+}

go/process/state_integration_test.go

@@ -0,0 +1,96 @@
+//go:build integration
+
+package process
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a
+// key, reads it, and updates it with the revision it read, then each succeeds and returns the new
+// revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails
+// as stdio.ErrChanged.
+func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	s := conn.State("sends")
+
+	rev, err := s.Create("key", map[string]any{"v": 1})
+	if err != nil {
+		t.Fatalf("Create on a granted bucket: %v", err)
+	}
+	if rev == 0 {
+		t.Fatal("Create returned no revision")
+	}
+
+	e, err := s.Get("key")
+	if err != nil || e == nil {
+		t.Fatalf("Get after Create: %+v, %v", e, err)
+	}
+	if e.Revision != rev || string(e.Value) != `{"v":1}` {
+		t.Fatalf("Get read back %+v, want revision %d and {\"v\":1}", e, rev)
+	}
+
+	rev2, err := s.Update("key", map[string]any{"v": 2}, rev)
+	if err != nil {
+		t.Fatalf("Update at the revision read: %v", err)
+	}
+	if rev2 == rev {
+		t.Fatal("Update did not return a new revision")
+	}
+
+	if _, err := s.Update("key", map[string]any{"v": 3}, rev); !errors.Is(err, stdio.ErrChanged) {
+		t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err)
+	}
+
+	if _, err := s.Create("key", map[string]any{"v": 4}); !errors.Is(err, stdio.ErrChanged) {
+		t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err)
+	}
+}
+
+// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes
+// back as an error naming the bucket. The bucket exists (made by the admin) so the only reason Create
+// can fail is the missing grant, never "bucket not found"; and Conn's own ErrorHandler (conn.go
+// withRefusal/publishRefused) catches the denied publish at once, so this returns promptly rather than
+// after a timeout.
+func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	s := conn.State("unrelated")
+	_, err = s.Create("key", map[string]any{"v": 1})
+	if err == nil {
+		t.Fatal("writing to an ungranted bucket was not refused")
+	}
+	if !errors.Is(err, nats.ErrPermissionViolation) {
+		t.Errorf("the refusal %v does not wrap nats.ErrPermissionViolation", err)
+	}
+	if !strings.Contains(err.Error(), "mesh-mail_unrelated") {
+		t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
+	}
+}

go/process/state_test.go

@@ -0,0 +1,105 @@
+package process
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asker"
+)
+
+// fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State.
+type fakeEntry struct {
+	jetstream.KeyValueEntry
+	key   string
+	value []byte
+	rev   uint64
+}
+
+func (e fakeEntry) Key() string      { return e.key }
+func (e fakeEntry) Value() []byte    { return e.value }
+func (e fakeEntry) Revision() uint64 { return e.rev }
+
+// fakeKV is an in-memory jetstream.KeyValue: embedding the nil interface satisfies every method this test
+// never calls, so only Get, Create and Update need a body (1188/AC2's "a fake KV").
+type fakeKV struct {
+	jetstream.KeyValue
+	mu     sync.Mutex
+	values map[string][]byte
+	revs   map[string]uint64
+	next   uint64
+}
+
+func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} }
+
+func (f *fakeKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	v, ok := f.values[key]
+	if !ok {
+		return nil, jetstream.ErrKeyNotFound
+	}
+	return fakeEntry{key: key, value: v, rev: f.revs[key]}, nil
+}
+
+func (f *fakeKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if _, ok := f.values[key]; ok {
+		return 0, jetstream.ErrKeyExists
+	}
+	f.next++
+	f.values[key], f.revs[key] = value, f.next
+	return f.next, nil
+}
+
+func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.revs[key] != revision {
+		return 0, jetstream.ErrKeyRevisionMismatch
+	}
+	f.next++
+	f.values[key], f.revs[key] = value, f.next
+	return f.next, nil
+}
+
+func stateOn(kv jetstream.KeyValue) State {
+	return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }}
+}
+
+// 1188/AC2: given the process connection's state type, when used as go/asker's Store, then it compiles and
+// behaves as the interface says, against a fake KV.
+func TestStateSatisfiesAskerStore(t *testing.T) {
+	var _ asker.Store = State{}
+
+	s := stateOn(newFakeKV())
+	if e, err := s.Get("ask.1"); err != nil || e != nil {
+		t.Fatalf("Get of an absent key: %v, %v", e, err)
+	}
+	rev, err := s.Create("ask.1", map[string]any{"a": 1})
+	if err != nil {
+		t.Fatalf("Create: %v", err)
+	}
+	e, err := s.Get("ask.1")
+	if err != nil || e == nil || e.Revision != rev {
+		t.Fatalf("Get after Create: %+v, %v", e, err)
+	}
+	if _, err := s.Create("ask.1", map[string]any{"a": 2}); !errors.Is(err, stdio.ErrChanged) {
+		t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err)
+	}
+	rev2, err := s.Update("ask.1", map[string]any{"a": 3}, rev)
+	if err != nil {
+		t.Fatalf("Update at the revision read: %v", err)
+	}
+	if _, err := s.Update("ask.1", map[string]any{"a": 4}, rev); !errors.Is(err, stdio.ErrChanged) {
+		t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err)
+	}
+	if rev2 == rev {
+		t.Fatal("Update did not advance the revision")
+	}
+}

go/stdio.go

@@ -57,7 +57,11 @@ type message struct {
 // channel is the one conversation with the runtime: what this bundle writes, serialised, and what
 // it asked and waits for.
 type channel struct {
-	mu        sync.Mutex
+	mu sync.Mutex
+	// wmu orders the writes to the runtime and nothing else: a write waits on the pipe for as long as
+	// the runtime does not read, and mu must never wait with it, or the loop that reads the runtime's
+	// answers stops (issue 1286).
+	wmu       sync.Mutex
 	enc       *json.Encoder
 	next      int
 	waiting   map[string]chan message
@@ -90,8 +94,8 @@ type Envelope struct {
 
 func (c *channel) say(m map[string]any) {
 	m["jsonrpc"] = "2.0"
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	c.wmu.Lock()
+	defer c.wmu.Unlock()
 	_ = c.enc.Encode(m)
 }
 
@@ -472,10 +476,15 @@ func matchFrom(p []string, pi int, k []string, ki int) bool {
 	return matchFrom(p, pi+1, k, ki+1)
 }
 
-// EmitOptions is the provenance an emitter may attach (ADR 0042).
+// EmitOptions is the provenance an emitter may attach (ADR 0042), and the message id it may choose.
 type EmitOptions struct {
 	CausationID string
 	Schema      string
+	// MessageID is the id this emit is sent under, as x-event-id (and, by the tool runner, as the bus's
+	// own Nats-Msg-Id): chosen by the emitter so a repeated publish inside the stream's duplicate window
+	// is kept to one, never generated once an id is given (novox/hq ADR 0436 clause 4). Empty is a random
+	// one, as on main.
+	MessageID string
 }
 
 // Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has
@@ -495,8 +504,17 @@ func Emit(eventType string, body any, opts ...EmitOptions) error {
 	if node == "" {
 		node = "unknown"
 	}
+	id := ""
+	for _, o := range opts {
+		if o.MessageID != "" {
+			id = o.MessageID
+		}
+	}
+	if id == "" {
+		id = eventID()
+	}
 	headers := map[string]string{
-		"x-event-id": eventID(), "x-source": source, "x-node": node,
+		"x-event-id": id, "x-source": source, "x-node": node,
 		"x-time": time.Now().UTC().Format(time.RFC3339Nano), "content-type": "application/json",
 	}
 	for _, o := range opts {

go/stdio_test.go

@@ -224,3 +224,54 @@ func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) {
 		t.Fatalf("one event was handled %d times", handled)
 	}
 }
+
+// A message the runtime has not read yet does not stop the channel taking its answers: with one tool
+// answer blocked on the pipe, a refused subscribe is still delivered to Subscribe, which returns the
+// refusal — no lock of the channel is held across a write to the pipe (novox/hq issue 1286/AC1, 1286/AC2).
+func TestARefusedSubscribeIsDeliveredWhileAMessageWaitsOnThePipe(t *testing.T) {
+	t.Setenv(ServedAs, "plex")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	defer func() { _ = inW.Close() }()
+	// Not the channel an earlier run left behind: Subscribe goes to the one this run serves.
+	currentMu.Lock()
+	current = nil
+	currentMu.Unlock()
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "quick", Run: func(map[string]any) (any, error) { return "ok", nil }}}, inR, outW)
+	}()
+	subscribed := make(chan error, 1)
+	go func() {
+		for {
+			if _, err := live(); err == nil {
+				break
+			}
+			time.Sleep(time.Millisecond)
+		}
+		subscribed <- Subscribe("*.download.completed", func(Envelope) error { return nil })
+	}()
+	go func() {
+		lines := bufio.NewScanner(outR)
+		if !lines.Scan() {
+			return
+		}
+		var req map[string]any
+		_ = json.Unmarshal(lines.Bytes(), &req)
+		send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) }
+		// A tool answer the runtime has not read: its write waits on the pipe, and no one reads further.
+		send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "quick"}})
+		time.Sleep(100 * time.Millisecond)
+		send(map[string]any{"jsonrpc": "2.0", "id": req["id"], "error": map[string]any{"code": -32000, "message": "not yet"}})
+	}()
+	deadline := time.After(2 * time.Second)
+	select {
+	case err := <-subscribed:
+		if err == nil || err.Error() != "not yet" {
+			t.Fatalf("the refused subscribe said %v", err)
+		}
+	case <-deadline:
+		t.Fatal("Subscribe did not return in 2 s: the channel is stuck on a write to a pipe nobody reads")
+	}
+	// Let the waiting answer go, so nothing is left blocked.
+	go func() { _, _ = io.Copy(io.Discard, outR) }()
+}

merge-check.sh

@@ -20,7 +20,7 @@ node node_modules/typescript/bin/tsc --noEmit
 npm test
 
 if command -v go >/dev/null 2>&1; then
-	(cd go && test -z "$(gofmt -l .)" && go vet ./... && go test -count=1 ./...)
+	sh go/merge-check.sh
 else
 	echo "NOT TESTED HERE: go/ (the Go binding) — the Go toolchain runs it from go/merge-check.sh"
 fi