diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go index ca15ee586..eed0e18c6 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 000000000..88c275b6d --- /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 000000000..a145dd202 --- /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 000000000..0a248cb9b --- /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 000000000..66cb8f8ac --- /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 f3b83a2a3..cd816d4a2 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 {