From 65dbbcfd59fb182ab1770be8e95003ff3be19207 Mon Sep 17 00:00:00 2001 From: Chandan Kumar Date: Sun, 27 Sep 2026 13:13:40 +0530 Subject: [PATCH 1/2] fix(vault): search document bodies and match any query word (#1554) vault_search only indexed title + path + the auto-summary, and the summary is written from the first 3000 runes of the file. Anything the summary left out (room names, prices, codes) could never be found. On top of that the FTS query used plainto_tsquery, which ANDs every word, so a normal question like "what is the price of the Deluxe Ocean Suite?" matched nothing even when the key words were indexed. - Add vault_document_chunks (migration 98): the file body split into chunks with their own tsvector and embedding. body_indexed_hash on vault_documents records which content_hash the chunks came from, so unchanged files are not re-chunked or re-embedded. - The enrich worker rebuilds chunks when a file changes. This needs no LLM, so it runs even when no provider is configured. - Rescan backfills chunks for docs indexed before this change, since they already have a summary and never go back through the worker. - FTS matches any query word; ts_rank still ranks docs with more matching words first. Both FTS and vector search look at the doc and its chunks and score each doc by its best hit. SQLite is unchanged: its vault search is LIKE on title/path only. --- docs/24-knowledge-vault.md | 20 +- internal/http/vault_handlers.go | 5 + internal/store/pg/vault_chunks.go | 121 +++++++++++ internal/store/pg/vault_documents.go | 176 ++++++++++----- internal/store/sqlitestore/vault_documents.go | 10 + internal/store/vault_store.go | 15 ++ internal/upgrade/version.go | 2 +- internal/vault/body_index.go | 81 +++++++ internal/vault/body_index_test.go | 66 ++++++ internal/vault/enrich_worker.go | 20 +- .../000098_vault_document_chunks.down.sql | 2 + .../000098_vault_document_chunks.up.sql | 29 +++ .../integration/v3_vault_body_search_test.go | 205 ++++++++++++++++++ 13 files changed, 683 insertions(+), 69 deletions(-) create mode 100644 internal/store/pg/vault_chunks.go create mode 100644 internal/vault/body_index.go create mode 100644 internal/vault/body_index_test.go create mode 100644 migrations/000098_vault_document_chunks.down.sql create mode 100644 migrations/000098_vault_document_chunks.up.sql create mode 100644 tests/integration/v3_vault_body_search_test.go 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..12ccbf7a --- /dev/null +++ b/internal/vault/body_index.go @@ -0,0 +1,81 @@ +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. +func indexBody(ctx context.Context, vs store.VaultStore, tenantID, docID, contentHash, fullPath string) { + raw, err := os.ReadFile(fullPath) + if err != nil { + slog.Warn("vault.body_index: read_file", "path", fullPath, "err", err) + return + } + if err := vs.ReplaceDocumentChunks(ctx, tenantID, docID, contentHash, chunkBody(raw)); err != nil { + slog.Warn("vault.body_index: replace_chunks", "doc", docID, "err", err) + } +} + +// 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_worker.go b/internal/vault/enrich_worker.go index 97e3fdc0..699128e3 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,20 @@ 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. + for _, item := range pending { + if doc := docMap[item.DocID]; doc != nil && doc.DocType != "media" && doc.DocType != "document" { + indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)) + } + } + + // 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] 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) + } +} From 77f4fdd7fbf735ddd3c25f226f08a544c6c76818 Mon Sep 17 00:00:00 2001 From: Chandan Kumar Date: Sun, 27 Sep 2026 17:56:38 +0530 Subject: [PATCH 2/2] fix(vault): retry body indexing after a failed attempt The enrich worker recorded a doc in its dedup map even when writing its body chunks failed, so the next event with the same content hash was skipped and the doc stayed without chunks until a manual rescan. indexBody now returns its error and processChunk leaves those docs out of the dedup map, so the next event for the same hash tries again. Docs whose chunks were written are still deduplicated as before. --- internal/vault/body_index.go | 9 +- internal/vault/enrich_body_retry_test.go | 107 +++++++++++++++++++++++ internal/vault/enrich_worker.go | 12 ++- 3 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 internal/vault/enrich_body_retry_test.go diff --git a/internal/vault/body_index.go b/internal/vault/body_index.go index 12ccbf7a..64cb264a 100644 --- a/internal/vault/body_index.go +++ b/internal/vault/body_index.go @@ -38,15 +38,18 @@ func chunkBody(raw []byte) []store.VaultChunk { // indexBody rebuilds the body chunks of one text document from its workspace // file. The store skips the write when the chunks already match contentHash. -func indexBody(ctx context.Context, vs store.VaultStore, tenantID, docID, contentHash, fullPath string) { +// 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 + 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. @@ -72,7 +75,7 @@ func IndexStaleBodies(ctx context.Context, vs store.VaultStore, tenantID, worksp if ctx.Err() != nil { return true } - indexBody(ctx, vs, tenantID, doc.ID, doc.ContentHash, filepath.Join(workspace, doc.Path)) + _ = 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)) 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 699128e3..af225ae7 100644 --- a/internal/vault/enrich_worker.go +++ b/internal/vault/enrich_worker.go @@ -324,9 +324,15 @@ func (w *EnrichWorker) processChunk(ctx context.Context, items []eventbus.VaultD } // 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" { - indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)) + if err := indexBody(ctx, w.vault, tenantID, item.DocID, item.ContentHash, filepath.Join(item.Workspace, item.Path)); err != nil { + bodyFailed[item.DocID] = true + } } } @@ -454,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) } }