Add mesh-knowledge, the knowledge base's first slice (hq issue 425, ADR 0290)

An agent learns what a decision says only by reading whole records. This module holds every decision record and research effort whole in a database in the store, with clauses that carry their own status and source, typed links read from both ends, and every version. It answers ask, get and holds within a stated bound, and takes a record through reserve, propose (refused on any of the 29 per-record rules), review by someone other than the author, and acceptance, on the operator's proven warrant when the record needs it. The 4 whole-set rules make the module unhealthy when they fail. The import reads hq at a commit through the records module, never overwrites, and changes nothing when run again. Prose decisions get drafted clauses that a second judgement must confirm before they are served.
2026-10-11 04:58:35 +02:00
parent 27cb95e98f
commit ba7b455719
50 changed files with 11883 additions and 0 deletions
@@ -0,0 +1,558 @@
+package main
+
+// The store on the mesh's postgres (novox/hq ADR 0290 rule 7): the database the mesh provisions for this module
+// (the postgres-database provision), its address in the file DATABASE_URL_FILE names. Its schema is numbered
+// migrations in migrations/, applied by the preparation step (`mesh-knowledge prepare`, ADR 0135) and recorded
+// in the database itself; a fresh database is given the consolidated schema, schema.sql, at once, and every
+// migration is recorded as applied. The serving bundle never changes the schema: it refuses to serve one that is
+// behind, and says so.
+
+import (
+	"context"
+	"crypto/sha256"
+	"embed"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+//go:embed migrations/*.sql
+var migrationFiles embed.FS
+
+//go:embed schema.sql
+var consolidated string
+
+// Migration is one numbered change of the schema.
+type Migration struct {
+	Number   int
+	Name     string
+	SQL      string
+	Checksum string
+}
+
+var migrationFile = regexp.MustCompile(`^(\d{4})-([a-z0-9-]+)\.sql$`)
+
+// Migrations are the migrations this binary carries, in order. A misnamed file, two of one number or an empty
+// one is refused: each would make the schema something nobody wrote down.
+func Migrations() ([]Migration, error) {
+	entries, err := fs.ReadDir(migrationFiles, "migrations")
+	if err != nil {
+		return nil, err
+	}
+	var out []Migration
+	seen := map[int]string{}
+	for _, e := range entries {
+		m := migrationFile.FindStringSubmatch(e.Name())
+		if m == nil {
+			return nil, fmt.Errorf("%s is not a migration's name (NNNN-what-it-does.sql)", e.Name())
+		}
+		n, _ := strconv.Atoi(m[1])
+		if other, twice := seen[n]; twice {
+			return nil, fmt.Errorf("two migrations are numbered %04d: %s and %s", n, other, e.Name())
+		}
+		seen[n] = e.Name()
+		body, err := fs.ReadFile(migrationFiles, path.Join("migrations", e.Name()))
+		if err != nil {
+			return nil, err
+		}
+		if strings.TrimSpace(string(body)) == "" {
+			return nil, fmt.Errorf("%s is empty", e.Name())
+		}
+		sum := sha256.Sum256(body)
+		out = append(out, Migration{Number: n, Name: m[2], SQL: string(body), Checksum: hex.EncodeToString(sum[:])})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Number < out[j].Number })
+	for i, m := range out {
+		if m.Number != i+1 {
+			return nil, fmt.Errorf("the migrations are not numbered 0001 on without a gap: %04d is the %d. one", m.Number, i+1)
+		}
+	}
+	return out, nil
+}
+
+// ledger is the record of what ran, kept in the database it describes.
+const ledger = `create table if not exists migration (
+    number   integer     primary key,
+    name     text        not null,
+    checksum text        not null,
+    applied  timestamptz not null default now())`
+
+// lockKey is the advisory lock every migration run takes, so two preparations never interleave.
+const lockKey int64 = 4_250_290_425
+
+// PgStore is the Store on postgres.
+type PgStore struct{ pool *pgxpool.Pool }
+
+// PgStoreFromEnv opens the store the mesh provisioned, its address in the file DATABASE_URL_FILE names.
+func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
+	file := os.Getenv("DATABASE_URL_FILE")
+	if file == "" {
+		return nil, errors.New("DATABASE_URL_FILE is not set: the knowledge base's database is a requirement the mesh resolves")
+	}
+	raw, err := os.ReadFile(file)
+	if err != nil {
+		return nil, fmt.Errorf("the database's address cannot be read: %w", err)
+	}
+	return OpenPgStore(ctx, strings.TrimSpace(string(raw)))
+}
+
+// OpenPgStore opens a store at an address.
+func OpenPgStore(ctx context.Context, url string) (*PgStore, error) {
+	pool, err := pgxpool.New(ctx, url)
+	if err != nil {
+		return nil, err
+	}
+	return &PgStore{pool: pool}, nil
+}
+
+// Close closes the store.
+func (s *PgStore) Close() { s.pool.Close() }
+
+// Prepare brings the schema to what this version needs, and says what it did: on a database that holds nothing
+// of it, the consolidated schema with every migration recorded; otherwise every migration not yet applied, each
+// in its own transaction with its row in the ledger. A migration that ran and whose file changed since, or one
+// numbered below the highest applied that never ran, stops it: the database's contents are then not known.
+func (s *PgStore) Prepare(ctx context.Context) ([]string, error) {
+	migrations, err := Migrations()
+	if err != nil {
+		return nil, err
+	}
+	conn, err := s.pool.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Release()
+	if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, lockKey); err != nil {
+		return nil, fmt.Errorf("the migration lock cannot be taken: %w", err)
+	}
+	defer func() { _, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock($1)`, lockKey) }()
+	if _, err := conn.Exec(ctx, ledger); err != nil {
+		return nil, err
+	}
+	applied, err := appliedMigrations(ctx, conn.Conn())
+	if err != nil {
+		return nil, err
+	}
+	var said []string
+	if len(applied) == 0 {
+		var exists bool
+		if err := conn.QueryRow(ctx, `select to_regclass('record') is not null`).Scan(&exists); err != nil {
+			return nil, err
+		}
+		if exists {
+			return nil, errors.New("the database holds the knowledge base's tables and no record of a migration: its contents are not known, so nothing is changed")
+		}
+		tx, err := conn.Begin(ctx)
+		if err != nil {
+			return nil, err
+		}
+		defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+		if _, err := tx.Exec(ctx, consolidated); err != nil {
+			return nil, fmt.Errorf("the consolidated schema failed, and nothing of it was kept: %w", err)
+		}
+		for _, m := range migrations {
+			if _, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum); err != nil {
+				return nil, err
+			}
+		}
+		if err := tx.Commit(ctx); err != nil {
+			return nil, err
+		}
+		return []string{fmt.Sprintf("a fresh database: the consolidated schema, through migration %04d", len(migrations))}, nil
+	}
+	highest := 0
+	var problems []string
+	byNumber := map[int]appliedMigration{}
+	for _, a := range applied {
+		byNumber[a.number] = a
+		highest = max(highest, a.number)
+	}
+	var pending []Migration
+	for _, m := range migrations {
+		a, ran := byNumber[m.Number]
+		switch {
+		case ran && a.checksum != m.Checksum:
+			problems = append(problems, fmt.Sprintf("migration %04d-%s ran, and its file has changed since: write a new one", m.Number, m.Name))
+		case !ran && m.Number < highest:
+			problems = append(problems, fmt.Sprintf("migration %04d-%s never ran, and %04d did: renumber it above %04d", m.Number, m.Name, highest, highest))
+		case !ran:
+			pending = append(pending, m)
+		}
+	}
+	if len(problems) > 0 {
+		return nil, errors.New("this database and these migrations disagree: " + strings.Join(problems, "; "))
+	}
+	for _, m := range pending {
+		err := func() error {
+			tx, err := conn.Begin(ctx)
+			if err != nil {
+				return err
+			}
+			defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+			if _, err := tx.Exec(ctx, m.SQL); err != nil {
+				return fmt.Errorf("migration %04d-%s failed, and nothing it did was kept: %w", m.Number, m.Name, err)
+			}
+			if _, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum); err != nil {
+				return err
+			}
+			return tx.Commit(ctx)
+		}()
+		if err != nil {
+			return said, err
+		}
+		said = append(said, fmt.Sprintf("migration %04d-%s applied", m.Number, m.Name))
+	}
+	if len(said) == 0 {
+		said = []string{fmt.Sprintf("the schema is at migration %04d, as this version needs", len(migrations))}
+	}
+	return said, nil
+}
+
+type appliedMigration struct {
+	number   int
+	checksum string
+}
+
+func appliedMigrations(ctx context.Context, conn *pgx.Conn) ([]appliedMigration, error) {
+	rows, err := conn.Query(ctx, `select number, checksum from migration order by number`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []appliedMigration
+	for rows.Next() {
+		var a appliedMigration
+		if err := rows.Scan(&a.number, &a.checksum); err != nil {
+			return nil, err
+		}
+		out = append(out, a)
+	}
+	return out, rows.Err()
+}
+
+// Ready says why the store cannot serve this version, or nil: the schema must be at the migrations it carries.
+func (s *PgStore) Ready(ctx context.Context) error {
+	migrations, err := Migrations()
+	if err != nil {
+		return err
+	}
+	var at int
+	err = s.pool.QueryRow(ctx, `select coalesce(max(number), 0) from migration`).Scan(&at)
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "42P01" {
+		at = 0
+	} else if err != nil {
+		return fmt.Errorf("the database cannot be reached: %w", err)
+	}
+	if at != len(migrations) {
+		return fmt.Errorf("the database's schema is at migration %04d and this version needs %04d: the preparation step "+
+			"(mesh-knowledge prepare) has not run", at, len(migrations))
+	}
+	return nil
+}
+
+const entryColumns = `r.kind, r.number, r.current_version, coalesce(r.pending_version, 0), r.revision,
+  (select count(*) from version v where v.kind = r.kind and v.number = r.number),
+  (select body from version v where v.kind = r.kind and v.number = r.number and v.n = r.current_version),
+  (select body from version v where v.kind = r.kind and v.number = r.number and v.n = r.pending_version)`
+
+func scanEntry(row pgx.Row) (*Entry, error) {
+	var e Entry
+	var kind string
+	var number int
+	var current, pending []byte
+	if err := row.Scan(&kind, &number, &e.CurrentVersion, &e.PendingVersion, &e.Revision, &e.Versions, &current, &pending); err != nil {
+		return nil, err
+	}
+	e.Current = &Record{}
+	if err := json.Unmarshal(current, e.Current); err != nil {
+		return nil, fmt.Errorf("%s's version %d cannot be read: %w", Key{kind, number}, e.CurrentVersion, err)
+	}
+	if len(pending) > 0 {
+		e.Pending = &Record{}
+		if err := json.Unmarshal(pending, e.Pending); err != nil {
+			return nil, fmt.Errorf("%s's version %d cannot be read: %w", Key{kind, number}, e.PendingVersion, err)
+		}
+	}
+	return &e, nil
+}
+
+func (s *PgStore) Get(ctx context.Context, k Key) (*Entry, error) {
+	e, err := scanEntry(s.pool.QueryRow(ctx, `select `+entryColumns+` from record r where r.kind = $1 and r.number = $2`, k.Kind, k.Number))
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	return e, err
+}
+
+func (s *PgStore) All(ctx context.Context) ([]*Entry, error) {
+	rows, err := s.pool.Query(ctx, `select `+entryColumns+` from record r order by r.kind, r.number`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []*Entry
+	for rows.Next() {
+		e, err := scanEntry(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+func sectionsText(r *Record) string {
+	var b strings.Builder
+	for _, s := range r.Sections {
+		b.WriteString(s.Text)
+		b.WriteString("\n")
+	}
+	return b.String()
+}
+
+// served writes the rows of the version a record serves: its status, topic and search vector, its clauses and
+// its links.
+func served(ctx context.Context, tx pgx.Tx, r *Record) error {
+	if _, err := tx.Exec(ctx, `update record set status = $3, topic = $4,
+	    search = setweight(to_tsvector('english', $5), 'A') || setweight(to_tsvector('english', $4), 'B') ||
+	             setweight(to_tsvector('english', $6), 'D')
+	  where kind = $1 and number = $2`, r.Kind, r.Number, r.Status, r.Topic, r.Title, sectionsText(r)); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `delete from clause where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
+		return err
+	}
+	for _, c := range r.Clauses {
+		if _, err := tx.Exec(ctx, `insert into clause (kind, number, n, text, status, source, search)
+		  values ($1, $2, $3, $4, $5, $6, setweight(to_tsvector('english', $4), 'A'))`,
+			r.Kind, r.Number, c.N, c.Text, c.Status, c.Source); err != nil {
+			return err
+		}
+	}
+	if _, err := tx.Exec(ctx, `delete from link where from_kind = $1 and from_number = $2`, r.Kind, r.Number); err != nil {
+		return err
+	}
+	for _, l := range r.Links {
+		body, _ := json.Marshal(l)
+		if _, err := tx.Exec(ctx, `insert into link (from_kind, from_number, from_clause, kind, to_kind, to_number, to_clause, to_path, body)
+		  values ($1, $2, $3, $4, $5, $6, $7, $8, $9)`, r.Kind, r.Number, l.From, l.Kind, l.To.Kind, l.To.Number, l.To.Clause, l.To.Path, body); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func insertVersion(ctx context.Context, tx pgx.Tx, v Version) error {
+	body, err := json.Marshal(v.Record)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `insert into version (kind, number, n, by, at, why, insight, body) values ($1, $2, $3, $4, $5, $6, $7, $8)`,
+		v.Record.Kind, v.Record.Number, v.N, v.By, v.At, v.Why, v.Insight, body)
+	return err
+}
+
+func (s *PgStore) Create(ctx context.Context, v Version) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+	r := v.Record
+	tag, err := tx.Exec(ctx, `insert into record (kind, number, status, topic, current_version) values ($1, $2, $3, $4, 1)
+	  on conflict do nothing`, r.Kind, r.Number, r.Status, r.Topic)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrExists
+	}
+	v.N = 1
+	if err := insertVersion(ctx, tx, v); err != nil {
+		return err
+	}
+	if err := served(ctx, tx, r); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+func (s *PgStore) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+	r := v.Record
+	var held int64
+	var versions int
+	err = tx.QueryRow(ctx, `select revision, (select count(*) from version v where v.kind = r.kind and v.number = r.number)
+	  from record r where kind = $1 and number = $2 for update`, r.Kind, r.Number).Scan(&held, &versions)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return fmt.Errorf("%s is not held", r.Key())
+	}
+	if err != nil {
+		return err
+	}
+	if held != revision {
+		return ErrChanged
+	}
+	v.N = versions + 1
+	if err := insertVersion(ctx, tx, v); err != nil {
+		return err
+	}
+	if current {
+		if _, err := tx.Exec(ctx, `update record set current_version = $3 where kind = $1 and number = $2`, r.Kind, r.Number, v.N); err != nil {
+			return err
+		}
+		if err := served(ctx, tx, r); err != nil {
+			return err
+		}
+	} else {
+		if _, err := tx.Exec(ctx, `update record set pending_version = $3 where kind = $1 and number = $2`, r.Kind, r.Number, v.N); err != nil {
+			return err
+		}
+	}
+	if clearPending {
+		if _, err := tx.Exec(ctx, `update record set pending_version = null where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
+			return err
+		}
+	}
+	if _, err := tx.Exec(ctx, `update record set revision = revision + 1 where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+func (s *PgStore) Versions(ctx context.Context, k Key) ([]Version, error) {
+	rows, err := s.pool.Query(ctx, `select n, by, at, why, insight, body from version where kind = $1 and number = $2 order by n`, k.Kind, k.Number)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Version
+	for rows.Next() {
+		var v Version
+		var body []byte
+		if err := rows.Scan(&v.N, &v.By, &v.At, &v.Why, &v.Insight, &body); err != nil {
+			return nil, err
+		}
+		v.Record = &Record{}
+		if err := json.Unmarshal(body, v.Record); err != nil {
+			return nil, err
+		}
+		out = append(out, v)
+	}
+	return out, rows.Err()
+}
+
+// Search ranks by the database's full-text search: any of the question's words, a clause weighing more than a
+// record's sections, a title most.
+func (s *PgStore) Search(ctx context.Context, question string, limit int) ([]Hit, error) {
+	words := questionWords(question)
+	if len(words) == 0 {
+		return nil, nil
+	}
+	if limit <= 0 {
+		limit = 200
+	}
+	rows, err := s.pool.Query(ctx, `with q as (select replace(plainto_tsquery('english', $1)::text, ' & ', ' | ')::tsquery as q)
+	  select kind, number, clause, rank from (
+	    select c.kind, c.number, c.n as clause, ts_rank_cd(c.search, q.q) as rank
+	      from clause c join record r on r.kind = c.kind and r.number = c.number, q
+	      where c.search @@ q.q and r.status not in ('reserved', 'withdrawn')
+	    union all
+	    select r.kind, r.number, 0, ts_rank_cd(r.search, q.q)
+	      from record r, q where r.search @@ q.q and r.status not in ('reserved', 'withdrawn')
+	  ) hits order by rank desc, number desc, clause limit $2`, strings.Join(words, " "), limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Hit
+	for rows.Next() {
+		var h Hit
+		if err := rows.Scan(&h.Key.Kind, &h.Key.Number, &h.Clause, &h.Rank); err != nil {
+			return nil, err
+		}
+		out = append(out, h)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) Inbound(ctx context.Context, k Key) ([]Inbound, error) {
+	rows, err := s.pool.Query(ctx, `select from_kind, from_number, body from link where to_kind = $1 and to_number = $2
+	  order by from_number, kind`, k.Kind, k.Number)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Inbound
+	for rows.Next() {
+		var in Inbound
+		var body []byte
+		if err := rows.Scan(&in.From.Kind, &in.From.Number, &body); err != nil {
+			return nil, err
+		}
+		if err := json.Unmarshal(body, &in.Link); err != nil {
+			return nil, err
+		}
+		out = append(out, in)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) SetFindings(ctx context.Context, f []Finding) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+	if _, err := tx.Exec(ctx, `delete from finding`); err != nil {
+		return err
+	}
+	for _, x := range f {
+		if _, err := tx.Exec(ctx, `insert into finding (rule, records, says, at) values ($1, $2, $3, $4)`, x.Rule, x.Records, x.Says, x.At); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
+}
+
+func (s *PgStore) Findings(ctx context.Context) ([]Finding, error) {
+	rows, err := s.pool.Query(ctx, `select rule, records, says, at from finding order by rule, says`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Finding
+	for rows.Next() {
+		var f Finding
+		if err := rows.Scan(&f.Rule, &f.Records, &f.Says, &f.At); err != nil {
+			return nil, err
+		}
+		out = append(out, f)
+	}
+	return out, rows.Err()
+}
+
+func (s *PgStore) Told(ctx context.Context, topic string) (bool, error) {
+	tag, err := s.pool.Exec(ctx, `insert into told (topic) values ($1) on conflict do nothing`, topic)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}