Join item 2: mesh-knowledge's ledger pipe, call ids and page heartbeat come from the SDK

The ledger pipe asks the runtime through stdio.BusState (it answered ErrNoPipe before), an
operation names the call the runtime proved (stdio.CallIDOf), and the page takes the
runtime's word on its consumer (stdio.OnHeartbeat): a beat is every subscription heard, a
missed heartbeat marks each subscribed source at once with the runtime's why. Rows in
TestASourceThatDoesNotAnswerIsSaidAndAskedAgainOnABackoff and
TestTheLedgerStoreRefusesWhatTheLedgerWouldHaveToTakeBack; each failed before the change.

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

modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
 )
 
@@ -92,6 +94,23 @@ func TestTheLedgerStoreRefusesWhatTheLedgerWouldHaveToTakeBack(t *testing.T) {
 	if err := s.Create(ctx, Version{Record: ledgerRecord(7, StatusProposed)}); err == nil {
 		t.Fatal("a refused apply was taken")
 	}
+	// The bundle's own pipe is the runtime's call on its bus state (stdio.BusState), and the call it serves is the one
+	// the runtime named (stdio.CallIDOf): the method and request go through whole, the answer comes back decoded.
+	if reflect.ValueOf(busState).Pointer() != reflect.ValueOf(stdio.BusState).Pointer() ||
+		reflect.ValueOf(callIDOf).Pointer() != reflect.ValueOf(stdio.CallIDOf).Pointer() {
+		t.Fatal("the pipe is not the SDK's bus-state call, or the call id not the runtime's")
+	}
+	defer func(was func(string, any) (json.RawMessage, error)) { busState = was }(busState)
+	var asked string
+	busState = func(method string, params any) (json.RawMessage, error) {
+		asked = method
+		return json.RawMessage(`{"instance":"1"}`), nil
+	}
+	var got struct{ Instance string }
+	if err := (sdkPipe{}).Call(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: "1"}, &got); err != nil ||
+		asked != buscalls.LedgerGet || got.Instance != "1" {
+		t.Fatalf("the pipe asked %s and answered %+v, %v", asked, got, err)
+	}
 }
 
 func TestTheOperationAChangeIsIsNamedByWhatChanged(t *testing.T) {

modules/mesh-knowledge/cmd/mesh-knowledge/overview.go

@@ -13,6 +13,8 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
 )
 
 // The sources of the page, besides the module's own ledger.
@@ -519,3 +521,36 @@ func (o *Overview) Calls(verb string) int {
 	defer o.mu.Unlock()
 	return o.verbCalls[verb]
 }
+
+// Beat takes the runtime's word on the module's consumer between events (stdio.OnHeartbeat): one consumer carries every
+// subscription, so a beat is heard by every source the page subscribes to, and a missed heartbeat marks each of them
+// from its subscription at once, with the runtime's why, instead of waiting out HeartbeatWithin.
+func (o *Overview) Beat(h stdio.Heartbeat) {
+	sources := make([]string, 0, len(OverviewEvents))
+	for s := range OverviewEvents {
+		sources = append(sources, s)
+	}
+	sort.Strings(sources)
+	if !h.Missed {
+		for _, s := range sources {
+			o.Heard(s, "", nil)
+		}
+		return
+	}
+	why := "its subscription's consumer missed the bus's heartbeat"
+	if h.Why != "" {
+		why += ": " + h.Why
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	for _, s := range sources {
+		st := o.state[s]
+		switch {
+		case st == nil:
+		case st.Mark == nil:
+			st.Answered, st.Mark = false, &sourceMark{Cause: MarkSubscription, Why: why, Since: o.now()}
+		case st.Mark.Cause == MarkSubscription:
+			st.Mark.Why = why
+		}
+	}
+}

modules/mesh-knowledge/cmd/mesh-knowledge/overview_test.go

@@ -7,6 +7,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
 )
 
 // overviewWorld is a page model over the sources played here: what each verb answers, and what fails.
@@ -170,6 +172,27 @@ func TestASourceThatDoesNotAnswerIsSaidAndAskedAgainOnABackoff(t *testing.T) {
 	if mk := w.o.state[SourceDelivery].Mark; mk == nil || mk.Cause != MarkVerb {
 		t.Fatalf("%+v", mk)
 	}
+	// The runtime's word on the module's consumer (stdio.OnHeartbeat): a missed heartbeat marks every subscribed source
+	// at once, before the bound passes, with the runtime's why; the next beat asks each verb once and clears the marks.
+	w.fail[DeliveryGroupsVerb] = false
+	w.now = w.o.state[SourceDelivery].Mark.Next
+	w.o.Due()
+	asked = w.o.Calls(AsksVerb)
+	w.o.Beat(stdio.Heartbeat{At: w.now, Missed: true, Why: "the bus's heartbeat stopped arriving"})
+	if u := w.o.Unanswered(); len(u) != 3 || u[0].Mark.Cause != MarkSubscription || !strings.Contains(u[0].Mark.Why, "stopped arriving") {
+		t.Fatalf("a missed heartbeat marked %+v", u)
+	}
+	w.o.Beat(stdio.Heartbeat{At: w.now})
+	if len(w.o.Unanswered()) != 0 || w.o.Calls(AsksVerb) != asked+1 {
+		t.Fatalf("a beat left %+v, asks asked %d", w.o.Unanswered(), w.o.Calls(AsksVerb)-asked)
+	}
+	// A beat keeps a quiet source from being marked silent: it is heard, not silent.
+	w.now = w.now.Add(HeartbeatWithin - time.Second)
+	w.o.Beat(stdio.Heartbeat{At: w.now})
+	w.now = w.now.Add(2 * time.Second)
+	if got := w.o.Silent(); len(got) != 0 {
+		t.Fatalf("a source heard by its beat was marked silent: %v", got)
+	}
 }
 
 func TestThePagesModelIsAToolAndItsRereadIsAHandAct(t *testing.T) {

modules/mesh-knowledge/cmd/mesh-knowledge/page.go

@@ -117,6 +117,11 @@ func ServePage(k *KB) {
 			}()
 		}
 	}
+	// The runtime's word on the module's consumer between events: a beat is each subscription heard, a missed heartbeat
+	// marks the sources at once (stdio.OnHeartbeat).
+	if err := stdio.OnHeartbeat(o.Beat); err != nil {
+		logf("[mesh-knowledge] the page hears no heartbeat of its subscriptions (%v): a silent source is marked after %s", err, HeartbeatWithin)
+	}
 	o.Start()
 	go func() {
 		for range time.Tick(5 * time.Second) {

modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go

@@ -7,7 +7,6 @@ package main
 import (
 	"context"
 	"encoding/json"
-	"errors"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
 )
@@ -55,23 +54,15 @@ func callOf(ctx context.Context) string {
 	return s
 }
 
-// ErrNoPipe is the answer of a pipe the pinned mesh-sdk does not give the bundle yet.
-var ErrNoPipe = errors.New("the pinned mesh-sdk gives a bundle no request on the tool runner's bus-state calls yet " +
-	"(a request to the SDK's owner: stdio.Request); the records cannot be reached over the bus")
-
-// sdkPipe is the pipe through the SDK. Request is set when the SDK offers it; until then the pipe answers ErrNoPipe,
-// loudly, and nothing is written.
+// sdkPipe is the pipe through the SDK: each call on the module's own ledger is asked of the runtime, which serves it
+// from the module's own bindings only (stdio.BusState, ADR 0595 §9).
 type sdkPipe struct{}
 
-// request is the SDK's raw request on the runtime's channel (stdio.Request(method, params) (json.RawMessage, error)),
-// wired in the one place the group's mesh-sdk pin brings it.
-var request func(method string, params any) (json.RawMessage, error)
+// busState is the runtime's call on the module's own bus state: stdio.BusState, a test's stand-in.
+var busState = stdio.BusState
 
 func (sdkPipe) Call(method string, req, answer any) error {
-	if request == nil {
-		return ErrNoPipe
-	}
-	raw, err := request(method, req)
+	raw, err := busState(method, req)
 	if err != nil {
 		return err
 	}
@@ -82,9 +73,8 @@ func (sdkPipe) Call(method string, req, answer any) error {
 }
 
 // callIDOf is the id of the tool call that was handed args, as the tool runner names it in the call's `_meta` under
-// buscalls.MetaCall: wired to the SDK where it offers it (a request to the SDK's owner: stdio.CallIDOf); until then no
-// call is named, and an operation carries the module's own account alone.
-var callIDOf = func(args map[string]any) string { return "" }
+// buscalls.MetaCall (stdio.CallIDOf): empty for the module's own act, whose operation carries the module's account alone.
+var callIDOf = stdio.CallIDOf
 
 // noteCall holds the call a caller's write serves for as long as it runs, and ctxFor names it to the ledger: the tool
 // runner stamps the caller it proved for that call as the operation's author (ADR 0597 §7), never a name the caller or