feat: add --merge flag to bd import for last-writer-wins conflict resolution

When importing a JSONL backup on a multi-machine setup, a stale snapshot
can reopen locally-closed issues because the upsert blindly overwrites
the status field. The new --merge flag compares updated_at timestamps
and skips updates where the local record is same age or newer, preventing
stale snapshots from overwriting local changes.

Default import behaviour (faithful snapshot restore) is unchanged.

Fixes pa-x34l.2.4

Author: rjc123

cmd/bd/auto_import_upgrade.go

@@ -38,7 +38,7 @@ func maybeAutoImportJSONL(ctx context.Context, s storage.DoltStorage, beadsDir s
 	// Database is empty but JSONL has data — auto-import.
 	fmt.Fprintf(os.Stderr, "auto-importing %d bytes from %s into empty database...\n", info.Size(), jsonlPath)
 
-	result, err := importFromLocalJSONLFull(ctx, s, jsonlPath)
+	result, err := importFromLocalJSONLFull(ctx, s, jsonlPath, false)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "warning: auto-import from %s failed: %v\n", jsonlPath, err)
 		fmt.Fprintf(os.Stderr, "\nYour issues are still safe in %s.\n", jsonlPath)

cmd/bd/import.go

@@ -28,17 +28,20 @@ brings new issues, 'bd import' loads them into the local Dolt database.
 EXAMPLES:
   bd import                        # Import from .beads/issues.jsonl
   bd import backup.jsonl           # Import from a specific file
-  bd import --dry-run              # Show what would be imported`,
+  bd import --dry-run              # Show what would be imported
+  bd import --merge                # Import, but skip issues where local is newer`,
 	GroupID: "sync",
 	RunE:    runImport,
 }
 
 var (
 	importDryRun bool
+	importMerge  bool
 )
 
 func init() {
 	importCmd.Flags().BoolVar(&importDryRun, "dry-run", false, "Show what would be imported without importing")
+	importCmd.Flags().BoolVar(&importMerge, "merge", false, "Use last-writer-wins: skip updates where the local record is newer (by updated_at)")
 	rootCmd.AddCommand(importCmd)
 }
 
@@ -78,7 +81,7 @@ func runImport(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("no database — run 'bd init' or 'bd bootstrap' first")
 	}
 
-	result, err := importFromLocalJSONLFull(ctx, store, jsonlPath)
+	result, err := importFromLocalJSONLFull(ctx, store, jsonlPath, importMerge)
 	if err != nil {
 		return fmt.Errorf("import failed: %w", err)
 	}

cmd/bd/import_from_jsonl_test.go

@@ -378,6 +378,180 @@ func TestImportFromLocalJSONL(t *testing.T) {
 	})
 }
 
+func TestImportMergeByTimestamp(t *testing.T) {
+	skipIfNoDolt(t)
+
+	t.Run("merge skips update when local record is newer", func(t *testing.T) {
+		tmpDir := t.TempDir()
+		dbPath := filepath.Join(tmpDir, "dolt")
+		store := newTestStore(t, dbPath)
+		ctx := context.Background()
+
+		// Step 1: Import an issue that is open, updated at T2.
+		initial := `{"id":"test-merge1","title":"Original","status":"open","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-06-01T00:00:00Z"}
+`
+		jsonlPath := filepath.Join(tmpDir, "issues.jsonl")
+		if err := os.WriteFile(jsonlPath, []byte(initial), 0644); err != nil {
+			t.Fatalf("write initial JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, false); err != nil {
+			t.Fatalf("initial import: %v", err)
+		}
+
+		// Step 2: Simulate local close — re-import with status=closed at a newer T3.
+		closed := `{"id":"test-merge1","title":"Original","status":"closed","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-09-01T00:00:00Z","closed_at":"2025-09-01T00:00:00Z","close_reason":"done"}
+`
+		if err := os.WriteFile(jsonlPath, []byte(closed), 0644); err != nil {
+			t.Fatalf("write closed JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, false); err != nil {
+			t.Fatalf("close import: %v", err)
+		}
+
+		// Verify issue is closed.
+		issue, err := store.GetIssue(ctx, "test-merge1")
+		if err != nil {
+			t.Fatalf("get issue after close: %v", err)
+		}
+		if issue.Status != "closed" {
+			t.Fatalf("expected status 'closed', got %q", issue.Status)
+		}
+
+		// Step 3: Import a STALE snapshot (T2, status=open) with --merge.
+		// This simulates pulling a backup from another machine that predates
+		// the local close.
+		stale := `{"id":"test-merge1","title":"Stale title","status":"open","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-06-01T00:00:00Z"}
+`
+		if err := os.WriteFile(jsonlPath, []byte(stale), 0644); err != nil {
+			t.Fatalf("write stale JSONL: %v", err)
+		}
+		result, err := importFromLocalJSONLFull(ctx, store, jsonlPath, true)
+		if err != nil {
+			t.Fatalf("merge import: %v", err)
+		}
+		if result.Issues != 1 {
+			t.Errorf("expected 1 issue processed, got %d", result.Issues)
+		}
+
+		// Verify the issue is STILL closed — the stale snapshot was skipped.
+		issue, err = store.GetIssue(ctx, "test-merge1")
+		if err != nil {
+			t.Fatalf("get issue after merge: %v", err)
+		}
+		if issue.Status != "closed" {
+			t.Errorf("merge allowed stale snapshot to reopen issue: status=%q, want 'closed'", issue.Status)
+		}
+		if issue.Title != "Original" {
+			t.Errorf("merge allowed stale snapshot to overwrite title: got %q, want 'Original'", issue.Title)
+		}
+	})
+
+	t.Run("merge allows update when incoming record is newer", func(t *testing.T) {
+		tmpDir := t.TempDir()
+		dbPath := filepath.Join(tmpDir, "dolt")
+		store := newTestStore(t, dbPath)
+		ctx := context.Background()
+
+		// Step 1: Import an issue at T1.
+		initial := `{"id":"test-merge2","title":"Old title","status":"open","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-01-01T00:00:00Z"}
+`
+		jsonlPath := filepath.Join(tmpDir, "issues.jsonl")
+		if err := os.WriteFile(jsonlPath, []byte(initial), 0644); err != nil {
+			t.Fatalf("write initial JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, false); err != nil {
+			t.Fatalf("initial import: %v", err)
+		}
+
+		// Step 2: Import a NEWER snapshot with --merge. This should update.
+		newer := `{"id":"test-merge2","title":"New title","status":"closed","priority":1,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-12-01T00:00:00Z","closed_at":"2025-12-01T00:00:00Z"}
+`
+		if err := os.WriteFile(jsonlPath, []byte(newer), 0644); err != nil {
+			t.Fatalf("write newer JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, true); err != nil {
+			t.Fatalf("merge import: %v", err)
+		}
+
+		issue, err := store.GetIssue(ctx, "test-merge2")
+		if err != nil {
+			t.Fatalf("get issue after merge: %v", err)
+		}
+		if issue.Status != "closed" {
+			t.Errorf("merge should have applied newer update: status=%q, want 'closed'", issue.Status)
+		}
+		if issue.Title != "New title" {
+			t.Errorf("merge should have applied newer title: got %q, want 'New title'", issue.Title)
+		}
+	})
+
+	t.Run("without merge flag stale snapshot overwrites", func(t *testing.T) {
+		// Control test: without --merge, the old behaviour applies.
+		tmpDir := t.TempDir()
+		dbPath := filepath.Join(tmpDir, "dolt")
+		store := newTestStore(t, dbPath)
+		ctx := context.Background()
+
+		// Import at T2 as closed.
+		closed := `{"id":"test-merge3","title":"Closed issue","status":"closed","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-09-01T00:00:00Z","closed_at":"2025-09-01T00:00:00Z"}
+`
+		jsonlPath := filepath.Join(tmpDir, "issues.jsonl")
+		if err := os.WriteFile(jsonlPath, []byte(closed), 0644); err != nil {
+			t.Fatalf("write JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, false); err != nil {
+			t.Fatalf("initial import: %v", err)
+		}
+
+		// Import stale (T1, open) WITHOUT merge — should overwrite.
+		stale := `{"id":"test-merge3","title":"Stale","status":"open","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-01-01T00:00:00Z"}
+`
+		if err := os.WriteFile(jsonlPath, []byte(stale), 0644); err != nil {
+			t.Fatalf("write stale JSONL: %v", err)
+		}
+		if _, err := importFromLocalJSONLFull(ctx, store, jsonlPath, false); err != nil {
+			t.Fatalf("stale import: %v", err)
+		}
+
+		issue, err := store.GetIssue(ctx, "test-merge3")
+		if err != nil {
+			t.Fatalf("get issue: %v", err)
+		}
+		if issue.Status != "open" {
+			t.Errorf("without merge, stale import should overwrite: status=%q, want 'open'", issue.Status)
+		}
+	})
+
+	t.Run("merge creates new issues normally", func(t *testing.T) {
+		tmpDir := t.TempDir()
+		dbPath := filepath.Join(tmpDir, "dolt")
+		store := newTestStore(t, dbPath)
+		ctx := context.Background()
+
+		jsonl := `{"id":"test-merge4","title":"Brand new","status":"open","priority":2,"issue_type":"task","created_at":"2025-01-01T00:00:00Z","updated_at":"2025-01-01T00:00:00Z"}
+`
+		jsonlPath := filepath.Join(tmpDir, "issues.jsonl")
+		if err := os.WriteFile(jsonlPath, []byte(jsonl), 0644); err != nil {
+			t.Fatalf("write JSONL: %v", err)
+		}
+		result, err := importFromLocalJSONLFull(ctx, store, jsonlPath, true)
+		if err != nil {
+			t.Fatalf("merge import of new issue: %v", err)
+		}
+		if result.Issues != 1 {
+			t.Errorf("expected 1 issue, got %d", result.Issues)
+		}
+
+		issue, err := store.GetIssue(ctx, "test-merge4")
+		if err != nil {
+			t.Fatalf("get issue: %v", err)
+		}
+		if issue.Title != "Brand new" {
+			t.Errorf("expected title 'Brand new', got %q", issue.Title)
+		}
+	})
+}
+
 func TestImportFromLocalJSONL_LegacyFormats(t *testing.T) {
 	skipIfNoDolt(t)
 

cmd/bd/import_shared.go

@@ -25,6 +25,7 @@ type ImportOptions struct {
 	DeletionIDs                []string
 	SkipPrefixValidation       bool
 	ProtectLocalExportIDs      map[string]time.Time
+	MergeByTimestamp           bool
 }
 
 // ImportResult describes what an import operation did.
@@ -53,6 +54,7 @@ func importIssuesCore(ctx context.Context, _ string, store storage.DoltStorage,
 	err := store.CreateIssuesWithFullOptions(ctx, issues, getActorWithGit(), storage.BatchCreateOptions{
 		OrphanHandling:       storage.OrphanAllow,
 		SkipPrefixValidation: opts.SkipPrefixValidation,
+		MergeByTimestamp:     opts.MergeByTimestamp,
 	})
 	if err != nil {
 		return nil, err
@@ -78,7 +80,7 @@ type memoryRecord struct {
 // into the Dolt store. Returns the number of issues imported and any error.
 // This is a convenience wrapper around importFromLocalJSONLFull.
 func importFromLocalJSONL(ctx context.Context, store storage.DoltStorage, localPath string) (int, error) {
-	result, err := importFromLocalJSONLFull(ctx, store, localPath)
+	result, err := importFromLocalJSONLFull(ctx, store, localPath, false)
 	if err != nil {
 		return 0, err
 	}
@@ -88,7 +90,9 @@ func importFromLocalJSONL(ctx context.Context, store storage.DoltStorage, localP
 // importFromLocalJSONLFull imports issues and memories from a local JSONL file.
 // It detects memory records (lines with "_type":"memory") and imports them
 // via SetConfig, while routing regular issue records through the normal path.
-func importFromLocalJSONLFull(ctx context.Context, store storage.DoltStorage, localPath string) (*importLocalResult, error) {
+// When merge is true, existing issues are only updated if the incoming record
+// has a strictly newer updated_at timestamp (last-writer-wins).
+func importFromLocalJSONLFull(ctx context.Context, store storage.DoltStorage, localPath string, merge bool) (*importLocalResult, error) {
 	//nolint:gosec // G304: path from user-provided CLI argument
 	data, err := os.ReadFile(localPath)
 	if err != nil {
@@ -182,6 +186,7 @@ func importFromLocalJSONLFull(ctx context.Context, store storage.DoltStorage, lo
 
 		opts := ImportOptions{
 			SkipPrefixValidation: true,
+			MergeByTimestamp:     merge,
 		}
 		_, err = importIssuesCore(ctx, "", store, issues, opts)
 		if err != nil {

internal/storage/batch.go

@@ -22,4 +22,10 @@ type BatchCreateOptions struct {
 	OrphanHandling OrphanHandling
 	// SkipPrefixValidation skips prefix validation for existing IDs (used during import)
 	SkipPrefixValidation bool
+	// MergeByTimestamp enables last-writer-wins conflict resolution during import.
+	// When true, existing issues are only updated if the incoming record has a
+	// newer updated_at timestamp than the local record. This prevents stale
+	// snapshots from overwriting locally-modified issues (e.g. reopening a
+	// locally-closed issue).
+	MergeByTimestamp bool
 }

internal/storage/issueops/create.go

@@ -84,7 +84,7 @@ func CreateIssueInTx(ctx context.Context, tx *sql.Tx, bc *BatchContext, issue *t
 		return nil
 	}
 
-	isNew, err := InsertIssueIfNew(ctx, tx, issueTable, issue)
+	isNew, err := InsertIssueIfNew(ctx, tx, issueTable, issue, bc.Opts.MergeByTimestamp)
 	if err != nil {
 		return err
 	}
@@ -237,19 +237,39 @@ func CheckOrphan(ctx context.Context, tx *sql.Tx, issue *types.Issue, issueTable
 }
 
 // InsertIssueIfNew inserts the issue and returns whether it was genuinely new.
+// When mergeByTimestamp is true, existing issues are only updated if the
+// incoming record's updated_at is strictly newer than the local record's.
 //
 //nolint:gosec // G201: table is a hardcoded constant
-func InsertIssueIfNew(ctx context.Context, tx *sql.Tx, issueTable string, issue *types.Issue) (isNew bool, err error) {
-	var existingCount int
+func InsertIssueIfNew(ctx context.Context, tx *sql.Tx, issueTable string, issue *types.Issue, mergeByTimestamp bool) (isNew bool, err error) {
 	if issue.ID != "" {
+		var existingCount int
 		if err := tx.QueryRowContext(ctx, fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE id = ?`, issueTable), issue.ID).Scan(&existingCount); err != nil {
 			return false, fmt.Errorf("failed to check issue existence for %s: %w", issue.ID, err)
 		}
+		if existingCount > 0 {
+			if mergeByTimestamp {
+				var localUpdatedAt time.Time
+				if err := tx.QueryRowContext(ctx, fmt.Sprintf(`SELECT updated_at FROM %s WHERE id = ?`, issueTable), issue.ID).Scan(&localUpdatedAt); err != nil {
+					return false, fmt.Errorf("failed to read updated_at for %s: %w", issue.ID, err)
+				}
+				if !issue.UpdatedAt.After(localUpdatedAt) {
+					// Local record is same age or newer — skip update.
+					return false, nil
+				}
+			}
+			// Existing issue, proceed with upsert.
+			if err := InsertIssueIntoTable(ctx, tx, issueTable, issue); err != nil {
+				return false, fmt.Errorf("failed to update issue %s: %w", issue.ID, err)
+			}
+			return false, nil
+		}
 	}
+	// New issue — insert.
 	if err := InsertIssueIntoTable(ctx, tx, issueTable, issue); err != nil {
 		return false, fmt.Errorf("failed to insert issue %s: %w", issue.ID, err)
 	}
-	return existingCount == 0, nil
+	return true, nil
 }
 
 // PersistLabels writes issue.Labels into the appropriate labels table.