go: a module appends to and reads its log through the runtime, so its record outlives a bucket's history (hq ADR 0297)

2026-10-11 19:16:09 +02:00
parent 4cf45b5ca0
commit eaed934629
3 changed files with 478 additions and 1 deletions
@@ -3,7 +3,7 @@
 `stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime
 names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a
 function; calls run concurrently. `stdio.CallerOf(args)` inside a tool is who called it: the bus user the runtime took from the subject the call arrived on, named in the call's `_meta`, never read from the arguments — or `stdio.Unproven` when the runtime named none (novox/hq issue 365). A tool named `<seat>.<verb>` is the seat's implementation.
-`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (taken when the handler returns nil, offered again when it returns an error — the one rule is in the repository README, "Taking an event") and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists and watches the module's declared state — all through the runtime (novox/hq ADR 0193, 0198, 0201). Standard library only.
+`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (taken when the handler returns nil, offered again when it returns an error — the one rule is in the repository README, "Taking an event") and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists and watches the module's declared state, and `stdio.Log(name)` appends to, reads, and measures a log of operations the module declared — all through the runtime (novox/hq ADR 0193, 0198, 0201, 0297). Standard library only.

 ## Installing

@@ -0,0 +1,123 @@
+package stdio
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// Log — a module's record of operations on the bus, kept as long as the log is (novox/hq ADR 0297). Not the
+// current value: that is state, and a log's state is its fold. Each entry is one operation on one key, its
+// sequence in the log is its identity, and the order of the log is the order of the operations. A module
+// declares each log it owns in its manifest (`logs`) and names it here as it named it there; a log is its
+// owner's alone, and the runtime refuses, with the reason, one it did not issue.
+
+// LogEntry is one operation on one key, as the log holds it.
+type LogEntry struct {
+	Sequence uint64          `json:"sequence"`
+	Key      string          `json:"key"`
+	Value    json.RawMessage `json:"value"`
+	At       time.Time       `json:"at"`
+}
+
+// LogPage is one page of a read: its entries, in order, the sequence to read on from (0 when nothing
+// follows) and the log's last sequence when it was read.
+type LogPage struct {
+	Entries []LogEntry `json:"entries"`
+	Next    uint64     `json:"next"`
+	Last    uint64     `json:"last"`
+}
+
+// LogInfo is how much a log holds: its entries and bytes, its cap in bytes, and its first and last sequence.
+type LogInfo struct {
+	Entries  uint64 `json:"entries"`
+	Bytes    uint64 `json:"bytes"`
+	MaxBytes uint64 `json:"max_bytes"`
+	First    uint64 `json:"first"`
+	Last     uint64 `json:"last"`
+}
+
+// KeptLog is one log this module owns.
+type KeptLog struct{ name string }
+
+// Log is a log this module owns, by its local name.
+func Log(name string) KeptLog { return KeptLog{name: name} }
+
+const logTimeout = 30 * time.Second
+
+func (l KeptLog) ask(verb string, params map[string]any) (json.RawMessage, error) {
+	c, err := live()
+	if err != nil {
+		return nil, err
+	}
+	params["log"] = l.name
+	return c.ask("mesh/log."+verb, params, logTimeout)
+}
+
+func (l KeptLog) append(params map[string]any) (uint64, error) {
+	raw, err := l.ask("append", params)
+	if err != nil {
+		return 0, err
+	}
+	var appended struct {
+		Sequence uint64 `json:"sequence"`
+	}
+	return appended.Sequence, json.Unmarshal(raw, &appended)
+}
+
+// Append writes an entry for key, whatever came before it, and returns its sequence. A key is one or more
+// names of letters, digits, `-` and `_`, joined by `.`; a value is JSON of at most 256 KiB. A full log
+// refuses the entry: it never drops an older one to make room.
+func (l KeptLog) Append(key string, value any) (uint64, error) {
+	return l.append(map[string]any{"key": key, "value": value})
+}
+
+// AppendAt writes an entry for key only if the key's last entry is still at expect — the sequence the caller
+// read, 0 for a key with no entry yet — and returns its sequence. When another entry for the key came in
+// between, nothing is written and the error is ErrChanged, as for a lost State update: read again and
+// decide again.
+func (l KeptLog) AppendAt(key string, value any, expect uint64) (uint64, error) {
+	return l.append(map[string]any{"key": key, "value": value, "expect": expect})
+}
+
+// Read is the entries from the sequence from on, oldest first, at most limit of them (0 is the runtime's
+// page, 100; it gives at most 1000). keys narrows it: a key, with `*` for one name and `>` for the rest;
+// empty is every key. Read on from the page's Next until it is 0.
+func (l KeptLog) Read(from uint64, keys string, limit int) (*LogPage, error) {
+	if limit < 0 {
+		return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", l.name, limit)
+	}
+	params := map[string]any{"from": from}
+	if keys != "" {
+		params["keys"] = keys
+	}
+	if limit > 0 {
+		params["limit"] = limit
+	}
+	raw, err := l.ask("read", params)
+	if err != nil {
+		return nil, err
+	}
+	var p LogPage
+	return &p, json.Unmarshal(raw, &p)
+}
+
+// Last is a key's last entry, or nil when the key has none.
+func (l KeptLog) Last(key string) (*LogEntry, error) {
+	raw, err := l.ask("last", map[string]any{"key": key})
+	if err != nil || len(raw) == 0 || string(raw) == "null" {
+		return nil, err
+	}
+	var e LogEntry
+	return &e, json.Unmarshal(raw, &e)
+}
+
+// Info is how much the log holds and how much it may.
+func (l KeptLog) Info() (*LogInfo, error) {
+	raw, err := l.ask("info", map[string]any{})
+	if err != nil {
+		return nil, err
+	}
+	var i LogInfo
+	return &i, json.Unmarshal(raw, &i)
+}
@@ -0,0 +1,354 @@
+package stdio
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// logRuntime is the runtime's side of mesh/log.append, read, last and info as the bus does them (novox/hq
+// ADR 0297): one sequence per entry across the log, an append with expect refused unless the key's last
+// entry is still at it (0 for a key with none), that refusal answered with CodeChanged as a lost state
+// update is, and a read paged by its limit with the sequence to read on from. asked keeps every ask's
+// method and params, as the bundle sent them.
+type logRuntime struct {
+	mu      sync.Mutex
+	entries []LogEntry
+	asked   []message
+}
+
+func (r *logRuntime) last(key string) uint64 {
+	for i := len(r.entries) - 1; i >= 0; i-- {
+		if r.entries[i].Key == key {
+			return r.entries[i].Sequence
+		}
+	}
+	return 0
+}
+
+// matches is a key pattern as the runtime reads it: `*` one name, `>` the rest.
+func matches(pattern, key string) bool {
+	if pattern == "" {
+		return true
+	}
+	p, k := strings.Split(pattern, "."), strings.Split(key, ".")
+	for i, t := range p {
+		if t == ">" {
+			return len(k) > i
+		}
+		if i >= len(k) || (t != "*" && t != k[i]) {
+			return false
+		}
+	}
+	return len(p) == len(k)
+}
+
+func (r *logRuntime) answer(m message) map[string]any {
+	var p struct {
+		Log    string          `json:"log"`
+		Key    string          `json:"key"`
+		Value  json.RawMessage `json:"value"`
+		Expect *uint64         `json:"expect"`
+		From   uint64          `json:"from"`
+		Keys   string          `json:"keys"`
+		Limit  int             `json:"limit"`
+	}
+	_ = json.Unmarshal(m.Params, &p)
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.asked = append(r.asked, m)
+	if p.Log != "changes" {
+		return map[string]any{"error": map[string]any{"code": -32000, "message": "issues was issued no log " + p.Log}}
+	}
+	switch strings.TrimPrefix(m.Method, "mesh/log.") {
+	case "append":
+		if p.Expect != nil && r.last(p.Key) != *p.Expect {
+			return map[string]any{"error": map[string]any{"code": CodeChanged, "message": p.Key + ": the key was written by another writer first"}}
+		}
+		seq := uint64(len(r.entries)) + 1
+		r.entries = append(r.entries, LogEntry{Sequence: seq, Key: p.Key, Value: p.Value,
+			At: time.Date(2026, 10, 11, 18, 0, int(seq), 0, time.UTC)})
+		return map[string]any{"result": map[string]any{"sequence": seq}}
+	case "read":
+		limit := p.Limit
+		if limit == 0 {
+			limit = 100
+		}
+		page := map[string]any{"entries": []LogEntry{}, "next": 0, "last": len(r.entries)}
+		var got []LogEntry
+		for _, e := range r.entries {
+			if e.Sequence < p.From || !matches(p.Keys, e.Key) {
+				continue
+			}
+			if len(got) == limit {
+				page["next"] = e.Sequence
+				break
+			}
+			got = append(got, e)
+		}
+		if got != nil {
+			page["entries"] = got
+		}
+		return map[string]any{"result": page}
+	case "last":
+		seq := r.last(p.Key)
+		if seq == 0 {
+			return map[string]any{"result": nil}
+		}
+		return map[string]any{"result": r.entries[seq-1]}
+	case "info":
+		var bytes int
+		for _, e := range r.entries {
+			bytes += len(e.Value)
+		}
+		info := map[string]any{"entries": len(r.entries), "bytes": bytes, "max_bytes": 1 << 30, "first": 0, "last": len(r.entries)}
+		if len(r.entries) > 0 {
+			info["first"] = 1
+		}
+		return map[string]any{"result": info}
+	}
+	return map[string]any{"error": map[string]any{"code": -32601, "message": "no " + m.Method}}
+}
+
+// logging serves a bundle whose one tool, "log", runs run against r, answering each mesh/log ask beside
+// the loop as the runtime does. It returns what the tool answered.
+func logging(t *testing.T, r *logRuntime, run func() (any, error)) map[string]any {
+	t.Helper()
+	t.Setenv(ServedAs, "issues")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "log", Run: func(map[string]any) (any, error) { return run() }}}, inR, outW)
+	}()
+	var wmu sync.Mutex
+	send := func(m map[string]any) {
+		m["jsonrpc"] = "2.0"
+		b, _ := json.Marshal(m)
+		wmu.Lock()
+		defer wmu.Unlock()
+		_, _ = inW.Write(append(b, '\n'))
+	}
+	done := make(chan map[string]any, 1)
+	go func() {
+		lines := bufio.NewScanner(outR)
+		lines.Buffer(make([]byte, 1<<20), 1<<24)
+		for lines.Scan() {
+			var m message
+			if json.Unmarshal(lines.Bytes(), &m) != nil {
+				continue
+			}
+			if strings.HasPrefix(m.Method, "mesh/log.") {
+				go func(m message) {
+					a := r.answer(m)
+					a["id"] = m.ID
+					send(a)
+				}(m)
+				continue
+			}
+			var answer map[string]any
+			_ = json.Unmarshal(lines.Bytes(), &answer)
+			if answer["id"] == float64(1) {
+				done <- answer
+				return
+			}
+		}
+	}()
+	send(map[string]any{"id": 1, "method": "tools/call", "params": map[string]any{"name": "log"}})
+	t.Cleanup(func() { _ = inW.Close() })
+	select {
+	case a := <-done:
+		return a
+	case <-time.After(10 * time.Second):
+		t.Fatal("the bundle did not answer")
+		return nil
+	}
+}
+
+// params is what the bundle sent with its n-th ask, and the ask's method.
+func (r *logRuntime) params(t *testing.T, n int) (string, map[string]any) {
+	t.Helper()
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if n >= len(r.asked) {
+		t.Fatalf("the bundle asked %d times, not %d", len(r.asked), n+1)
+	}
+	var p map[string]any
+	_ = json.Unmarshal(r.asked[n].Params, &p)
+	return r.asked[n].Method, p
+}
+
+// A bundle appends to, reads, and measures its log through the runtime (novox/hq ADR 0297): each verb is
+// asked as mesh/log.<verb> naming the log, its answer is handed back whole, a key with no entry is nil,
+// and a read pages on from Next until it is 0.
+func TestABundleKeepsALogThroughTheRuntime(t *testing.T) {
+	r := &logRuntime{}
+	var (
+		first, second, third uint64
+		pages                []*LogPage
+		last, none           *LogEntry
+		info                 *LogInfo
+		narrowed             *LogPage
+	)
+	got := logging(t, r, func() (any, error) {
+		changes := Log("changes")
+		var err error
+		if first, err = changes.Append("401", map[string]any{"op": "opened", "by": "fix@g14"}); err != nil {
+			return nil, err
+		}
+		if second, err = changes.Append("402", map[string]any{"op": "opened"}); err != nil {
+			return nil, err
+		}
+		if third, err = changes.AppendAt("401", map[string]any{"op": "moved", "to": "fixing"}, first); err != nil {
+			return nil, err
+		}
+		for _, k := range []string{"403", "404", "405"} {
+			if _, err = changes.Append(k, map[string]any{"op": "opened"}); err != nil {
+				return nil, err
+			}
+		}
+		for from := uint64(1); ; {
+			p, err := changes.Read(from, "", 2)
+			if err != nil {
+				return nil, err
+			}
+			pages = append(pages, p)
+			if p.Next == 0 {
+				break
+			}
+			from = p.Next
+		}
+		if narrowed, err = changes.Read(0, "401", 0); err != nil {
+			return nil, err
+		}
+		if last, err = changes.Last("401"); err != nil {
+			return nil, err
+		}
+		if none, err = changes.Last("999"); err != nil {
+			return nil, err
+		}
+		info, err = changes.Info()
+		return "done", err
+	})
+	if out, isError := text(t, got); isError || out != `"done"` {
+		t.Fatalf("the tool answered %s", out)
+	}
+	if first != 1 || second != 2 || third != 3 {
+		t.Fatalf("appended at %d, %d, %d", first, second, third)
+	}
+
+	// Asked as the contract names them: the verb, the log, and only what was given.
+	method, p := r.params(t, 0)
+	if method != "mesh/log.append" || p["log"] != "changes" || p["key"] != "401" || p["value"].(map[string]any)["by"] != "fix@g14" {
+		t.Fatalf("an append asked %s %v", method, p)
+	}
+	if _, has := p["expect"]; has {
+		t.Fatalf("an Append without expect named one: %v", p)
+	}
+	if method, p = r.params(t, 2); method != "mesh/log.append" || p["expect"] != float64(1) {
+		t.Fatalf("an AppendAt asked %s %v", method, p)
+	}
+	if method, p = r.params(t, 6); method != "mesh/log.read" || p["from"] != float64(1) || p["limit"] != float64(2) || p["keys"] != nil {
+		t.Fatalf("a read asked %s %v", method, p)
+	}
+
+	// Paged two at a time through six entries: three pages, every entry once, in order.
+	if len(pages) != 3 {
+		t.Fatalf("read %d pages, not 3", len(pages))
+	}
+	var seqs []uint64
+	for _, pg := range pages {
+		if pg.Last != 6 {
+			t.Fatalf("a page said the last sequence is %d", pg.Last)
+		}
+		for _, e := range pg.Entries {
+			seqs = append(seqs, e.Sequence)
+		}
+	}
+	if fmt.Sprint(seqs) != "[1 2 3 4 5 6]" || pages[0].Next != 3 || pages[2].Next != 0 {
+		t.Fatalf("paged %v, next %d then %d", seqs, pages[0].Next, pages[2].Next)
+	}
+	if e := pages[1].Entries[0]; e.Key != "401" || string(e.Value) != `{"op":"moved","to":"fixing"}` ||
+		!e.At.Equal(time.Date(2026, 10, 11, 18, 0, 3, 0, time.UTC)) {
+		t.Fatalf("an entry read as %+v", e)
+	}
+
+	// Narrowed to a key, with the runtime's page: no limit sent.
+	method, p = r.params(t, 9)
+	if method != "mesh/log.read" || p["keys"] != "401" || p["limit"] != nil {
+		t.Fatalf("a narrowed read asked %s %v", method, p)
+	}
+	if len(narrowed.Entries) != 2 || narrowed.Entries[0].Sequence != 1 || narrowed.Entries[1].Sequence != 3 || narrowed.Next != 0 {
+		t.Fatalf("narrowed to 401: %+v", narrowed)
+	}
+
+	if last == nil || last.Sequence != 3 || last.Key != "401" {
+		t.Fatalf("the last of 401 is %+v", last)
+	}
+	if method, p = r.params(t, 10); method != "mesh/log.last" || p["key"] != "401" {
+		t.Fatalf("a last asked %s %v", method, p)
+	}
+	if none != nil {
+		t.Fatalf("a key with no entry has a last: %+v", none)
+	}
+	if info == nil || info.Entries != 6 || info.MaxBytes != 1<<30 || info.First != 1 || info.Last != 6 || info.Bytes == 0 {
+		t.Fatalf("the log's info is %+v", info)
+	}
+	if method, p = r.params(t, 12); method != "mesh/log.info" || p["log"] != "changes" || len(p) != 1 {
+		t.Fatalf("an info asked %s %v", method, p)
+	}
+}
+
+// An AppendAt on a stale expect is ErrChanged and writes nothing: another entry for the key came in
+// between. Expect 0 is a key with no entry yet, and is ErrChanged once the key has one. A log not issued
+// is refused with the runtime's reason, and is not ErrChanged.
+func TestAnAppendAtAStaleSequenceIsErrChangedAndWritesNothing(t *testing.T) {
+	r := &logRuntime{}
+	var errs []error
+	got := logging(t, r, func() (any, error) {
+		changes := Log("changes")
+		read, err := changes.AppendAt("401", map[string]any{"op": "opened"}, 0)
+		if err != nil {
+			return nil, err
+		}
+		// Another writer is first.
+		if _, err := changes.Append("401", map[string]any{"op": "noted"}); err != nil {
+			return nil, err
+		}
+		_, stale := changes.AppendAt("401", map[string]any{"op": "moved"}, read)
+		_, again := changes.AppendAt("401", map[string]any{"op": "opened"}, 0)
+		_, other := Log("elsewhere").Append("401", map[string]any{})
+		errs = []error{stale, again, other}
+		return "done", nil
+	})
+	if out, isError := text(t, got); isError || out != `"done"` {
+		t.Fatalf("the tool answered %s", out)
+	}
+	if !errors.Is(errs[0], ErrChanged) || !errors.Is(errs[1], ErrChanged) {
+		t.Fatalf("a stale expect answered %v and %v, not ErrChanged", errs[0], errs[1])
+	}
+	if errs[2] == nil || errors.Is(errs[2], ErrChanged) || !strings.Contains(errs[2].Error(), "no log elsewhere") {
+		t.Fatalf("a log not issued answered %v", errs[2])
+	}
+	if len(r.entries) != 2 {
+		t.Fatalf("the log holds %d entries, not 2: a refused append wrote", len(r.entries))
+	}
+}
+
+// A read asking for fewer than no entries is refused before the runtime is asked.
+func TestAReadOfANegativeLimitIsRefused(t *testing.T) {
+	r := &logRuntime{}
+	var err error
+	logging(t, r, func() (any, error) {
+		_, err = Log("changes").Read(0, "", -1)
+		return "done", nil
+	})
+	if err == nil || len(r.asked) != 0 {
+		t.Fatalf("a negative limit answered %v after %d asks", err, len(r.asked))
+	}
+}