brain

Documentation

Overview

Потоковая/чанкованная запись корпуса (issue #237): корпус не накапливается в одном срезе — CountCorpus считает leafs, WriteCorpusChunked стримит источники и пишет чанками по size. cgo-free: БД-запись (WriteCorpus) приходит извне через write, поэтому логика тестируется без Ladybug.

Parallel embedding and batching helpers for corpus writes. Kept cgo-free so they unit-test without the Ladybug library.

Package brain is deduction search over Ladybug (FTS + HNSW). Query/embed code that needs cgo lives behind the system_ladybug tag.

Freshness state for the gator→graph→index cycle (issue #292).

The periodic compose service writes a small YAML state file after each successful cycle; the read contract (/stats, scripts/stack/status) reads it back and compares it with the kb.lbug mtime and the newest gator parquet pack. Cgo-free so the read path unit-tests without the Ladybug library.

Live-process detection for the db file. Cgo-free so it unit-tests without the Ladybug library.

YAML emitter ported from bin/kb/yamlout.py — preserves insertion order.

Index

• Constants
• func BrainAPIAlive(addr string) bool
• func Configure(c *config.Config)
• func FreshnessPath(root string) string
• func LiveHolders(path string) ([]string, error)
• func SaveFreshness(root string, f Freshness) error
• func WriteCorpusChunked(ctx context.Context, sources []contract.Source, size, limit int, ...) (int, error)
• type CommitInput
• type CorpusStats
  • func CountCorpus(ctx context.Context, sources []contract.Source, limit int) (CorpusStats, error)
• type Dict
• type Freshness
  • func LoadFreshness(root string) Freshness
• type FreshnessView
  • func ViewFreshness(root, kbPath string) FreshnessView
  • func ViewFreshnessForDB(dbPath string) FreshnessView
• type Hit
• type KV
• type MessageInput
• type ProgressReporter
  • func NewProgressReporter(w io.Writer, interval time.Duration) *ProgressReporter
  • func (r *ProgressReporter) Finish(done, total int)
  • func (r *ProgressReporter) Report(done, total int)

Constants

const DefaultStaleAfter = 26 * time.Hour

DefaultStaleAfter is the fallback staleness horizon when the state file does not carry one: a cycle is expected well inside a day, so a day plus slack flags a silently broken loop.

const ModelID = "minishlab/potion-multilingual-128M"

Functions

func BrainAPIAlive

func BrainAPIAlive(addr string) bool

BrainAPIAlive reports whether a brain HTTP API answers /stats on addr (e.g. "127.0.0.1:8630"). The compose brain runs in its own container as another uid, so its fds are invisible to LiveHolders from the host; a healthy /stats response is the reliable cross-namespace signal.

func Configure

func Configure(c *config.Config)

Configure sets the process-wide typed config used by the brain services and forwards the SearXNG settings to the web-search defaults.

func FreshnessPath

func FreshnessPath(root string) string

FreshnessPath resolves the state file under a repo root.

func LiveHolders

func LiveHolders(path string) ([]string, error)

LiveHolders returns one "pid N (cmdline)" string per process currently holding path open. Detection is by device+inode of each /proc/<pid>/fd target, so holders are found across bind mounts and container PID namespaces: a compose brain service holding /data/var/kb.lbug is flagged when a host-side run checks the same file at ~/2dph/var/kb.lbug. Holders of a deleted inode whose path (in their own namespace) equals path are also reported — that state means a server is up but serving stale data.

Best effort: unreadable /proc entries are skipped, and on non-Linux systems the result is always empty.

func SaveFreshness

func SaveFreshness(root string, f Freshness) error

SaveFreshness writes the state file atomically (tmp + rename) so a reader never sees a half-written file. YAML, not JSON (repo convention).

func WriteCorpusChunked

func WriteCorpusChunked(ctx context.Context, sources []contract.Source, size, limit int, stats CorpusStats, write func(chunk []contract.Leaf, base, total int) (int, error)) (int, error)

WriteCorpusChunked стримит источники и пишет leafs чанками по size (<=0 → 2048), не удерживая корпус целиком: чанк заполняется из стрима, отдаётся write, буфер освобождается. Порядок источников и leafs сохраняется (по первым вхождениям). limit>0 обрывает стрим после limit стримнутых leafs (семантика --limit, применяется до чанков). stats — результат CountCorpus (pass 1): Total (уникальных) идёт в сквозной прогресс, BySource — в отчёт.

Кросс-чанковый dedup (issue #248, A1) включён ВСЕГДА (не только --skip): leaf с ContentHash, уже виденным в этом прогоне, пропускается — до эмбеддинга/записи не доходит. Детерминизм финального набора не меняется: id = ContentHash, MERGE-семантика БД схлопнула бы дубликаты к тому же id, здесь они отсекаются раньше (и не делают C-работу). При --skip поверх dedup'а остаётся existing-сет (WriteOptions.Existing / filterExistingLeafs): уже существующие в БД и так пропускаются.

write получает (чанк, base=уже записанные до чанка leafs, total=уникальных по pass 1) и возвращает число записанных; base/total нужны WriteCorpus для сквозного прогресса (done не сбрасывается на чанк).

Types

type CommitInput

type CommitInput struct {
	ID      string // канон узла: repo:sha
	Repo    string // имя репозитория (как leaf.Repo git-корпуса)
	SHA     string // полный commit sha (40 hex)
	Subject string // первая строка message
	Author  string // имя автора из git author
	Email   string // email автора (lowercase)
	Date    string // авторская дата RFC3339
}

CommitInput — один git-коммит для записи. Коннектор (internal/gitgraph) мапит gitlog.Commit в эти поля: Email уже lowercase, ID = repo:sha.

type CorpusStats

type CorpusStats struct {
	Streamed int            // всего leafs от источников, с дублями (raw стрим)
	Total    int            // уникальных leafs (кросс-чанковый dedup по ContentHash)
	BySource map[string]int // уникальных по имени источника
}

CorpusStats — итоги pass 1 (подсчёт): сколько leafs даёт каждый источник.

issue #248 (A1): счёт уникальный — pass 1 и pass 2 дедуплицируют одинаково (см. leafCounter), поэтому Total (уникальных) честно равен числу записей в БД и служит total сквозного прогресса; Streamed показывает «сырой» стрим (307k streamed → ~105k unique на полном rebuild) для прозрачности отчёта.

func CountCorpus

func CountCorpus(ctx context.Context, sources []contract.Source, limit int) (CorpusStats, error)

CountCorpus стримит источники в порядке списка и считает уникальные leafs, не накапливая их. Pass 1 для dry-run и для сквозного total/по-источникам. Тот же limit, что у WriteCorpusChunked, даёт число, равное записанному (dry-run == фактической записи, acceptance #248).

type Dict

type Dict []KV

Dict is an ordered map (slice of KV).

type Freshness

type Freshness struct {
	ImportAt   string   `yaml:"import_at,omitempty"`
	IndexAt    string   `yaml:"index_at,omitempty"`
	KBMtime    string   `yaml:"kb_mtime,omitempty"`
	PackMtime  string   `yaml:"pack_mtime,omitempty"`
	Channels   []string `yaml:"channels,omitempty"`
	StaleAfter string   `yaml:"stale_after,omitempty"`
	LastError  string   `yaml:"last_error,omitempty"`
	UpdatedAt  string   `yaml:"updated_at,omitempty"`
}

Freshness is the persisted last-good state of the import/index cycle. Times are RFC3339 UTC strings, empty when that stage never ran.

func LoadFreshness

func LoadFreshness(root string) Freshness

LoadFreshness reads the state file; a missing or malformed file yields the zero state (staleness is then reported, never a hard error).

type FreshnessView

type FreshnessView struct {
	IndexAt    string `json:"index_at" yaml:"index_at"`
	ImportAt   string `json:"import_at" yaml:"import_at"`
	KBMtime    string `json:"kb_mtime" yaml:"kb_mtime"`
	PackMtime  string `json:"pack_mtime" yaml:"pack_mtime"`
	StaleAfter string `json:"stale_after" yaml:"stale_after"`
	Stale      bool   `json:"stale" yaml:"stale"`
	LastError  string `json:"last_error,omitempty" yaml:"last_error,omitempty"`
}

FreshnessView is the derived read-contract shape (state + live kb mtime).

func ViewFreshness

func ViewFreshness(root, kbPath string) FreshnessView

ViewFreshness merges the persisted state with the live kb.lbug mtime and reports staleness. kbPath empty defaults to <root>/var/kb.lbug.

func ViewFreshnessForDB

func ViewFreshnessForDB(dbPath string) FreshnessView

ViewFreshnessForDB derives the read-contract view for a kb.lbug path. The repo root is the grandparent of the db (var/kb.lbug), where the state file and the pack mtime live.

type Hit

type Hit = rank.Hit

Hit is the search hit type; ranking lives in package rank so CI can test it without the native ladybug library.

type KV

type KV struct {
	K string
	V any
}

KV is an ordered key-value pair for maps.

type MessageInput

type MessageInput struct {
	canon.Message
	Folder   string
	Subject  string
	GatorRef string
}

MessageInput — canon.Message + атрибуты gator-узла, которых нет в canon #99 (folder/subject из row gator kind=mail и gator_ref deeplink `kind=mail#v-<hash8>`). canon.Message встраивается as-is: рёбра берутся из его Edges(), ID/адреса — из полей.

type ProgressReporter

type ProgressReporter struct {
	// contains filtered or unexported fields
}

ProgressReporter prints a human progress line (done, rate, ETA) to w at most once per interval. Thread-safe; zero value is ready to use.

func NewProgressReporter

func NewProgressReporter(w io.Writer, interval time.Duration) *ProgressReporter

NewProgressReporter returns a reporter that prints to w every interval.

func (*ProgressReporter) Finish

func (r *ProgressReporter) Finish(done, total int)

Finish prints a final line regardless of interval.

func (*ProgressReporter) Report

func (r *ProgressReporter) Report(done, total int)

Report records one completion and prints a line if the interval elapsed.