jschoubben
c394ecd9fd
mesh/merge-gate pass: builds mesh-knowledge → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery failed: its walk failed, carrying its own commit: a gate on a first machine (what it carried put back), a build, a machine
The operator's channel edits a cleared message with the resolved line it was raised with, so one sent only with the clear was never shown.
495 lines
15 KiB
Go
495 lines
15 KiB
Go
package main
|
|
|
|
// What the tests stand on: the knowledge base over each store — in memory always, and on a real postgres when one
|
|
// can be had (MESH_TEST_POSTGRES, or a private server started here from the machine's initdb on a socket of its own,
|
|
// listening on no port) — hq as real files copied from it at a commit (testdata/hq, and testdata/hq-main.tar.gz,
|
|
// the whole of main), and the issue tracker, the agent's judgement and the router played here.
|
|
|
|
import (
|
|
"archive/tar"
|
|
"compress/gzip"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/url"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
)
|
|
|
|
const bundleVar = "MESH_KNOWLEDGE_TEST_BUNDLE"
|
|
|
|
func TestMain(m *testing.M) {
|
|
if os.Getenv(bundleVar) == "1" {
|
|
// The bundle the bus test starts: the knowledge base as the runtime launches it, its store in memory.
|
|
k := newKB(newMemStore())
|
|
serve(k, os.Getenv("MESH_KNOWLEDGE_STATUS"))
|
|
os.Exit(0)
|
|
}
|
|
code := m.Run()
|
|
stopPostgres()
|
|
os.Exit(code)
|
|
}
|
|
|
|
// ---- postgres ------------------------------------------------------------------------------------------------
|
|
|
|
var (
|
|
pgOnce sync.Once
|
|
pgBase string
|
|
pgDir string
|
|
pgWhy string
|
|
pgCount int
|
|
pgMu sync.Mutex
|
|
)
|
|
|
|
func startPostgres() {
|
|
if url := os.Getenv("MESH_TEST_POSTGRES"); url != "" {
|
|
pgBase = url
|
|
return
|
|
}
|
|
initdb, err := exec.LookPath("initdb")
|
|
if err != nil {
|
|
pgWhy = "no postgres: MESH_TEST_POSTGRES is unset and the machine has no initdb"
|
|
return
|
|
}
|
|
pgctl := filepath.Join(filepath.Dir(initdb), "pg_ctl")
|
|
dir, err := os.MkdirTemp("", "mkpg")
|
|
if err != nil {
|
|
pgWhy = err.Error()
|
|
return
|
|
}
|
|
pgDir = dir
|
|
if out, err := exec.Command(initdb, "-D", dir+"/data", "-U", "postgres", "-A", "trust", "-E", "UTF8", "--no-locale").CombinedOutput(); err != nil {
|
|
pgWhy = fmt.Sprintf("initdb: %v: %s", err, out)
|
|
return
|
|
}
|
|
// A socket in its own directory and no port: nothing on this machine can reach it but these tests.
|
|
if out, err := exec.Command(pgctl, "-D", dir+"/data", "-l", dir+"/log", "-w", "-o",
|
|
"-k "+dir+" -c listen_addresses='' -F", "start").CombinedOutput(); err != nil {
|
|
pgWhy = fmt.Sprintf("pg_ctl start: %v: %s", err, out)
|
|
return
|
|
}
|
|
pgBase = "postgres://postgres@/postgres?host=" + dir
|
|
}
|
|
|
|
func stopPostgres() {
|
|
if pgDir == "" {
|
|
return
|
|
}
|
|
if initdb, err := exec.LookPath("initdb"); err == nil {
|
|
_ = exec.Command(filepath.Join(filepath.Dir(initdb), "pg_ctl"), "-D", pgDir+"/data", "-m", "immediate", "stop").Run()
|
|
}
|
|
_ = os.RemoveAll(pgDir)
|
|
}
|
|
|
|
// freshDatabase is a database of its own for one test, on the test postgres, or a skip saying why there is none.
|
|
func freshDatabase(t *testing.T) string {
|
|
t.Helper()
|
|
pgOnce.Do(startPostgres)
|
|
if pgBase == "" {
|
|
t.Skipf("NOT TESTED ON POSTGRES: %s", pgWhy)
|
|
}
|
|
pgMu.Lock()
|
|
pgCount++
|
|
name := fmt.Sprintf("kb_%d_%d", os.Getpid(), pgCount)
|
|
pgMu.Unlock()
|
|
ctx := context.Background()
|
|
admin, err := OpenPgStore(ctx, pgBase)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer admin.Close()
|
|
if _, err := admin.pool.Exec(ctx, "create database "+name); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return replaceDatabase(pgBase, name)
|
|
}
|
|
|
|
func replaceDatabase(base, name string) string {
|
|
u, err := url.Parse(base)
|
|
if err != nil {
|
|
return base
|
|
}
|
|
u.Path = "/" + name
|
|
return u.String()
|
|
}
|
|
|
|
// pgStore is a prepared store on a fresh database.
|
|
func pgStore(t *testing.T) *PgStore {
|
|
t.Helper()
|
|
url := freshDatabase(t)
|
|
s, err := OpenPgStore(context.Background(), url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(s.Close)
|
|
if _, err := s.Prepare(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return s
|
|
}
|
|
|
|
// eachStore runs a test over the store in memory 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("postgres", func(t *testing.T) { run(t, pgStore(t)) })
|
|
}
|
|
|
|
// ---- hq ---------------------------------------------------------------------------------------------------------
|
|
|
|
// fixture is the handful of real hq files copied at one commit.
|
|
func fixture() dirRecords { return dirRecords{dir: filepath.Join("testdata", "hq")} }
|
|
|
|
var (
|
|
mainOnce sync.Once
|
|
mainDir string
|
|
mainErr error
|
|
)
|
|
|
|
// hqMain is the whole of hq's main at the commit testdata/hq-main.commit names, unpacked once.
|
|
func hqMain(t *testing.T) dirRecords {
|
|
t.Helper()
|
|
mainOnce.Do(func() {
|
|
dir, err := os.MkdirTemp("", "kb-hq-main")
|
|
if err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
mainDir = dir
|
|
f, err := os.Open(filepath.Join("testdata", "hq-main.tar.gz"))
|
|
if err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
defer f.Close()
|
|
gz, err := gzip.NewReader(f)
|
|
if err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
tr := tar.NewReader(gz)
|
|
for {
|
|
h, err := tr.Next()
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
path := filepath.Join(dir, filepath.FromSlash(h.Name))
|
|
if h.Typeflag == tar.TypeDir {
|
|
_ = os.MkdirAll(path, 0o755)
|
|
continue
|
|
}
|
|
_ = os.MkdirAll(filepath.Dir(path), 0o755)
|
|
raw, err := io.ReadAll(tr)
|
|
if err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
if err := os.WriteFile(path, raw, 0o644); err != nil {
|
|
mainErr = err
|
|
return
|
|
}
|
|
}
|
|
commit, _ := os.ReadFile(filepath.Join("testdata", "hq-main.commit"))
|
|
mainErr = os.WriteFile(filepath.Join(dir, "hq", "COMMIT"), commit, 0o644)
|
|
})
|
|
if mainErr != nil {
|
|
t.Fatal(mainErr)
|
|
}
|
|
return dirRecords{dir: filepath.Join(mainDir, "hq")}
|
|
}
|
|
|
|
// ---- what the knowledge base reaches, played here ------------------------------------------------------------
|
|
|
|
// tracker is the issue tracker's counters, played: the next number of a kind, from where hq's main stands.
|
|
type tracker struct {
|
|
mu sync.Mutex
|
|
next map[string]int
|
|
asked []map[string]string
|
|
}
|
|
|
|
func newTracker() *tracker {
|
|
return &tracker{next: map[string]int{KindDecision: 300, KindResearch: 50}}
|
|
}
|
|
|
|
func (tr *tracker) reserve(kind, title string) (int, error) {
|
|
tr.mu.Lock()
|
|
defer tr.mu.Unlock()
|
|
tr.asked = append(tr.asked, map[string]string{"kind": kind, "for": title})
|
|
n := tr.next[kind]
|
|
tr.next[kind]++
|
|
return n, nil
|
|
}
|
|
|
|
// judge is the agent's judgement, played: it drafts a clause from each sentence of a section that says "is", "are"
|
|
// or "decid", confirms every clause but those whose text holds "NOT EXACT", names clause 1 for a change, and
|
|
// approves every record it reviews unless told otherwise.
|
|
type judge struct {
|
|
mu sync.Mutex
|
|
questions []string
|
|
fail bool
|
|
// review is the answer a review gets, raw; approve when empty. reviewFail fails every review. during runs
|
|
// inside a review's judgement, while the knowledge base waits for it.
|
|
review string
|
|
reviewFail bool
|
|
reviews []string
|
|
during func()
|
|
}
|
|
|
|
func (j *judge) judge(question string, schema map[string]any, model string) (json.RawMessage, string, error) {
|
|
j.mu.Lock()
|
|
defer j.mu.Unlock()
|
|
j.questions = append(j.questions, question)
|
|
if j.fail {
|
|
return nil, "", fmt.Errorf("the agent holds no licence")
|
|
}
|
|
props := schema["properties"].(map[string]any)
|
|
switch {
|
|
case props["verdict"] != nil:
|
|
j.reviews = append(j.reviews, question)
|
|
if during := j.during; during != nil {
|
|
j.during = nil
|
|
j.mu.Unlock()
|
|
during()
|
|
j.mu.Lock()
|
|
}
|
|
if j.reviewFail {
|
|
return nil, "", fmt.Errorf("nothing answers claude-code.claude_code_judge")
|
|
}
|
|
if j.review != "" {
|
|
return json.RawMessage(j.review), "the agent on test (" + model + ")", nil
|
|
}
|
|
return json.RawMessage(`{"verdict":"approve","findings":[]}`), "the agent on test (" + model + ")", nil
|
|
case props["verdicts"] != nil:
|
|
var verdicts []map[string]any
|
|
for _, line := range strings.Split(question[strings.Index(question, "The drafted clauses:"):], "\n") {
|
|
m := regexp.MustCompile(`^(\d+)\. `).FindStringSubmatch(line)
|
|
if m == nil {
|
|
continue
|
|
}
|
|
exact := !strings.Contains(line, "NOT EXACT")
|
|
v := map[string]any{"clause": atoi(m[1]), "exact": exact}
|
|
if !exact {
|
|
v["reason"] = "the section says it of the controller only"
|
|
}
|
|
verdicts = append(verdicts, v)
|
|
}
|
|
raw, _ := json.Marshal(map[string]any{"verdicts": verdicts})
|
|
return raw, "the agent on test (" + model + ")", nil
|
|
case props["clauses"] != nil && props["reason"] != nil:
|
|
return json.RawMessage(`{"clauses":[1],"reason":"it replaces the first"}`), "the agent on test (" + model + ")", nil
|
|
case props["exact"] != nil:
|
|
return json.RawMessage(`{"exact":true,"reason":"it does"}`), "the agent on test (" + model + ")", nil
|
|
}
|
|
marker := "The Decision section:"
|
|
if !strings.Contains(question, marker) {
|
|
marker = "The record's sections:"
|
|
}
|
|
section := question[strings.Index(question, marker)+len(marker):]
|
|
var clauses []map[string]string
|
|
for _, s := range regexp.MustCompile(`[^.]+\.`).FindAllString(strings.Join(strings.Fields(section), " "), -1) {
|
|
s = strings.TrimSpace(s)
|
|
if len(clauses) < 4 && len(s) > 20 && regexp.MustCompile(`(?i)\b(is|are|decid)`).MatchString(s) {
|
|
clauses = append(clauses, map[string]string{"lead": "Drafted " + fmt.Sprint(len(clauses)+1), "text": s})
|
|
}
|
|
}
|
|
if len(clauses) == 0 {
|
|
clauses = append(clauses, map[string]string{"lead": "Drafted 1", "text": "what the section decides."})
|
|
}
|
|
raw, _ := json.Marshal(map[string]any{"clauses": clauses})
|
|
return raw, "the agent on test (" + model + ")", nil
|
|
}
|
|
|
|
// router is the operator channel, played: the asks published, and its own record of each, which only it writes.
|
|
type router struct {
|
|
mu sync.Mutex
|
|
asked []asks.Ask
|
|
records map[string]routerRecord
|
|
}
|
|
|
|
func newRouter() *router { return &router{records: map[string]routerRecord{}} }
|
|
|
|
func (r *router) Publish(subject string, body any, id string) error {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if subject != asks.AskSubject(Module) {
|
|
return fmt.Errorf("published on %s", subject)
|
|
}
|
|
a := body.(asks.Ask)
|
|
r.asked = append(r.asked, a)
|
|
cp := a
|
|
r.records[a.ID] = routerRecord{State: "open", Ask: &cp}
|
|
return nil
|
|
}
|
|
|
|
func (r *router) Record(id string) (*stdio.StateEntry, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
rec, ok := r.records[id]
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
raw, _ := json.Marshal(rec)
|
|
return &stdio.StateEntry{Key: id, Value: raw, Revision: 1}, nil
|
|
}
|
|
|
|
// answer closes an ask as the router does when the operator taps an option on a channel that proves who answered.
|
|
func (r *router) answer(id, option string, at time.Time) asks.Warrant {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
rec := r.records[id]
|
|
o, _ := rec.Ask.Option(option)
|
|
w := asks.Warrant{Ask: id, Asker: Module, About: rec.Ask.About, Outcome: asks.OutcomeChosen, Option: o.ID, Label: o.Label,
|
|
Level: o.Level, By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "1", Verified: "user id verified"},
|
|
Channel: "telegram", At: at, AskDigest: rec.Ask.Digest()}
|
|
rec.State, rec.Warrant = "closed", &w
|
|
r.records[id] = rec
|
|
return w
|
|
}
|
|
|
|
// said is the events said.
|
|
type said struct {
|
|
mu sync.Mutex
|
|
events []map[string]any
|
|
}
|
|
|
|
func (s *said) emit(event string, body any) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
b := body.(map[string]any)
|
|
cp := map[string]any{"event": event}
|
|
for k, v := range b {
|
|
cp[k] = v
|
|
}
|
|
s.events = append(s.events, cp)
|
|
return nil
|
|
}
|
|
|
|
func (s *said) of(event string) []map[string]any {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var out []map[string]any
|
|
for _, e := range s.events {
|
|
if e["event"] == event {
|
|
out = append(out, e)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// world is one knowledge base and everything it reaches, played.
|
|
type world struct {
|
|
k *KB
|
|
tracker *tracker
|
|
judge *judge
|
|
router *router
|
|
said *said
|
|
told []string
|
|
// toldKeys are the keys told under, and cleared what was cleared: key, then its resolved line.
|
|
toldKeys []string
|
|
cleared [][2]string
|
|
resolved []string // the resolved line given with each notice raised
|
|
clock time.Time
|
|
}
|
|
|
|
func newWorld(t *testing.T, s Store, records Records) *world {
|
|
t.Helper()
|
|
w := &world{tracker: newTracker(), judge: &judge{}, router: newRouter(), said: &said{},
|
|
clock: time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)}
|
|
w.k = &KB{Store: s, Records: records, Reserve: w.tracker.reserve, Judge: w.judge.judge, Asks: w.router,
|
|
Emit: w.said.emit, Settings: func() (Settings, error) { return Settings{}, nil }, Now: func() time.Time { return w.clock },
|
|
Logf: func(f string, a ...any) { t.Logf(f, a...) },
|
|
Notify: func(key, summary, headline, explanation, resolved string) error {
|
|
w.told, w.toldKeys, w.resolved = append(w.told, summary), append(w.toldKeys, key), append(w.resolved, resolved)
|
|
return nil
|
|
},
|
|
Clear: func(key, resolved string) error { w.cleared = append(w.cleared, [2]string{key, resolved}); return nil }}
|
|
return w
|
|
}
|
|
|
|
// valid is a proposal every rule takes, for the record numbered n, against the fixture's records.
|
|
func valid(n int) ProposeAsk {
|
|
return ProposeAsk{Kind: KindDecision, Number: n, Title: "A merge window closes after ninety seconds of quiet",
|
|
Topic: "the mesh", Date: "2026-10-11", Deciders: []string{"jochen"},
|
|
Clauses: []string{"**The window closes after ninety seconds of quiet.** A merge opens or extends it.",
|
|
"**A batch is delivered by one walk.** One commit per repository."},
|
|
Sections: []Section{
|
|
{Name: "Context", Text: "The merge window of [ADR 0276](0276-merges-are-assembled-in-a-rolling-window-and-each-batch-is-delivered-by-one-walk-with-one-commit-per-repository.md) waits too long."},
|
|
{Name: "Considered Options", Text: "1. **Keep it.** Rejected."},
|
|
{Name: "Consequences", Text: "Merges land sooner."},
|
|
{Name: "How it is checked", Text: "The controller's merge-window tests."},
|
|
},
|
|
Links: []LinkAsk{{Kind: LinkExtends, To: "0276"}, {Kind: LinkChanges, To: "0276 §1", From: 1}}}
|
|
}
|
|
|
|
// imported is a world whose knowledge base holds the fixture's records.
|
|
func imported(t *testing.T, s Store) *world {
|
|
t.Helper()
|
|
w := newWorld(t, s, fixture())
|
|
if _, err := w.k.Import("import@test", ImportAsk{}, fixture()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return w
|
|
}
|
|
|
|
// reviewAll takes every review waiting, as the background work does, and answers what each step said.
|
|
func (w *world) reviewAll(t *testing.T) []string {
|
|
t.Helper()
|
|
var said []string
|
|
for i := 0; i < 20; i++ {
|
|
s, err := w.k.ReviewStep()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if s == "" {
|
|
return said
|
|
}
|
|
said = append(said, s)
|
|
}
|
|
t.Fatalf("the reviews never ended: %v", said)
|
|
return nil
|
|
}
|
|
|
|
// accepted proposes a valid record under a reserved number and has the knowledge base's review approve it.
|
|
func (w *world) accepted(t *testing.T, author string, p ProposeAsk) {
|
|
t.Helper()
|
|
if _, err := w.k.Propose(author, p); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
w.reviewAll(t)
|
|
if e := mustGet(t, w.k.Store, Key{p.Kind, p.Number}); e.Pending != nil || !e.Current.Accepted() {
|
|
t.Fatalf("%s not accepted after its review: %s, %+v", Key{p.Kind, p.Number}, e.Current.Status, e.Working().Judging)
|
|
}
|
|
}
|
|
|
|
// reserved reserves a decision number and answers it.
|
|
func (w *world) reserved(t *testing.T, caller string) int {
|
|
t.Helper()
|
|
r, err := w.k.ReserveNumber(caller, KindDecision, "a test record")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return r.Number
|
|
}
|
|
|
|
func mustGet(t *testing.T, s Store, k Key) *Entry {
|
|
t.Helper()
|
|
e, err := s.Get(context.Background(), k)
|
|
if err != nil || e == nil {
|
|
t.Fatalf("%s: %v %v", k, e, err)
|
|
}
|
|
return e
|
|
}
|