From 2c4931751a3fe1daf9c69d60c1ef27b6a51d99ec Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:40:57 +0100 Subject: [PATCH 1/2] A-176: a helper writes a durable consumer's facts into the module's state consumers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit process.ConsumerLedger writes stream, name, acknowledged sequence, pending, last event applied with its at and the delivery of the oldest unacknowledged message into one key per consumer, after each message delivered and applied, so a tool reads them. Builds: design 47 G §579, A-166, A-176 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/consumers.go | 122 +++++++++++++++++++++++++++++++++++ go/process/consumers_test.go | 52 +++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 go/process/consumers.go create mode 100644 go/process/consumers_test.go diff --git a/go/process/consumers.go b/go/process/consumers.go new file mode 100644 index 00000000..80ffb536 --- /dev/null +++ b/go/process/consumers.go @@ -0,0 +1,122 @@ +package process + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// ConsumersState is the name of the state a module keeps its durable consumers' facts in, one key per consumer by its name, +// history 1 (design 47 G A-166, A-176): what a module's tool reads of a consumer, since a tool runs in a runtime of its own and +// cannot read the consumer's information. +const ConsumersState = "consumers" + +// ConsumerFacts is one durable consumer as ConsumerLedger keeps it. +type ConsumerFacts struct { + // Stream and Name are the consumer's stream and its name. + Stream string `json:"stream"` + Name string `json:"name"` + // Acknowledged is the stream sequence of the last message applied. + Acknowledged uint64 `json:"acknowledged"` + // Pending is the messages the consumer has not been given yet, as the last message said. + Pending uint64 `json:"pending"` + // LastEvent is the subject of the last event applied and LastAt its own `at`, or when the bus stored it where it has none. + LastEvent string `json:"last-event,omitempty"` + LastAt time.Time `json:"last-at,omitempty"` + // OldestUnacknowledged is when the oldest message delivered and not yet applied was delivered; zero when none is. + OldestUnacknowledged time.Time `json:"oldest-unacknowledged,omitempty"` +} + +// ConsumerSink stores one key of the module's state consumers. +type ConsumerSink func(ctx context.Context, key string, value []byte) error + +// ConsumerLedger writes a durable consumer's facts into the module's state consumers after each message delivered and each message +// applied, so a tool reads them and nothing is read on a timer (design 47 G A-166, A-176). One ledger serves one consumer. +type ConsumerLedger struct { + stream, name string + sink ConsumerSink + now func() time.Time + logf func(string, ...any) + + mu sync.Mutex + facts ConsumerFacts + unacks map[uint64]time.Time // delivered and not applied, by stream sequence +} + +// NewConsumerLedger is the ledger of the consumer name on stream, writing through sink; a nil now is the clock and a nil logf says +// nothing. +func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.Time, logf func(string, ...any)) *ConsumerLedger { + if now == nil { + now = time.Now + } + if logf == nil { + logf = func(string, ...any) {} + } + return &ConsumerLedger{stream: stream, name: name, sink: sink, now: now, logf: logf, unacks: map[uint64]time.Time{}, + facts: ConsumerFacts{Stream: stream, Name: name}} +} + +// Delivered notes that m was given to the consumer, and writes the facts. +func (l *ConsumerLedger) Delivered(m jetstream.Msg) { + md, err := m.Metadata() + if err != nil { + l.logf("consumers: %s's message has no metadata, its delivery is not kept: %v", l.name, err) + return + } + l.mu.Lock() + l.unacks[md.Sequence.Stream] = l.now().UTC() + l.facts.Pending = md.NumPending + l.mu.Unlock() + l.write() +} + +// Applied notes that m was applied and acknowledged, at the event's own `at` (zero: when the bus stored it), and writes the facts. +func (l *ConsumerLedger) Applied(m jetstream.Msg, at time.Time) { + md, err := m.Metadata() + if err != nil { + l.logf("consumers: %s's message has no metadata, its application is not kept: %v", l.name, err) + return + } + if at.IsZero() { + at = md.Timestamp + } + l.mu.Lock() + delete(l.unacks, md.Sequence.Stream) + if md.Sequence.Stream > l.facts.Acknowledged { + l.facts.Acknowledged = md.Sequence.Stream + } + l.facts.Pending = md.NumPending + l.facts.LastEvent, l.facts.LastAt = m.Subject(), at.UTC() + l.mu.Unlock() + l.write() +} + +// Facts are the facts now. +func (l *ConsumerLedger) Facts() ConsumerFacts { + l.mu.Lock() + defer l.mu.Unlock() + f := l.facts + f.OldestUnacknowledged = time.Time{} + for _, t := range l.unacks { + if f.OldestUnacknowledged.IsZero() || t.Before(f.OldestUnacknowledged) { + f.OldestUnacknowledged = t + } + } + return f +} + +func (l *ConsumerLedger) write() { + if l.sink == nil { + return + } + data, err := json.Marshal(l.Facts()) + if err == nil { + err = l.sink(context.Background(), l.name, data) + } + if err != nil { + l.logf("consumers: the facts of %s are not written to the state %s: %v", l.name, ConsumersState, err) + } +} diff --git a/go/process/consumers_test.go b/go/process/consumers_test.go new file mode 100644 index 00000000..5f5fb9ce --- /dev/null +++ b/go/process/consumers_test.go @@ -0,0 +1,52 @@ +package process + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +type ledgerMsg struct { + jetstream.Msg + subject string + seq uint64 + pending uint64 + stored time.Time +} + +func (m ledgerMsg) Subject() string { return m.subject } +func (m ledgerMsg) Metadata() (*jetstream.MsgMetadata, error) { + return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq, Consumer: m.seq}, NumPending: m.pending, Timestamp: m.stored}, nil +} + +func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) { + kept := map[string][]byte{} + now := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + l := NewConsumerLedger("EVENTS", "n1_m", func(_ context.Context, key string, value []byte) error { kept[key] = value; return nil }, func() time.Time { return now }, nil) + read := func() ConsumerFacts { + var f ConsumerFacts + if err := json.Unmarshal(kept["n1_m"], &f); err != nil { + t.Fatal(err) + } + return f + } + a, b := ledgerMsg{subject: "mesh.seat.s.event.a", seq: 7, pending: 1}, ledgerMsg{subject: "mesh.seat.s.event.b", seq: 8, pending: 0} + l.Delivered(a) + now = now.Add(time.Minute) + l.Delivered(b) + if f := read(); f.Pending != 0 || !f.OldestUnacknowledged.Equal(now.Add(-time.Minute)) { + t.Errorf("after two deliveries: %+v, want pending 0 and the first one's delivery as the oldest unacknowledged", f) + } + at := now.Add(-time.Hour) + l.Applied(a, at) + if f := read(); f.Acknowledged != 7 || f.LastEvent != "mesh.seat.s.event.a" || !f.LastAt.Equal(at) || !f.OldestUnacknowledged.Equal(now) { + t.Errorf("after the first applied: %+v, want sequence 7, its event and at, and the second one's delivery as the oldest", f) + } + l.Applied(b, time.Time{}) + if f := read(); f.Acknowledged != 8 || !f.OldestUnacknowledged.IsZero() { + t.Errorf("after both applied: %+v, want sequence 8 and none unacknowledged", f) + } +} From 7af6fdedf5dbd0f55638ea0c90fbed5c7cddbbc0 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 03:08:25 +0100 Subject: [PATCH 2/2] RR2-4: the manifest reader reads the long form of an invokes entry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tool runner's invokes hold * beside {verb, fields} entries; Invokes keeps every verb, so none of its readers changes, and InvokeFields maps a verb to the fields it names, written back in the long form. Builds: design 47 G §621, §617 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/catalogue/manifest.go | 84 ++++++++++++++++++++++++++++- go/catalogue/reader_release_test.go | 39 +++++++++++++- 2 files changed, 121 insertions(+), 2 deletions(-) diff --git a/go/catalogue/manifest.go b/go/catalogue/manifest.go index 38c08f1f..d09ca9a5 100644 --- a/go/catalogue/manifest.go +++ b/go/catalogue/manifest.go @@ -587,6 +587,11 @@ type Manifest struct { // person's account (design 25 §7) already had the same shape. Invokes []string `json:"invokes,omitempty"` + // InvokeFields maps a verb of Invokes to the fields it names, read from an entry of `invokes` in its long form + // {verb, fields}, which may stand beside `*` (design 47 G §617): Invokes keeps the verb, so none of its readers + // changes. A long-form entry naming no field is a verb of Invokes alone. + InvokeFields map[string][]string `json:"-"` + // Waits are the event families a session may wait on through the mesh MCP server's `mesh_wait`, each // `.` as `consumes` names one, a seat's or a module's (novox/hq ADR 0367 §2). Only the tool // runner's module declares them, one by one and never by a wildcard. **A grant apart from `consumes`**: the @@ -1578,6 +1583,10 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error { if err := consumesEntryProblem(keys["consumes"]); err != nil { return err } + invokeFields, err := readInvokes(keys) + if err != nil { + return err + } plain := map[string]string{} many := map[string]map[string]string{} if secrets, ok := keys["secrets"]; ok && string(secrets) != "null" { @@ -1732,6 +1741,7 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error { m.forms = forms m.OptionalRequires = optionalRequires m.RequireShares = requireShares + m.InvokeFields = invokeFields if len(plain) > 0 { m.Secrets = plain } @@ -1750,6 +1760,61 @@ func (m *Manifest) UnmarshalJSON(raw []byte) error { return nil } +// invokeLongForm is an entry of `invokes` in its long form: the verb, and the fields of its answer the caller relies on +// (design 47 G §617). +type invokeLongForm struct { + Verb string `json:"verb"` + Fields []string `json:"fields,omitempty"` +} + +// readInvokes reads `invokes` whose entries are verbs or, in their long form, {verb, fields}: it puts the verbs back in +// keys as plain names and answers the fields each long-form entry names. +func readInvokes(keys map[string]json.RawMessage) (map[string][]string, error) { + raw, ok := keys["invokes"] + if !ok || string(raw) == "null" { + return nil, nil + } + var entries []json.RawMessage + if err := json.Unmarshal(raw, &entries); err != nil { + return nil, nil // not a list: the field's own decoder says so + } + var verbs []string + var fields map[string][]string + for i, e := range entries { + trimmed := bytes.TrimSpace(e) + if len(trimmed) == 0 || trimmed[0] != '{' { + var verb string + if err := json.Unmarshal(trimmed, &verb); err != nil { + return nil, fmt.Errorf("invokes[%d]: a verb, or {verb, fields}: %w", i, err) + } + verbs = append(verbs, verb) + continue + } + var long invokeLongForm + dec := json.NewDecoder(bytes.NewReader(trimmed)) + dec.DisallowUnknownFields() + if err := dec.Decode(&long); err != nil { + return nil, fmt.Errorf("invokes[%d]: a verb, or {verb, fields}: %w", i, err) + } + if long.Verb == "" { + return nil, fmt.Errorf("invokes[%d]: a long-form entry names its verb", i) + } + verbs = append(verbs, long.Verb) + if len(long.Fields) > 0 { + if fields == nil { + fields = map[string][]string{} + } + fields[long.Verb] = long.Fields + } + } + plain, err := json.Marshal(verbs) + if err != nil { + return nil, err + } + keys["invokes"] = plain + return fields, nil +} + // checkSharesNotContributed refuses a module whose own `contributes` to s3-bucket names `shares` (ADR 0468 §8). func (m Manifest) checkSharesNotContributed() error { if len(m.RequireShares[BucketRequirement]) == 0 { @@ -1803,13 +1868,30 @@ func (m Manifest) marshalFields() ([]byte, error) { if err != nil { return nil, err } - if len(m.SecretsMany) == 0 && len(m.ContributesMany) == 0 && len(m.OptionalRequires) == 0 && len(m.RequireShares) == 0 { + if len(m.SecretsMany) == 0 && len(m.ContributesMany) == 0 && len(m.OptionalRequires) == 0 && len(m.RequireShares) == 0 && + len(m.InvokeFields) == 0 { return raw, nil } var keys map[string]json.RawMessage if err := json.Unmarshal(raw, &keys); err != nil { return nil, err } + if len(m.InvokeFields) > 0 { + // Each verb that names fields is written back in its long form (design 47 G §617). + entries := make([]any, 0, len(m.Invokes)) + for _, verb := range m.Invokes { + if fields, named := m.InvokeFields[verb]; named { + entries = append(entries, invokeLongForm{Verb: verb, Fields: fields}) + continue + } + entries = append(entries, verb) + } + listed, err := json.Marshal(entries) + if err != nil { + return nil, err + } + keys["invokes"] = listed + } if optional := optionalRequiresList(m); len(optional) > 0 { // Beside `requires`, never inside it: an older controller reads `requires` as plain names, and one // manifest it cannot read stops the whole catalogue (OptionalRequiresKey). diff --git a/go/catalogue/reader_release_test.go b/go/catalogue/reader_release_test.go index fccd9ba5..87c4f35b 100644 --- a/go/catalogue/reader_release_test.go +++ b/go/catalogue/reader_release_test.go @@ -1,6 +1,11 @@ package catalogue -import "testing" +import ( + "encoding/json" + "slices" + "strings" + "testing" +) // A manifest carrying the words design 47 G writes is read whole by the reader release (design 47 G §340, §409): // `version` as MAJOR.MINOR (§368), a `*` emitter in `consumes` (§386), an observer's finding stream (§387) and an @@ -23,3 +28,35 @@ func TestTheReaderReleaseReadsTheManifestWordsGWrites(t *testing.T) { t.Fatalf("read as %+v", m) } } + +// The second reader release reads the words G's manifests write beyond the first's (design 47 G §617): each row a +// manifest read whole, or refused naming what is wrong, and what the reader gives of it. +func TestTheSecondReaderReleaseReadsGsManifestWords(t *testing.T) { + for _, c := range []struct { + name, raw, refused string + check func(Manifest) bool + }{ + {name: "* with two long-form invokes (§617)", + raw: `{"module":"m","version":"1.0","invokes":["*",{"verb":"seat:mesh-catalog.revision","fields":["none-yet"]},` + + `{"verb":"seat:mesh-catalog.entry"}]}`, + check: func(m Manifest) bool { + out, _ := json.Marshal(m) + return slices.Equal(m.Invokes, []string{"*", "seat:mesh-catalog.revision", "seat:mesh-catalog.entry"}) && + len(m.InvokeFields) == 1 && slices.Equal(m.InvokeFields["seat:mesh-catalog.revision"], []string{"none-yet"}) && + strings.Contains(string(out), `"invokes":["*",{"verb":"seat:mesh-catalog.revision","fields":["none-yet"]},"seat:mesh-catalog.entry"]`) + }}, + {name: "a long-form invoke with a word it does not know (§617)", + raw: `{"module":"m","version":"1.0","invokes":[{"verb":"seat:mesh-catalog.revision","field":["none-yet"]}]}`, + refused: `invokes[0]: a verb, or {verb, fields}`}, + } { + m, err := decodeForTest([]byte(c.raw)) + switch { + case c.refused != "" && (err == nil || !strings.Contains(err.Error(), c.refused)): + t.Errorf("%s: read as %v, want refused naming %q", c.name, err, c.refused) + case c.refused == "" && err != nil: + t.Errorf("%s: refused: %v", c.name, err) + case c.refused == "" && (len(m.UnknownField()) != 0 || !c.check(m)): + t.Errorf("%s: read as %+v (set aside %v)", c.name, m, m.UnknownField()) + } + } +}