From e0d7644bb62555b2b11e325cf2c74657a8ad270c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 11:47:31 +0100 Subject: [PATCH] CAT-K2: a Store on the ledger, the records' heads without text and each section's text once ledgerStore implements Store on the tool runner's ledger calls: a record is the instance decision/0290 or research/081, a Save is the operation named by what changed, a section's text is written once in its write-section and named by digest and sequence in the head, a version is read back from the history, the whole-set findings and the told marks are keys of the state findings. Every test that runs over each store now also runs over the ledger, played with the manifest's own types. The bundle's pipe to the tool runner needs the SDK's raw request (request to A1); until then it answers loudly. Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-knowledge/kit_test.go | 4 +- .../cmd/mesh-knowledge/ledger_fake_test.go | 257 +++++++++ .../cmd/mesh-knowledge/ledgerstore.go | 519 ++++++++++++++++++ .../cmd/mesh-knowledge/ledgerstore_test.go | 162 ++++++ .../mesh-knowledge/cmd/mesh-knowledge/pipe.go | 82 +++ .../cmd/mesh-knowledge/store.go | 44 +- 6 files changed, 1057 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go create mode 100644 modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go create mode 100644 modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go create mode 100644 modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go index ca15ee58..eed0e18c 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go @@ -146,9 +146,11 @@ func pgStore(t *testing.T) *PgStore { return s } -// eachStore runs a test over the store in memory and over postgres. +// eachStore runs a test over the store in memory, over the ledger (played here, with the manifest's own types) and over +// postgres. func eachStore(t *testing.T, run func(t *testing.T, s Store)) { t.Run("memory", func(t *testing.T) { run(t, newMemStore()) }) + t.Run("ledger", func(t *testing.T) { run(t, newLedgerStore(newFakeLedger(t), newFakeStates())) }) t.Run("postgres", func(t *testing.T) { run(t, pgStore(t)) }) } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go new file mode 100644 index 00000000..88c275b6 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_fake_test.go @@ -0,0 +1,257 @@ +package main + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// fakeLedger plays the tool runner's side of the ledger for the module's own types (read from its manifest), as design +// 25 §12.4 says: apply checks an operation against the type, appends it with the sequence expected and derives the +// state by the operation's effect. It is a pipe, so a test of the store runs the same rules the manifest declares. +type fakeLedger struct { + t *testing.T + types map[string]catalogue.LedgerType + mu sync.Mutex + hist map[string][]buscalls.Operation + state map[string]map[string]any + // calls are the methods called, in order; applied are the call ids each apply named. + calls []string + applied []buscalls.ApplyRequest + // failApply, when set, refuses the apply of that operation. + failApply string + account string +} + +func newFakeLedger(t *testing.T) *fakeLedger { + t.Helper() + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + m, err := catalogue.Decode(raw) + if err != nil { + t.Fatal(err) + } + f := &fakeLedger{t: t, types: map[string]catalogue.LedgerType{}, hist: map[string][]buscalls.Operation{}, state: map[string]map[string]any{}, account: "mesh-knowledge"} + for _, e := range m.BusState { + if e.Name == LedgerName { + for _, ty := range e.Types { + f.types[ty.Name] = ty + } + } + } + if len(f.types) != 2 { + t.Fatalf("the manifest's ledger has types %v", f.types) + } + return f +} + +func (f *fakeLedger) Call(method string, req, answer any) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, method) + raw, _ := json.Marshal(req) + var out any + switch method { + case buscalls.LedgerApply: + var r buscalls.ApplyRequest + _ = json.Unmarshal(raw, &r) + a, err := f.apply(r) + if err != nil { + return err + } + out = a + case buscalls.LedgerGet: + var r buscalls.InstanceRequest + _ = json.Unmarshal(raw, &r) + st := f.state[r.Instance] + is := buscalls.InstanceState{Instance: r.Instance, State: json.RawMessage("null")} + if st != nil { + b, _ := json.Marshal(st) + is.State, is.Sequence = b, uint64(len(f.hist[r.Instance])) + } + out = is + case buscalls.LedgerHistory: + var r buscalls.HistoryRequest + _ = json.Unmarshal(raw, &r) + ops := f.hist[r.Instance] + page := buscalls.HistoryPage{Last: uint64(len(ops))} + for _, o := range ops { + if o.Sequence < r.From { + continue + } + if r.Limit > 0 && len(page.Operations) == r.Limit { + page.Next = o.Sequence + break + } + page.Operations = append(page.Operations, o) + } + out = page + default: + return fmt.Errorf("the fake ledger answers no %s", method) + } + b, _ := json.Marshal(out) + return json.Unmarshal(b, answer) +} + +func (f *fakeLedger) apply(r buscalls.ApplyRequest) (buscalls.Applied, error) { + if r.Ledger != LedgerName { + return buscalls.Applied{}, fmt.Errorf("no ledger %q", r.Ledger) + } + if f.failApply == r.Operation { + return buscalls.Applied{}, fmt.Errorf("refused: %s", r.Operation) + } + typeName, _, _ := strings.Cut(r.Instance, "/") + ty, ok := f.types[typeName] + if !ok { + return buscalls.Applied{}, fmt.Errorf("no type %q", typeName) + } + var op *catalogue.LedgerOperation + for i := range ty.Operations { + if ty.Operations[i].Name == r.Operation { + op = &ty.Operations[i] + } + } + if op == nil { + return buscalls.Applied{}, fmt.Errorf("the type %s allows no operation %q", typeName, r.Operation) + } + body := map[string]any{} + if len(r.Body) > 0 { + if err := json.Unmarshal(r.Body, &body); err != nil { + return buscalls.Applied{}, err + } + } + for name, field := range ty.Schema { + v, has := body[name] + if !has { + continue + } + okType := map[string]func(any) bool{ + "string": func(v any) bool { _, ok := v.(string); return ok }, "integer": func(v any) bool { _, ok := v.(float64); return ok }, + "object": func(v any) bool { _, ok := v.(map[string]any); return ok }, + }[field.Type] + if okType != nil && !okType(v) { + return buscalls.Applied{}, fmt.Errorf("the field %s is not a %s", name, field.Type) + } + if field.DigestOf != "" { + if txt, _ := body[field.DigestOf].(string); v != SectionDigest(txt) { + return buscalls.Applied{}, fmt.Errorf("the %s is not the digest of its %s", name, field.DigestOf) + } + } + } + seq := uint64(len(f.hist[r.Instance]) + 1) + author := buscalls.Author{Account: f.account} + if r.Call != "" { + author.Caller = "caller-of-" + r.Call + } + f.hist[r.Instance] = append(f.hist[r.Instance], buscalls.Operation{Sequence: seq, Instance: r.Instance, Operation: r.Operation, Body: r.Body, Author: author, At: time.Unix(int64(seq), 0).UTC()}) + f.applied = append(f.applied, r) + st := f.state[r.Instance] + if st == nil { + st = map[string]any{} + } + copied := map[string]any{} + for _, name := range op.Fields { + if v, ok := body[name]; ok { + copied[name] = v + } + } + if op.Sequence { + copied["sequence"] = seq + } + if op.Author { + copied["author"] = author + } + switch op.Effect { + case catalogue.EffectSetFields: + for k, v := range copied { + st[k] = v + } + case catalogue.EffectSetEntry: + m, _ := st[op.Into].(map[string]any) + if m == nil { + m = map[string]any{} + } + m[body[op.Key].(string)] = copied + st[op.Into] = m + default: + f.t.Fatalf("the fake ledger plays no effect %s", op.Effect) + } + f.state[r.Instance] = st + return buscalls.Applied{Taken: true, Sequence: seq}, nil +} + +// fakeStates plays the module's states. +type fakeStates struct { + mu sync.Mutex + kv map[string]map[string][]byte +} + +func newFakeStates() *fakeStates { return &fakeStates{kv: map[string]map[string][]byte{}} } + +func (f *fakeStates) State(name string) KV { return fakeKV{f, name} } + +type fakeKV struct { + f *fakeStates + name string +} + +func (k fakeKV) bucket() map[string][]byte { + if k.f.kv[k.name] == nil { + k.f.kv[k.name] = map[string][]byte{} + } + return k.f.kv[k.name] +} + +func (k fakeKV) Get(key string) (*stdio.StateEntry, error) { + k.f.mu.Lock() + defer k.f.mu.Unlock() + v, ok := k.bucket()[key] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: key, Value: v, Revision: 1}, nil +} +func (k fakeKV) Put(key string, value any) (uint64, error) { + k.f.mu.Lock() + defer k.f.mu.Unlock() + b, _ := json.Marshal(value) + k.bucket()[key] = b + return 1, nil +} +func (k fakeKV) Create(key string, value any) (uint64, error) { + k.f.mu.Lock() + defer k.f.mu.Unlock() + if _, ok := k.bucket()[key]; ok { + return 0, stdio.ErrChanged + } + b, _ := json.Marshal(value) + k.bucket()[key] = b + return 1, nil +} +func (k fakeKV) Update(key string, value any, _ uint64) (uint64, error) { return k.Put(key, value) } +func (k fakeKV) Delete(key string) error { + k.f.mu.Lock() + defer k.f.mu.Unlock() + delete(k.bucket(), key) + return nil +} +func (k fakeKV) Keys() ([]string, error) { + k.f.mu.Lock() + defer k.f.mu.Unlock() + var out []string + for key := range k.bucket() { + out = append(out, key) + } + return out, nil +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go new file mode 100644 index 00000000..a145dd20 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go @@ -0,0 +1,519 @@ +package main + +// The records kept in the ledger (novox/hq ADR 0597 §11, to-be 55 §2.1): the Store the rules of the knowledge base are +// written against, on the tool runner's ledger calls. A record is one instance, `decision/0290` or `research/081`; every +// change to it is one operation appended to its history; its head, the entry, is the state apply derives from the +// operations and is never written by the module; a section's text is in the `write-section` operation that wrote it and +// nowhere else, the head naming it by its title, its digest and that operation's sequence. + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// MostSectionBytes is the most a section holds (to-be 55 §4), within a history entry's 256 KiB. +const MostSectionBytes = 40 << 10 + +// sectionRef is how a head names a section: its title, the digest of its text and the sequence of the write-section that +// wrote it. +type sectionRef struct { + Name string `json:"name"` + Digest string `json:"digest"` + Sequence uint64 `json:"sequence"` +} + +// storedRecord is a record as the head and a version keep it: the record whole but for its sections' text. +type storedRecord struct { + Record + Sections []sectionRef `json:"sections,omitempty"` +} + +// entryHead is the head of an instance: the version served, the one proposed over it, and how many versions there are. +type entryHead struct { + Current *storedRecord `json:"current"` + CurrentVersion int `json:"current-version"` + Pending *storedRecord `json:"pending,omitempty"` + PendingVersion int `json:"pending-version,omitempty"` + Versions int `json:"versions"` +} + +// versionHead is a version as its operation keeps it. +type versionHead struct { + N int `json:"n"` + By string `json:"by"` + At time.Time `json:"at"` + Why string `json:"why"` + Insight bool `json:"insight,omitempty"` + Record storedRecord `json:"record"` +} + +// headState is the state of an instance, as apply derives it from the operations. +type headState struct { + Kind string `json:"kind"` + Number int `json:"number"` + Entry *entryHead `json:"entry"` + Sequence uint64 `json:"sequence"` +} + +// SectionDigest is the digest a section's text is kept under; the ledger refuses a write-section whose digest is another. +func SectionDigest(text string) string { + sum := sha256.Sum256([]byte(text)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// InstanceOf is the instance a record is: `decision/0290`, `research/081`. +func InstanceOf(k Key) string { + if k.Kind == KindResearch { + return fmt.Sprintf("research/%03d", k.Number) + } + return fmt.Sprintf("decision/%04d", k.Number) +} + +// ledgerStore is the Store on the ledger. +type ledgerStore struct { + pipe Pipe + states States + + mu sync.Mutex + // heads are the instances read, by key, each with the sequence it was read at; texts are the sections' text by digest. + heads map[Key]*headState + texts map[string]string + scanned bool +} + +// newLedgerStore is the store over a pipe and the module's states. +func newLedgerStore(p Pipe, s States) *ledgerStore { + return &ledgerStore{pipe: p, states: s, heads: map[Key]*headState{}, texts: map[string]string{}} +} + +// stripped is r as a stored record naming its sections by ref: the refs of the sections already written are kept, the +// new ones are filled by the caller. +func stripped(r *Record, refs []sectionRef) storedRecord { + c := r.Clone() + c.Sections = nil + return storedRecord{Record: *c, Sections: refs} +} + +// read is one instance's head from the ledger, or nil when it has none. +func (s *ledgerStore) read(k Key) (*headState, error) { + var got buscalls.InstanceState + if err := s.pipe.Call(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: InstanceOf(k)}, &got); err != nil { + return nil, err + } + if len(got.State) == 0 || string(got.State) == "null" { + return nil, nil + } + var h headState + if err := json.Unmarshal(got.State, &h); err != nil { + return nil, fmt.Errorf("the head of %s is not a record's: %w", k, err) + } + h.Sequence = got.Sequence + return &h, nil +} + +// fresh is the head, read again unless this process wrote it or read it and nothing else writes: one instance of the +// module holds the seat, so what it holds is what the ledger has. +func (s *ledgerStore) fresh(k Key) (*headState, error) { + if h := s.heads[k]; h != nil { + return h, nil + } + h, err := s.read(k) + if err != nil || h == nil { + return nil, err + } + s.heads[k] = h + return h, nil +} + +// scan reads every instance the ledger holds, once: the numbers are probed in order until a long run of none, because the +// ledger answers an instance by its path and not a list of them. +func (s *ledgerStore) scan() error { + if s.scanned { + return nil + } + const gap = 50 + for _, kind := range Kinds { + misses := 0 + for n := 1; misses < gap; n++ { + k := Key{kind, n} + h, err := s.fresh(k) + if err != nil { + return err + } + if h == nil { + misses++ + continue + } + misses = 0 + } + } + s.scanned = true + return nil +} + +// text is the text of a section by its ref, read from the write-section that wrote it when it is not held, and held to its +// digest. +func (s *ledgerStore) text(ref sectionRef, k Key) (string, error) { + if t, ok := s.texts[ref.Digest]; ok { + return t, nil + } + var page buscalls.HistoryPage + if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: ref.Sequence, Limit: 1}, &page); err != nil { + return "", err + } + if len(page.Operations) == 0 || page.Operations[0].Sequence != ref.Sequence || page.Operations[0].Operation != OpWriteSection { + return "", fmt.Errorf("the section %q of %s names operation %d, which is no write-section of it", ref.Name, k, ref.Sequence) + } + var body struct { + Text string `json:"text"` + } + if err := json.Unmarshal(page.Operations[0].Body, &body); err != nil { + return "", err + } + if SectionDigest(body.Text) != ref.Digest { + return "", fmt.Errorf("the section %q of %s has a text whose digest is not %s: the history is damaged", ref.Name, k, ref.Digest) + } + s.texts[ref.Digest] = body.Text + return body.Text, nil +} + +// whole is a stored record with its sections' text. +func (s *ledgerStore) whole(sr *storedRecord, k Key) (*Record, error) { + if sr == nil { + return nil, nil + } + r := sr.Record.Clone() + r.Sections = nil + for _, ref := range sr.Sections { + t, err := s.text(ref, k) + if err != nil { + return nil, err + } + r.Sections = append(r.Sections, Section{Name: ref.Name, Text: t}) + } + return r, nil +} + +// entryOf is the entry a head is. +func (s *ledgerStore) entryOf(k Key, h *headState) (*Entry, error) { + cur, err := s.whole(h.Entry.Current, k) + if err != nil { + return nil, err + } + pend, err := s.whole(h.Entry.Pending, k) + if err != nil { + return nil, err + } + return &Entry{Current: cur, CurrentVersion: h.Entry.CurrentVersion, Pending: pend, PendingVersion: h.Entry.PendingVersion, + Versions: h.Entry.Versions, Revision: int64(h.Sequence)}, nil +} + +func (s *ledgerStore) Get(_ context.Context, k Key) (*Entry, error) { + s.mu.Lock() + defer s.mu.Unlock() + h, err := s.fresh(k) + if err != nil || h == nil { + return nil, err + } + return s.entryOf(k, h) +} + +func (s *ledgerStore) All(context.Context) ([]*Entry, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.scan(); err != nil { + return nil, err + } + keys := make([]Key, 0, len(s.heads)) + for k := range s.heads { + keys = append(keys, k) + } + sortKeys(keys) + out := make([]*Entry, 0, len(keys)) + for _, k := range keys { + e, err := s.entryOf(k, s.heads[k]) + if err != nil { + return nil, err + } + out = append(out, e) + } + return out, nil +} + +// write appends a write-section for each section of r that is new, and answers the refs of all its sections. known are +// the refs already written, by name and digest. +func (s *ledgerStore) write(ctx context.Context, k Key, r *Record, known []sectionRef) ([]sectionRef, error) { + var refs []sectionRef + for _, sec := range r.Sections { + d := SectionDigest(sec.Text) + found := false + for _, kr := range known { + if kr.Name == sec.Name && kr.Digest == d { + refs, found = append(refs, kr), true + break + } + } + if found { + continue + } + if len(sec.Text) > MostSectionBytes { + return nil, fmt.Errorf("the section %q of %s is %d bytes; a section holds at most %d: split it", sec.Name, k, len(sec.Text), MostSectionBytes) + } + seq, err := s.apply(ctx, k, OpWriteSection, map[string]any{"title": sec.Name, "text": sec.Text, "digest": d}) + if err != nil { + return nil, err + } + s.texts[d] = sec.Text + refs = append(refs, sectionRef{Name: sec.Name, Digest: d, Sequence: seq}) + } + return refs, nil +} + +// apply hands one operation to the ledger and answers its sequence. +func (s *ledgerStore) apply(ctx context.Context, k Key, op string, body map[string]any) (uint64, error) { + raw, err := json.Marshal(body) + if err != nil { + return 0, err + } + var got buscalls.Applied + if err := s.pipe.Call(buscalls.LedgerApply, buscalls.ApplyRequest{Ledger: LedgerName, Instance: InstanceOf(k), Operation: op, Body: raw, Call: callOf(ctx)}, &got); err != nil { + return 0, err + } + if !got.Taken { + return 0, fmt.Errorf("the ledger did not take %s on %s", op, k) + } + return got.Sequence, nil +} + +// operationFor is the operation a change of an entry is, named by what changed, so the history says it. +func operationFor(old *Entry, v Version, current bool) string { + if old == nil { + switch { + case v.Record.Status == StatusReserved: + return OpReserve + case v.Record.Source != "" && v.Record.Author == "" && len(v.Record.Writers) == 0: + return OpImport + } + return OpPropose + } + prev, next := old.Working(), v.Record + switch { + case next.Status == StatusWithdrawn && prev.Status != StatusWithdrawn: + return OpWithdraw + case next.Accepted() && !prev.Accepted(), next.Status == StatusSuperseded && prev.Status != StatusSuperseded: + return OpAccept + case (next.Retracting != nil) != (prev.Retracting != nil), len(next.Retractions) != len(prev.Retractions), next.Status == StatusRetracted && prev.Status != StatusRetracted: + return OpRetract + case next.Ask != nil && (prev.Ask == nil || prev.Ask.Said != next.Ask.Said) && next.Ask.Warrant != "": + return OpWarrant + case next.Ask != nil && (prev.Ask == nil || prev.Ask.ID != next.Ask.ID): + return OpAsk + case len(next.Reviews) != len(prev.Reviews): + return OpReview + case (next.Drafting != nil) != (prev.Drafting != nil) || (next.Drafting != nil && prev.Drafting != nil && *next.Drafting != *prev.Drafting): + return OpDraft + case next.Author != prev.Author: + return OpTakeOver + } + return OpSubmit +} + +func (s *ledgerStore) Create(ctx context.Context, v Version) error { + s.mu.Lock() + defer s.mu.Unlock() + k := v.Record.Key() + h, err := s.fresh(k) + if err != nil { + return err + } + if h != nil { + return ErrExists + } + v.N = 1 + refs, err := s.write(ctx, k, v.Record, nil) + if err != nil { + return err + } + return s.commit(ctx, k, operationFor(nil, v, true), &entryHead{CurrentVersion: 1, Versions: 1, Current: ptr(stripped(v.Record, refs))}, v, refs) +} + +func ptr[T any](v T) *T { return &v } + +// commit appends the operation that carries the new head and the version, and holds the head it derived. +func (s *ledgerStore) commit(ctx context.Context, k Key, op string, e *entryHead, v Version, refs []sectionRef) error { + body := map[string]any{"kind": k.Kind, "number": k.Number, "entry": e, + "version": versionHead{N: v.N, By: v.By, At: v.At, Why: v.Why, Insight: v.Insight, Record: stripped(v.Record, refs)}, + "pending-sections": map[string]any{}} + seq, err := s.apply(ctx, k, op, body) + if err != nil { + return err + } + s.heads[k] = &headState{Kind: k.Kind, Number: k.Number, Entry: e, Sequence: seq} + return nil +} + +func (s *ledgerStore) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error { + s.mu.Lock() + defer s.mu.Unlock() + k := v.Record.Key() + h, err := s.fresh(k) + if err != nil { + return err + } + if h == nil { + return fmt.Errorf("%s is not held", k) + } + if int64(h.Sequence) != revision { + return ErrChanged + } + old, err := s.entryOf(k, h) + if err != nil { + return err + } + var known []sectionRef + for _, sr := range []*storedRecord{h.Entry.Current, h.Entry.Pending} { + if sr != nil { + known = append(known, sr.Sections...) + } + } + refs, err := s.write(ctx, k, v.Record, known) + if err != nil { + return err + } + e := *h.Entry + v.N = e.Versions + 1 + e.Versions = v.N + sr := ptr(stripped(v.Record, refs)) + if current { + e.Current, e.CurrentVersion = sr, v.N + } else { + e.Pending, e.PendingVersion = sr, v.N + } + if clearPending { + e.Pending, e.PendingVersion = nil, 0 + } + return s.commit(ctx, k, operationFor(old, v, current), &e, v, refs) +} + +func (s *ledgerStore) Versions(_ context.Context, k Key) ([]Version, error) { + s.mu.Lock() + defer s.mu.Unlock() + h, err := s.fresh(k) + if err != nil || h == nil { + return nil, err + } + var out []Version + for from := uint64(0); ; { + var page buscalls.HistoryPage + if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: from}, &page); err != nil { + return nil, err + } + for _, op := range page.Operations { + var body struct { + Version *versionHead `json:"version"` + } + if op.Operation == OpWriteSection || json.Unmarshal(op.Body, &body) != nil || body.Version == nil { + continue + } + r, err := s.whole(&body.Version.Record, k) + if err != nil { + return nil, err + } + out = append(out, Version{N: body.Version.N, By: body.Version.By, At: body.Version.At, Why: body.Version.Why, Record: r, Insight: body.Version.Insight}) + } + if page.Next == 0 { + break + } + from = page.Next + } + sort.SliceStable(out, func(i, j int) bool { return out[i].N < out[j].N }) + return out, nil +} + +// Search ranks the served versions as the in-memory store does, over every record; the index of the module's memory +// replaces it (the search index of to-be 55 §1). +func (s *ledgerStore) Search(ctx context.Context, question string, limit int) ([]Hit, error) { + all, err := s.All(ctx) + if err != nil { + return nil, err + } + return searchEntries(all, question, limit), nil +} + +func (s *ledgerStore) Inbound(_ context.Context, k Key) ([]Inbound, error) { + s.mu.Lock() + defer s.mu.Unlock() + if err := s.scan(); err != nil { + return nil, err + } + // The links are in the heads, which hold no text: nothing is read of a section. + var served []*Record + for _, h := range s.heads { + served = append(served, &h.Entry.Current.Record) + } + return inboundOf(served, k), nil +} + +// The whole-set findings and what the operator was told are keys of the state `findings`. +const ( + findingsState = "findings" + findingsKey = "set" + toldPrefix = "told." +) + +func (s *ledgerStore) SetFindings(_ context.Context, f []Finding) error { + _, err := s.states.State(findingsState).Put(findingsKey, f) + return err +} + +func (s *ledgerStore) Findings(context.Context) ([]Finding, error) { + e, err := s.states.State(findingsState).Get(findingsKey) + if err != nil || e == nil { + return nil, err + } + var out []Finding + return out, json.Unmarshal(e.Value, &out) +} + +func (s *ledgerStore) Told(_ context.Context, topic string) (bool, error) { + _, err := s.states.State(findingsState).Create(toldPrefix+topic, time.Now().UTC()) + switch { + case err == nil: + return true, nil + case isChanged(err): + return false, nil + } + return false, err +} + +func (s *ledgerStore) ToldOf(_ context.Context, prefix string) ([]string, error) { + keys, err := s.states.State(findingsState).Keys() + if err != nil { + return nil, err + } + var out []string + for _, k := range keys { + if t, ok := strings.CutPrefix(k, toldPrefix); ok && strings.HasPrefix(t, prefix) { + out = append(out, t) + } + } + sort.Strings(out) + return out, nil +} + +func (s *ledgerStore) Untold(_ context.Context, key string) error { + return s.states.State(findingsState).Delete(toldPrefix + key) +} + +func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go new file mode 100644 index 00000000..0a248cb9 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore_test.go @@ -0,0 +1,162 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +func ledgerRecord(n int, status string, secs ...Section) *Record { + return &Record{Kind: KindResearch, Number: n, Title: "A record", Status: status, Date: "2026-11-04", Sections: secs} +} + +// A record is one instance; a section's text is written once, in its own operation, and a new version names it by its +// digest and sequence; the head holds no text (to-be 55 §2.1, AC46). +func TestTheLedgerStoreKeepsEachSectionOnceAndTheHeadWithoutText(t *testing.T) { + ctx := context.Background() + f := newFakeLedger(t) + s := newLedgerStore(f, newFakeStates()) + at := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) + if err := s.Create(ctx, Version{By: "a@n", At: at, Why: "proposed", Record: ledgerRecord(81, StatusProposed, Section{"Overview", "one"}, Section{"Direction", "two"})}); err != nil { + t.Fatal(err) + } + e, _ := s.Get(ctx, Key{KindResearch, 81}) + r2 := e.Current.Clone() + r2.Sections[1].Text = "two, revised" + if err := s.Save(ctx, Version{By: "a@n", At: at, Why: "revised", Record: r2}, true, false, e.Revision); err != nil { + t.Fatal(err) + } + // Three texts written (one, two, two revised), never "one" again; the operations are named by what they did. + var names []string + for _, a := range f.applied { + names = append(names, a.Operation) + } + if got := strings.Join(names, " "); got != "write-section write-section propose write-section submit" { + t.Fatalf("operations %q", got) + } + head := string(mustJSON(t, f.state["research/081"]["entry"])) + if strings.Contains(head, "two, revised") || strings.Contains(head, `"text"`) { + t.Fatalf("the head holds a section's text: %s", head) + } + got, _ := s.Get(ctx, Key{KindResearch, 81}) + if got.Current.Sections[1].Text != "two, revised" || got.Current.Sections[0].Text != "one" || got.Versions != 2 || got.CurrentVersion != 2 { + t.Fatalf("%+v", got.Current.Sections) + } + // A reader that holds nothing (a restart) hydrates from the history, every version. + fresh := newLedgerStore(f, newFakeStates()) + vs, err := fresh.Versions(ctx, Key{KindResearch, 81}) + if err != nil || len(vs) != 2 || vs[0].Record.Sections[1].Text != "two" || vs[1].Record.Sections[1].Text != "two, revised" || vs[1].Why != "revised" { + t.Fatalf("%v %+v", err, vs) + } +} + +func TestTheLedgerStoreRefusesWhatTheLedgerWouldHaveToTakeBack(t *testing.T) { + ctx := context.Background() + f := newFakeLedger(t) + s := newLedgerStore(f, newFakeStates()) + v := Version{By: "a@n", At: time.Now(), Why: "x", Record: ledgerRecord(5, StatusProposed, Section{"Overview", "one"})} + if err := s.Create(withCall(ctx, "call-9"), v); err != nil { + t.Fatal(err) + } + // The tool call in flight is named to the ledger, so the author stamped is the caller it proved; a background act names none. + if last := f.applied[len(f.applied)-1]; last.Call != "call-9" || f.hist["research/005"][1].Author.Caller != "caller-of-call-9" { + t.Fatalf("%+v", last) + } + if err := s.Create(ctx, v); !errors.Is(err, ErrExists) { + t.Fatalf("a record held is created again: %v", err) + } + e, _ := s.Get(ctx, v.Record.Key()) + if err := s.Save(ctx, v, true, false, e.Revision-1); !errors.Is(err, ErrChanged) { + t.Fatalf("a write on a head already moved: %v", err) + } + big := ledgerRecord(6, StatusProposed, Section{"Overview", strings.Repeat("x", MostSectionBytes+1)}) + if err := s.Create(ctx, Version{Record: big}); err == nil || !strings.Contains(err.Error(), "split it") { + t.Fatalf("a section over its bound: %v", err) + } + if h, _ := s.Get(ctx, big.Key()); h != nil { + t.Fatal("a refused record is held") + } + // A text altered in the history is caught by its digest. + op := &f.hist["research/005"][0] + op.Body = mustJSON(t, map[string]any{"title": "Overview", "text": "altered", "digest": SectionDigest("one")}) + if _, err := newLedgerStore(f, newFakeStates()).Get(ctx, v.Record.Key()); err == nil || !strings.Contains(err.Error(), "digest") { + t.Fatalf("a damaged history was read: %v", err) + } + // The ledger's refusal is the call's refusal. + f.failApply = OpPropose + if err := s.Create(ctx, Version{Record: ledgerRecord(7, StatusProposed)}); err == nil { + t.Fatal("a refused apply was taken") + } +} + +func TestTheOperationAChangeIsIsNamedByWhatChanged(t *testing.T) { + base := ledgerRecord(1, StatusProposed) + base.Author = "a@n" + with := func(f func(*Record)) *Record { c := base.Clone(); f(c); return c } + held := &Entry{Current: base, CurrentVersion: 1, Versions: 1} + for _, c := range []struct { + name string + old *Entry + next *Record + want string + }{ + {"a reserved number", nil, ledgerRecord(1, StatusReserved), OpReserve}, + {"a proposal", nil, base, OpPropose}, + {"an import", nil, with(func(r *Record) { r.Author, r.Source = "", "hq@abc:x.md" }), OpImport}, + {"a revision", held, with(func(r *Record) { r.Title = "B" }), OpSubmit}, + {"a review", held, with(func(r *Record) { r.Reviews = []Review{{By: "j"}} }), OpReview}, + {"an acceptance", held, with(func(r *Record) { r.Status = StatusAccepted }), OpAccept}, + {"a withdrawal", held, with(func(r *Record) { r.Status = StatusWithdrawn }), OpWithdraw}, + {"an ask", held, with(func(r *Record) { r.Ask = &OperatorAsk{ID: "x", Said: "waiting"} }), OpAsk}, + {"a warrant", &Entry{Current: with(func(r *Record) { r.Ask = &OperatorAsk{ID: "x", Said: "waiting"} })}, with(func(r *Record) { r.Ask = &OperatorAsk{ID: "x", Said: "approved", Warrant: "w"} }), OpWarrant}, + {"a take-over", held, with(func(r *Record) { r.Author = "b@n" }), OpTakeOver}, + {"a drafting step", held, with(func(r *Record) { r.Drafting = &Drafting{State: DraftWaiting} }), OpDraft}, + {"a retraction asked", held, with(func(r *Record) { r.Retracting = &Retraction{Why: "x"} }), OpRetract}, + } { + if got := operationFor(c.old, Version{Record: c.next}, true); got != c.want { + t.Errorf("%s: %s, want %s", c.name, got, c.want) + } + } +} + +func TestTheFindingsAndTheToldMarksAreKeysOfTheState(t *testing.T) { + ctx := context.Background() + st := newFakeStates() + s := newLedgerStore(newFakeLedger(t), st) + if err := s.SetFindings(ctx, []Finding{{Rule: "cited", Says: "x"}}); err != nil { + t.Fatal(err) + } + if f, _ := s.Findings(ctx); len(f) != 1 || f[0].Rule != "cited" { + t.Fatalf("%v", f) + } + if first, _ := s.Told(ctx, "clauses.0290"); !first { + t.Fatal("told once") + } + if again, _ := s.Told(ctx, "clauses.0290"); again { + t.Fatal("told twice") + } + _, _ = s.Told(ctx, "review.0291") + if got, _ := s.ToldOf(ctx, "clauses."); len(got) != 1 || got[0] != "clauses.0290" { + t.Fatalf("%v", got) + } + _ = s.Untold(ctx, "clauses.0290") + if first, _ := s.Told(ctx, "clauses.0290"); !first { + t.Fatal("told again after it was over") + } +} + +func mustJSON(t *testing.T, v any) json.RawMessage { + t.Helper() + b, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return b +} + +var _ = buscalls.LedgerApply diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go b/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go new file mode 100644 index 00000000..66cb8f8a --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/pipe.go @@ -0,0 +1,82 @@ +package main + +// The module's reach to its bus state (novox/hq ADR 0595 §9, ADR 0597 §10): the calls of mesh-sdk's buscalls, made over +// the bundle's pipe to the tool runner, and the module's states, made through the SDK's State. Both are interfaces here, +// so the ledger's store is tested against a ledger played in the test and not against a bus. + +import ( + "context" + "encoding/json" + "errors" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Pipe makes one call of buscalls on the tool runner: the method, its request, and where its answer is decoded. +type Pipe interface { + Call(method string, request, answer any) error +} + +// KV is one of the module's states (stdio.KeptState is one). +type KV interface { + Get(key string) (*stdio.StateEntry, error) + Put(key string, value any) (uint64, error) + Create(key string, value any) (uint64, error) + Update(key string, value any, revision uint64) (uint64, error) + Delete(key string) error + Keys() ([]string, error) +} + +// States are the module's states by their local names in `requires`. +type States interface { + State(name string) KV +} + +// sdkStates are the module's states through the SDK. +type sdkStates struct{} + +func (sdkStates) State(name string) KV { return stdio.State(name) } + +// callKey carries, in a context, the id of the tool call an operation serves: the tool runner stamps the caller it +// proved for that call as the operation's author (buscalls.MetaCall, ADR 0597 §7). +type callKey struct{} + +// withCall is ctx naming the tool call in flight. +func withCall(ctx context.Context, call string) context.Context { + if call == "" { + return ctx + } + return context.WithValue(ctx, callKey{}, call) +} + +// callOf is the tool call ctx names, or "" for the module's own act. +func callOf(ctx context.Context) string { + s, _ := ctx.Value(callKey{}).(string) + return s +} + +// ErrNoPipe is the answer of a pipe the pinned mesh-sdk does not give the bundle yet. +var ErrNoPipe = errors.New("the pinned mesh-sdk gives a bundle no request on the tool runner's bus-state calls yet " + + "(a request to the SDK's owner: stdio.Request); the records cannot be reached over the bus") + +// sdkPipe is the pipe through the SDK. Request is set when the SDK offers it; until then the pipe answers ErrNoPipe, +// loudly, and nothing is written. +type sdkPipe struct{} + +// request is the SDK's raw request on the runtime's channel (stdio.Request(method, params) (json.RawMessage, error)), +// wired in the one place the group's mesh-sdk pin brings it. +var request func(method string, params any) (json.RawMessage, error) + +func (sdkPipe) Call(method string, req, answer any) error { + if request == nil { + return ErrNoPipe + } + raw, err := request(method, req) + if err != nil { + return err + } + if answer == nil || len(raw) == 0 { + return nil + } + return json.Unmarshal(raw, answer) +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/store.go b/modules/mesh-knowledge/cmd/mesh-knowledge/store.go index f3b83a2a..cd816d4a 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/store.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/store.go @@ -190,12 +190,26 @@ func (s *memStore) Versions(_ context.Context, k Key) ([]Version, error) { // and sections; the title and a clause weigh more. Enough to test the answers' shape; the store's ranking is the // database's full-text search. func (s *memStore) Search(_ context.Context, question string, limit int) ([]Hit, error) { - words := questionWords(question) - if len(words) == 0 { - return nil, nil - } s.mu.Lock() defer s.mu.Unlock() + keys := make([]Key, 0, len(s.entries)) + for k := range s.entries { + keys = append(keys, k) + } + sortKeys(keys) + var all []*Entry + for _, k := range keys { + all = append(all, s.entries[k].entry()) + } + return searchEntries(all, question, limit), nil +} + +// searchEntries ranks the served versions of entries against a question. +func searchEntries(entries []*Entry, question string, limit int) []Hit { + words := questionWords(question) + if len(words) == 0 { + return nil + } var hits []Hit score := func(text string) float64 { n := 0.0 @@ -208,8 +222,9 @@ func (s *memStore) Search(_ context.Context, question string, limit int) ([]Hit, } return n } - for k, e := range s.entries { - r := e.versions[e.current-1].Record + for _, e := range entries { + r := e.Current + k := r.Key() if r.Status == StatusReserved || r.Status == StatusWithdrawn || r.Status == StatusRetracted { continue } @@ -241,15 +256,24 @@ func (s *memStore) Search(_ context.Context, question string, limit int) ([]Hit, if limit > 0 && len(hits) > limit { hits = hits[:limit] } - return hits, nil + return hits } func (s *memStore) Inbound(_ context.Context, k Key) ([]Inbound, error) { s.mu.Lock() defer s.mu.Unlock() + var served []*Record + for _, e := range s.entries { + served = append(served, e.versions[e.current-1].Record) + } + return inboundOf(served, k), nil +} + +// inboundOf is every link of another served record in served that points at k. +func inboundOf(served []*Record, k Key) []Inbound { var out []Inbound - for from, e := range s.entries { - r := e.versions[e.current-1].Record + for _, r := range served { + from := r.Key() for _, l := range r.Links { if to, ok := l.To.Record(); ok && to == k { out = append(out, Inbound{From: from, Link: l}) @@ -262,7 +286,7 @@ func (s *memStore) Inbound(_ context.Context, k Key) ([]Inbound, error) { } return out[i].Link.String() < out[j].Link.String() }) - return out, nil + return out } func (s *memStore) SetFindings(_ context.Context, f []Finding) error {