Stop the forge bridge performing mesh/check: the forge module writes it (R2-1)

R4e round 2 finding R2-1, ADR 0521: nothing in the delivery plan orders the
forge module before the bridge, and the forge module's durable consumer keeps
its deliver policy when its filters change, so a check.status emitted before
it consumed the event was lost. The forge module now watches the checker's
owed state and writes mesh/check itself, once per revision.

The bridge emits no check.status, declares no emits, and neither reads the
checker's owed nor hears its stream; OwedAct still refuses mesh/check. The
integration test shows the bridge calls no tool and publishes nothing of its
own when the checker owes mesh/check.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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