diff --git a/go/README.md b/go/README.md index 680c496..412efb7 100644 --- a/go/README.md +++ b/go/README.md @@ -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 `.` 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 diff --git a/go/log.go b/go/log.go new file mode 100644 index 0000000..7b1030d --- /dev/null +++ b/go/log.go @@ -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) +} diff --git a/go/log_test.go b/go/log_test.go new file mode 100644 index 0000000..b650925 --- /dev/null +++ b/go/log_test.go @@ -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. 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)) + } +}