fix(duckdb-watch): tombstone deletions on unwatched/deferred roots (#31)

* fix(duckdb-watch): add unwatched-root poller for local watch mode

Local duckdb watch omitted the probe-gated authoritative poller that pg
watch uses for deferred/unwatched roots. Deletions under those scopes were
never tombstoned in SQLite or the DuckDB mirror because interval pushes
only run SyncAll, which does not prove missed deletions.

Mirror pg watch: share one sync engine with the watcher, register polling
obligations for unwatched roots, and reconcile them via ReconcileWatchRoots.

* refactor(duckdb-watch): address PR review nits — comment + shared test helper

Author: diazMelgarejo

cmd/periscope/archive_write_backend.go

@@ -75,6 +75,10 @@ type archivePushWatchHooks struct {
 		context.Context, *syncpkg.Engine, bool,
 	) (bool, error)
 	newPGPusher        func(*syncpkg.Engine) *pgPusher
+	newDuckDBPusher    func(*syncpkg.Engine) *duckDBPusher
+	duckDBStartupSync  func(
+		context.Context, *syncpkg.Engine, bool,
+	) (bool, error)
 	newUnwatchedPoller func(context.Context, unwatchedPollSyncer) unwatchedRootPoller
 }
 
@@ -87,7 +91,7 @@ type unwatchedRootPoller interface {
 	Stop()
 }
 
-// newArchivePushUnwatchedPoller builds the pg watch polling owner for
+// newArchivePushUnwatchedPoller builds the pg/duckdb watch polling owner for
 // deferred scopes. The watcher's full recovery and rename promotion defer
 // unavailable scopes to their polling probes, and the interval push runs a
 // plain SyncAll that never tombstones missed deletions, so without this owner
@@ -723,53 +727,143 @@ func (b *localArchiveWriteBackend) DuckDBPushWatch(
 	if debounce <= 0 {
 		debounce = defaultWatchDebounce
 	}
-	push := func(pctx context.Context, reason pushReason, full bool) error {
-		pushCfg := cfg
-		pushCfg.Full = full
-		// Watch pushes are automatic: a mirror held by a live serve
-		// process defers instead of rebuilding the whole archive on
-		// every changed batch, and archive-scale diagnostics are
-		// skipped. Push ignores the defer behavior when full is set.
-		pushCfg.Automatic = true
-		var res duckdbsync.PushResult
-		var err error
-		if b.watchHooks != nil && b.watchHooks.duckDBPush != nil {
-			res, err = b.watchHooks.duckDBPush(pctx, reason, full)
-		} else {
-			res, err = b.DuckDBPush(
-				pctx, duckCfg, pushCfg, projects, exclude,
-			)
-		}
-		if err != nil {
-			return err
+	for _, def := range parser.Registry {
+		if !b.appCfg.IsUserConfigured(def.Type) {
+			continue
 		}
-		return completeDuckDBWatchPush(res, reason)
+		warnMissingDirs(b.appCfg.ResolveDirs(def.Type), string(def.Type))
 	}
+	cleanResyncTemp(b.appCfg.DBPath)
+
+	engine := syncpkg.NewEngine(b.database, syncpkg.EngineConfig{
+		AgentDirs:               b.appCfg.AgentDirs,
+		IncludeCwdPrefixes:      b.appCfg.SyncIncludeCwdPrefixes,
+		Machine:                 b.appCfg.LocalMachineName,
+		BlockedResultCategories: b.appCfg.ResultContentBlockedCategories,
+	})
+	defer engine.Close()
+
+	var pusher *duckDBPusher
+	if b.watchHooks != nil && b.watchHooks.newDuckDBPusher != nil {
+		pusher = b.watchHooks.newDuckDBPusher(engine)
+	} else {
+		pusher = b.newDuckDBPusher(
+			func(c context.Context) error {
+				stats := engine.SyncAll(c, nil)
+				if err := c.Err(); err != nil {
+					return err
+				}
+				if !stats.AuthoritativeDiscoveryComplete() {
+					return errors.New("local sync discovery incomplete")
+				}
+				engine.FlushSignals()
+				return nil
+			},
+			duckCfg, projects, exclude,
+		)
+	}
+
+	fmt.Printf(
+		"periscope duckdb watch: pushing to DuckDB "+
+			"(debounce %s, floor %s)\n",
+		debounce, interval,
+	)
+
 	loop, stopLoop := newArchivePushLoop(
 		b.watchHooks,
 		"duckdb watch", debounce, interval,
 		func(c context.Context, r pushReason) error {
-			return push(c, r, false)
+			pushCfg := cfg
+			pushCfg.Automatic = true
+			if b.watchHooks != nil && b.watchHooks.duckDBPush != nil {
+				res, err := b.watchHooks.duckDBPush(c, r, false)
+				if err != nil {
+					return err
+				}
+				return completeDuckDBWatchPush(res, r)
+			}
+			return pusher.push(c, r, false, pushCfg)
 		},
 	)
 	defer stopLoop()
 
+	poller := newArchivePushUnwatchedPoller(ctx, b.watchHooks, engine)
+	defer poller.Stop()
+
 	stopWatcher, openDispatch, unwatchedDirs := startArchivePushWatcher(
-		b.watchHooks, b.appCfg, nil,
+		b.watchHooks, b.appCfg, engine,
 		func(callbackCtx context.Context, batch syncpkg.WatchBatch) error {
+			scope := func() watchRecoveryScope {
+				return probeWatchRecoveryScope(b.appCfg)
+			}
+			if err := syncWatchBatch(callbackCtx, engine, batch, scope); err != nil {
+				return err
+			}
 			return notifyPushForWatchBatch(callbackCtx, loop, batch)
 		},
-		syncpkg.WatcherOptions{OnCoverageDegraded: loop.NotifyCoverageDegraded},
+		syncpkg.WatcherOptions{
+			OnCoverageDegraded: func(roots []string) error {
+				if err := poller.AddObligation(pollingObligation{
+					Key: "watcher-fallback", Roots: roots,
+				}); err != nil {
+					return err
+				}
+				return loop.NotifyCoverageDegraded(roots)
+			},
+			OnPollingRequired: func(obligation syncpkg.PollingObligation) error {
+				return poller.AddObligation(pollingObligation{
+					Key:   obligation.Key,
+					Roots: obligation.Roots,
+					Probe: obligation.Probe,
+				})
+			},
+			OnPollingReleased: poller.RemoveObligation,
+		},
 	)
 	defer stopWatcher()
 	if len(unwatchedDirs) > 0 {
 		log.Printf(
-			"duckdb watch: %d root(s) not watched; relying on the %s floor for coverage",
-			len(unwatchedDirs), interval,
+			"duckdb watch: %d root(s) not watched; polling every %s",
+			len(unwatchedDirs), unwatchedPollInterval,
 		)
 	}
-	initialErr := push(ctx, reasonStartup, cfg.Full)
+
+	startupSync := runPGWatchStartupSync
+	// DuckDB watch shares the same SyncAll-based startup path as pg watch;
+	// runPGWatchStartupSync is the shared default when no duckDBStartupSync
+	// hook overrides it (pgStartupSync is only a test harness fallback).
+	if b.watchHooks != nil && b.watchHooks.duckDBStartupSync != nil {
+		startupSync = b.watchHooks.duckDBStartupSync
+	} else if b.watchHooks != nil && b.watchHooks.pgStartupSync != nil {
+		startupSync = b.watchHooks.pgStartupSync
+	}
+	didResync, startupErr := startupSync(ctx, engine, cfg.Full)
+	if startupErr != nil && errors.Is(startupErr, context.Canceled) {
+		return nil
+	}
+	initialErr := startupErr
+	if initialErr == nil {
+		pushCfg := cfg
+		pushCfg.Automatic = true
+		if b.watchHooks != nil && b.watchHooks.duckDBPush != nil {
+			res, err := b.watchHooks.duckDBPush(
+				ctx, reasonStartup, cfg.Full || didResync,
+			)
+			if err != nil {
+				initialErr = err
+			} else {
+				initialErr = completeDuckDBWatchPush(res, reasonStartup)
+			}
+		} else {
+			initialErr = pusher.push(
+				ctx, reasonStartup, cfg.Full || didResync, pushCfg,
+			)
+		}
+	}
 	if initialErr != nil {
+		if errors.Is(initialErr, context.Canceled) && ctx.Err() != nil {
+			return nil
+		}
 		log.Printf("duckdb watch: initial push failed: %v", initialErr)
 	}
 	completePushWatchStartup(ctx, initialErr, loop, openDispatch)

cmd/periscope/archive_write_backend_test.go

@@ -810,15 +810,21 @@ func (noopPGTarget) PushWithOptions(
 func (noopPGTarget) Close() error { return nil }
 
 // The watcher's full recovery and rename promotion defer unavailable scopes
-// to their polling probes, and pg watch's interval push is a plain SyncAll
-// that never tombstones missed deletions. Local pg watch must therefore own
-// those deferred scopes with a probe-gated authoritative poller: a root the
-// watcher cannot cover (here a symlinked recursive root, the same obligation
-// machinery that owns roots missing at startup on portable backends)
-// registers a polling obligation, and the poller reconciles its scope
-// authoritatively on ticks — with no watcher event and no floor push
-// involved.
-func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
+// to their polling probes, and local push-watch interval pushes are a plain
+// SyncAll that never tombstones missed deletions. Local pg/duckdb watch must
+// therefore own those deferred scopes with a probe-gated authoritative poller.
+type localPushWatchPollingOwnerSpec struct {
+	obligationMsg string
+	exitMsg       string
+	sessionUUID   string
+	wireHooks     func(*archivePushWatchHooks)
+	runWatch      func(context.Context, *localArchiveWriteBackend) error
+}
+
+func testLocalPushWatchGivesDeferredScopesAPollingOwner(
+	t *testing.T, spec localPushWatchPollingOwnerSpec,
+) {
+	t.Helper()
 	dataDir := t.TempDir()
 	dbPath := filepath.Join(dataDir, "sessions.db")
 	database := dbtest.OpenTestDBAt(t, dbPath)
@@ -856,15 +862,6 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
 				label:    label,
 			}, func() {}
 		},
-		pgStartupSync: func(context.Context, *syncpkg.Engine, bool) (bool, error) {
-			return false, nil
-		},
-		newPGPusher: func(*syncpkg.Engine) *pgPusher {
-			return &pgPusher{
-				localSync: func(context.Context) error { return nil },
-				connect:   func() (pgTarget, error) { return noopPGTarget{}, nil },
-			}
-		},
 		newUnwatchedPoller: func(
 			ctx context.Context, engine unwatchedPollSyncer,
 		) unwatchedRootPoller {
@@ -876,41 +873,37 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
 			)
 		},
 	}
+	spec.wireHooks(backend.watchHooks)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 	done := make(chan error, 1)
-	go func() {
-		done <- backend.PGPushWatch(
-			ctx, pgTargetSelection{}, PGPushConfig{}, nil, nil,
-			time.Hour, time.Hour,
-		)
-	}()
+	go func() { done <- spec.runWatch(ctx, backend) }()
 
 	select {
 	case roots := <-owned:
-		assert.Contains(t, roots, codexRoot,
-			"the unwatchable root's polling obligation must reach the pg watch poller")
+		assert.Contains(t, roots, codexRoot, spec.obligationMsg)
 	case err := <-done:
-		t.Fatalf("pg watch exited before registering obligations: %v", err)
+		t.Fatalf("watch exited before registering obligations: %v", err)
 	case <-time.After(10 * time.Second):
 		t.Fatal("no polling obligation was registered for the unwatchable root")
 	}
 
 	// A session lands under the unwatched scope; only the poller's
 	// authoritative tick can bring it into the archive.
-	uuid := "d4e5f6a7-4444-4555-8666-777788889999"
 	day := filepath.Join(codexRoot, "2026", "05", "04")
 	require.NoError(t, os.MkdirAll(day, 0o755))
 	content := testjsonl.NewSessionBuilder().
 		AddCodexMeta(
-			"2026-05-04T14:00:00Z", uuid, "/home/user/code/api",
+			"2026-05-04T14:00:00Z", spec.sessionUUID, "/home/user/code/api",
 			"codex_cli_rs",
 		).
 		AddCodexMessage("2026-05-04T14:00:01Z", "user", "hello").
 		String()
 	require.NoError(t, os.WriteFile(
-		filepath.Join(day, "rollout-2026-05-04T14-31-58-"+uuid+".jsonl"),
+		filepath.Join(
+			day, "rollout-2026-05-04T14-31-58-"+spec.sessionUUID+".jsonl",
+		),
 		[]byte(content), 0o644,
 	))
 
@@ -919,7 +912,9 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
 		case ticks <- time.Now():
 		default:
 		}
-		session, err := database.GetSession(context.Background(), "codex:"+uuid)
+		session, err := database.GetSession(
+			context.Background(), "codex:"+spec.sessionUUID,
+		)
 		return err == nil && session != nil
 	}, 10*time.Second, 20*time.Millisecond,
 		"the returned root must be reconciled by the poller without watcher events or floor pushes")
@@ -929,10 +924,78 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
 	case err := <-done:
 		require.NoError(t, err)
 	case <-time.After(30 * time.Second):
-		t.Fatal("pg watch did not shut down")
+		t.Fatal(spec.exitMsg)
 	}
 }
 
+func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
+	testLocalPushWatchGivesDeferredScopesAPollingOwner(
+		t, localPushWatchPollingOwnerSpec{
+			obligationMsg: "the unwatchable root's polling obligation must reach the pg watch poller",
+			exitMsg:       "pg watch did not shut down",
+			sessionUUID:   "d4e5f6a7-4444-4555-8666-777788889999",
+			wireHooks: func(hooks *archivePushWatchHooks) {
+				hooks.pgStartupSync = func(
+					context.Context, *syncpkg.Engine, bool,
+				) (bool, error) {
+					return false, nil
+				}
+				hooks.newPGPusher = func(*syncpkg.Engine) *pgPusher {
+					return &pgPusher{
+						localSync: func(context.Context) error { return nil },
+						connect: func() (pgTarget, error) {
+							return noopPGTarget{}, nil
+						},
+					}
+				}
+			},
+			runWatch: func(
+				ctx context.Context, backend *localArchiveWriteBackend,
+			) error {
+				return backend.PGPushWatch(
+					ctx, pgTargetSelection{}, PGPushConfig{}, nil, nil,
+					time.Hour, time.Hour,
+				)
+			},
+		},
+	)
+}
+
+func TestLocalDuckDBPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) {
+	testLocalPushWatchGivesDeferredScopesAPollingOwner(
+		t, localPushWatchPollingOwnerSpec{
+			obligationMsg: "the unwatchable root's polling obligation must reach the duckdb watch poller",
+			exitMsg:       "duckdb watch did not shut down",
+			sessionUUID:   "e5f6a7b8-5555-4666-8777-888899990000",
+			wireHooks: func(hooks *archivePushWatchHooks) {
+				hooks.duckDBStartupSync = func(
+					context.Context, *syncpkg.Engine, bool,
+				) (bool, error) {
+					return false, nil
+				}
+				hooks.newDuckDBPusher = func(*syncpkg.Engine) *duckDBPusher {
+					return &duckDBPusher{
+						localSync: func(context.Context) error { return nil },
+						pushMirror: func(
+							context.Context, DuckDBPushConfig, bool,
+						) (duckdbsync.PushResult, error) {
+							return duckdbsync.PushResult{}, nil
+						},
+					}
+				}
+			},
+			runWatch: func(
+				ctx context.Context, backend *localArchiveWriteBackend,
+			) error {
+				return backend.DuckDBPushWatch(
+					ctx, config.DuckDBConfig{}, DuckDBPushConfig{}, nil, nil,
+					time.Hour, time.Hour,
+				)
+			},
+		},
+	)
+}
+
 func testLocalArchiveWriteBackend(t *testing.T) *localArchiveWriteBackend {
 	t.Helper()
 	dataDir := t.TempDir()