feat(brain): parallel corpus write + sort-by-date

.gitignore

@@ -23,3 +23,11 @@ mnt/
 
 # native brain binaries built by bin/stack/serve-brain
 /bin-build/
+
+# local compiled binaries dropped in repo root (go run -o . / go build)
+/brain
+/contacts
+/web
+
+# knowledge-mesh corpus dump (data, not source)
+/corpus/

bin/brain/index.go

@@ -1,5 +1,6 @@
-//usr/bin/env bash -c 'exec "${0%/*}/../cgo/zig" go run -tags=system_ladybug,brain_index "$0" "$@"' "$0" "$@"; exit
 //go:build cgo && system_ladybug && brain_index
+
+// usr/bin/env bash -c 'exec "${0%/*}/../cgo/zig" go run -tags=system_ladybug,brain_index "$0" "$@"' "$0" "$@"; exit
 //
 // bin/brain/index.go - rebuild the Ladybug graph (Go+Zig write path).
 //
@@ -9,6 +10,7 @@
 //
 // Bulk mail/corpus still --rebuild (fresh file, indexes last).
 // NOTE: never run gofmt -w on this file — it breaks the shebang.
+//
 package main
 
 import (
@@ -18,20 +20,21 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"time"
 
-	cliparse "github.com/eSlider/2dph/internal/cli"
 	"github.com/eSlider/2dph/internal/brain"
+	cliparse "github.com/eSlider/2dph/internal/cli"
 )
 
 func main() {
 	os.Exit(run(os.Args[1:]))
 }
 
 type indexFlags struct {
-	db, factsJSON, withChats, since string
-	corpus                          []string
-	rebuild, noDefaults, withMail, withFacts, dryRun, skipIndexes, jsonOut bool
-	limit                           int
+	db, factsJSON, withChats, since                                              string
+	corpus                                                                       []string
+	rebuild, noDefaults, withMail, withFacts, dryRun, skipIndexes, jsonOut, skip bool
+	limit, workers, batch, progress                                              int
 }
 
 func run(args []string) int {
@@ -50,6 +53,10 @@ func run(args []string) int {
 	p.Bool(&v.dryRun, "", "dry-run", "count leafs, write nothing")
 	p.Bool(&v.skipIndexes, "", "skip-indexes", "write leafs only")
 	p.Int(&v.limit, "", "limit", "max leafs to embed")
+	p.Int(&v.workers, "", "workers", "parallel embedding workers (default 4)")
+	p.Int(&v.batch, "", "batch", "leafs per transaction (default 64)")
+	p.Int(&v.progress, "", "progress", "progress/ETA line every N seconds")
+	p.Bool(&v.skip, "", "skip", "skip leafs already in the db (resume)")
 	p.Bool(&v.jsonOut, "", "json", "JSON stats")
 	// --corpus may repeat: parse manually from args leftovers after flaggy
 	if err := cliparse.Parse(p, filterCorpusArgs(args, &v.corpus)); err != nil {
@@ -130,13 +137,15 @@ func run(args []string) int {
 		}
 		return 0
 	}
-	if !v.rebuild {
-		fmt.Fprintln(os.Stderr, "brain/index: refuse non-rebuild write; pass --rebuild (fresh db)")
+	if !v.rebuild && !v.skip {
+		fmt.Fprintln(os.Stderr, "brain/index: refuse write; pass --rebuild (fresh db) or --skip (resume existing db)")
 		return 2
 	}
 
-	_ = os.Remove(dbpath)
-	_ = os.Remove(dbpath + ".wal")
+	if !v.skip {
+		_ = os.Remove(dbpath)
+		_ = os.Remove(dbpath + ".wal")
+	}
 
 	model, err := brain.LoadModel()
 	if err != nil {
@@ -157,7 +166,12 @@ func run(args []string) int {
 		return 1
 	}
 
-	infoN, err := brain.WriteCorpus(conn, leafs, model, v.limit)
+	prog := brain.NewProgressReporter(os.Stderr, time.Duration(v.progress)*time.Second)
+	opt := brain.WriteOptions{
+		Limit: v.limit, Workers: v.workers, Batch: v.batch, Skip: v.skip,
+		Progress: prog,
+	}
+	infoN, err := brain.WriteCorpus(conn, leafs, model, opt)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "brain/index: write info: %v\n", err)
 		return 1

docs/brain-rebuild.md

@@ -0,0 +1,63 @@
+# Brain corpus rebuild (runbook)
+
+Optimized corpus (re)index. Parallel embedding, batched writes, resume, and a
+live progress/ETA monitor.
+
+## Commands
+
+Build the native binaries (also runs the cgo brain tests):
+
+    bin/stack/serve-brain --build        # or ./bin/cgo/zig go build ...
+
+Fresh rebuild of info + mail + facts with control flags:
+
+    KB_BUFFER_POOL=10737418240 bin-build/brain-index \
+        --rebuild --with-mail --with-facts \
+        --workers 12 --batch 256 --progress 5 --skip
+
+- `--workers N`   parallel embedding goroutines (default 4; use ~cores on the box)
+- `--batch N`     leafs per upsert transaction (default 64)
+- `--progress N`  print rate + ETA every N seconds to stderr
+- `--skip`        resume: skip leafs whose id is already in the db
+
+Because leaf ids are deterministic (`LeafID(text, source)`), `--skip` makes a
+re-run cheap: it filters existing ids before embedding, so it embeds only new
+leafs. After a partial/aborted run, re-running with `--skip` skips the
+already-written corpus and goes straight to index build.
+
+> Note: `--rebuild` deletes the db, so `--skip` + `--rebuild` always restarts
+> fresh. To resume a crashed build (leafs already written, indexes missing),
+> run `--skip` **without** `--rebuild` so the db is preserved and only missing
+> indexes are built:
+>
+>     KB_BUFFER_POOL=10737418240 bin-build/brain-index --skip \
+>         --with-mail --with-facts --workers 12 --batch 256 --progress 5
+
+Observed rates: embedding is fast (~16k/s, 256-dim); the db **write** phase is
+the bottleneck (~100/s, ~40 min for 242k leafs).
+
+Before a fresh rebuild, remove the old db (daemons must be down first):
+
+    bin/stack/serve-brain --stop          # or pkill bin-build/brain-*
+    rm -f var/kb.lbug var/kb.lbug.wal
+
+## Buffer pool
+
+`OpenWritable` defaults to a 1 GB buffer pool. Override via `KB_BUFFER_POOL`.
+History: a 1 GB-pool run OOM'd at `CREATE_FTS_INDEX`; a 10 GB pool held a full
+242,275-leaf embedding (DB ~1.4 GB). Use 10 GB (`10737418240`).
+
+## Control / monitoring
+
+`ProgressReporter` (cgo-free, `internal/brain/corpus_pool.go`) prints
+`index: done/total rate/s eta=...` to stderr at `--progress` interval. The worker
+pool (`parallelEmbed`) is order-preserving, respects `ctx` cancellation, and
+collects embed errors per-result.
+
+## After the index-build phase
+
+    bin/stack/serve-brain                 # builds + starts brain-serve + brain-search
+
+Verify:
+
+    bin-build/brain-search ... 17830      # or the API :8630, sort-by-date

internal/brain/corpus.go

@@ -3,6 +3,7 @@
 package brain
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -17,15 +18,6 @@ import (
 var corpusDefaults = []string{"README.md", "PLAN.md", "AGENTS.md", "docs", "skills"}
 
 // CorpusLeaf is a markdown-derived info leaf before embed/write.
-type CorpusLeaf struct {
-	Source  string
-	Repo    string
-	Heading string
-	Text    string
-	Type    string
-	How     string
-}
-
 // LoadDefaultCorpus walks README/PLAN/AGENTS/docs/skills.
 func LoadDefaultCorpus(root string) ([]CorpusLeaf, error) {
 	var files []string
@@ -192,6 +184,12 @@ func LoadMailLeafs(root, since string, limit int) ([]CorpusLeaf, error) {
 	}
 	var out []CorpusLeaf
 	for _, md := range mds {
+		date := ""
+		if raw, err := os.ReadFile(md); err == nil {
+			if fm, _ := mdleaves.ExtractFrontmatter(string(raw)); len(fm["date"]) >= 10 {
+				date = fm["date"][:10]
+			}
+		}
 		files := []string{md}
 		att := filepath.Join(filepath.Dir(md), "attachments")
 		if entries, err := os.ReadDir(att); err == nil {
@@ -211,7 +209,7 @@ func LoadMailLeafs(root, since string, limit int) ([]CorpusLeaf, error) {
 				src := fmt.Sprintf("ooMail:%s:%s", id, filepath.Base(f))
 				out = append(out, CorpusLeaf{
 					Source: src, Repo: "ooMail", Heading: lf.Heading,
-					Text: lf.Text, Type: lf.Type, How: "mail/import",
+					Text: lf.Text, Type: lf.Type, How: "mail/import", Date: date,
 				})
 			}
 		}
@@ -237,22 +235,73 @@ func msgDate(md string) string {
 	return ""
 }
 
-// WriteCorpus embeds and upserts corpus leafs, linking FROM_FILE.
-func WriteCorpus(conn *lbug.Connection, leafs []CorpusLeaf, model *StaticModel, limit int) (int, error) {
-	n := 0
+// WriteOptions controls corpus writing (worker concurrency, batch size, resume).
+type WriteOptions struct {
+	Limit    int               // max leafs to embed/write (0 = all)
+	Workers  int               // parallel embedding workers (0 = 4)
+	Batch    int               // leafs per transaction (0 = 64)
+	Skip     bool              // skip leafs whose id already exists (resume)
+	Progress *ProgressReporter // optional progress/ETA monitor
+}
+
+// WriteCorpus embeds (in parallel) and upserts corpus leafs in batches, linking
+// FROM_FILE. Embedding errors abort; per-batch writes use one transaction.
+func WriteCorpus(conn *lbug.Connection, leafs []CorpusLeaf, model *StaticModel, opt WriteOptions) (int, error) {
+	if opt.Workers <= 0 {
+		opt.Workers = 4
+	}
+	if opt.Batch <= 0 {
+		opt.Batch = 64
+	}
+	if opt.Limit > 0 && len(leafs) > opt.Limit {
+		leafs = leafs[:opt.Limit]
+	}
+
+	// Resume: drop leafs already present before embedding, so a re-run skips
+	// the costly embedding step entirely.
+	if opt.Skip {
+		existing, err := existingLeafIDSet(conn)
+		if err != nil {
+			return 0, err
+		}
+		leafs = filterExistingLeafs(leafs, existing)
+	}
+	if opt.Progress != nil {
+		opt.Progress.Report(0, len(leafs))
+	}
+	if len(leafs) == 0 {
+		if opt.Progress != nil {
+			opt.Progress.Finish(0, 0)
+		}
+		return 0, nil
+	}
+
+	items := make([]poolItem, len(leafs))
 	for i, lf := range leafs {
-		if limit > 0 && i >= limit {
-			break
+		items[i] = poolItem{i: i, text: lf.Heading + "\n\n" + lf.Text}
+	}
+	embed := func(text string) ([]float64, error) {
+		if model == nil || text == "" {
+			return nil, nil
 		}
-		query := strings.ToValidUTF8(lf.Heading+"\n\n"+lf.Text, "\uFFFD")
-		var emb []float64
-		if model != nil && lf.Text != "" {
-			var err error
-			emb, err = model.Embed(lf.Text)
-			if err != nil {
-				return n, err
-			}
+		return model.Embed(text)
+	}
+	var progress func(int, int)
+	if opt.Progress != nil {
+		progress = opt.Progress.Report
+	}
+	results, err := parallelEmbed(context.Background(), items, embed, opt.Workers, progress)
+	if err != nil {
+		return 0, err
+	}
+
+	inputs := make([]LeafInput, 0, len(results))
+	repos := make([]string, 0, len(results))
+	for i, r := range results {
+		if r.err != nil {
+			return 0, fmt.Errorf("embed %d: %w", i, r.err)
 		}
+		lf := leafs[r.i]
 		typ := lf.Type
 		if typ == "" {
 			typ = "reference"
@@ -261,18 +310,55 @@ func WriteCorpus(conn *lbug.Connection, leafs []CorpusLeaf, model *StaticModel,
 		if how == "" {
 			how = "kb/index"
 		}
-		id, err := UpsertLeaf(conn, LeafInput{
-			Text: query, Root: "info", Confidence: "confirmed",
-			Source: lf.Source, SourceRev: "working-tree", How: how,
-			Loc: lf.Source, Type: typ, Embedding: emb,
+		inputs = append(inputs, LeafInput{
+			Text: strings.ToValidUTF8(items[r.i].text, "\uFFFD"), Root: "info",
+			Confidence: "confirmed", Source: lf.Source, SourceRev: "working-tree",
+			How: how, Loc: lf.Source, Type: typ, Embedding: r.emb, ValidFrom: lf.Date,
 		})
+		repos = append(repos, lf.Repo)
+	}
+
+	n := 0
+	for _, b := range chunkBounds(len(inputs), opt.Batch) {
+		ids, err := AddLeafs(conn, inputs[b[0]:b[1]])
 		if err != nil {
 			return n, err
 		}
-		if _, err := LinkFromFile(conn, id, lf.Source, lf.Repo, ""); err != nil {
-			return n, err
+		for j, id := range ids {
+			in := inputs[b[0]+j]
+			if _, err := LinkFromFile(conn, id, in.Source, repos[b[0]+j], ""); err != nil {
+				return n, err
+			}
+			n++
+		}
+		if opt.Progress != nil {
+			opt.Progress.Report(n, len(inputs))
 		}
-		n++
+	}
+	if opt.Progress != nil {
+		opt.Progress.Finish(n, len(leafs))
 	}
 	return n, nil
 }
+
+// existingLeafIDSet returns the set of all current leaf ids (for resume).
+func existingLeafIDSet(conn *lbug.Connection) (map[string]bool, error) {
+	res, err := conn.Query("MATCH (l:Leaf) RETURN l.id")
+	if err != nil {
+		return nil, err
+	}
+	defer res.Close()
+	set := map[string]bool{}
+	for res.HasNext() {
+		row, err := res.Next()
+		if err != nil {
+			return nil, err
+		}
+		vals, err := row.GetAsSlice()
+		if err != nil || len(vals) < 1 {
+			continue
+		}
+		set[fmt.Sprint(vals[0])] = true
+	}
+	return set, nil
+}