mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.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"

// bundleDay is the bundle's clock: the day the fixture's proposals are dated (valid's Date), so a reservation and the
// proposal after it are of one day whatever day the test runs on (issue 540).
var bundleDay = time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)

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())
		k.Now = func() time.Time { return bundleDay }
		serve(k, os.Getenv("MESH_KNOWLEDGE_STATUS"))
		os.Exit(0)
	}
	// Every test runs in a zone that is not UTC, so a time said locally is proven to be, wherever the tests run. It
	// is set once, before any goroutine starts: a test that set it while a bus test's server read it raced.
	time.Local = testZone
	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}},
		Homes: []string{testHome}}
}

// testHome is the hq design valid names as its home: a decision is accepted only cited (novox/hq issue 504).
const testHome = "03-DESIGN/01-to-be/55-the-knowledge-base.md"

// 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
}