M-10: the manager hears what each node holds as holdings-changed (issue 1409)

The manager consumes the agent module's holdings-changed, keeping each report
as its node's latest and considering them at once, in place of reading the
agent module's state; once it hears them, at every start after its
carry-over, it says holdings.asked with its time and who asks, so every node
says what it holds again.

Builds: design 39 bus-and-switch §23, §46, B12, B22

Claude-Session: https://claude.ai/code/session_01NcbgfHb9d4T9nq5Q6WqdiJ
This commit is contained in:
2026-11-11 23:53:10 +01:00
parent 31ce47d1cc
commit bd65b023bd
4 changed files with 114 additions and 51 deletions
@@ -0,0 +1,68 @@
package main
// What every node holds, heard as events (design 39 bus-and-switch §23, §46, B12): the agent module says
// `holdings-changed` with its holdings report whole, at its start, on every change and when asked; the manager says
// `holdings.asked` at every start, after its carry-over, and on its resync, and takes each report it hears as that
// node's latest.
import (
"encoding/json"
"fmt"
"sort"
"sync"
"time"
)
// HoldingsChanged is the agent module's event saying what one node holds, which the manager consumes.
const HoldingsChanged = "claude-code.holdings-changed"
// HoldingsAsked is the manager's event asking every node to say what it holds again.
const HoldingsAsked = "holdings.asked"
// Heard is what every node last said it holds, by node.
type Heard struct {
mu sync.Mutex
reports map[string]Holdings
}
// Take keeps one `holdings-changed` as its node's latest report.
func (h *Heard) Take(body []byte) (Holdings, error) {
var r Holdings
if err := json.Unmarshal(body, &r); err != nil {
return r, fmt.Errorf("%s is not a holdings report: %w", HoldingsChanged, err)
}
if r.Node == "" {
return r, fmt.Errorf("%s names no node", HoldingsChanged)
}
h.mu.Lock()
defer h.mu.Unlock()
if h.reports == nil {
h.reports = map[string]Holdings{}
}
h.reports[r.Node] = r
return r, nil
}
// All is every node's latest report, by node's name.
func (h *Heard) All() []Holdings {
h.mu.Lock()
all := make([]Holdings, 0, len(h.reports))
for _, r := range h.reports {
all = append(all, r)
}
h.mu.Unlock()
sort.Slice(all, func(i, j int) bool { return all[i].Node < all[j].Node })
return all
}
// Of is the generation node last reported applying.
func (h *Heard) Of(node string) int64 {
h.mu.Lock()
defer h.mu.Unlock()
return h.reports[node].Generation
}
// AskHoldings says `holdings.asked`, its time and who asks (§23).
func (m *Manager) AskHoldings(by string) error {
return m.Emit(HoldingsAsked, map[string]any{"at": m.Now().UTC().Format(time.RFC3339Nano), "by": by})
}
@@ -13,7 +13,6 @@ import (
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
@@ -87,11 +86,7 @@ func manager() (*Manager, error) {
}
return a, json.Unmarshal(raw, &a)
},
Reported: func(node string) int64 {
heardMu.Lock()
defer heardMu.Unlock()
return heard[node].Generation
},
Reported: heard.Of,
Emit: func(event string, body map[string]any) error { return stdio.Emit(event, body) },
Now: time.Now,
Log: say,
@@ -128,11 +123,8 @@ func keypair(dir string) (KeyPair, error) {
// ---- the long-running half ---------------------------------------------------------------------------
// heard is what every node last reported holding (ADR 0206), by node.
var (
heardMu sync.Mutex
heard = map[string]Holdings{}
)
// heard is what every node last said it holds (ADR 0206, design 39 bus-and-switch §23), by node.
var heard = &Heard{}
// carryOverFirst runs the carry-over at the start, before the manager hears an event, refreshes or reads usage (§11), and
// answers once it has ended: at once when `carried.postgres` exists, else when a run writes it, this one or one asked
@@ -153,7 +145,6 @@ func carryOverFirst() {
func daemon() {
carryOverFirst()
mu, reports := &heardMu, heard
var passing sync.Mutex
pass := func() {
passing.Lock() // one pass at a time: each account is leased, and a pass is cheap
@@ -163,14 +154,7 @@ func daemon() {
say("not ready: %v", err)
return
}
mu.Lock()
all := make([]Holdings, 0, len(reports))
for _, r := range reports {
all = append(all, r)
}
mu.Unlock()
sort.Slice(all, func(i, j int) bool { return all[i].Node < all[j].Node })
adopted, err := m.Consider(context.Background(), all)
adopted, err := m.Consider(context.Background(), heard.All())
if err != nil {
say("considering the reports failed: %v", err)
}
@@ -179,42 +163,34 @@ func daemon() {
}
}
// What every node holds (ADR 0206): the whole current set, then each change. Asked again until it
// answers — the channel to the runtime opens as the bundle starts, and the agent module's state may
// arrive after this one.
// What every node holds (ADR 0206, design 39 bus-and-switch §23): each `holdings-changed` heard is that node's
// latest report and starts a pass; once heard, every node is asked to say it again (B12: at every start, after the
// carry-over). Asked again until it answers — the channel to the runtime opens as the bundle starts.
go func() {
waits := []time.Duration{2 * time.Second, 5 * time.Second, 10 * time.Second, 30 * time.Second}
for attempt := 0; ; attempt++ {
err := stdio.State("claude-code.holdings").Watch("", func(c stdio.StateChange) error {
mu.Lock()
if c.Op == "put" {
var h Holdings
if json.Unmarshal(c.Value, &h) == nil {
reports[c.Key] = h
}
} else {
delete(reports, c.Key)
}
mu.Unlock()
// During the current values the pass waits for the whole set: newest login first needs all.
if !c.Current {
go pass()
err := stdio.Subscribe(HoldingsChanged, func(e stdio.Envelope) error {
if _, err := heard.Take(e.Body); err != nil {
say("%v", err)
return nil
}
go pass()
return nil
})
if err == nil {
mu.Lock()
n := len(reports)
mu.Unlock()
say("watching what %d node(s) hold", n)
pass()
say("hearing what the nodes hold")
if m, err := manager(); err != nil {
say("not ready to ask what the nodes hold: %v", err)
} else if err := m.AskHoldings(Itself); err != nil {
say("asking what the nodes hold could not be said: %v", err)
}
return
}
pause := time.Minute
if attempt < len(waits) {
pause = waits[attempt]
}
say("what the nodes hold cannot be watched yet (%v); asking again in %s", err, pause)
say("what the nodes hold cannot be heard yet (%v); asking again in %s", err, pause)
time.Sleep(pause)
}
}()
@@ -388,22 +388,40 @@ func TestANodeThatDoesNotAnswerIsAskedAgainAndOneWithNoLoginIsSettled(t *testing
}
}
func TestAReportIsReadAsTheAgentModuleWritesIt(t *testing.T) {
// The agent module's own report shape (claude-code's holdingsOf), parsed here.
raw := `{"node":"laptop","identity":{"accountUuid":"u-1","emailAddress":"a@example.org"},"kind":"subscription",
"refresh":{"present":true,"fingerprint":"sha256:0123456789abcdef","expiresAt":null},
"access":{"fingerprint":"sha256:fedcba9876543210","expiresAt":1},"licence":null,"generation":0,
"changedAt":"2026-10-04T11:00:00.000Z"}`
var h Holdings
if err := json.Unmarshal([]byte(raw), &h); err != nil {
// design 39 bus-and-switch §23, §46, B12, B22: a holdings-changed, decoded from a copy of the agent module's form, is
// kept as its node's latest report, and one carrying a fingerprint the manager does not hold is a candidate; the
// manager asks every node with holdings.asked, its time and who asks.
func TestAHoldingsChangedIsHeardAndMakesACandidate(t *testing.T) {
mm := newMesh(t)
raw, err := os.ReadFile("testdata/holdings-changed.json")
if err != nil {
t.Fatal(err)
}
heard := &Heard{}
if _, err := heard.Take(raw); err != nil {
t.Fatal(err)
}
all := heard.All()
if len(all) != 1 || heard.Of("laptop") != 4 {
t.Fatalf("heard %+v", all)
}
h := all[0]
if h.Node != "laptop" || h.Identity.AccountUUID != "u-1" || !h.Refresh.Present || h.Refresh.Fingerprint != "sha256:0123456789abcdef" {
t.Fatalf("%+v", h)
}
if _, err := time.Parse(time.RFC3339Nano, h.ChangedAt); err != nil {
t.Fatalf("the report's time does not parse: %v", err)
}
if c, err := mm.m.CandidatesIn(context.Background(), all); err != nil || len(c["u-1"]) != 1 {
t.Fatalf("candidates %v %v", c, err)
}
if _, err := heard.Take([]byte(`{"kind":"subscription"}`)); err == nil {
t.Fatal("a report naming no node was kept")
}
if err := mm.m.AskHoldings(Itself); err != nil || len(mm.events) != 1 ||
mm.events[0] != HoldingsAsked+` {"at":"`+t0.UTC().Format(time.RFC3339Nano)+`","by":"claude-licence-manager"}` {
t.Fatalf("said %v %v", mm.events, err)
}
}
// A node whose report arrives after its account was adopted — with an older login, so never a candidate —
@@ -0,0 +1 @@
{"node":"laptop","identity":{"accountUuid":"u-1","emailAddress":"a@example.org"},"kind":"subscription","refresh":{"present":true,"fingerprint":"sha256:0123456789abcdef","expiresAt":null},"access":{"fingerprint":"sha256:fedcba9876543210","expiresAt":1},"licence":null,"generation":4,"changedAt":"2026-10-04T11:00:00.000Z"}