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

@@ -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))

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)
+	}
+}

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)
 	}
 }