From da0c5d9591b2afd97e4aabde38030a0a558fd591 Mon Sep 17 00:00:00 2001 From: Christopher Hlubek Date: Mon, 7 Sep 2026 15:07:38 +0200 Subject: [PATCH 1/5] sdd: decision tactical Local explicit indexing, CLI lazy fill and MCP backfill share one ... SDD-Mutation: entry-20260907-150733-d-tac-qiq --- .sdd/graph/2026/09/07-150733-d-tac-qiq.md | 37 +++++++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 .sdd/graph/2026/09/07-150733-d-tac-qiq.md diff --git a/.sdd/graph/2026/09/07-150733-d-tac-qiq.md b/.sdd/graph/2026/09/07-150733-d-tac-qiq.md new file mode 100644 index 00000000..dfdb6318 --- /dev/null +++ b/.sdd/graph/2026/09/07-150733-d-tac-qiq.md @@ -0,0 +1,37 @@ +--- +type: decision +layer: tactical +kind: plan +refs: + - id: 20260906-121218-d-tac-ccm + kind: builds-on + - id: 20260902-154750-d-tac-o1s + kind: builds-on + - id: 20260413-142536-d-cpt-ah1 + kind: grounded-in +participants: + - Christopher +confidence: high +topics: + - search/index + - cli/ux + - portability/runtime + - llm/providers +summary: This tactical plan commits to one synchronous indexing path in which local explicit indexing, CLI lazy fill, and MCP backfill pull chunks entry-by-entry until an embedding batch fills, then publish complete entry versions, eliminating sequential underfilled requests and eager project-wide chunk preparation. It extends the incremental entry-indexing plan (20260906-121218-d-tac-ccm), keeps the embed.Batched decorator from the batching directive (20260902-154750-d-tac-o1s) with no new public batching API, and decomposes handlers, finders, and shells per the CQRS contract (20260413-142536-d-cpt-ah1). Concurrent batching moves to consumers via Embedder, with embed.NewBatcher exports removed after migration. +--- + +Local explicit indexing, CLI lazy fill and MCP backfill share one synchronous path that fills embedding batches across entry boundaries and publishes complete entry versions. + +This extends 20260906-121218-d-tac-ccm to eliminate sequential underfilled requests and eager project-wide chunk preparation. Consumers own concurrent batching through Embedder; remove embed.NewBatcher and its exclusive exports after consumer migration. Keep embed.Batched from 20260902-154750-d-tac-o1s; add no public batching or scheduling API. + +Following 20260413-142536-d-cpt-ah1, existing commands carry intent and progress callbacks, handlers share preparation, embedding and publication, and pure packing and ownership mapping live below handlers. Application composition supplies source authority and configuration; finders retain discovery and coverage reads; shells present progress. Single-entry indexing reuses the same semantic logic. + +## Acceptance criteria + +- [ ] Pull chunks one entry at a time until a batch fills; flush the final tail. Retain only active-batch and unfinished-entry chunks and vectors, without preparing the project's chunks upfront. +- [ ] Publish complete versions, including empty and oversized entries; preserve published siblings on failure and skip them on retry. Partial entries remain invisible; failures surface explicitly. +- [ ] Show published entries against entries needing indexing, processed chunks and current batch activity. Count entries during selection; avoid a repeated chunk-derivation pass or changing chunk-total estimate. +- [ ] Preserve fixed source identity, search synchronization scope, force rebuild, version retention, eligibility and compatibility-store behavior. +- [ ] Remove the concurrent batcher exports while retaining Batched and the existing Embedder boundary. + +Verify incremental consumption, vector ownership and interruption recovery through behavior tests. Within-entry checkpoints and lazy splitting inside one entry remain outside scope. From 349dc73d11d211c4079df167d2da81701dd0103e Mon Sep 17 00:00:00 2001 From: Christopher Hlubek Date: Mon, 7 Sep 2026 15:08:30 +0200 Subject: [PATCH 2/5] sdd: wip start 20260907-150733-d-tac-qiq (Christopher) SDD-Mutation: wip-start-20260907-150829-christopher --- .sdd/graph/wip/20260907-150829-christopher.md | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 .sdd/graph/wip/20260907-150829-christopher.md diff --git a/.sdd/graph/wip/20260907-150829-christopher.md b/.sdd/graph/wip/20260907-150829-christopher.md new file mode 100644 index 00000000..a2c5467a --- /dev/null +++ b/.sdd/graph/wip/20260907-150829-christopher.md @@ -0,0 +1,7 @@ +--- +entry: 20260907-150733-d-tac-qiq +participant: Christopher +exclusive: true +--- + +Unify synchronous indexing with incremental cross-entry packing and entry progress From 05f740fb87ff572b43034e307e6f29eae858617c Mon Sep 17 00:00:00 2001 From: Christopher Hlubek Date: Mon, 7 Sep 2026 15:20:05 +0200 Subject: [PATCH 3/5] refactor(search): unify incremental synchronous indexing Implements 20260907-150733-d-tac-qiq. --- cmd/sdd/progress.go | 23 +- cmd/sdd/progress_test.go | 19 +- cmd/sdd/serve.go | 10 +- internal/command/index.go | 24 +- internal/handlers/handler_index.go | 316 +++------------ internal/handlers/handler_index_test.go | 23 +- internal/handlers/handler_repo_test.go | 6 +- internal/handlers/index_pipeline.go | 111 ++++++ internal/handlers/index_search_entry.go | 54 +-- internal/handlers/index_stream_test.go | 204 ++++++++++ internal/handlers/search_index.go | 148 +++----- internal/model/batches.go | 33 ++ internal/model/batches_test.go | 52 +++ internal/model/index_work.go | 52 +++ .../model/vectors/validate.go | 14 +- pkg/application/doc.go | 9 +- pkg/application/search_index_exercise_test.go | 22 +- pkg/application/types/search_entry.go | 29 +- pkg/llm/embed/batcher.go | 298 --------------- pkg/llm/embed/batcher_test.go | 359 ------------------ 20 files changed, 659 insertions(+), 1147 deletions(-) create mode 100644 internal/handlers/index_pipeline.go create mode 100644 internal/handlers/index_stream_test.go create mode 100644 internal/model/batches.go create mode 100644 internal/model/batches_test.go create mode 100644 internal/model/index_work.go rename pkg/llm/embed/vectors.go => internal/model/vectors/validate.go (52%) delete mode 100644 pkg/llm/embed/batcher.go delete mode 100644 pkg/llm/embed/batcher_test.go diff --git a/cmd/sdd/progress.go b/cmd/sdd/progress.go index a187acd2..9378c790 100644 --- a/cmd/sdd/progress.go +++ b/cmd/sdd/progress.go @@ -1,6 +1,8 @@ package main import ( + "fmt" + "github.com/networkteam/sdd/internal/cliout" "github.com/networkteam/sdd/internal/command" "github.com/networkteam/sdd/internal/model" @@ -9,20 +11,22 @@ import ( // embedProgress bridges the embedding and cache-freshening command callbacks // onto one cliout.Reporter, shared by `sdd index` and `sdd search`. type embedProgress struct { - reporter *cliout.Reporter - total int - curRepo string + reporter *cliout.Reporter + total int + curRepo string + chunks int + batchNote string } func newEmbedProgress() *embedProgress { r := cliout.NewReporter() - r.SetUnit("chunks") + r.SetUnit("entries") return &embedProgress{reporter: r} } // onPlanned grows the running total (member work is only known after each cache // is fresh) and declares the indexing phase once real embedding work is planned -// — a zero-chunk (warm) plan neither advances the bar nor arms a footer. +// — an empty (warm) plan neither advances the bar nor arms a footer. func (p *embedProgress) onPlanned(n int) { if n > 0 { p.reporter.SetPhase(model.PhaseIndexing) @@ -36,10 +40,15 @@ func (p *embedProgress) onBatchStart(ids []string, chunks int) { if p.curRepo != "" { note = p.curRepo + " · " + note } - p.reporter.SetNote(note) + p.batchNote = note + p.reporter.SetNote(fmt.Sprintf("%s · %d chunks published", note, p.chunks)) } -func (p *embedProgress) onEntryIndexed(_ string, chunks int) { p.reporter.Add(chunks) } +func (p *embedProgress) onEntryIndexed(_ string, chunks int) { + p.chunks += chunks + p.reporter.Add(1) + p.reporter.SetNote(fmt.Sprintf("%s · %d chunks published", p.batchNote, p.chunks)) +} func (p *embedProgress) onRepoStart(id string) { p.curRepo = id } diff --git a/cmd/sdd/progress_test.go b/cmd/sdd/progress_test.go index 32da7b5c..74672407 100644 --- a/cmd/sdd/progress_test.go +++ b/cmd/sdd/progress_test.go @@ -1,14 +1,15 @@ package main import ( + "strings" "testing" "github.com/networkteam/sdd/internal/model" ) // The shared wiring helper maps every command's callbacks onto one reporter: a -// zero-chunk (warm) plan reports no phase, real embedding work reports indexing, -// and a freshen phase (syncing) is not clobbered by a subsequent zero-chunk +// zero-entry (warm) plan reports no phase, real embedding work reports indexing, +// and a freshen phase (syncing) is not clobbered by a subsequent zero-entry // plan — so a text-only cross-repo read stays labeled syncing, never indexing. func TestEmbedProgress_PhaseMapping(t *testing.T) { p := newEmbedProgress() @@ -25,7 +26,7 @@ func TestEmbedProgress_PhaseMapping(t *testing.T) { p.onPlanned(0) // a warm member fill must not overwrite syncing with indexing if got := latestPhase(t, p); got != model.PhaseSyncing { - t.Errorf("zero-chunk plan must not clobber syncing; got %q", got) + t.Errorf("zero-entry plan must not clobber syncing; got %q", got) } p.onPlanned(5) // real embedding work @@ -42,3 +43,15 @@ func latestPhase(t *testing.T, p *embedProgress) model.Phase { } return snap.Phase } + +func TestEmbedProgressCountsPublishedEntries(t *testing.T) { + p := newEmbedProgress() + cmd := p.localBuild(false, nil) + cmd.OnPlanned(2) + cmd.OnBatchStart([]string{"a", "b"}, 32) + cmd.OnEntryIndexed("a", 20) + progress, ok := p.reporter.Recv() + if !ok || progress.Done != 1 || progress.Total != 2 || progress.Unit != "entries" || !strings.Contains(progress.Note, "20 chunks published") { + t.Fatalf("progress = %+v", progress) + } +} diff --git a/cmd/sdd/serve.go b/cmd/sdd/serve.go index 7d6da7ef..1cf6b5c1 100644 --- a/cmd/sdd/serve.go +++ b/cmd/sdd/serve.go @@ -282,7 +282,10 @@ func buildLocalApplication(ctx context.Context, cmd *cli.Command, graphDir, sddD cacheRoot := registry.CacheRoot() baseRepoKey := persistentIndexRepoKey(cfg, stableRepoRoot) baseIndex := localadapter.NewPersistentSearchIndexStore(project, cacheRoot, baseRepoKey) - embeddings := localEmbedder.Embedder + var embeddings embed.Embedder + if localEmbedder.Embedder != nil { + embeddings = localEmbedder + } targets, err := newLocalMutationTargets(project, filepath.Dir(sddDir)) if err != nil { return nil, "", sdd.RequestIdentity{}, err @@ -332,7 +335,10 @@ func buildLocalApplication(ctx context.Context, cmd *cli.Command, graphDir, sddD // connected-repository storage contract) and exclude embedded entries, // so binary-shipped base facts embed once per machine in the base store, // not once per connected repo. - memberEmbedder := crossEmbedder.Embedder + var memberEmbedder embed.Embedder + if crossEmbedder.Embedder != nil { + memberEmbedder = crossEmbedder + } memberIndex := localadapter.NewPersistentSearchIndexStore(sdd.ProjectID(dependency), cacheRoot, dependency) options := sdd.ProjectRuntimeOptions{ Project: sdd.ProjectRef{ID: sdd.ProjectID(dependency), DisplayName: dependency}, Graph: memberGraph, diff --git a/internal/command/index.go b/internal/command/index.go index abc361d8..29e95d47 100644 --- a/internal/command/index.go +++ b/internal/command/index.go @@ -13,13 +13,8 @@ type BuildIndexCmd struct { // an up-to-date row set. Force bool - // OnPlanned is called once, after the skip pass decides what to embed and - // before the first round-trip, with the total chunk count across all - // entries to be embedded. Optional; the authoritative progress total — - // embedding time scales with chunks, and this count comes from the same - // skip logic that produces the work, so the bar's denominator matches what - // actually runs. - OnPlanned func(totalChunks int) + // OnPlanned reports the number of entries needing indexing after selection. + OnPlanned func(totalEntries int) // OnBatchStart is called before each embedding round-trip with the entry // IDs in that batch and their combined chunk count. Optional; names the @@ -29,8 +24,7 @@ type BuildIndexCmd struct { OnBatchStart func(entryIDs []string, chunkCount int) // OnEntryIndexed is called once per entry after its rows are upserted, with - // the entry's chunk count. Optional; advances the progress bar by that many - // chunks as work completes. + // the entry's chunk count. Progress advances by one published entry. OnEntryIndexed func(entryID string, chunkCount int) // OnEntrySkipped is called for entries whose manifest record matches @@ -49,8 +43,8 @@ type BuildIndexCmd struct { // switch is paid lazily rather than requiring an explicit warm-up. type LazyFillIndexCmd struct { // OnPlanned mirrors BuildIndexCmd's callback — fired once with the total - // chunk count to embed, the authoritative progress total. - OnPlanned func(totalChunks int) + // entry count to index, the authoritative progress total. + OnPlanned func(totalEntries int) // OnBatchStart mirrors BuildIndexCmd's callback — fired before each // embedding round-trip with the batch's entry IDs and combined chunk count, @@ -58,7 +52,7 @@ type LazyFillIndexCmd struct { OnBatchStart func(entryIDs []string, chunkCount int) // OnEntryIndexed mirrors BuildIndexCmd's callback — the per-entry chunk - // count that advances the bar as work completes. + // count; the bar advances by one entry as publication completes. OnEntryIndexed func(entryID string, chunkCount int) // OnComplete is called once after lazy-fill finishes, with the count @@ -91,10 +85,10 @@ type BuildConnectedIndexesCmd struct { OnPhase func(phase model.Phase) // OnPlanned fires once per repo, after that repo's skip pass, with the - // chunk count to embed for it. The caller accumulates these into a + // entry count to index for it. The caller accumulates these into a // running total — the bar's denominator grows as each repo is reached, // because member work is only known after its cache is fresh. - OnPlanned func(chunks int) + OnPlanned func(entries int) // OnBatchStart mirrors BuildIndexCmd's callback — fired before each // embedding round-trip with the batch's entry IDs and combined chunk @@ -102,6 +96,6 @@ type BuildConnectedIndexesCmd struct { OnBatchStart func(entryIDs []string, chunkCount int) // OnEntryIndexed mirrors BuildIndexCmd's callback — the per-entry chunk - // count that advances the bar as work completes. + // count; the bar advances by one entry as publication completes. OnEntryIndexed func(entryID string, chunkCount int) } diff --git a/internal/handlers/handler_index.go b/internal/handlers/handler_index.go index 4bfabce1..72601d65 100644 --- a/internal/handlers/handler_index.go +++ b/internal/handlers/handler_index.go @@ -13,6 +13,7 @@ import ( "github.com/networkteam/sdd/internal/index" "github.com/networkteam/sdd/internal/model" "github.com/networkteam/sdd/internal/textsplitter" + "github.com/networkteam/sdd/pkg/application/types" "github.com/networkteam/sdd/pkg/llm/embed" "github.com/networkteam/slogutils" ) @@ -46,8 +47,7 @@ type IndexHandler struct { excludeEmbedded bool } -// IndexEmbedder is the embedder the CLI index path runs on, with the transport -// batch size it buckets work by so progress advances once per round-trip. +// IndexEmbedder carries the local composition size through CLI and MCP indexing. type IndexEmbedder struct { embed.Embedder BatchSize int @@ -96,12 +96,7 @@ func NewIndexHandler(opts IndexHandlerOptions) *IndexHandler { return h } -// Build is the sdd-index warm-up. It loads the graph, derives chunks for -// every entry (skipping unchanged ones unless cmd.Force), embeds in -// batches across entries (one Embed call per outer batch), and upserts -// per-entry into the index. The manifest is saved after every batch — a -// crash mid-build leaves a partially-populated index that lazy-fill can -// finish later. +// Build warms the index, persisting each complete entry before reporting progress. func (h *IndexHandler) Build(ctx context.Context, cmd *command.BuildIndexCmd) error { if cmd == nil { return errors.New("BuildIndexCmd is required") @@ -124,20 +119,6 @@ func (h *IndexHandler) LazyFill(ctx context.Context, cmd *command.LazyFillIndexC return h.indexEntries(ctx, false, cmd.OnPlanned, cmd.OnBatchStart, cmd.OnEntryIndexed, nil, onComplete) } -// indexEntries is the shared core for Build and LazyFill. The -// up-to-date check (manifest hash + fingerprint match) skips converged -// entries; force bypasses that check. -// -// Work is packed into buckets sized to the embedder's BatchSize so each -// Embed call corresponds to a single transport round-trip on -// the model — progress callbacks fire per bucket (≈ per HTTP call) -// instead of waiting for one giant cross-entry batch to return. Each -// bucket holds entries whose total chunk count fits within BatchSize; -// an entry whose own chunks exceed BatchSize gets its own oversized -// bucket (the embedder splits internally). -// -// The manifest is saved after every bucket so a crash mid-build leaves -// a resumable state. func (h *IndexHandler) indexEntries(ctx context.Context, force bool, onPlanned func(int), onBatchStart func([]string, int), onIndexed func(string, int), onSkipped func(string), onComplete func(int, int)) error { @@ -158,120 +139,98 @@ func (h *IndexHandler) indexEntries(ctx context.Context, force bool, func (h *IndexHandler) indexEntriesLocked(ctx context.Context, idx *index.Index, g *model.Graph, force bool, onPlanned func(int), onBatchStart func([]string, int), onIndexed func(string, int), onSkipped func(string), onComplete func(int, int)) error { - - logger := slogutils.FromContext(ctx) - manifest, err := index.LoadManifest(h.indexDir) if err != nil { return fmt.Errorf("loading manifest: %w", err) } - fingerprint := h.embedder.Fingerprint() - batchSize := h.embedder.BatchSize - if batchSize <= 0 { - batchSize = 1 // defensive — no embedder should report 0, but a bucket of 1 still terminates - } - - var ( - work []entryWithChunks - skipped int - ) - - // The writer's current state hash for every entry it processed — the input - // to version GC below, which keeps a version only when it is a current - // version here or was indexed within the retention window. currentHashes := map[string]string{} - - for _, e := range g.Entries { - if !chunking.IncludeEntry(e, h.excludeEmbedded) { + var work []indexInput + skipped := 0 + for _, entry := range g.Entries { + if err := ctx.Err(); err != nil { + return err + } + if !chunking.IncludeEntry(entry, h.excludeEmbedded) { continue } - hash, err := chunking.EntryStateHash(ctx, e, h.attachments) + hash, err := chunking.EntryStateHash(ctx, entry, h.attachments) if err != nil { - logger.Warn("hash failure, skipping entry", "entry", e.ID, "err", err) - continue + return err } - currentHashes[e.ID] = hash - if !force && manifest.Entries[e.ID].HasVersion(hash, fingerprint) { - logger.Debug("skipped, up to date", "entry", e.ID) + currentHashes[entry.ID] = hash + if !force && manifest.Entries[entry.ID].HasVersion(hash, fingerprint) { + skipped++ if onSkipped != nil { - onSkipped(e.ID) + onSkipped(entry.ID) } - skipped++ continue } - chunks, err := chunking.DeriveChunks(ctx, e, hash, h.splitter, h.attachments) - if err != nil { - return fmt.Errorf("deriving chunks for %s: %w", e.ID, err) - } - work = append(work, entryWithChunks{entry: e, hash: hash, chunks: chunks}) - } - - // The authoritative progress total: total chunks across the work set, - // from the same skip logic that produced it — so the bar's denominator - // can't drift from what actually embeds. Reported before any round-trip. - totalChunks := 0 - for _, w := range work { - totalChunks += len(w.chunks) + work = append(work, indexInput{entry: entry, version: types.SearchEntryVersion{ + Namespace: types.IndexNamespace{Project: types.ProjectID(h.graphDir), Fingerprint: fingerprint, Metric: "cosine"}, EntryID: entry.ID, EntryHash: hash, + }}) } if onPlanned != nil { - onPlanned(totalChunks) + onPlanned(len(work)) } - - if len(work) == 0 { - // No entries to embed, but stale versions may still be collectable — - // this write session already holds the exclusive lock. - if err := h.collectGarbage(ctx, idx, manifest, currentHashes); err != nil { - return err - } - if onComplete != nil { - onComplete(0, skipped) + inputs := func(yield func(indexInput, error) bool) { + for _, input := range work { + if !yield(input, nil) { + return + } } - return nil } - indexed := 0 - bucketStart := 0 - bucketChunks := 0 - for i, w := range work { - // Flush the current bucket when adding this entry's chunks would - // exceed batchSize. Empty bucket case (single entry larger than - // batchSize) takes the entry on its own — the embedder will - // split internally on its way to the wire. - if bucketChunks > 0 && bucketChunks+len(w.chunks) > batchSize { - if err := h.indexBucket(ctx, idx, work[bucketStart:i], manifest, fingerprint, force, onBatchStart, onIndexed); err != nil { + err = indexStream(ctx, inputs, h.embedder, h.attachments, h.splitter, nil, + func(ctx context.Context, entry *model.IndexWork) error { + if err := h.publishIndexEntry(ctx, idx, manifest, entry, force); err != nil { return err } - if err := manifest.Save(h.indexDir); err != nil { - return fmt.Errorf("save manifest after bucket [%d:%d]: %w", bucketStart, i, err) + indexed++ + if onIndexed != nil { + onIndexed(entry.Version.EntryID, len(entry.Rows)) } - indexed += i - bucketStart - bucketStart = i - bucketChunks = 0 - } - bucketChunks += len(w.chunks) - } - // Flush the trailing bucket. - if bucketStart < len(work) { - if err := h.indexBucket(ctx, idx, work[bucketStart:], manifest, fingerprint, force, onBatchStart, onIndexed); err != nil { - return err - } - if err := manifest.Save(h.indexDir); err != nil { - return fmt.Errorf("save manifest after final bucket: %w", err) - } - indexed += len(work) - bucketStart + return nil + }, onBatchStart) + if err != nil { + return err } - if err := h.collectGarbage(ctx, idx, manifest, currentHashes); err != nil { return err } - if onComplete != nil { onComplete(indexed, skipped) } return nil } +func (h *IndexHandler) publishIndexEntry(ctx context.Context, idx *index.Index, manifest *index.Manifest, entry *model.IndexWork, force bool) error { + if err := types.ValidateEntryPublication(entry.Version, entry.Rows); err != nil { + return err + } + rows := make([]index.Row, len(entry.Rows)) + ids := make([]string, len(rows)) + for i, row := range entry.Rows { + c := row.Chunk + ids[i] = c.ID + rows[i] = index.Row{EntryID: c.EntryID, EntryHash: c.EntryHash, ChunkID: c.ID, Text: c.Text, Body: c.Body, Breadcrumb: c.Breadcrumb, Depth: c.Depth, IsSummary: c.IsSummary, IsAttachment: c.IsAttachment, SourceAttachmentPath: c.SourceAttachmentPath, ContentHash: c.ContentHash, ModelFingerprint: entry.Version.Namespace.Fingerprint, Embedding: row.Vector} + } + var old []string + if force { + old = manifest.Entries[entry.Version.EntryID].AllChunkIDs() + } + if err := idx.UpsertEntry(ctx, entry.Version.EntryID, old, rows); err != nil { + return err + } + version := index.EntryVersion{Hash: entry.Version.EntryHash, Fingerprint: entry.Version.Namespace.Fingerprint, ChunkIDs: ids, IndexedAt: h.now()} + if force { + manifest.SetSingleVersion(entry.Version.EntryID, version) + } else { + manifest.AddVersion(entry.Version.EntryID, version) + } + return manifest.Save(h.indexDir) +} + // collectGarbage drops stored versions that are neither a current version (in // currentHashes, the writer's graph) nor within the retention window, deleting // their rows from the index and persisting the pruned manifest. It runs inside @@ -293,154 +252,3 @@ func (h *IndexHandler) collectGarbage(ctx context.Context, idx *index.Index, man slogutils.FromContext(ctx).Info("garbage-collected stale index versions", "chunks", len(dropped)) return nil } - -// indexBucket embeds and upserts a single bucket of entries. All chunks -// across the bucket are embedded in one Embed call — that's -// one transport round-trip on the model when the bucket is sized to -// the embedder's BatchSize. After the call returns, every entry in -// the bucket has all its embeddings ready and is upserted as a unit; -// the manifest entry follows. -func (h *IndexHandler) indexBucket(ctx context.Context, idx *index.Index, bucket []entryWithChunks, - manifest *index.Manifest, fingerprint string, force bool, - onBatchStart func([]string, int), onIndexed func(string, int)) error { - - logger := slogutils.FromContext(ctx) - - // Announce the batch (entry IDs + combined chunk count) before the - // embedding round-trip so the view can name what's in flight while the - // call runs. This does not advance the bar — that happens per entry as - // work completes (report → onIndexed), so the bar never reads done before - // the work is. - bucketIDs := make([]string, len(bucket)) - bucketChunks := 0 - for i, w := range bucket { - bucketIDs[i] = w.entry.ID - bucketChunks += len(w.chunks) - } - if onBatchStart != nil { - onBatchStart(bucketIDs, bucketChunks) - } - - // report logs the entry at Info (the operational record routed to the - // transient view or the leveled stderr handler) and fires the progress - // callback. Keeps the log and the count advance together at each site. - report := func(id string, n int) { - logger.Info("indexed", "entry", id, "chunks", n) - if onIndexed != nil { - onIndexed(id, n) - } - } - - // Flatten chunks across the bucket while remembering which entry // embedding belongs to. - type ownedChunk struct { - entryID string - entryHash string - chunkID string - chunk textsplitter.Chunk - } - var allChunks []ownedChunk - for _, w := range bucket { - for _, c := range w.chunks { - allChunks = append(allChunks, ownedChunk{entryID: w.entry.ID, entryHash: w.hash, chunkID: c.ChunkID, chunk: c.Chunk}) - } - } - - if len(allChunks) == 0 { - // Every entry in this bucket has no chunks (empty summary, no - // body). Record them in the manifest anyway so the up-to-date - // check skips them next pass. - for _, w := range bucket { - version := index.EntryVersion{ - Hash: w.hash, - Fingerprint: fingerprint, - ChunkIDs: nil, - IndexedAt: h.now(), - } - if force { - if old := manifest.Entries[w.entry.ID].AllChunkIDs(); len(old) > 0 { - if err := idx.DeleteEntry(ctx, old); err != nil { - return fmt.Errorf("delete old chunks for %s: %w", w.entry.ID, err) - } - } - manifest.SetSingleVersion(w.entry.ID, version) - } else { - manifest.AddVersion(w.entry.ID, version) - } - report(w.entry.ID, 0) - } - return nil - } - - texts := make([]string, len(allChunks)) - for i, c := range allChunks { - texts[i] = c.chunk.Text - } - embedded, err := h.embedder.Embed(ctx, embed.Request{Purpose: embed.PurposeDocument, Texts: texts}) - if err != nil { - return fmt.Errorf("embed bucket: %w", err) - } - embeddings := embedded.Vectors - if len(embeddings) != len(texts) { - return fmt.Errorf("embedder returned %d embeddings for %d inputs", len(embeddings), len(texts)) - } - - rowsByEntry := map[string][]index.Row{} - for i, c := range allChunks { - rowsByEntry[c.entryID] = append(rowsByEntry[c.entryID], index.Row{ - EntryID: c.entryID, - EntryHash: c.entryHash, - ChunkID: c.chunkID, - Text: c.chunk.Text, - Body: c.chunk.Body, - Breadcrumb: c.chunk.Breadcrumb, - Depth: c.chunk.Depth, - IsSummary: c.chunk.IsSummary, - IsAttachment: c.chunk.IsAttachment, - SourceAttachmentPath: c.chunk.SourceAttachmentPath, - ContentHash: index.HashContent(c.chunk.Text), - ModelFingerprint: fingerprint, - Embedding: embeddings[i], - }) - } - - for _, w := range bucket { - rows := rowsByEntry[w.entry.ID] - newChunkIDs := make([]string, 0, len(rows)) - for _, r := range rows { - newChunkIDs = append(newChunkIDs, r.ChunkID) - } - version := index.EntryVersion{ - Hash: w.hash, - Fingerprint: fingerprint, - ChunkIDs: newChunkIDs, - IndexedAt: h.now(), - } - // Force is the destructive repair path: drop every stored version's - // rows and write this one as the entry's sole version. The lazy path - // adds this version without deleting — a changed entry accumulates a - // version so a shared store never flip-flops between two branches. - if force { - old := manifest.Entries[w.entry.ID].AllChunkIDs() - if err := idx.UpsertEntry(ctx, w.entry.ID, old, rows); err != nil { - return fmt.Errorf("upsert %s: %w", w.entry.ID, err) - } - manifest.SetSingleVersion(w.entry.ID, version) - } else { - if err := idx.UpsertEntry(ctx, w.entry.ID, nil, rows); err != nil { - return fmt.Errorf("upsert %s: %w", w.entry.ID, err) - } - manifest.AddVersion(w.entry.ID, version) - } - report(w.entry.ID, len(rows)) - } - return nil -} - -// entryWithChunks pairs an entry with its derived chunks and the hash -// recorded in the manifest. Lifted to a top-level type so indexBucket -// can take a slice without re-declaring the shape. -type entryWithChunks struct { - entry *model.Entry - hash string - chunks []chunking.Chunk -} diff --git a/internal/handlers/handler_index_test.go b/internal/handlers/handler_index_test.go index 6a496d44..fd670e89 100644 --- a/internal/handlers/handler_index_test.go +++ b/internal/handlers/handler_index_test.go @@ -193,15 +193,16 @@ func TestIndexHandler_BuildFiresOnBatchStart(t *testing.T) { var batches [][]string var batchChunks []int - plannedChunks := -1 + plannedEntries := -1 indexedChunks := 0 + indexedEntries := 0 cmd := &command.BuildIndexCmd{ - OnPlanned: func(total int) { plannedChunks = total }, + OnPlanned: func(total int) { plannedEntries = total }, OnBatchStart: func(ids []string, chunks int) { batches = append(batches, ids) batchChunks = append(batchChunks, chunks) }, - OnEntryIndexed: func(_ string, chunks int) { indexedChunks += chunks }, + OnEntryIndexed: func(_ string, chunks int) { indexedChunks += chunks; indexedEntries++ }, } if err := h.Build(context.Background(), cmd); err != nil { t.Fatalf("Build: %v", err) @@ -219,19 +220,13 @@ func TestIndexHandler_BuildFiresOnBatchStart(t *testing.T) { if got := withoutEmbedded(t, batchedIDs); len(got) != 2 { t.Errorf("batches carried %d project entry IDs, want 2 (%v)", len(got), batchedIDs) } - // The planned total is the chunk sum, and it must equal both the batch's - // announced chunk count and the chunks reported as entries complete — the - // bar's denominator and numerator come from the same work set, so it lands - // on 100% exactly when the work does. - if plannedChunks <= 0 { - t.Errorf("OnPlanned reported %d chunks, want > 0", plannedChunks) + if plannedEntries <= 0 || plannedEntries != indexedEntries { + t.Errorf("planned %d entries, published %d", plannedEntries, indexedEntries) } - if announcedChunks != plannedChunks { - t.Errorf("announced batch chunks %d != planned total %d", announcedChunks, plannedChunks) - } - if indexedChunks != plannedChunks { - t.Errorf("indexed chunks %d != planned total %d", indexedChunks, plannedChunks) + if announcedChunks != indexedChunks { + t.Errorf("announced %d chunks, published %d", announcedChunks, indexedChunks) } + } func TestIndexHandler_BuildSkipsUnchanged(t *testing.T) { diff --git a/internal/handlers/handler_repo_test.go b/internal/handlers/handler_repo_test.go index 3ae67b34..b5a30988 100644 --- a/internal/handlers/handler_repo_test.go +++ b/internal/handlers/handler_repo_test.go @@ -228,7 +228,7 @@ func TestBuildConnectedIndexes_FreshensAndFills(t *testing.T) { fill := &command.BuildConnectedIndexesCmd{ OnRepoStart: func(id string) { startedRepos = append(startedRepos, id) }, OnPlanned: func(n int) { planned += n }, - OnEntryIndexed: func(_ string, n int) { indexed += n }, + OnEntryIndexed: func(_ string, _ int) { indexed++ }, } if err := h.BuildConnectedIndexes(context.Background(), []string{repoID}, indexEmbedder(emb), fill); err != nil { t.Fatalf("BuildConnectedIndexes: %v", err) @@ -241,10 +241,10 @@ func TestBuildConnectedIndexes_FreshensAndFills(t *testing.T) { t.Errorf("OnRepoStart calls = %v, want [%s]", startedRepos, repoID) } if planned == 0 { - t.Error("expected planned chunks > 0") + t.Error("expected planned entries > 0") } if indexed != planned { - t.Errorf("indexed %d chunks, planned %d — every planned chunk should land", indexed, planned) + t.Errorf("indexed %d entries, planned %d — every planned entry should land", indexed, planned) } // The member index lives under the machine-global (repo-id, fingerprint) diff --git a/internal/handlers/index_pipeline.go b/internal/handlers/index_pipeline.go new file mode 100644 index 00000000..ee75d7da --- /dev/null +++ b/internal/handlers/index_pipeline.go @@ -0,0 +1,111 @@ +package handlers + +import ( + "context" + "fmt" + "iter" + + "github.com/networkteam/sdd/internal/chunking" + "github.com/networkteam/sdd/internal/model" + "github.com/networkteam/sdd/internal/textsplitter" + "github.com/networkteam/sdd/pkg/application/types" + "github.com/networkteam/sdd/pkg/llm/embed" +) + +type indexInput struct { + entry *model.Entry + version types.SearchEntryVersion +} + +func indexBatchSize(emb embed.Embedder) int { + if local, ok := emb.(IndexEmbedder); ok && local.BatchSize > 0 { + return local.BatchSize + } + return 32 +} + +func prepareIndexEntry(ctx context.Context, input indexInput, reader chunking.AttachmentReader, splitter *textsplitter.Splitter, skip func(types.CanonicalChunk) bool) (*model.IndexWork, error) { + attachments := &chunking.CachedAttachments{Reader: reader} + hash, err := chunking.EntryStateHash(ctx, input.entry, attachments) + if err != nil { + return nil, err + } + if input.version.EntryHash != "" && hash != input.version.EntryHash { + return nil, fmt.Errorf("sdd: pinned entry content does not match descriptor") + } + input.version.EntryHash = hash + chunks, err := chunking.DeriveChunks(ctx, input.entry, hash, splitter, attachments) + if err != nil { + return nil, err + } + prepared := &model.IndexWork{Version: input.version} + for _, chunk := range chunks { + canonical := chunking.CanonicalChunk(input.entry.ID, hash, chunk) + if skip == nil || !skip(canonical) { + prepared.Rows = append(prepared.Rows, types.IndexedChunk{Chunk: canonical}) + } + } + return prepared, nil +} + +func indexStream(ctx context.Context, inputs iter.Seq2[indexInput, error], emb embed.Embedder, reader chunking.AttachmentReader, splitter *textsplitter.Splitter, skip func(types.CanonicalChunk) bool, publish func(context.Context, *model.IndexWork) error, onBatch func([]string, int)) error { + items := func(yield func(model.IndexItem, error) bool) { + for input, err := range inputs { + if err == nil { + err = ctx.Err() + } + if err != nil { + yield(model.IndexItem{}, err) + return + } + entry, err := prepareIndexEntry(ctx, input, reader, splitter, skip) + if err != nil { + yield(model.IndexItem{}, err) + return + } + if len(entry.Rows) == 0 { + if err := publish(ctx, entry); err != nil { + yield(model.IndexItem{}, err) + return + } + } + for i := range entry.Rows { + if !yield(model.IndexItem{Owner: entry, Position: i}, nil) { + return + } + } + } + } + for batch, err := range model.Batches(items, indexBatchSize(emb)) { + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + work := model.IndexBatch(batch) + if onBatch != nil { + onBatch(work.EntryIDs(), len(batch)) + } + result, err := emb.Embed(ctx, embed.Request{Purpose: embed.PurposeDocument, Texts: work.Texts()}) + if err != nil { + return err + } + completed, err := work.Complete(result.Vectors) + if err != nil { + return err + } + for _, entry := range completed { + if err := types.ValidateEntryPublication(entry.Version, entry.Rows); err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + if err := publish(ctx, entry); err != nil { + return err + } + } + } + return ctx.Err() +} diff --git a/internal/handlers/index_search_entry.go b/internal/handlers/index_search_entry.go index aec23ce7..233d3a51 100644 --- a/internal/handlers/index_search_entry.go +++ b/internal/handlers/index_search_entry.go @@ -2,7 +2,6 @@ package handlers import ( "context" - "fmt" "github.com/networkteam/sdd/internal/chunking" "github.com/networkteam/sdd/internal/command" @@ -33,43 +32,18 @@ func (h SearchEntryHandler) Index(ctx context.Context, cmd command.IndexSearchEn if err != nil || published { return err } - attachments := &chunking.CachedAttachments{Reader: h.Attachments} - hash, err := chunking.EntryStateHash(ctx, h.Entry, attachments) - if err != nil { - return err - } - if hash != key.EntryHash { - return fmt.Errorf("sdd: pinned entry content does not match descriptor") - } - chunks, err := chunking.DeriveChunks(ctx, h.Entry, hash, textsplitter.NewSplitter(), attachments) - if err != nil { - return err - } - rows := make([]types.IndexedChunk, len(chunks)) - if len(chunks) > 0 { - texts := make([]string, len(chunks)) - for i, chunk := range chunks { - texts[i] = chunk.Chunk.Text - } - result, err := h.Embedder.Embed(ctx, embed.Request{Purpose: embed.PurposeDocument, Texts: texts}) - if err != nil { - return err - } - if len(result.Vectors) != len(chunks) { - return fmt.Errorf("sdd: embedder returned %d vectors for %d chunks", len(result.Vectors), len(chunks)) - } - for i, chunk := range chunks { - rows[i] = types.IndexedChunk{Chunk: chunking.CanonicalChunk(h.Entry.ID, hash, chunk), Vector: result.Vectors[i]} - } - } - if err := types.ValidateEntryPublication(key, rows); err != nil { - return err - } - if err := h.Store.PublishEntry(ctx, key, rows); err != nil { - return err - } - if cmd.OnPublished != nil { - cmd.OnPublished(key.EntryID, len(rows)) - } - return nil + inputs := func(yield func(indexInput, error) bool) { yield(indexInput{entry: h.Entry, version: key}, nil) } + return indexStream(ctx, inputs, h.Embedder, h.Attachments, textsplitter.NewSplitter(), nil, + func(ctx context.Context, entry *model.IndexWork) error { + if err := types.ValidateEntryPublication(entry.Version, entry.Rows); err != nil { + return err + } + if err := h.Store.PublishEntry(ctx, entry.Version, entry.Rows); err != nil { + return err + } + if cmd.OnPublished != nil { + cmd.OnPublished(key.EntryID, len(entry.Rows)) + } + return nil + }, nil) } diff --git a/internal/handlers/index_stream_test.go b/internal/handlers/index_stream_test.go new file mode 100644 index 00000000..3a3a599f --- /dev/null +++ b/internal/handlers/index_stream_test.go @@ -0,0 +1,204 @@ +package handlers_test + +import ( + "context" + "errors" + "fmt" + "reflect" + "testing" + + "github.com/networkteam/sdd/internal/chunking" + "github.com/networkteam/sdd/internal/command" + "github.com/networkteam/sdd/internal/handlers" + "github.com/networkteam/sdd/internal/index" + "github.com/networkteam/sdd/internal/model" + "github.com/networkteam/sdd/pkg/application/types" + "github.com/networkteam/sdd/pkg/llm/embed" + "github.com/networkteam/sdd/pkg/local" +) + +type streamAttachments struct{ reads map[string]int } + +func (a streamAttachments) ReadAttachment(_ context.Context, e *model.Entry, _ string) ([]byte, error) { + a.reads[e.ID]++ + return []byte("Attachment for " + e.ID), nil +} + +type streamStore struct { + *local.MemorySearchIndexStore + rows map[string][]types.IndexedChunk +} + +func (s streamStore) PublishEntry(ctx context.Context, version types.SearchEntryVersion, rows []types.IndexedChunk) error { + if err := s.MemorySearchIndexStore.PublishEntry(ctx, version, rows); err != nil { + return err + } + s.rows[version.EntryID] = rows + return nil +} + +func TestReconcilePacksIncrementallyAndRetainsPublishedEntries(t *testing.T) { + ctx := t.Context() + ns := types.IndexNamespace{Project: "test", Fingerprint: "test", Metric: "cosine"} + var entries []*model.Entry + hashes := map[string]string{} + attachments := streamAttachments{reads: map[string]int{}} + for _, id := range []string{"a", "b", "c"} { + e := &model.Entry{ID: id, Summary: "Summary " + id, Content: "Body " + id, Attachments: []string{id + ".md"}} + entries = append(entries, e) + var err error + hashes[id], err = chunking.EntryStateHash(ctx, e, attachments) + if err != nil { + t.Fatal(err) + } + } + clear(attachments.reads) + store := streamStore{MemorySearchIndexStore: local.NewMemorySearchIndexStore(), rows: map[string][]types.IndexedChunk{}} + failed := errors.New("provider unavailable") + var sizes []int + vectors := map[string]float32{} + fail := true + inner := embed.EmbedderFunc{Space: "test", Run: func(_ context.Context, req embed.Request) (embed.Result, error) { + sizes = append(sizes, len(req.Texts)) + if fail && len(sizes) == 1 && attachments.reads["c"] != 0 { + t.Fatal("derived later entry before consuming first batch") + } + if fail && len(sizes) == 2 { + return embed.Result{}, failed + } + result := embed.Result{} + for _, text := range req.Texts { + if vectors[text] == 0 { + vectors[text] = float32(len(vectors) + 1) + } + result.Vectors = append(result.Vectors, []float32{vectors[text], 1}) + } + return result, nil + }} + h := handlers.SearchIndexHandler{Graph: model.NewGraph(entries), Namespace: ns, Hashes: hashes, Store: store, Attachments: attachments, Embedder: handlers.IndexEmbedder{Embedder: inner, BatchSize: 4}} + if err := h.Reconcile(ctx, command.ReconcileSearchIndexCmd{}); !errors.Is(err, failed) { + t.Fatalf("error = %v", err) + } + if len(store.rows) != 1 || len(store.rows["a"]) != 3 { + t.Fatalf("partial publication: %v", store.rows) + } + if !reflect.DeepEqual(sizes, []int{4, 4}) { + t.Fatalf("batches = %v", sizes) + } + fail = false + sizes = nil + clear(attachments.reads) + if err := h.Reconcile(ctx, command.ReconcileSearchIndexCmd{}); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(sizes, []int{4, 2}) { + t.Fatalf("resumed batches = %v", sizes) + } + if attachments.reads["a"] != 0 { + t.Fatal("published entry prepared again") + } + for id, rows := range store.rows { + if len(rows) != 3 { + t.Fatalf("entry %s has %d rows", id, len(rows)) + } + for _, row := range rows { + if row.Chunk.EntryID != id || row.Vector[0] != vectors[row.Chunk.Text] { + t.Fatalf("misrouted vector: %+v", row) + } + } + } + sizes = nil + if err := h.Reconcile(ctx, command.ReconcileSearchIndexCmd{}); err != nil || len(sizes) != 0 { + t.Fatalf("warm reconcile: %v, %v", sizes, err) + } +} + +func TestReconcileEmptyAndOversizedEntry(t *testing.T) { + ns := types.IndexNamespace{Project: "test", Fingerprint: "test", Metric: "cosine"} + entries := []*model.Entry{{ID: "empty"}, {ID: "large", Summary: "Summary", Content: "## One\nFirst.\n\n## Two\nSecond.\n\n## Three\nThird."}} + store := streamStore{MemorySearchIndexStore: local.NewMemorySearchIndexStore(), rows: map[string][]types.IndexedChunk{}} + calls := 0 + inner := embed.EmbedderFunc{Space: "test", Run: func(_ context.Context, req embed.Request) (embed.Result, error) { + calls++ + if len(req.Texts) > 2 { + t.Fatal("oversized provider request") + } + if _, exists := store.rows["large"]; exists { + t.Fatal("entry published before embedding finished") + } + out := embed.Result{} + for range req.Texts { + out.Vectors = append(out.Vectors, []float32{1, 2}) + } + return out, nil + }} + h := handlers.SearchIndexHandler{Graph: model.NewGraph(entries), Namespace: ns, Store: store, Embedder: handlers.IndexEmbedder{Embedder: inner, BatchSize: 2}} + if err := h.Reconcile(t.Context(), command.ReconcileSearchIndexCmd{}); err != nil { + t.Fatal(err) + } + if _, ok := store.rows["empty"]; !ok || calls < 2 { + t.Fatalf("empty publication or oversized splitting missing: %v, %d", store.rows, calls) + } + before := calls + if err := h.Reconcile(t.Context(), command.ReconcileSearchIndexCmd{}); err != nil || calls != before { + t.Fatalf("retry: %v, calls %d", err, calls) + } +} + +func TestReconcileRejectsInvalidVectors(t *testing.T) { + for _, vectors := range [][][]float32{nil, {{}}, {{0, 0}}} { + t.Run(fmt.Sprint(vectors), func(t *testing.T) { + store := local.NewMemorySearchIndexStore() + h := handlers.SearchIndexHandler{Graph: model.NewGraph([]*model.Entry{{ID: "a", Summary: "One"}}), Namespace: types.IndexNamespace{Project: "test", Fingerprint: "test", Metric: "cosine"}, Store: store, Embedder: embed.EmbedderFunc{Space: "test", Run: func(context.Context, embed.Request) (embed.Result, error) { return embed.Result{Vectors: vectors}, nil }}} + if err := h.Reconcile(t.Context(), command.ReconcileSearchIndexCmd{}); err == nil { + t.Fatal("invalid vectors accepted") + } + }) + } +} + +type streamGraphReader struct { + handlers.Reader + graph *model.Graph +} + +func (r streamGraphReader) CurrentGraph(string) (*model.Graph, error) { return r.graph, nil } + +func TestCLIIndexPublishesBeforeProgressAndResumes(t *testing.T) { + entries := []*model.Entry{{ID: "a", Summary: "Summary A", Content: "Body A"}, {ID: "b", Summary: "Summary B", Content: "Body B"}, {ID: "c", Summary: "Summary C", Content: "Body C"}} + dir := t.TempDir() + calls := 0 + stop := errors.New("interrupted") + emb := embed.EmbedderFunc{Space: "test", Run: func(_ context.Context, req embed.Request) (embed.Result, error) { + calls++ + if calls == 2 { + return embed.Result{}, stop + } + result := embed.Result{} + for range req.Texts { + result.Vectors = append(result.Vectors, []float32{1, 2}) + } + return result, nil + }} + h := handlers.NewIndexHandler(handlers.IndexHandlerOptions{GraphDir: t.TempDir(), IndexDir: dir, Reader: streamGraphReader{graph: model.NewGraph(entries)}, Embedder: handlers.IndexEmbedder{Embedder: emb, BatchSize: 3}}) + var planned, published int + cmd := &command.BuildIndexCmd{OnPlanned: func(n int) { planned = n }, OnEntryIndexed: func(id string, _ int) { + manifest, err := index.LoadManifest(dir) + if err != nil || len(manifest.Entries[id].Versions) == 0 { + t.Fatalf("progress preceded durable manifest: %s %v", id, err) + } + published++ + }} + if err := h.Build(t.Context(), cmd); !errors.Is(err, stop) { + t.Fatalf("error = %v", err) + } + if planned != 3 || published != 1 { + t.Fatalf("planned %d published %d", planned, published) + } + if err := h.Build(t.Context(), cmd); err != nil { + t.Fatal(err) + } + if planned != 2 || published != 3 { + t.Fatalf("resumed planned %d published %d", planned, published) + } +} diff --git a/internal/handlers/search_index.go b/internal/handlers/search_index.go index 4654ba30..5db37d3b 100644 --- a/internal/handlers/search_index.go +++ b/internal/handlers/search_index.go @@ -2,7 +2,6 @@ package handlers import ( "context" - "fmt" "github.com/networkteam/slogutils" @@ -36,37 +35,62 @@ type SearchIndexHandler struct { type versionKey struct{ entryID, entryHash string } func (h *SearchIndexHandler) complete(ctx context.Context, entries []*model.Entry, hashes map[string]string, skip func(types.CanonicalChunk) bool) error { - if store, ok := h.Store.(EntryPublisher); ok { + publisher, complete := h.Store.(EntryPublisher) + inputs := func(yield func(indexInput, error) bool) { for _, entry := range entries { - hash := hashes[entry.ID] - if hash == "" { - var err error - hash, err = chunking.EntryStateHash(ctx, entry, h.Attachments) + version := types.SearchEntryVersion{Namespace: h.Namespace, EntryID: entry.ID, EntryHash: hashes[entry.ID]} + if version.EntryHash == "" { + hash, err := chunking.EntryStateHash(ctx, entry, h.Attachments) if err != nil { - return err + yield(indexInput{}, err) + return } + version.EntryHash = hash } - handler := SearchEntryHandler{Store: store, Embedder: h.Embedder, Entry: entry, Attachments: h.Attachments} - cmd := command.IndexSearchEntryCmd{ - Entry: types.SearchEntryDescriptor{Version: types.SearchEntryVersion{Namespace: h.Namespace, EntryID: entry.ID, EntryHash: hash}, SourceRevision: h.Revision}, - OnPublished: func(id string, count int) { - h.entries++ - h.chunks += count - if h.cmd.OnEntryIndexed != nil { - h.cmd.OnEntryIndexed(id, count) - } - }, + if complete { + present, err := publisher.EntryPublished(ctx, version) + if err != nil { + yield(indexInput{}, err) + return + } + if present { + continue + } } - if err := handler.Index(ctx, cmd); err != nil { - return err + if !yield(indexInput{entry: entry, version: version}, nil) { + return } } - } else { - for _, entry := range entries { - if err := h.embedEntries(ctx, h.Namespace, []*model.Entry{entry}, hashes, skip); err != nil { - return err + } + if complete { + skip = nil + } + err := indexStream(ctx, inputs, h.Embedder, h.Attachments, textsplitter.NewSplitter(), skip, + func(ctx context.Context, entry *model.IndexWork) error { + if complete { + if err := types.ValidateEntryPublication(entry.Version, entry.Rows); err != nil { + return err + } + if err := publisher.PublishEntry(ctx, entry.Version, entry.Rows); err != nil { + return err + } + } else { + if len(entry.Rows) == 0 { + return nil + } + if err := h.Store.Reconcile(ctx, h.Namespace, h.Revision, entry.Rows, nil); err != nil { + return err + } } - } + h.entries++ + h.chunks += len(entry.Rows) + if h.cmd.OnEntryIndexed != nil { + h.cmd.OnEntryIndexed(entry.Version.EntryID, len(entry.Rows)) + } + return nil + }, nil) + if err != nil { + return err } if h.cmd.OnComplete != nil { @@ -125,79 +149,3 @@ func (h *SearchIndexHandler) reconcileByChunkIdentity(ctx context.Context, names } return h.complete(ctx, entries, hashes, keep) } - -func (h *SearchIndexHandler) embedEntries(ctx context.Context, namespace types.IndexNamespace, entries []*model.Entry, hashes map[string]string, skip func(types.CanonicalChunk) bool) error { - if len(entries) == 0 { - return nil - } - attachments := h.Attachments - splitter := textsplitter.NewSplitter() - - var pending []types.CanonicalChunk - for _, entry := range entries { - hash := hashes[entry.ID] - if hash == "" { - h, err := chunking.EntryStateHash(ctx, entry, attachments) - if err != nil { - return err - } - hash = h - } - chunks, err := chunking.DeriveChunks(ctx, entry, hash, splitter, attachments) - if err != nil { - return err - } - for _, c := range chunks { - chunk := chunking.CanonicalChunk(entry.ID, hash, c) - if skip != nil && skip(chunk) { - continue - } - pending = append(pending, chunk) - } - } - if len(pending) == 0 { - return nil - } - texts := make([]string, len(pending)) - for i, chunk := range pending { - texts[i] = chunk.Text - } - embedded, err := h.Embedder.Embed(ctx, embed.Request{Purpose: embed.PurposeDocument, Texts: texts}) - if err != nil { - return err - } - if len(embedded.Vectors) != len(pending) { - return fmt.Errorf("sdd: embedder returned %d vectors for %d inputs", len(embedded.Vectors), len(pending)) - } - dims := 0 - upserts := make([]types.IndexedChunk, 0, len(pending)) - for i, vector := range embedded.Vectors { - if len(vector) == 0 { - return fmt.Errorf("sdd: embedding vector %d is empty", i) - } - if dims == 0 { - dims = len(vector) - } - if len(vector) != dims { - return fmt.Errorf("sdd: embedding vector %d has %d dimensions, want %d", i, len(vector), dims) - } - upserts = append(upserts, types.IndexedChunk{Chunk: pending[i], Vector: vector}) - } - if err := h.Store.Reconcile(ctx, namespace, h.Revision, upserts, nil); err != nil { - return err - } - counts := make(map[string]int) - for _, chunk := range pending { - counts[chunk.EntryID]++ - } - for _, entry := range entries { - if count := counts[entry.ID]; count > 0 { - h.entries++ - if h.cmd.OnEntryIndexed != nil { - h.cmd.OnEntryIndexed(entry.ID, count) - } - } - } - h.chunks += len(pending) - return nil -} diff --git a/internal/model/batches.go b/internal/model/batches.go new file mode 100644 index 00000000..a87b2f61 --- /dev/null +++ b/internal/model/batches.go @@ -0,0 +1,33 @@ +package model + +import "iter" + +// Batches consumes only enough input for the next batch. A source error follows +// any pending batch so its completed work can be retained by the caller. +func Batches[T any](source iter.Seq2[T, error], size int) iter.Seq2[[]T, error] { + if size < 1 { + panic("batch size must be positive") + } + return func(yield func([]T, error) bool) { + batch := make([]T, 0, size) + for item, err := range source { + if err != nil { + if len(batch) > 0 && !yield(batch, nil) { + return + } + yield(nil, err) + return + } + batch = append(batch, item) + if len(batch) == size { + if !yield(batch, nil) { + return + } + batch = make([]T, 0, size) + } + } + if len(batch) > 0 { + yield(batch, nil) + } + } +} diff --git a/internal/model/batches_test.go b/internal/model/batches_test.go new file mode 100644 index 00000000..163b70cc --- /dev/null +++ b/internal/model/batches_test.go @@ -0,0 +1,52 @@ +package model_test + +import ( + "errors" + "reflect" + "testing" + + "github.com/networkteam/sdd/internal/model" +) + +func TestBatchesStopsPullingWhenConsumerStops(t *testing.T) { + pulled := 0 + source := func(yield func(int, error) bool) { + for i := range 100 { + pulled++ + if !yield(i, nil) { + return + } + } + } + for batch, err := range model.Batches(source, 3) { + if err != nil || !reflect.DeepEqual(batch, []int{0, 1, 2}) { + t.Fatalf("%v %v", batch, err) + } + break + } + if pulled != 3 { + t.Fatalf("pulled %d", pulled) + } +} + +func TestBatchesRetainsTailBeforeSourceError(t *testing.T) { + want := errors.New("source failed") + source := func(yield func(int, error) bool) { + if !yield(1, nil) { + return + } + yield(0, want) + } + var batches [][]int + var got error + for batch, err := range model.Batches(source, 3) { + if err != nil { + got = err + break + } + batches = append(batches, batch) + } + if !errors.Is(got, want) || !reflect.DeepEqual(batches, [][]int{{1}}) { + t.Fatalf("%v %v", batches, got) + } +} diff --git a/internal/model/index_work.go b/internal/model/index_work.go new file mode 100644 index 00000000..30409d82 --- /dev/null +++ b/internal/model/index_work.go @@ -0,0 +1,52 @@ +package model + +import ( + "github.com/networkteam/sdd/internal/model/vectors" + "github.com/networkteam/sdd/pkg/application/types" +) + +type IndexWork struct { + Version types.SearchEntryVersion + Rows []types.IndexedChunk +} + +type IndexItem struct { + Owner *IndexWork + Position int +} + +type IndexBatch []IndexItem + +func (b IndexBatch) Texts() []string { + texts := make([]string, len(b)) + for i, item := range b { + texts[i] = item.Owner.Rows[item.Position].Chunk.Text + } + return texts +} + +func (b IndexBatch) EntryIDs() []string { + var ids []string + for i, item := range b { + if i == 0 || b[i-1].Owner != item.Owner { + ids = append(ids, item.Owner.Version.EntryID) + } + } + return ids +} + +func (b IndexBatch) Complete(embeddings [][]float32) ([]*IndexWork, error) { + if err := vectors.Validate(embeddings, len(b)); err != nil { + return nil, err + } + for i, item := range b { + item.Owner.Rows[item.Position].Vector = embeddings[i] + } + var completed []*IndexWork + for _, item := range b { + if item.Position == len(item.Owner.Rows)-1 { + completed = append(completed, item.Owner) + } + } + return completed, nil +} diff --git a/pkg/llm/embed/vectors.go b/internal/model/vectors/validate.go similarity index 52% rename from pkg/llm/embed/vectors.go rename to internal/model/vectors/validate.go index fe5fd53b..78e0c7ed 100644 --- a/pkg/llm/embed/vectors.go +++ b/internal/model/vectors/validate.go @@ -1,34 +1,34 @@ -package embed +package vectors import ( "fmt" "math" ) -func validateBatchVectors(vectors [][]float32, count int) error { +func Validate(vectors [][]float32, count int) error { if len(vectors) != count { - return fmt.Errorf("embed: got %d vectors for %d texts", len(vectors), count) + return fmt.Errorf("sdd: embedder returned %d vectors for %d chunks", len(vectors), count) } dims := 0 for _, vector := range vectors { if len(vector) == 0 { - return fmt.Errorf("embed: empty vector") + return fmt.Errorf("sdd: empty vector") } if dims == 0 { dims = len(vector) } if len(vector) != dims { - return fmt.Errorf("embed: inconsistent vector dimensions") + return fmt.Errorf("sdd: inconsistent vector dimensions") } norm := float64(0) for _, v := range vector { if math.IsNaN(float64(v)) || math.IsInf(float64(v), 0) { - return fmt.Errorf("embed: non-finite vector") + return fmt.Errorf("sdd: non-finite vector") } norm += float64(v) * float64(v) } if norm == 0 { - return fmt.Errorf("embed: zero vector") + return fmt.Errorf("sdd: zero vector") } } return nil diff --git a/pkg/application/doc.go b/pkg/application/doc.go index b49ed486..5e5dc110 100644 --- a/pkg/application/doc.go +++ b/pkg/application/doc.go @@ -117,10 +117,11 @@ // enqueue. Deduplicate indexing by full SearchEntryVersion, and run IndexSearchEntry // with source retention through retries. Queue state never establishes coverage. // -// Share one document batcher per embedding configuration and process. Compose -// query routing separately and provider deadlines and observation inside it. -// Configure explicit limits and measure provider/query latency in the consumer's -// workload; cross-process limits belong to the consumer. See embed.Batcher. +// Consumers own concurrent document batching behind embed.Embedder. Compose +// query routing separately and provider deadlines and observation per provider +// call. embed.Batched splits oversized requests; it does not combine callers. +// Local CLI and MCP indexing share incremental synchronous packing and publish +// complete entries as batches finish. Chunk preparation retains only active work. // // Deploy publication-aware retrieval before asynchronous writers. The derivation // schema participates in entry hashes, so prior rows can remain stored while diff --git a/pkg/application/search_index_exercise_test.go b/pkg/application/search_index_exercise_test.go index 6c90f3d5..bed01510 100644 --- a/pkg/application/search_index_exercise_test.go +++ b/pkg/application/search_index_exercise_test.go @@ -28,7 +28,6 @@ type exerciseStats struct { mu sync.Mutex Calls int `json:"calls"` Tokens int `json:"tokens"` - Shared int `json:"shared_batches"` MaxItems int `json:"max_items"` } @@ -38,9 +37,6 @@ func (s *exerciseStats) RecordCall(ctx context.Context, stat llm.CallStat) { s.Calls++ s.Tokens += stat.Usage.InputTokens s.MaxItems = max(s.MaxItems, stat.Items) - if len(embed.Attribution(ctx).Callers) > 1 { - s.Shared++ - } } type exerciseStore struct { @@ -61,7 +57,6 @@ type exerciseReport struct { Failed int `json:"failed"` Calls int `json:"calls"` Tokens int `json:"tokens"` - Shared int `json:"shared_batches"` MaxItems int `json:"max_items"` Peak int32 `json:"peak_provider_calls"` Elapsed time.Duration `json:"elapsed"` @@ -115,17 +110,9 @@ func TestEntryIndexingExerciseWorker(t *testing.T) { return inner.Embed(ctx, req) }} observed := embed.Observed(embed.Bounded(tracked, 2*time.Minute), stats) - batcher, err := embed.NewBatcher(t.Context(), observed, embed.BatchOptions{MaxItems: 32, MaxBytes: 128 * 1024, BufferItems: 4, Window: 10 * time.Millisecond, Concurrency: concurrency}) - if err != nil { - t.Fatal(err) - } - defer func() { - if err := batcher.Close(t.Context()); err != nil { - t.Error(err) - } - }() + documents := embed.Batched(observed, 32) store := exerciseStore{PersistentSearchIndexStore: local.NewPersistentSearchIndexStore("exercise", filepath.Join(root, "index"), "exercise"), fail: os.Getenv("SDD_INDEX_EXERCISE_FAIL") == "1"} - runtime, err := sdd.NewProjectRuntime(sdd.ProjectRuntimeOptions{Project: sdd.ProjectRef{ID: "exercise"}, Graph: graph, SearchIndex: store, Embedder: batcher, ExcludeEmbeddedFromIndex: true, LLM: llm.RunnerFunc(func(context.Context, llm.Request) (llm.Result, error) { return llm.Result{}, nil })}) + runtime, err := sdd.NewProjectRuntime(sdd.ProjectRuntimeOptions{Project: sdd.ProjectRef{ID: "exercise"}, Graph: graph, SearchIndex: store, Embedder: documents, ExcludeEmbeddedFromIndex: true, LLM: llm.RunnerFunc(func(context.Context, llm.Request) (llm.Result, error) { return llm.Result{}, nil })}) if err != nil { t.Fatal(err) } @@ -143,10 +130,10 @@ func TestEntryIndexingExerciseWorker(t *testing.T) { report := exerciseReport{} started := time.Now() var workers sync.WaitGroup - for range 8 { + for range concurrency { workers.Go(func() { for entry := range jobs { - ctx := embed.WithCaller(t.Context(), entry.Version.EntryID) + ctx := t.Context() err := runtime.IndexSearchEntry(ctx, sdd.IndexSearchEntryCmd{Entry: entry, OnPublished: func(id string, _ int) { mu.Lock(); report.Published++; fmt.Println("COMMITTED", id); mu.Unlock() }}) if err != nil { mu.Lock() @@ -180,7 +167,6 @@ func TestEntryIndexingExerciseWorker(t *testing.T) { stats.mu.Lock() report.Calls = stats.Calls report.Tokens = stats.Tokens - report.Shared = stats.Shared report.MaxItems = stats.MaxItems stats.mu.Unlock() report.Peak = peak.Load() diff --git a/pkg/application/types/search_entry.go b/pkg/application/types/search_entry.go index 0bac217c..2f24d0ea 100644 --- a/pkg/application/types/search_entry.go +++ b/pkg/application/types/search_entry.go @@ -2,7 +2,8 @@ package types import ( "fmt" - "math" + + "github.com/networkteam/sdd/internal/model/vectors" ) // SearchEntryVersion is the publication and deduplication key. Revision is @@ -31,31 +32,13 @@ func ValidateEntryPublication(version SearchEntryVersion, chunks []IndexedChunk) return fmt.Errorf("sdd: incomplete entry version") } seen := make(map[string]bool, len(chunks)) - dims := 0 - for _, row := range chunks { + embeddings := make([][]float32, len(chunks)) + for i, row := range chunks { if row.Chunk.EntryID != version.EntryID || row.Chunk.EntryHash != version.EntryHash || row.Chunk.ID == "" || seen[row.Chunk.ID] { return fmt.Errorf("sdd: invalid or duplicate chunk identity %q", row.Chunk.ID) } seen[row.Chunk.ID] = true - if len(row.Vector) == 0 { - return fmt.Errorf("sdd: empty vector for %s", row.Chunk.ID) - } - if dims == 0 { - dims = len(row.Vector) - } - if len(row.Vector) != dims { - return fmt.Errorf("sdd: inconsistent vector dimensions") - } - norm := float64(0) - for _, v := range row.Vector { - if math.IsNaN(float64(v)) || math.IsInf(float64(v), 0) { - return fmt.Errorf("sdd: non-finite vector") - } - norm += float64(v) * float64(v) - } - if norm == 0 { - return fmt.Errorf("sdd: zero vector") - } + embeddings[i] = row.Vector } - return nil + return vectors.Validate(embeddings, len(chunks)) } diff --git a/pkg/llm/embed/batcher.go b/pkg/llm/embed/batcher.go deleted file mode 100644 index be8db7d6..00000000 --- a/pkg/llm/embed/batcher.go +++ /dev/null @@ -1,298 +0,0 @@ -package embed - -import ( - "context" - "crypto/rand" - "errors" - "fmt" - "sync" - "sync/atomic" - "time" - - "github.com/networkteam/sdd/pkg/llm" -) - -var ErrBatcherClosed = errors.New("embed: batcher closed") - -// BatchOptions bounds admitted document work separately from provider calls. -// MaxBytes is a UTF-8 input bound, not a provider token or wire-payload bound. -// Measure and MaxUnits optionally impose a provider-specific aggregate limit; -// the adapter must still enforce its exact wire contract. A single text above -// either limit fails explicitly. Entries with many texts are admitted one at -// a time and may span arbitrarily many batches. -type BatchOptions struct { - MaxItems int - MaxBytes int - BufferItems int - Window time.Duration - Concurrency int - MaxUnits int - Measure func(string) (int, error) -} - -// Batcher combines document requests over one fixed embedder configuration. -// Route queries separately and compose Bounded inside the batcher for -// provider deadlines. Shared calls use the batcher's lifetime, never a caller's -// context. Place Observed inside Batcher: document results carry zero Usage -// because usage belongs to provider batches, not to participating callers. -type Batcher struct { - inner Embedder - options BatchOptions - ctx context.Context - cancel context.CancelFunc - items chan batchItem - batches chan []batchItem - done chan struct{} - wg sync.WaitGroup - sequence atomic.Uint64 - instance string -} - -type batchItem struct { - text string - index int - units int - queued time.Time - caller context.Context - reply chan batchReply -} - -type batchReply struct { - index int - vector []float32 - identity llm.Identity - err error -} - -type callerKey struct{} -type attributionKey struct{} - -type BatchAttribution struct { - ID string - Callers []string -} - -// WithCaller attaches an opaque correlation ID, such as a host's job attempt. -// It affects observation only; no queue or retry semantics enter the batcher. -func WithCaller(ctx context.Context, id string) context.Context { - return context.WithValue(ctx, callerKey{}, id) -} - -// Attribution returns the shared-call identity available to an inner StatsSink. -// Callers is a set; usage must be recorded once against ID, never per caller. -func Attribution(ctx context.Context) BatchAttribution { - a, _ := ctx.Value(attributionKey{}).(BatchAttribution) - a.Callers = append([]string(nil), a.Callers...) - return a -} - -func NewBatcher(ctx context.Context, inner Embedder, options BatchOptions) (*Batcher, error) { - if inner == nil || options.MaxItems < 1 || options.MaxBytes < 1 || options.BufferItems < 1 || options.Concurrency < 1 || options.Window <= 0 || (options.Measure == nil) != (options.MaxUnits == 0) || options.MaxUnits < 0 { - return nil, fmt.Errorf("embed: invalid batch options") - } - lifetime, cancel := context.WithCancel(ctx) - b := &Batcher{instance: rand.Text(), inner: inner, options: options, ctx: lifetime, cancel: cancel, items: make(chan batchItem, options.BufferItems), batches: make(chan []batchItem), done: make(chan struct{})} - b.wg.Add(1 + options.Concurrency) - go b.collect() - for range options.Concurrency { - go b.work() - } - go func() { b.wg.Wait(); close(b.done) }() - return b, nil -} - -func (b *Batcher) Fingerprint() string { return b.inner.Fingerprint() } - -func (b *Batcher) Embed(ctx context.Context, req Request) (Result, error) { - if err := ctx.Err(); err != nil { - return Result{}, err - } - if b.ctx.Err() != nil { - return Result{}, ErrBatcherClosed - } - if req.Purpose != PurposeDocument { - return Result{}, fmt.Errorf("embed: unsupported batch purpose %q", req.Purpose) - } - units := make([]int, len(req.Texts)) - for i, text := range req.Texts { - if err := ctx.Err(); err != nil { - return Result{}, err - } - if b.ctx.Err() != nil { - return Result{}, ErrBatcherClosed - } - if len(text) > b.options.MaxBytes { - return Result{}, fmt.Errorf("embed: text exceeds batch byte limit") - } - if b.options.Measure != nil { - value, err := b.options.Measure(text) - if err != nil { - return Result{}, err - } - if value < 0 || value > b.options.MaxUnits { - return Result{}, fmt.Errorf("embed: text exceeds provider unit limit") - } - units[i] = value - } - } - result := Result{Vectors: make([][]float32, len(req.Texts))} - replies := make(chan batchReply, min(len(req.Texts), b.options.MaxItems)) - caller, cancel := context.WithCancel(ctx) - defer cancel() - submitted, received := 0, 0 - var next *batchItem - for received < len(req.Texts) { - var admission chan batchItem - var item batchItem - if submitted < len(req.Texts) { - if next == nil { - text := req.Texts[submitted] - - next = &batchItem{text: text, index: submitted, units: units[submitted], caller: caller, reply: replies, queued: time.Now()} - } - admission, item = b.items, *next - } - select { - case admission <- item: - submitted++ - next = nil - case reply := <-replies: - if reply.err != nil { - return Result{}, reply.err - } - result.Vectors[reply.index] = reply.vector - result.Identity = reply.identity - received++ - case <-ctx.Done(): - return Result{}, ctx.Err() - case <-b.ctx.Done(): - return Result{}, ErrBatcherClosed - } - } - - return result, nil -} - -func (b *Batcher) collect() { - defer b.wg.Done() - var batch []batchItem - var timer *time.Timer - var tick <-chan time.Time - bytes, units := 0, 0 - stop := func() { - if timer != nil { - timer.Stop() - } - timer = nil - tick = nil - } - defer stop() - flush := func() bool { - stop() - if len(batch) == 0 { - return true - } - select { - case b.batches <- batch: - batch = nil - bytes, units = 0, 0 - return true - case <-b.ctx.Done(): - return false - } - } - for { - select { - case <-b.ctx.Done(): - return - case <-tick: - if !flush() { - return - } - case item := <-b.items: - if item.caller.Err() != nil { - continue - } - if len(batch) > 0 && (bytes+len(item.text) > b.options.MaxBytes || (b.options.Measure != nil && units+item.units > b.options.MaxUnits)) { - if !flush() { - return - } - } - if len(batch) == 0 { - timer = time.NewTimer(max(0, time.Until(item.queued.Add(b.options.Window)))) - tick = timer.C - } - batch = append(batch, item) - bytes += len(item.text) - units += item.units - if len(batch) == b.options.MaxItems { - if !flush() { - return - } - } - } - } -} - -func (b *Batcher) work() { - defer b.wg.Done() - for { - select { - case <-b.ctx.Done(): - return - case items := <-b.batches: - b.dispatch(items) - } - } -} - -func (b *Batcher) dispatch(items []batchItem) { - active := items[:0] - texts := make([]string, 0, len(items)) - callers := []string{} - seen := map[string]bool{} - for _, item := range items { - if item.caller.Err() != nil { - continue - } - active = append(active, item) - texts = append(texts, item.text) - id, _ := item.caller.Value(callerKey{}).(string) - if id != "" && !seen[id] { - seen[id] = true - callers = append(callers, id) - } - } - if len(active) == 0 { - return - } - id := fmt.Sprintf("%s-%d", b.instance, b.sequence.Add(1)) - call := context.WithValue(b.ctx, attributionKey{}, BatchAttribution{ID: id, Callers: callers}) - result, err := b.inner.Embed(call, Request{Purpose: PurposeDocument, Texts: texts}) - if err == nil { - err = validateBatchVectors(result.Vectors, len(active)) - } - for i, item := range active { - reply := batchReply{index: item.index, identity: result.Identity, err: err} - if err == nil { - reply.vector = result.Vectors[i] - } - select { - case item.reply <- reply: - case <-item.caller.Done(): - case <-b.ctx.Done(): - } - } -} - -// Close cancels queued and in-flight work. Callers receive explicit failure; -// no durable retry is performed. ctx bounds waiting for providers to stop. -func (b *Batcher) Close(ctx context.Context) error { - b.cancel() - select { - case <-b.done: - return nil - case <-ctx.Done(): - return ctx.Err() - } -} diff --git a/pkg/llm/embed/batcher_test.go b/pkg/llm/embed/batcher_test.go deleted file mode 100644 index 2178fde9..00000000 --- a/pkg/llm/embed/batcher_test.go +++ /dev/null @@ -1,359 +0,0 @@ -package embed_test - -import ( - "context" - "errors" - "fmt" - "strconv" - "sync" - "sync/atomic" - "testing" - "testing/synctest" - "time" - - "github.com/networkteam/sdd/pkg/llm" - "github.com/networkteam/sdd/pkg/llm/embed" -) - -func batchOptions() embed.BatchOptions { - return embed.BatchOptions{MaxItems: 8, MaxBytes: 1024, BufferItems: 2, Window: 15 * time.Millisecond, Concurrency: 2} -} -func newBatcher(t *testing.T, run func(context.Context, embed.Request) (embed.Result, error), options embed.BatchOptions) *embed.Batcher { - t.Helper() - b, err := embed.NewBatcher(t.Context(), embed.EmbedderFunc{Space: "fixed", Run: run}, options) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - if err := b.Close(ctx); err != nil { - t.Error(err) - } - }) - return b -} -func numberedVectors(req embed.Request) embed.Result { - r := embed.Result{Vectors: make([][]float32, len(req.Texts)), Usage: llm.Usage{InputTokens: len(req.Texts)}} - for i, text := range req.Texts { - n, _ := strconv.Atoi(text) - r.Vectors[i] = []float32{float32(n + 1), 1} - } - return r -} - -func TestBatcherCrossCallerRoutingAndOversizedAdmission(t *testing.T) { - var mu sync.Mutex - calls, active, peak := 0, 0, 0 - var attributed []embed.BatchAttribution - b := newBatcher(t, func(ctx context.Context, req embed.Request) (embed.Result, error) { - mu.Lock() - calls++ - active++ - peak = max(peak, active) - attributed = append(attributed, embed.Attribution(ctx)) - mu.Unlock() - if len(req.Texts) > 8 { - t.Error("oversized provider batch") - } - timer := time.NewTimer(2 * time.Millisecond) - defer timer.Stop() - select { - case <-timer.C: - case <-ctx.Done(): - return embed.Result{}, ctx.Err() - } - mu.Lock() - active-- - mu.Unlock() - return numberedVectors(req), nil - }, batchOptions()) - var wg sync.WaitGroup - for caller := range 12 { - wg.Go(func() { - texts := make([]string, 25) - for i := range texts { - texts[i] = strconv.Itoa(caller*25 + i) - } - result, err := b.Embed(embed.WithCaller(t.Context(), fmt.Sprintf("job-%d", caller)), embed.Request{Purpose: embed.PurposeDocument, Texts: texts}) - if err != nil { - t.Error(err) - return - } - if result.Usage.InputTokens != 0 { - t.Error("usage duplicated across callers") - } - for i, v := range result.Vectors { - if v[0] != float32(caller*25+i+1) { - t.Errorf("misrouted vector %d: %v", i, v) - } - } - }) - } - wg.Wait() - if calls >= 300 || peak > 2 || peak < 2 { - t.Fatalf("calls=%d peak=%d", calls, peak) - } - shared := false - for _, a := range attributed { - if a.ID == "" { - t.Error("missing batch ID") - } - shared = shared || len(a.Callers) > 1 - } - if !shared { - t.Fatal("no cross-caller batch") - } -} - -func TestBatcherTailWindowAndPayloadFlushing(t *testing.T) { - options := batchOptions() - options.MaxBytes = 3 - options.Window = 20 * time.Millisecond - var calls atomic.Int32 - b := newBatcher(t, func(_ context.Context, req embed.Request) (embed.Result, error) { - calls.Add(1) - size := 0 - for _, s := range req.Texts { - size += len(s) - } - if size > 3 { - t.Error("payload overflow") - } - return numberedVectors(req), nil - }, options) - started := time.Now() - result, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"11", "22", "33"}}) - if err != nil || len(result.Vectors) != 3 { - t.Fatalf("%v %v", result, err) - } - if calls.Load() != 3 || time.Since(started) > time.Second { - t.Fatalf("tail not flushed: calls=%d elapsed=%s", calls.Load(), time.Since(started)) - } - if _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"too long"}}); err == nil { - t.Fatal("oversized text accepted") - } -} - -func TestBatcherCancellationDoesNotCancelSharedProvider(t *testing.T) { - started := make(chan struct{}) - release := make(chan struct{}) - options := batchOptions() - options.MaxItems = 2 - options.Concurrency = 1 - options.Window = time.Second - b := newBatcher(t, func(ctx context.Context, req embed.Request) (embed.Result, error) { - close(started) - select { - case <-release: - return numberedVectors(req), nil - case <-ctx.Done(): - return embed.Result{}, ctx.Err() - } - }, options) - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - first, second := make(chan error, 1), make(chan error, 1) - go func() { - _, err := b.Embed(ctx, embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1"}}) - first <- err - }() - go func() { - _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"2"}}) - second <- err - }() - <-started - cancel() - if err := <-first; !errors.Is(err, context.Canceled) { - t.Fatalf("first=%v", err) - } - close(release) - if err := <-second; err != nil { - t.Fatalf("second=%v", err) - } -} - -func TestBatcherShutdownReleasesBlockedAdmission(t *testing.T) { - started := make(chan struct{}) - options := batchOptions() - options.MaxItems = 1 - options.Concurrency = 1 - options.BufferItems = 1 - b := newBatcher(t, func(ctx context.Context, _ embed.Request) (embed.Result, error) { - close(started) - <-ctx.Done() - return embed.Result{}, ctx.Err() - }, options) - result := make(chan error, 1) - go func() { - _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1", "2", "3", "4", "5"}}) - result <- err - }() - <-started - if err := b.Close(t.Context()); err != nil { - t.Fatal(err) - } - if err := <-result; !errors.Is(err, embed.ErrBatcherClosed) && !errors.Is(err, context.Canceled) { - t.Fatalf("shutdown=%v", err) - } -} - -func TestBatcherPropagatesProviderAndVectorFailures(t *testing.T) { - providerFailure := errors.New("provider failure") - for _, kind := range []string{"provider", "count", "empty", "dimensions", "zero"} { - t.Run(kind, func(t *testing.T) { - b := newBatcher(t, func(_ context.Context, req embed.Request) (embed.Result, error) { - r := numberedVectors(req) - switch kind { - case "provider": - return embed.Result{}, providerFailure - case "count": - r.Vectors = nil - case "empty": - r.Vectors[0] = nil - case "dimensions": - r.Vectors[0] = []float32{1} - case "zero": - r.Vectors[0] = []float32{0, 0} - } - return r, nil - }, batchOptions()) - _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1", "2"}}) - if err == nil { - t.Fatal("failure swallowed") - } - if kind == "provider" && !errors.Is(err, providerFailure) { - t.Fatalf("provider cause lost: %v", err) - } - }) - } -} - -func TestBatcherComposesDeadlineAndQueryRouting(t *testing.T) { - provider := embed.EmbedderFunc{Space: "fixed", Run: func(ctx context.Context, req embed.Request) (embed.Result, error) { - if req.Purpose == embed.PurposeQuery { - return numberedVectors(req), nil - } - <-ctx.Done() - return embed.Result{}, ctx.Err() - }} - documents, err := embed.NewBatcher(t.Context(), embed.Bounded(provider, 30*time.Millisecond), batchOptions()) - if err != nil { - t.Fatal(err) - } - defer func() { - if err := documents.Close(t.Context()); err != nil { - t.Error(err) - } - }() - routed := embed.EmbedderFunc{Space: provider.Fingerprint(), Run: func(ctx context.Context, req embed.Request) (embed.Result, error) { - if req.Purpose == embed.PurposeQuery { - return provider.Embed(ctx, req) - } - return documents.Embed(ctx, req) - }} - docResult := make(chan error, 1) - go func() { - _, err := routed.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1"}}) - docResult <- err - }() - if _, err := routed.Embed(t.Context(), embed.Request{Purpose: embed.PurposeQuery, Texts: []string{"2"}}); err != nil { - t.Fatal(err) - } - if err := <-docResult; !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("deadline=%v", err) - } -} - -func TestBatcherProviderUnitLimit(t *testing.T) { - options := batchOptions() - options.MaxUnits = 3 - options.Measure = func(text string) (int, error) { return len(text), nil } - b := newBatcher(t, func(_ context.Context, req embed.Request) (embed.Result, error) { - units := 0 - for _, text := range req.Texts { - units += len(text) - } - if units > 3 { - t.Errorf("provider units=%d", units) - } - return numberedVectors(req), nil - }, options) - result, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"11", "22", "33"}}) - if err != nil || len(result.Vectors) != 3 { - t.Fatalf("result=%+v error=%v", result, err) - } - if _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1234"}}); err == nil { - t.Fatal("oversized units accepted") - } -} - -func TestBatcherWindowStartsAtOldestItem(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - options := batchOptions() - options.Window = time.Second - called := make(chan int, 2) - b := newBatcher(t, func(_ context.Context, req embed.Request) (embed.Result, error) { - called <- len(req.Texts) - return numberedVectors(req), nil - }, options) - var wg sync.WaitGroup - submit := func(text string) { - wg.Go(func() { - if _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{text}}); err != nil { - t.Error(err) - } - }) - } - submit("1") - synctest.Wait() - time.Sleep(750 * time.Millisecond) - submit("2") - synctest.Wait() - time.Sleep(250 * time.Millisecond) - synctest.Wait() - select { - case count := <-called: - if count != 2 { - t.Fatalf("items=%d", count) - } - default: - t.Fatal("later arrival reset the flush window") - } - wg.Wait() - }) -} - -func TestBatcherRejectsEntireRequestBeforeAdmission(t *testing.T) { - for _, mode := range []string{"bytes", "units", "measurement"} { - t.Run(mode, func(t *testing.T) { - options := batchOptions() - options.MaxItems = 1 - options.MaxBytes = 2 - if mode != "bytes" { - options.MaxBytes = 100 - options.MaxUnits = 2 - options.Measure = func(text string) (int, error) { - if mode == "measurement" && text == "bad" { - return 0, errors.New("cannot measure") - } - return len(text), nil - } - } - var calls atomic.Int32 - b := newBatcher(t, func(_ context.Context, req embed.Request) (embed.Result, error) { - calls.Add(1) - return numberedVectors(req), nil - }, options) - if _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"1", "2", "bad"}}); err == nil { - t.Fatal("invalid trailing text accepted") - } - if _, err := b.Embed(t.Context(), embed.Request{Purpose: embed.PurposeDocument, Texts: []string{"3"}}); err != nil { - t.Fatal(err) - } - if calls.Load() != 1 { - t.Fatalf("invalid request dispatched work: calls=%d", calls.Load()) - } - }) - } -} From 8112ab2b89f6f24eec9df76d7150b51f19da532c Mon Sep 17 00:00:00 2001 From: Christopher Hlubek Date: Mon, 7 Sep 2026 15:42:16 +0200 Subject: [PATCH 4/5] sdd: signal tactical Implemented incremental synchronous indexing in commit 05f740fb, ... SDD-Mutation: entry-20260907-154209-s-tac-1j3 --- .sdd/graph/2026/09/07-154209-s-tac-1j3.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 .sdd/graph/2026/09/07-154209-s-tac-1j3.md diff --git a/.sdd/graph/2026/09/07-154209-s-tac-1j3.md b/.sdd/graph/2026/09/07-154209-s-tac-1j3.md new file mode 100644 index 00000000..2200cf12 --- /dev/null +++ b/.sdd/graph/2026/09/07-154209-s-tac-1j3.md @@ -0,0 +1,19 @@ +--- +type: signal +layer: tactical +kind: done +closes: + - 20260907-150733-d-tac-qiq +participants: + - Christopher +confidence: high +topics: + - implementation/search +summary: Implemented incremental synchronous indexing in commit 05f740fb, with CLI indexing, lazy fill, and MCP backfill sharing cross-entry chunk packing, durable published versions, and retry skipping, while removing NewBatcher and its exclusive exports. This completes the one-synchronous-indexing-path tactical plan (20260907-150733-d-tac-qiq). Tests cover incremental consumption, vector ownership, tail batches, progress callbacks, and interruption recovery; merge and release remain pending. +--- + +Implemented incremental synchronous indexing in commit 05f740fb, completing 20260907-150733-d-tac-qiq. + +CLI indexing, lazy fill and MCP backfill share cross-entry packing that derives chunks on demand and publishes complete versions, including empty and oversized entries. Published siblings survive failure and are skipped on retry. CLI progress counts published entries and reports chunks and batch activity. Fixed sources, synchronization scope, force rebuild, retention, eligibility and compatibility stores remain supported. NewBatcher and its exclusive exports are removed; Batched and Embedder remain. + +Behavior tests cover incremental consumption, vector ownership, tail batches, invalid vectors, durable progress callbacks and interruption recovery. The implementation is committed on worktree-incremental-indexing; merge and release remain pending. From ef2357a13c03b98923de13b4af3bd14a570ef1ab Mon Sep 17 00:00:00 2001 From: Christopher Hlubek Date: Mon, 7 Sep 2026 16:48:46 +0200 Subject: [PATCH 5/5] fix(cli): clear stale progress for empty entries --- cmd/sdd/progress.go | 18 ++++++++++++++++-- cmd/sdd/progress_test.go | 41 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/cmd/sdd/progress.go b/cmd/sdd/progress.go index 9378c790..f7c62fe1 100644 --- a/cmd/sdd/progress.go +++ b/cmd/sdd/progress.go @@ -47,15 +47,29 @@ func (p *embedProgress) onBatchStart(ids []string, chunks int) { func (p *embedProgress) onEntryIndexed(_ string, chunks int) { p.chunks += chunks p.reporter.Add(1) - p.reporter.SetNote(fmt.Sprintf("%s · %d chunks published", p.batchNote, p.chunks)) + prefix := p.batchNote + if chunks == 0 { + prefix = p.curRepo + p.batchNote = "" + } + note := fmt.Sprintf("%d chunks published", p.chunks) + if prefix != "" { + note = prefix + " · " + note + } + p.reporter.SetNote(note) } -func (p *embedProgress) onRepoStart(id string) { p.curRepo = id } +func (p *embedProgress) onRepoStart(id string) { + p.curRepo = id + p.batchNote = "" + p.reporter.SetNote("") +} // onPhase maps a handler-reported freshening phase onto the footer and clears // the stale embed note — a cache pull is not embedding any batch. func (p *embedProgress) onPhase(ph model.Phase) { if ph == model.PhaseConnecting || ph == model.PhaseSyncing { + p.batchNote = "" p.reporter.SetNote("") } p.reporter.SetPhase(ph) diff --git a/cmd/sdd/progress_test.go b/cmd/sdd/progress_test.go index 74672407..25ac1d62 100644 --- a/cmd/sdd/progress_test.go +++ b/cmd/sdd/progress_test.go @@ -55,3 +55,44 @@ func TestEmbedProgressCountsPublishedEntries(t *testing.T) { t.Fatalf("progress = %+v", progress) } } + +func TestEmbedProgressEmptyEntries(t *testing.T) { + for _, tc := range []struct { + name string + previousBatch bool + repo string + want string + }{ + {name: "first local entry", want: "0 chunks published"}, + {name: "local entry after batch", previousBatch: true, want: "3 chunks published"}, + {name: "first connected entry", repo: "current", want: "current · 0 chunks published"}, + {name: "new repository after batch", previousBatch: true, repo: "current", want: "current · 3 chunks published"}, + } { + t.Run(tc.name, func(t *testing.T) { + p := newEmbedProgress() + done := 1 + if tc.previousBatch { + cmd := p.connected(false) + if tc.repo != "" { + cmd.OnRepoStart("previous") + } + cmd.OnPlanned(1) + cmd.OnBatchStart([]string{"a"}, 3) + cmd.OnEntryIndexed("a", 3) + done++ + } + if tc.repo != "" { + cmd := p.connected(false) + cmd.OnRepoStart(tc.repo) + cmd.OnPhase(model.PhaseSyncing) + } + cmd := p.lazyFill() + cmd.OnPlanned(1) + cmd.OnEntryIndexed("empty", 0) + progress, ok := p.reporter.Recv() + if !ok || progress.Note != tc.want || progress.Done != done || progress.Total != done { + t.Fatalf("progress = %+v, want note %q and %d completed entries", progress, tc.want, done) + } + }) + } +}