refactor(search): unify incremental synchronous indexing

.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.

.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.

.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

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,17 +40,36 @@ 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)
+	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)

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,56 @@ 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)
+	}
+}
+
+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)
+			}
+		})
+	}
+}

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,

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,16 +43,16 @@ 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,
 	// naming the work in flight.
 	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,17 +85,17 @@ 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
 	// count, naming the work in flight.
 	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)
 }