diff --git a/docs/24-knowledge-vault.md b/docs/24-knowledge-vault.md index 76d6c94c..ec2586fd 100644 --- a/docs/24-knowledge-vault.md +++ b/docs/24-knowledge-vault.md @@ -86,7 +86,8 @@ Document registry: metadata pointers. Content lives on filesystem; registry hold | `doc_type` | TEXT | context, memory, note, skill, episodic | | `content_hash` | TEXT | SHA-256 of file content (detects changes) | | `embedding` | vector(1536) | pgvector: semantic similarity | -| `tsv` | tsvector | Generated: FTS index on title+path | +| `tsv` | tsvector | Generated: FTS index on title+path+summary | +| `body_indexed_hash` | TEXT | `content_hash` the body chunks were built from (NULL = not chunked yet) | | `metadata` | JSONB | Optional custom fields | | `created_at`, `updated_at` | TIMESTAMPTZ | Timestamps | | **Unique constraint** | (agent_id, scope, path) | One doc per path per scope | @@ -99,6 +100,19 @@ Document registry: metadata pointers. Content lives on filesystem; registry hold - `idx_vault_docs_embedding` — HNSW vector (semantic search) - `idx_vault_docs_tsv` — GIN FTS index (keyword search) +### vault_document_chunks + +The body of each text document, split into ~1000-char chunks (200-char overlap), so search reaches content that the auto-summary leaves out. The enrich worker rebuilds the chunks from the workspace file whenever `content_hash` moves past `body_indexed_hash`. `POST /v1/vault/rescan` also backfills docs that have no chunks yet. Media and `document` types are not chunked. + +| Column | Type | Notes | +|--------|------|-------| +| `document_id` | UUID | Parent doc (cascade delete) | +| `chunk_index` | INT | Position in the body | +| `start_line`, `end_line` | INT | Line range in the file | +| `text` | TEXT | Chunk content | +| `embedding` | vector(1536) | Only the first 64 chunks of a doc are embedded | +| `tsv` | tsvector | Generated: FTS on `text` ('simple' config) | + ### vault_links Bidirectional links between documents (wikilinks, explicit references). @@ -192,8 +206,8 @@ Hybrid search integrates vault FTS, vector embeddings, episodic memory, and know `VaultStore.Search(ctx, opts VaultSearchOptions)` on single vault: -- **FTS**: PostgreSQL `plainto_tsquery()` on tsv (title+path keywords) -- **Vector**: pgvector cosine similarity on embedding (semantic) +- **FTS**: matches any query word (OR), so natural-language questions work; docs matching more words rank higher. Runs on the doc `tsv` and on body chunks, and a doc scores by its best hit +- **Vector**: pgvector cosine similarity on the doc embedding and on chunk embeddings, best hit per doc - **Combined scoring**: Normalize each method's scores (0–1), then apply query-time weights - **Results:** Top N documents with score diff --git a/internal/http/vault_handlers.go b/internal/http/vault_handlers.go index 9bb3f868..93599bdf 100644 --- a/internal/http/vault_handlers.go +++ b/internal/http/vault_handlers.go @@ -218,6 +218,11 @@ func (h *VaultHandler) handleRescan(w http.ResponseWriter, r *http.Request) { } } + // Backfill body chunks for docs indexed before chunking existed, or whose + // file changed without the chunks catching up. Detached from the request + // so a big vault is not cut off by the rescan timeout. + go vault.IndexStaleBodies(context.WithoutCancel(r.Context()), h.store, tenantID, wsPath) + if h.enrichProgress != nil && total > 0 { h.enrichProgress.Start(total, store.TenantIDFromContext(r.Context())) } diff --git a/internal/store/pg/vault_chunks.go b/internal/store/pg/vault_chunks.go new file mode 100644 index 00000000..0906c373 --- /dev/null +++ b/internal/store/pg/vault_chunks.go @@ -0,0 +1,121 @@ +package pg + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log/slog" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// vaultMaxEmbeddedChunks caps how many chunks of one document get an embedding. +// Every chunk is still full-text indexed; chunks past the cap only lack a vector. +const vaultMaxEmbeddedChunks = 64 + +// ReplaceDocumentChunks swaps the body chunks of a document and embeds them. +// No-op when the stored chunks were already built from contentHash. +func (s *PGVaultStore) ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []store.VaultChunk) error { + tid, err := parseUUID(tenantID) + if err != nil { + return fmt.Errorf("vault replace chunks: tenant: %w", err) + } + did, err := parseUUID(docID) + if err != nil { + return fmt.Errorf("vault replace chunks: doc: %w", err) + } + + var indexedHash sql.NullString + err = s.db.QueryRowContext(ctx, + `SELECT body_indexed_hash FROM vault_documents WHERE id = $1 AND tenant_id = $2`, + did, tid, + ).Scan(&indexedHash) + if err != nil { + return fmt.Errorf("vault replace chunks: fetch doc: %w", err) + } + if indexedHash.Valid && indexedHash.String == contentHash { + return nil + } + + // Embed outside the transaction; a failed embed still leaves the chunks + // searchable by keyword. + embeddings := make([]*string, len(chunks)) + if s.embProvider != nil && len(chunks) > 0 { + n := min(len(chunks), vaultMaxEmbeddedChunks) + texts := make([]string, n) + for i := range n { + texts[i] = chunks[i].Text + } + vecs, embErr := s.embProvider.Embed(ctx, texts) + if embErr != nil { + slog.Warn("vault.chunks: embed", "doc", docID, "err", embErr) + } + for i := range min(len(vecs), n) { + v := vectorToString(vecs[i]) + embeddings[i] = &v + } + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("vault replace chunks: begin: %w", err) + } + defer tx.Rollback() + + if _, err := tx.ExecContext(ctx, + `DELETE FROM vault_document_chunks WHERE document_id = $1 AND tenant_id = $2`, did, tid, + ); err != nil { + return fmt.Errorf("vault replace chunks: delete: %w", err) + } + for i, c := range chunks { + if _, err := tx.ExecContext(ctx, ` + INSERT INTO vault_document_chunks + (id, tenant_id, document_id, chunk_index, start_line, end_line, text, embedding) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`, + uuid.Must(uuid.NewV7()), tid, did, c.Index, c.StartLine, c.EndLine, c.Text, embeddings[i], + ); err != nil { + return fmt.Errorf("vault replace chunks: insert: %w", err) + } + } + res, err := tx.ExecContext(ctx, + `UPDATE vault_documents SET body_indexed_hash = $1 WHERE id = $2 AND tenant_id = $3`, + contentHash, did, tid, + ) + if err != nil { + return fmt.Errorf("vault replace chunks: mark indexed: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return errors.New("vault replace chunks: document deleted") + } + return tx.Commit() +} + +// ListDocsNeedingBodyIndex returns text documents whose chunks are missing or stale. +// Media and binary documents are skipped: they have no text body to index. +func (s *PGVaultStore) ListDocsNeedingBodyIndex(ctx context.Context, tenantID string, limit int) ([]store.VaultDocument, error) { + tid, err := parseUUID(tenantID) + if err != nil { + return nil, fmt.Errorf("vault list needing body index: tenant: %w", err) + } + + q := `SELECT id, tenant_id, agent_id, team_id, chat_id, scope, custom_scope, path, path_basename, title, doc_type, + content_hash, summary, metadata, created_at, updated_at + FROM vault_documents + WHERE tenant_id = $1 AND doc_type NOT IN ('media', 'document') + AND body_indexed_hash IS DISTINCT FROM content_hash + ORDER BY created_at ASC` + args := []any{tid} + if limit > 0 { + q += " LIMIT $2" + args = append(args, limit) + } + + var rows []vaultDocRow + if err := pkgSqlxDB.SelectContext(ctx, &rows, q, args...); err != nil { + return nil, fmt.Errorf("vault.list_needing_body_index: %w", err) + } + return vaultDocRowsToDocs(rows), nil +} diff --git a/internal/store/pg/vault_documents.go b/internal/store/pg/vault_documents.go index a8ac8ef9..e7220d58 100644 --- a/internal/store/pg/vault_documents.go +++ b/internal/store/pg/vault_documents.go @@ -440,7 +440,7 @@ func (s *PGVaultStore) UpdateHash(ctx context.Context, tenantID, id, newHash str // UpdateSummaryAndReembed updates summary and re-generates embedding from title+path+summary. // UpdateSummaryAndReembed and FindSimilarDocs moved to vault_documents_enrichment.go. -// Search performs hybrid FTS + vector search on vault_documents. +// Search performs hybrid FTS + vector search on vault documents and their body chunks. func (s *PGVaultStore) Search(ctx context.Context, opts store.VaultSearchOptions) ([]store.VaultSearchResult, error) { tid, err := parseUUID(opts.TenantID) if err != nil { @@ -451,10 +451,14 @@ func (s *PGVaultStore) Search(ctx context.Context, opts store.VaultSearchOptions return nil, fmt.Errorf("vault search: agent: %w", err) } - // Build team filter for search sub-queries. - tf := buildSearchTeamFilter(opts.TeamID, opts.TeamIDs) - // Chat-scope filter (applies only when team is isolated + chat_id non-nil/non-empty). - cf := buildSearchChatFilter(opts.ChatID, opts.TeamIsolated) + f := vaultSearchScope{ + agentID: aid, + team: buildSearchTeamFilter(opts.TeamID, opts.TeamIDs), + // Chat-scope filter (applies only when team is isolated + chat_id non-nil/non-empty). + chat: buildSearchChatFilter(opts.ChatID, opts.TeamIsolated), + scope: opts.Scope, + docTypes: opts.DocTypes, + } maxResults := opts.MaxResults if maxResults <= 0 { @@ -462,7 +466,7 @@ func (s *PGVaultStore) Search(ctx context.Context, opts store.VaultSearchOptions } // FTS search - ftsResults, err := s.ftsSearch(ctx, opts.Query, tid, aid, tf, cf, opts.Scope, opts.DocTypes, maxResults*2) + ftsResults, err := s.ftsSearch(ctx, opts.Query, tid, f, maxResults*2) if err != nil { return nil, err } @@ -473,7 +477,7 @@ func (s *PGVaultStore) Search(ctx context.Context, opts store.VaultSearchOptions vecs, embErr := s.embProvider.Embed(ctx, []string{opts.Query}) if embErr == nil && len(vecs) > 0 { var vecErr error - vecResults, vecErr = s.vectorSearch(ctx, vecs[0], tid, aid, tf, cf, opts.Scope, opts.DocTypes, maxResults*2) + vecResults, vecErr = s.vectorSearch(ctx, vecs[0], tid, f, maxResults*2) if vecErr != nil { slog.Debug("vault.vector_search_fallback", "err", vecErr) vecResults = nil @@ -569,35 +573,55 @@ func (cf searchChatFilter) append(q string, args []any, p int) (string, []any, i return q, args, p } -func (s *PGVaultStore) ftsSearch(ctx context.Context, query string, tenantID uuid.UUID, agentID *uuid.UUID, tf searchTeamFilter, cf searchChatFilter, scope string, docTypes []string, limit int) ([]store.VaultSearchResult, error) { - q := `SELECT id, tenant_id, agent_id, team_id, chat_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, - ts_rank(tsv, plainto_tsquery('simple', $1)) AS score - FROM vault_documents - WHERE tenant_id = $2 AND tsv @@ plainto_tsquery('simple', $1)` - args := []any{query, tenantID} - p := 3 +// vaultAnyTermQuery matches docs containing any word of $1. plainto_tsquery +// ANDs every word, and the 'simple' config keeps filler words like "what" or +// "the", so a plain question never matched anything. Lexemes come out of +// plainto_tsquery already quoted and never contain spaces, so swapping the +// operators is safe. ts_rank still puts docs matching more words first. +const vaultAnyTermQuery = `replace(plainto_tsquery('simple', $1)::text, ' & ', ' | ')::tsquery` - if agentID != nil { +// vaultDocCols is the vault_documents column list scanned into vaultDocRow, +// qualified for queries that join the chunks table. +const vaultDocCols = `d.id, d.tenant_id, d.agent_id, d.team_id, d.chat_id, d.scope, d.custom_scope, d.path, d.path_basename, + d.title, d.doc_type, d.content_hash, d.summary, d.metadata, d.created_at, d.updated_at` + +// vaultSearchScope holds the visibility filters shared by every search sub-query. +type vaultSearchScope struct { + agentID *uuid.UUID + team searchTeamFilter + chat searchChatFilter + scope string + docTypes []string +} + +// append adds the filters to q. Columns are unqualified; that works for the +// chunk joins too because none of them exist on vault_document_chunks. +func (f vaultSearchScope) append(q string, args []any, p int) (string, []any, int) { + if f.agentID != nil { q += fmt.Sprintf(" AND (agent_id = $%d OR agent_id IS NULL)", p) - args = append(args, *agentID) + args = append(args, *f.agentID) p++ } - - q, args, p = tf.append(q, args, p) - q, args, p = cf.append(q, args, p) - - if scope != "" { + q, args, p = f.team.append(q, args, p) + q, args, p = f.chat.append(q, args, p) + if f.scope != "" { q += fmt.Sprintf(" AND scope = $%d", p) - args = append(args, scope) + args = append(args, f.scope) p++ } - if len(docTypes) > 0 { + if len(f.docTypes) > 0 { q += fmt.Sprintf(" AND doc_type = ANY($%d)", p) - args = append(args, pqStringArray(docTypes)) + args = append(args, pqStringArray(f.docTypes)) p++ } + return q, args, p +} - q += fmt.Sprintf(" ORDER BY score DESC LIMIT $%d", p) +// selectSearchRows applies the scope filters, appends tail (which gets the +// LIMIT placeholder number) and runs the query. +func selectSearchRows(ctx context.Context, q string, args []any, f vaultSearchScope, tail string, limit int) ([]store.VaultSearchResult, error) { + q, args, p := f.append(q, args, len(args)+1) + q += fmt.Sprintf(tail, p) args = append(args, limit) var scanned []vaultSearchRow @@ -607,43 +631,77 @@ func (s *PGVaultStore) ftsSearch(ctx context.Context, query string, tenantID uui return vaultSearchRowsToResults(scanned, "vault"), nil } -func (s *PGVaultStore) vectorSearch(ctx context.Context, embedding []float32, tenantID uuid.UUID, agentID *uuid.UUID, tf searchTeamFilter, cf searchChatFilter, scope string, docTypes []string, limit int) ([]store.VaultSearchResult, error) { - vecStr := vectorToString(embedding) - q := `SELECT id, tenant_id, agent_id, team_id, chat_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, - 1 - (embedding <=> $1) AS score +// ftsSearch matches the query against the document header (title, path, +// summary) and the body chunks, scoring each doc by its best hit. +func (s *PGVaultStore) ftsSearch(ctx context.Context, query string, tenantID uuid.UUID, f vaultSearchScope, limit int) ([]store.VaultSearchResult, error) { + docHits, err := selectSearchRows(ctx, `SELECT id, tenant_id, agent_id, team_id, chat_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, + ts_rank(tsv, `+vaultAnyTermQuery+`) AS score FROM vault_documents - WHERE tenant_id = $2 AND embedding IS NOT NULL` - args := []any{vecStr, tenantID} - p := 3 - - if agentID != nil { - q += fmt.Sprintf(" AND (agent_id = $%d OR agent_id IS NULL)", p) - args = append(args, *agentID) - p++ - } - - q, args, p = tf.append(q, args, p) - q, args, p = cf.append(q, args, p) - - if scope != "" { - q += fmt.Sprintf(" AND scope = $%d", p) - args = append(args, scope) - p++ - } - if len(docTypes) > 0 { - q += fmt.Sprintf(" AND doc_type = ANY($%d)", p) - args = append(args, pqStringArray(docTypes)) - p++ - } - - q += fmt.Sprintf(" ORDER BY embedding <=> $1 LIMIT $%d", p) - args = append(args, limit) - - var scanned []vaultSearchRow - if err := pkgSqlxDB.SelectContext(ctx, &scanned, q, args...); err != nil { + WHERE tenant_id = $2 AND tsv @@ `+vaultAnyTermQuery, + []any{query, tenantID}, f, " ORDER BY score DESC LIMIT $%d", limit) + if err != nil { return nil, err } - return vaultSearchRowsToResults(scanned, "vault"), nil + + // GROUP BY d.id covers the other d columns since id is the primary key. + chunkHits, err := selectSearchRows(ctx, `SELECT `+vaultDocCols+`, + MAX(ts_rank(c.tsv, `+vaultAnyTermQuery+`)) AS score + FROM vault_document_chunks c + JOIN vault_documents d ON d.id = c.document_id + WHERE c.tenant_id = $2 AND c.tsv @@ `+vaultAnyTermQuery, + []any{query, tenantID}, f, " GROUP BY d.id ORDER BY score DESC LIMIT $%d", limit) + if err != nil { + return nil, err + } + return bestHitPerDoc(docHits, chunkHits), nil +} + +// vectorSearch ranks docs by cosine similarity of the header embedding or of +// the closest body chunk, whichever is higher. +func (s *PGVaultStore) vectorSearch(ctx context.Context, embedding []float32, tenantID uuid.UUID, f vaultSearchScope, limit int) ([]store.VaultSearchResult, error) { + vecStr := vectorToString(embedding) + docHits, err := selectSearchRows(ctx, `SELECT id, tenant_id, agent_id, team_id, chat_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, + 1 - (embedding <=> $1) AS score + FROM vault_documents + WHERE tenant_id = $2 AND embedding IS NOT NULL`, + []any{vecStr, tenantID}, f, " ORDER BY embedding <=> $1 LIMIT $%d", limit) + if err != nil { + return nil, err + } + + // One doc can own several of the nearest chunks, so fetch extra rows to + // still end up with enough distinct docs after bestHitPerDoc. + chunkHits, err := selectSearchRows(ctx, `SELECT `+vaultDocCols+`, + 1 - (c.embedding <=> $1) AS score + FROM vault_document_chunks c + JOIN vault_documents d ON d.id = c.document_id + WHERE c.tenant_id = $2 AND c.embedding IS NOT NULL`, + []any{vecStr, tenantID}, f, " ORDER BY c.embedding <=> $1 LIMIT $%d", limit*3) + if err != nil { + return nil, err + } + return bestHitPerDoc(docHits, chunkHits), nil +} + +// bestHitPerDoc keeps the highest score seen for each document, sorted by +// score. mergeResults adds scores per doc, so duplicates must go first. +func bestHitPerDoc(lists ...[]store.VaultSearchResult) []store.VaultSearchResult { + best := make(map[string]int) + var out []store.VaultSearchResult + for _, list := range lists { + for _, r := range list { + if i, ok := best[r.Document.ID]; ok { + if r.Score > out[i].Score { + out[i] = r + } + continue + } + best[r.Document.ID] = len(out) + out = append(out, r) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Score > out[j].Score }) + return out } // vaultSearchRowsToResults converts a slice of vaultSearchRow to store.VaultSearchResult. diff --git a/internal/store/sqlitestore/vault_documents.go b/internal/store/sqlitestore/vault_documents.go index b1fbf8c6..10fedfa3 100644 --- a/internal/store/sqlitestore/vault_documents.go +++ b/internal/store/sqlitestore/vault_documents.go @@ -345,6 +345,16 @@ func (s *SQLiteVaultStore) FindSimilarDocs(ctx context.Context, tenantID, agentI return nil, nil } +// ReplaceDocumentChunks is a no-op in SQLite (search is LIKE on title/path, no FTS). +func (s *SQLiteVaultStore) ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []store.VaultChunk) error { + return nil +} + +// ListDocsNeedingBodyIndex is a no-op in SQLite (body chunks are not stored). +func (s *SQLiteVaultStore) ListDocsNeedingBodyIndex(ctx context.Context, tenantID string, limit int) ([]store.VaultDocument, error) { + return nil, nil +} + // Search performs LIKE-based search on vault documents (no FTS/vector in lite). func (s *SQLiteVaultStore) Search(ctx context.Context, opts store.VaultSearchOptions) ([]store.VaultSearchResult, error) { query := opts.Query diff --git a/internal/store/vault_store.go b/internal/store/vault_store.go index dec586ac..55aa6623 100644 --- a/internal/store/vault_store.go +++ b/internal/store/vault_store.go @@ -67,6 +67,14 @@ type VaultSearchOptions struct { MinScore float64 // default 0.0 } +// VaultChunk is one piece of a vault document body, indexed for search. +type VaultChunk struct { + Index int + StartLine int + EndLine int + Text string +} + // VaultListOptions configures a list query for vault documents. type VaultListOptions struct { TeamID *string // nil = no filter, ptr-to-empty = personal (NULL team_id), ptr-to-uuid = specific team @@ -150,6 +158,13 @@ type VaultStore interface { // FindSimilarDocs finds documents with similar embeddings to the given docID. // Returns top-N neighbors excluding the source doc. Score = cosine similarity. FindSimilarDocs(ctx context.Context, tenantID, agentID, docID string, limit int) ([]VaultSearchResult, error) + // ReplaceDocumentChunks swaps the indexed body chunks of a document and + // embeds them. It is a no-op when the chunks were already built from + // contentHash, so unchanged files are not re-embedded. + ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []VaultChunk) error + // ListDocsNeedingBodyIndex returns text documents whose body chunks are + // missing or were built from an older content_hash. + ListDocsNeedingBodyIndex(ctx context.Context, tenantID string, limit int) ([]VaultDocument, error) // BatchFindByDelegationIDs returns vault docs sharing any of the given // delegation_ids in their metadata, keyed by delegation_id. Each // delegation's bucket is capped at `limit` (ordered by created_at DESC). diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index f143a0fd..6a98870b 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 97 +const RequiredSchemaVersion uint = 98 diff --git a/internal/vault/body_index.go b/internal/vault/body_index.go new file mode 100644 index 00000000..64cb264a --- /dev/null +++ b/internal/vault/body_index.go @@ -0,0 +1,84 @@ +package vault + +import ( + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/nextlevelbuilder/goclaw/internal/memory" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +const ( + // Same defaults as memory chunks. + vaultBodyChunkLen = 1000 + vaultBodyChunkOverlap = 200 + // SHORTCUT: bodies past 2 MiB are indexed only up to the cap. Stream the + // file into chunks if people start keeping large exports in the vault. + vaultBodyMaxBytes = 2 << 20 +) + +// chunkBody splits a file body into search chunks. Postgres TEXT rejects NUL +// bytes and invalid UTF-8, and a cut at vaultBodyMaxBytes can land mid-rune. +func chunkBody(raw []byte) []store.VaultChunk { + if len(raw) > vaultBodyMaxBytes { + raw = raw[:vaultBodyMaxBytes] + } + text := strings.ToValidUTF8(strings.ReplaceAll(string(raw), "\x00", ""), "") + parts := memory.ChunkText(text, vaultBodyChunkLen, vaultBodyChunkOverlap) + chunks := make([]store.VaultChunk, len(parts)) + for i, p := range parts { + chunks[i] = store.VaultChunk{Index: i, StartLine: p.StartLine, EndLine: p.EndLine, Text: p.Text} + } + return chunks +} + +// indexBody rebuilds the body chunks of one text document from its workspace +// file. The store skips the write when the chunks already match contentHash. +// Failures are logged here; the error only tells the caller to retry later. +func indexBody(ctx context.Context, vs store.VaultStore, tenantID, docID, contentHash, fullPath string) error { + raw, err := os.ReadFile(fullPath) + if err != nil { + slog.Warn("vault.body_index: read_file", "path", fullPath, "err", err) + return err + } + if err := vs.ReplaceDocumentChunks(ctx, tenantID, docID, contentHash, chunkBody(raw)); err != nil { + slog.Warn("vault.body_index: replace_chunks", "doc", docID, "err", err) + return err + } + return nil +} + +// bodyIndexRuns keeps one IndexStaleBodies pass per tenant. +var bodyIndexRuns sync.Map + +// IndexStaleBodies chunks every text document whose body index is missing or +// older than its content. Docs indexed before chunking existed are never sent +// to the enrich worker again because their summary is already set, so rescan +// calls this to backfill them. No LLM calls: only reads and embeddings. +// Returns false when a pass for the tenant is already running. +func IndexStaleBodies(ctx context.Context, vs store.VaultStore, tenantID, workspace string) bool { + if _, running := bodyIndexRuns.LoadOrStore(tenantID, struct{}{}); running { + return false + } + defer bodyIndexRuns.Delete(tenantID) + + docs, err := vs.ListDocsNeedingBodyIndex(ctx, tenantID, 0) + if err != nil { + slog.Warn("vault.body_index: list", "tenant", tenantID, "err", err) + return true + } + for _, doc := range docs { + if ctx.Err() != nil { + return true + } + _ = indexBody(ctx, vs, tenantID, doc.ID, doc.ContentHash, filepath.Join(workspace, doc.Path)) + } + if len(docs) > 0 { + slog.Info("vault.body_index: backfilled", "tenant", tenantID, "count", len(docs)) + } + return true +} diff --git a/internal/vault/body_index_test.go b/internal/vault/body_index_test.go new file mode 100644 index 00000000..6c4d5cb4 --- /dev/null +++ b/internal/vault/body_index_test.go @@ -0,0 +1,66 @@ +package vault + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type fakeVaultStoreBody struct { + store.VaultStore + stale []store.VaultDocument + replaced map[string][]store.VaultChunk // docID -> chunks + hashes map[string]string // docID -> contentHash passed in +} + +func (f *fakeVaultStoreBody) ListDocsNeedingBodyIndex(ctx context.Context, tenantID string, limit int) ([]store.VaultDocument, error) { + return f.stale, nil +} + +func (f *fakeVaultStoreBody) ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []store.VaultChunk) error { + f.replaced[docID] = chunks + f.hashes[docID] = contentHash + return nil +} + +func TestIndexStaleBodies_ChunksWorkspaceFiles(t *testing.T) { + ws := t.TempDir() + body := "# Prices\n| Deluxe Ocean Suite | 4,200,000 |\x00\nDeposit 30%\xff\n" + if err := os.WriteFile(filepath.Join(ws, "prices.md"), []byte(body), 0o644); err != nil { + t.Fatal(err) + } + fs := &fakeVaultStoreBody{ + stale: []store.VaultDocument{ + {ID: "d1", Path: "prices.md", ContentHash: "h1"}, + {ID: "d2", Path: "gone.md", ContentHash: "h2"}, + }, + replaced: map[string][]store.VaultChunk{}, + hashes: map[string]string{}, + } + + if !IndexStaleBodies(context.Background(), fs, "t1", ws) { + t.Fatal("first pass should run") + } + + chunks := fs.replaced["d1"] + if len(chunks) != 1 { + t.Fatalf("chunks = %+v, want 1", chunks) + } + want := "# Prices\n| Deluxe Ocean Suite | 4,200,000 |\nDeposit 30%" + if chunks[0].Text != want { + t.Fatalf("chunk text = %q, want %q", chunks[0].Text, want) + } + if chunks[0].StartLine != 1 || chunks[0].EndLine != 4 { + t.Fatalf("lines = %d-%d, want 1-4", chunks[0].StartLine, chunks[0].EndLine) + } + if fs.hashes["d1"] != "h1" { + t.Fatalf("hash = %q, want h1", fs.hashes["d1"]) + } + // A missing file must not mark the doc as indexed. + if _, ok := fs.replaced["d2"]; ok { + t.Fatal("missing file should be skipped") + } +} diff --git a/internal/vault/enrich_body_retry_test.go b/internal/vault/enrich_body_retry_test.go new file mode 100644 index 00000000..a53b26f4 --- /dev/null +++ b/internal/vault/enrich_body_retry_test.go @@ -0,0 +1,107 @@ +package vault + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" + "golang.org/x/sync/semaphore" +) + +// fakeVaultStoreBodyRetry fails ReplaceDocumentChunks a set number of times, +// then stores the chunks. Everything else the enrich pipeline touches is a no-op. +type fakeVaultStoreBodyRetry struct { + store.VaultStore + doc store.VaultDocument + failsLeft int + calls int + chunks []store.VaultChunk +} + +func (f *fakeVaultStoreBodyRetry) GetDocumentsByIDs(ctx context.Context, tenantID string, ids []string) ([]store.VaultDocument, error) { + return []store.VaultDocument{f.doc}, nil +} + +func (f *fakeVaultStoreBodyRetry) ReplaceDocumentChunks(ctx context.Context, tenantID, docID, contentHash string, chunks []store.VaultChunk) error { + f.calls++ + if f.failsLeft > 0 { + f.failsLeft-- + return errors.New("connection reset") + } + f.chunks = chunks + return nil +} + +func (f *fakeVaultStoreBodyRetry) UpdateSummaryAndReembed(ctx context.Context, tenantID, docID, summary string) error { + return nil +} + +func (f *fakeVaultStoreBodyRetry) FindSimilarDocs(ctx context.Context, tenantID, agentID, docID string, limit int) ([]store.VaultSearchResult, error) { + return nil, nil +} + +func (f *fakeVaultStoreBodyRetry) GetDocumentByID(ctx context.Context, tenantID, id string) (*store.VaultDocument, error) { + return nil, errors.New("not needed") +} + +// A transient body-index failure must not mark the doc as processed, or the +// next event with the same hash is dropped and the doc never gets chunks. +func TestEnrichWorker_RetriesBodyIndexAfterFailure(t *testing.T) { + ws := t.TempDir() + if err := os.WriteFile(filepath.Join(ws, "prices.md"), []byte("Deluxe Ocean Suite 4,200,000\n"), 0o644); err != nil { + t.Fatal(err) + } + tenantID := providers.MasterTenantID.String() + docID := uuid.NewString() + fs := &fakeVaultStoreBodyRetry{ + doc: store.VaultDocument{ + ID: docID, TenantID: tenantID, Path: "prices.md", DocType: "note", + ContentHash: "h1", Summary: "Hotel room rates.", + }, + failsLeft: 1, + } + reg := providers.NewRegistry(nil) + reg.Register(&mockClassifyProvider{}) + w := &EnrichWorker{ + vault: fs, + registry: reg, + dedup: make(map[string]string), + sem: semaphore.NewWeighted(enrichMaxConcurrent), + progress: NewEnrichProgress(nil), + cancelFuncs: &sync.Map{}, + } + event := eventbus.DomainEvent{Payload: eventbus.VaultDocUpsertedPayload{ + DocID: docID, TenantID: tenantID, Path: "prices.md", ContentHash: "h1", Workspace: ws, + }} + + ctx := context.Background() + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("first Handle: %v", err) + } + if fs.chunks != nil { + t.Fatal("chunks written although ReplaceDocumentChunks failed") + } + + // Same event again, e.g. the next file save or sync tick. + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("second Handle: %v", err) + } + if len(fs.chunks) != 1 || fs.chunks[0].Text != "Deluxe Ocean Suite 4,200,000" { + t.Fatalf("chunks after replay = %+v, want the file body", fs.chunks) + } + + // Once indexed, the dedup gate applies again. + if err := w.Handle(ctx, event); err != nil { + t.Fatalf("third Handle: %v", err) + } + if fs.calls != 2 { + t.Fatalf("ReplaceDocumentChunks calls = %d, want 2", fs.calls) + } +} diff --git a/internal/vault/enrich_worker.go b/internal/vault/enrich_worker.go index 97e3fdc0..af225ae7 100644 --- a/internal/vault/enrich_worker.go +++ b/internal/vault/enrich_worker.go @@ -309,12 +309,6 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD } } - // Resolve provider once per chunk (all items share tenantID) - provider, model := w.resolveProviderForTenant(ctx, tenantID) - if provider == nil { - slog.Warn("vault.enrich: no provider available", "tenant", tenantID) - return - } docIDs := make([]string, len(pending)) for i, item := range pending { docIDs[i] = item.DocID @@ -329,6 +323,26 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD docMap[existingDocs[i].ID] = &existingDocs[i] } + // Body chunks need no LLM, so they are indexed even without a provider. + // Docs whose body failed to index are kept out of the dedup map below, + // otherwise the next event for the same hash would be skipped and the + // doc would stay without chunks until someone runs a rescan. + bodyFailed := make(map[string]bool) + for _, item := range pending { + if doc := docMap[item.DocID]; doc != nil && doc.DocType != "media" && doc.DocType != "document" { + if err := indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)); err != nil { + bodyFailed[item.DocID] = true + } + } + } + + // Resolve provider once per chunk (all items share tenantID) + provider, model := w.resolveProviderForTenant(ctx, tenantID) + if provider == nil { + slog.Warn("vault.enrich: no provider available", "tenant", tenantID) + return + } + var all []prepared for _, item := range pending { existing := docMap[item.DocID] @@ -446,7 +460,9 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD // Phase 4 — Record dedup + wikilinks. for _, r := range embedded { - w.recordDedup(r.payload.DocID, r.payload.ContentHash) + if !bodyFailed[r.payload.DocID] { + w.recordDedup(r.payload.DocID, r.payload.ContentHash) + } w.syncWikilinks(ctx, r.payload) } } diff --git a/migrations/000098_vault_document_chunks.down.sql b/migrations/000098_vault_document_chunks.down.sql new file mode 100644 index 00000000..d11d12c7 --- /dev/null +++ b/migrations/000098_vault_document_chunks.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE vault_documents DROP COLUMN IF EXISTS body_indexed_hash; +DROP TABLE IF EXISTS vault_document_chunks; diff --git a/migrations/000098_vault_document_chunks.up.sql b/migrations/000098_vault_document_chunks.up.sql new file mode 100644 index 00000000..bbcb5f60 --- /dev/null +++ b/migrations/000098_vault_document_chunks.up.sql @@ -0,0 +1,29 @@ +-- Index vault document bodies in chunks (#1554). vault_documents.tsv and +-- embedding only cover title + path + summary, so terms that the auto-summary +-- drops (item names, prices, codes) could never be found by vault_search. +-- +-- Chunks are derived data: the enrich worker rebuilds them from the workspace +-- file. body_indexed_hash records the content_hash the chunks were built from, +-- so unchanged files are not re-chunked or re-embedded, and rows with a NULL +-- or stale hash are picked up again by POST /v1/vault/rescan. +CREATE TABLE IF NOT EXISTS vault_document_chunks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + document_id UUID NOT NULL REFERENCES vault_documents(id) ON DELETE CASCADE, + chunk_index INT NOT NULL, + start_line INT NOT NULL DEFAULT 0, + end_line INT NOT NULL DEFAULT 0, + text TEXT NOT NULL, + embedding vector(1536), + -- 'simple' config (no stemming), same as vault_documents.tsv + tsv tsvector GENERATED ALWAYS AS (to_tsvector('simple', text)) STORED, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (document_id, chunk_index) +); + +CREATE INDEX IF NOT EXISTS idx_vault_chunks_tenant ON vault_document_chunks(tenant_id); +CREATE INDEX IF NOT EXISTS idx_vault_chunks_tsv ON vault_document_chunks USING GIN(tsv); +CREATE INDEX IF NOT EXISTS idx_vault_chunks_embedding ON vault_document_chunks + USING hnsw(embedding vector_cosine_ops); + +ALTER TABLE vault_documents ADD COLUMN IF NOT EXISTS body_indexed_hash TEXT; diff --git a/tests/integration/v3_vault_body_search_test.go b/tests/integration/v3_vault_body_search_test.go new file mode 100644 index 00000000..17652a7d --- /dev/null +++ b/tests/integration/v3_vault_body_search_test.go @@ -0,0 +1,205 @@ +//go:build integration + +package integration + +import ( + "context" + "testing" + + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/store/pg" +) + +// Regression coverage for #1554: search must reach terms that only appear in the +// document body, and natural-language queries must not require every word. + +// seedPriceSheet stores a doc whose summary generalizes away the room names, +// with the full table indexed as body chunks. +func seedPriceSheet(t *testing.T, ctx context.Context, vs *pg.PGVaultStore, tid, aid, path string) *store.VaultDocument { + t.Helper() + doc := makeVaultDoc(tid, aid, path, "Hotel price list") + doc.Summary = "Lists room rates for 6 categories and the deposit policy." + if err := vs.UpsertDocument(ctx, doc); err != nil { + t.Fatalf("UpsertDocument: %v", err) + } + chunks := []store.VaultChunk{ + {Index: 0, StartLine: 1, EndLine: 3, Text: "| Room | Price |\n| Garden Twin | 1,800,000 |\n| Deluxe Ocean Suite | 4,200,000 |"}, + {Index: 1, StartLine: 4, EndLine: 5, Text: "Deposit is 30 percent, refundable up to 7 days before arrival."}, + } + if err := vs.ReplaceDocumentChunks(ctx, tid, doc.ID, doc.ContentHash, chunks); err != nil { + t.Fatalf("ReplaceDocumentChunks: %v", err) + } + return doc +} + +func searchPaths(t *testing.T, vs *pg.PGVaultStore, opts store.VaultSearchOptions) []string { + t.Helper() + if opts.MaxResults == 0 { + opts.MaxResults = 10 + } + results, err := vs.Search(context.Background(), opts) + if err != nil { + t.Fatalf("Search %q: %v", opts.Query, err) + } + paths := make([]string, len(results)) + for i, r := range results { + paths[i] = r.Document.Path + } + return paths +} + +func TestStoreVault_Search_FindsBodyOnlyTerms(t *testing.T) { + db := testDB(t) + tenantID, agentID := seedTenantAgent(t, db) + ctx := tenantCtx(tenantID) + vs := pg.NewPGVaultStore(db) // no embedder: keyword search only + tid, aid := tenantID.String(), agentID.String() + + seedPriceSheet(t, ctx, vs, tid, aid, "body/prices.md") + other := makeVaultDoc(tid, aid, "body/other.md", "Staff rota") + other.Summary = "Weekly shifts for front desk staff." + if err := vs.UpsertDocument(ctx, other); err != nil { + t.Fatalf("UpsertDocument other: %v", err) + } + + paths := searchPaths(t, vs, store.VaultSearchOptions{Query: "Ocean Suite", TenantID: tid, AgentID: aid}) + if len(paths) != 1 || paths[0] != "body/prices.md" { + t.Fatalf("search %q = %v, want only body/prices.md", "Ocean Suite", paths) + } +} + +func TestStoreVault_Search_NaturalLanguageQuery(t *testing.T) { + db := testDB(t) + tenantID, agentID := seedTenantAgent(t, db) + ctx := tenantCtx(tenantID) + vs := pg.NewPGVaultStore(db) + tid, aid := tenantID.String(), agentID.String() + + seedPriceSheet(t, ctx, vs, tid, aid, "nl/prices.md") + notes := makeVaultDoc(tid, aid, "nl/go.md", "Go Programming") + notes.Summary = "Go concurrency goroutines channels" + if err := vs.UpsertDocument(ctx, notes); err != nil { + t.Fatalf("UpsertDocument notes: %v", err) + } + + // Filler words are not in any document, so strict AND matching found nothing. + paths := searchPaths(t, vs, store.VaultSearchOptions{Query: "what is the price of the Deluxe Ocean Suite?", TenantID: tid, AgentID: aid}) + if len(paths) == 0 || paths[0] != "nl/prices.md" { + t.Fatalf("body query = %v, want nl/prices.md first", paths) + } + paths = searchPaths(t, vs, store.VaultSearchOptions{Query: "can you show me my notes about goroutines", TenantID: tid, AgentID: aid}) + if len(paths) == 0 || paths[0] != "nl/go.md" { + t.Fatalf("summary query = %v, want nl/go.md first", paths) + } +} + +func TestStoreVault_Search_BodyChunksKeepFilters(t *testing.T) { + db := testDB(t) + tenantA, agentA := seedTenantAgent(t, db) + tenantB, agentB := seedTenantAgent(t, db) + agentA2 := seedAgentInTenant(t, db, tenantA) + vs := pg.NewPGVaultStore(db) + + seedPriceSheet(t, tenantCtx(tenantA), vs, tenantA.String(), agentA.String(), "filters/prices.md") + + if paths := searchPaths(t, vs, store.VaultSearchOptions{Query: "Ocean Suite", TenantID: tenantB.String(), AgentID: agentB.String()}); len(paths) != 0 { + t.Fatalf("other tenant sees body match: %v", paths) + } + if paths := searchPaths(t, vs, store.VaultSearchOptions{Query: "Ocean Suite", TenantID: tenantA.String(), AgentID: agentA2.String()}); len(paths) != 0 { + t.Fatalf("other agent sees personal body match: %v", paths) + } +} + +func TestStoreVault_Search_ChunkEmbeddingMatch(t *testing.T) { + db := testDB(t) + tenantID, agentID := seedTenantAgent(t, db) + ctx := tenantCtx(tenantID) + vs := newVaultStore(db) // mock embedder: identical text gives identical vectors + tid, aid := tenantID.String(), agentID.String() + + seedPriceSheet(t, ctx, vs, tid, aid, "vec/prices.md") + for _, p := range []string{"vec/a.md", "vec/b.md", "vec/c.md"} { + d := makeVaultDoc(tid, aid, p, "Filler "+p) + d.Summary = "unrelated summary " + p + if err := vs.UpsertDocument(ctx, d); err != nil { + t.Fatalf("UpsertDocument %s: %v", p, err) + } + } + + query := "Deposit is 30 percent, refundable up to 7 days before arrival." + paths := searchPaths(t, vs, store.VaultSearchOptions{Query: query, TenantID: tid, AgentID: aid}) + if len(paths) == 0 || paths[0] != "vec/prices.md" { + t.Fatalf("chunk vector query = %v, want vec/prices.md first", paths) + } +} + +func TestStoreVault_ReplaceDocumentChunks_FollowsContentHash(t *testing.T) { + db := testDB(t) + tenantID, agentID := seedTenantAgent(t, db) + ctx := tenantCtx(tenantID) + vs := pg.NewPGVaultStore(db) + tid, aid := tenantID.String(), agentID.String() + + doc := makeVaultDoc(tid, aid, "hash/doc.md", "Menu") + if err := vs.UpsertDocument(ctx, doc); err != nil { + t.Fatalf("UpsertDocument: %v", err) + } + needing := func() bool { + docs, err := vs.ListDocsNeedingBodyIndex(ctx, tid, 0) + if err != nil { + t.Fatalf("ListDocsNeedingBodyIndex: %v", err) + } + for _, d := range docs { + if d.ID == doc.ID { + return true + } + } + return false + } + replace := func(hash, text string) { + if err := vs.ReplaceDocumentChunks(ctx, tid, doc.ID, hash, []store.VaultChunk{{Text: text}}); err != nil { + t.Fatalf("ReplaceDocumentChunks: %v", err) + } + } + query := func(q string) bool { + return containsPath(searchPaths(t, vs, store.VaultSearchOptions{Query: q, TenantID: tid, AgentID: aid}), "hash/doc.md") + } + + if !needing() { + t.Fatal("new doc should need a body index") + } + replace(doc.ContentHash, "pho bo special") + if needing() { + t.Fatal("indexed doc should not need a body index") + } + + // Same hash: chunks are kept, so the new text is not indexed. + replace(doc.ContentHash, "banh mi special") + if query("banh") || !query("pho") { + t.Fatal("unchanged content_hash must keep the existing chunks") + } + + // New content: the doc needs indexing again, then the old text is gone. + if err := vs.UpdateHash(ctx, tid, doc.ID, "def456"); err != nil { + t.Fatalf("UpdateHash: %v", err) + } + if !needing() { + t.Fatal("changed doc should need a body index") + } + replace("def456", "banh mi special") + if !query("banh") || query("pho") { + t.Fatal("new content_hash must replace the chunks") + } + + // Deleting the document removes its chunks with it. + if err := vs.DeleteDocument(ctx, tid, aid, "hash/doc.md"); err != nil { + t.Fatalf("DeleteDocument: %v", err) + } + var n int + if err := db.QueryRow(`SELECT count(*) FROM vault_document_chunks WHERE document_id = $1`, doc.ID).Scan(&n); err != nil { + t.Fatalf("count chunks: %v", err) + } + if n != 0 { + t.Fatalf("chunks left after delete: %d", n) + } +}