diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/holdings.go b/modules/claude-licence-manager/cmd/claude-licence-manager/holdings.go new file mode 100644 index 000000000..1925b9613 --- /dev/null +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/holdings.go @@ -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}) +} diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/main.go b/modules/claude-licence-manager/cmd/claude-licence-manager/main.go index dc3a4292e..ec5e2015e 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/main.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/main.go @@ -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) } }() diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go b/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go index b2ec649cd..c0e5da2a6 100644 --- a/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/manager_test.go @@ -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 — diff --git a/modules/claude-licence-manager/cmd/claude-licence-manager/testdata/holdings-changed.json b/modules/claude-licence-manager/cmd/claude-licence-manager/testdata/holdings-changed.json new file mode 100644 index 000000000..2a45f1859 --- /dev/null +++ b/modules/claude-licence-manager/cmd/claude-licence-manager/testdata/holdings-changed.json @@ -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"}