feat(snapshot): add safe merge import plans (#42)

Author: steipete

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+- Add explicit monotonic snapshot merge planning and import-impact reporting so cache consumers can apply changed shards without silently replacing local rows, while exact mirrors retain replacement semantics.
+
 ## v0.13.1 - 2026-06-23
 
 - Harden crawlkit scheduler, output, release-check, vector ranking, and CI

README.md

@@ -27,7 +27,7 @@ See `docs/boundary.md` for the crawlkit-versus-app ownership boundary and
 - `config`: standard TOML config paths, opt-in platform-native runtime dirs,
   migration-safe legacy path fallback, and token diagnostics.
 - `store`: SQLite open/read-only/transaction/query helpers plus safe FTS5 term and optimization helpers.
-- `snapshot`: `manifest.json` plus JSONL/Gzip table snapshot export, file fingerprints, incremental import, and managed sidecar trees.
+- `snapshot`: `manifest.json` plus JSONL/Gzip table snapshot export, file fingerprints, exact or monotonic-merge import planning, impact classification, and managed sidecar trees.
 - `backup`: age-encrypted JSONL/Gzip shards, backup manifests, recipient/identity helpers, history listing, and historical-ref restore verification.
 - `mirror`: clone/init/pull/commit/push helpers plus non-mutating fetch, immutable tags, Git-object reads, and history inspection for private snapshot repos.
 - `state`: generic crawler cursor and freshness records, including mapped adapters for existing app table layouts.

snapshot/snapshot.go

@@ -111,6 +111,14 @@ type ImportPlan struct {
 	Tables []TableImportPlan
 }
 
+type ImportImpact string
+
+const (
+	ImportImpactNone    ImportImpact = "none"
+	ImportImpactMerge   ImportImpact = "merge"
+	ImportImpactReplace ImportImpact = "replace"
+)
+
 type TableImportPlan struct {
 	Table  TableManifest
 	Mode   TableImportMode
@@ -240,6 +248,17 @@ func Import(ctx context.Context, opts ImportOptions) (Manifest, error) {
 }
 
 func PlanIncrementalImport(previous, current Manifest) ImportPlan {
+	return planIncrementalImport(previous, current, false)
+}
+
+// PlanMergeImport prepares a monotonic cache refresh. Changed snapshot files
+// are upserted without deleting rows that disappeared from the snapshot. Use
+// PlanIncrementalImport when the destination must exactly mirror the source.
+func PlanMergeImport(previous, current Manifest) ImportPlan {
+	return planIncrementalImport(previous, current, true)
+}
+
+func planIncrementalImport(previous, current Manifest, merge bool) ImportPlan {
 	if current.Version != previous.Version {
 		return ImportPlan{Full: true, Reason: "manifest version changed"}
 	}
@@ -260,15 +279,21 @@ func PlanIncrementalImport(previous, current Manifest) ImportPlan {
 	for _, table := range current.Tables {
 		previousTable, ok := previousTables[table.Name]
 		if !ok {
+			mode := TableImportReplace
+			reason := "new table"
+			if merge {
+				mode = TableImportFiles
+				reason = "merge new table"
+			}
 			plan.Tables = append(plan.Tables, TableImportPlan{
 				Table:  table,
-				Mode:   TableImportReplace,
+				Mode:   mode,
 				Files:  tableFileManifests(table),
-				Reason: "new table",
+				Reason: reason,
 			})
 			continue
 		}
-		tablePlan := planTableIncrement(previousTable, table)
+		tablePlan := planTableIncrement(previousTable, table, merge)
 		plan.Tables = append(plan.Tables, tablePlan)
 	}
 	return plan
@@ -286,6 +311,22 @@ func (p ImportPlan) Changed() bool {
 	return false
 }
 
+func (p ImportPlan) Impact() ImportImpact {
+	if p.Full {
+		return ImportImpactReplace
+	}
+	impact := ImportImpactNone
+	for _, table := range p.Tables {
+		switch table.Mode {
+		case TableImportReplace:
+			return ImportImpactReplace
+		case TableImportFiles:
+			impact = ImportImpactMerge
+		}
+	}
+	return impact
+}
+
 func ImportIncremental(ctx context.Context, opts IncrementalImportOptions) (Manifest, ImportPlan, error) {
 	if opts.DB == nil {
 		return Manifest{}, ImportPlan{}, errors.New("db is required")
@@ -691,7 +732,7 @@ func exportValue(value any) any {
 	}
 }
 
-func planTableIncrement(previous, current TableManifest) TableImportPlan {
+func planTableIncrement(previous, current TableManifest, merge bool) TableImportPlan {
 	if !sameStrings(previous.Columns, current.Columns) {
 		return TableImportPlan{Table: current, Mode: TableImportReplace, Files: tableFileManifests(current), Reason: "columns changed"}
 	}
@@ -706,6 +747,9 @@ func planTableIncrement(previous, current TableManifest) TableImportPlan {
 	if sameFileManifests(previousFiles, currentFiles) {
 		return TableImportPlan{Table: current, Mode: TableImportSkip, Reason: "unchanged"}
 	}
+	if merge {
+		return planTableMerge(previousFiles, currentFiles, current)
+	}
 	if len(currentFiles) < len(previousFiles) {
 		return TableImportPlan{Table: current, Mode: TableImportReplace, Files: currentFiles, Reason: "files removed"}
 	}
@@ -734,6 +778,31 @@ func planTableIncrement(previous, current TableManifest) TableImportPlan {
 	return TableImportPlan{Table: current, Mode: TableImportFiles, Files: changed, Reason: "tail files changed"}
 }
 
+func planTableMerge(previousFiles, currentFiles []FileManifest, current TableManifest) TableImportPlan {
+	previousByPath := make(map[string]FileManifest, len(previousFiles))
+	for _, file := range previousFiles {
+		previousByPath[file.Path] = file
+	}
+	currentPaths := make(map[string]struct{}, len(currentFiles))
+	changed := make([]FileManifest, 0, len(currentFiles))
+	for _, file := range currentFiles {
+		currentPaths[file.Path] = struct{}{}
+		previous, ok := previousByPath[file.Path]
+		if !ok || !sameFileManifest(previous, file) {
+			changed = append(changed, file)
+		}
+	}
+	for _, file := range previousFiles {
+		if _, ok := currentPaths[file.Path]; !ok {
+			return TableImportPlan{Table: current, Mode: TableImportReplace, Files: currentFiles, Reason: "files removed"}
+		}
+	}
+	if len(changed) == 0 {
+		return TableImportPlan{Table: current, Mode: TableImportSkip, Reason: "unchanged"}
+	}
+	return TableImportPlan{Table: current, Mode: TableImportFiles, Files: changed, Reason: "merge changed files"}
+}
+
 func tableShardDir(table string) (string, error) {
 	table = strings.TrimSpace(table)
 	if table == "" {

snapshot/snapshot_test.go

@@ -268,6 +268,82 @@ func TestPlanIncrementalImportReplacesUnsafeTailChanges(t *testing.T) {
 	}
 }
 
+func TestPlanMergeImportUsesChangedFilesWithoutReplacement(t *testing.T) {
+	previous := Manifest{
+		Version: 1,
+		Tables: []TableManifest{{
+			Name:    "things",
+			Columns: []string{"id", "body"},
+			Files:   []string{"tables/things/000000.jsonl.gz", "tables/things/000001.jsonl.gz"},
+			FileManifests: []FileManifest{
+				{Path: "tables/things/000000.jsonl.gz", Rows: 2, Size: 100, SHA256: "same"},
+				{Path: "tables/things/000001.jsonl.gz", Rows: 1, Size: 50, SHA256: "old"},
+			},
+		}},
+	}
+	current := previous
+	current.Tables = append([]TableManifest(nil), previous.Tables...)
+	current.Tables[0].Files = append([]string(nil), previous.Tables[0].Files...)
+	current.Tables[0].FileManifests = []FileManifest{
+		{Path: "tables/things/000000.jsonl.gz", Rows: 2, Size: 100, SHA256: "same"},
+		{Path: "tables/things/000001.jsonl.gz", Rows: 2, Size: 75, SHA256: "new"},
+		{Path: "tables/things/000002.jsonl.gz", Rows: 1, Size: 25, SHA256: "added"},
+	}
+	current.Tables[0].Files = append(current.Tables[0].Files, "tables/things/000002.jsonl.gz")
+
+	plan := PlanMergeImport(previous, current)
+	if plan.Impact() != ImportImpactMerge {
+		t.Fatalf("impact = %q, plan = %+v", plan.Impact(), plan)
+	}
+	if len(plan.Tables) != 1 || plan.Tables[0].Mode != TableImportFiles {
+		t.Fatalf("plan = %+v", plan)
+	}
+	files := plan.Tables[0].Files
+	if len(files) != 2 || files[0].SHA256 != "new" || files[1].SHA256 != "added" {
+		t.Fatalf("files = %+v", files)
+	}
+}
+
+func TestPlanMergeImportStillRequiresReplacementForRemovedFiles(t *testing.T) {
+	previous := Manifest{Version: 1, Tables: []TableManifest{{
+		Name: "things", Columns: []string{"id"}, Files: []string{"one", "two"},
+		FileManifests: []FileManifest{{Path: "one", SHA256: "one"}, {Path: "two", SHA256: "two"}},
+	}}}
+	current := Manifest{Version: 1, Tables: []TableManifest{{
+		Name: "things", Columns: []string{"id"}, Files: []string{"one"},
+		FileManifests: []FileManifest{{Path: "one", SHA256: "one"}},
+	}}}
+
+	plan := PlanMergeImport(previous, current)
+	if plan.Impact() != ImportImpactReplace || plan.Tables[0].Reason != "files removed" {
+		t.Fatalf("plan = %+v", plan)
+	}
+}
+
+func TestPlanMergeImportMergesNewTables(t *testing.T) {
+	current := Manifest{Version: 1, Tables: []TableManifest{{
+		Name: "things", Columns: []string{"id"}, Files: []string{"one"},
+		FileManifests: []FileManifest{{Path: "one", SHA256: "one"}},
+	}}}
+
+	plan := PlanMergeImport(Manifest{Version: 1}, current)
+	if plan.Impact() != ImportImpactMerge || plan.Tables[0].Mode != TableImportFiles {
+		t.Fatalf("plan = %+v", plan)
+	}
+}
+
+func TestImportPlanImpact(t *testing.T) {
+	if impact := (ImportPlan{}).Impact(); impact != ImportImpactNone {
+		t.Fatalf("empty impact = %q", impact)
+	}
+	if impact := (ImportPlan{Tables: []TableImportPlan{{Mode: TableImportFiles}}}).Impact(); impact != ImportImpactMerge {
+		t.Fatalf("merge impact = %q", impact)
+	}
+	if impact := (ImportPlan{Full: true}).Impact(); impact != ImportImpactReplace {
+		t.Fatalf("full impact = %q", impact)
+	}
+}
+
 func TestImportIncrementalImportsOnlyPlannedFiles(t *testing.T) {
 	ctx := context.Background()
 	src, err := store.Open(ctx, store.Options{
@@ -390,6 +466,62 @@ func TestImportIncrementalReplacesChangedTailShard(t *testing.T) {
 	}
 }
 
+func TestImportIncrementalMergePreservesDestinationOnlyRows(t *testing.T) {
+	ctx := context.Background()
+	src, err := store.Open(ctx, store.Options{
+		Path:   filepath.Join(t.TempDir(), "src.db"),
+		Schema: `create table things(id text primary key, body text not null);`,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer src.Close()
+	mustExec(t, src.DB(), `insert into things(id, body) values('one', 'old')`)
+	root := t.TempDir()
+	previous, err := Export(ctx, ExportOptions{DB: src.DB(), RootDir: root, Tables: []string{"things"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	dst, err := store.Open(ctx, store.Options{
+		Path:   filepath.Join(t.TempDir(), "dst.db"),
+		Schema: `create table things(id text primary key, body text not null);`,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer dst.Close()
+	if _, err := Import(ctx, ImportOptions{DB: dst.DB(), RootDir: root}); err != nil {
+		t.Fatal(err)
+	}
+	mustExec(t, dst.DB(), `insert into things(id, body) values('local', 'keep')`)
+	mustExec(t, src.DB(), `update things set body = 'new' where id = 'one'`)
+	mustExec(t, src.DB(), `insert into things(id, body) values('two', 'added')`)
+	current, err := Export(ctx, ExportOptions{DB: src.DB(), RootDir: root, Tables: []string{"things"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	plan := PlanMergeImport(previous, current)
+	if plan.Impact() != ImportImpactMerge {
+		t.Fatalf("plan = %+v", plan)
+	}
+	if _, _, err := ImportIncremental(ctx, IncrementalImportOptions{
+		DB:      dst.DB(),
+		RootDir: root,
+		Current: current,
+		Plan:    plan,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var got string
+	if err := dst.DB().QueryRowContext(ctx, `select group_concat(id || ':' || body, ',') from (select id, body from things order by id)`).Scan(&got); err != nil {
+		t.Fatal(err)
+	}
+	if got != "local:keep,one:new,two:added" {
+		t.Fatalf("things = %q", got)
+	}
+}
+
 func TestImportHooks(t *testing.T) {
 	ctx := context.Background()
 	src, err := store.Open(ctx, store.Options{