fix(store): persist periodic SQLite cleanup and retry failed deletes

docs/architecture/02-edge-server.md

@@ -66,7 +66,7 @@ ProcessExecutor 配置 `RunTimeout`（默认 30 分钟）、`ShutdownGracePeriod
 
 `FileStore`（`internal/store/file_store.go`）对写入信号做 debounce，再保存全量 JSON 快照；显式 `Flush` 同步执行快照、编码和文件 Sync/rename。`SQLiteStore`（`internal/store/sqlite_store.go`）在写入后同步持久化快照差分到 SQLite（WAL 模式，定期 checkpoint），支持崩溃恢复。SQL 连接初始化与持久化串行化独立于普通业务读面，不能用额外 SQL 读者的争用直接代替业务读取测量。
 
-终端状态 runs（finished/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items/checkpoints。Checkpoint 在 run 完成时保留，在 run 清理或所属 thread 删除时随 run 移除；这不删除工作区文件。
+终端状态 runs（finished/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items/checkpoints。Checkpoint 在 run 完成时保留，在 run 清理或所属 thread 删除时随 run 移除；这不删除工作区文件。 SQLite 后台清理有删除时同步提交；失败通过 `LastPersistError` 和日志留痕，下一周期即使没有新删除也会重试。关闭时先停止并等待后台清理/checkpoint 任务，再做最终持久化与数据库关闭。
 
 `EventBus`（`internal/events/bus.go`）是基于 channel 的发布/订阅模型：4 worker 并发 observer、子 channel 缓冲（256）、gap detection（`system.gap` 事件）；通过 `PersistFn` 钩子先持久化再广播。`EventLog` 是 append-only JSON-lines 事件日志（默认 50 MiB 上限，超限截断保留尾部 75%）。
 

edge-server/internal/store/sqlite_periodic_cleanup_test.go

@@ -0,0 +1,199 @@
+package store
+
+import (
+	"path/filepath"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+func TestSQLitePeriodicCleanupIsDurableBeforeClose(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newSQLitePeriodicCleanupFixture(t, false)
+		advanceSQLiteCleanupTick()
+		if _, ok := s.GetRun("remove-run"); ok {
+			t.Fatal("periodic cleanup did not remove the expired run from memory")
+		}
+		if got := sqliteCleanupFixtureRows(t, s); got != 0 {
+			t.Errorf("periodic cleanup left %d durable run/evidence rows after removing them from memory", got)
+		}
+		stopSQLiteWithoutFinalPersist(t, s)
+		assertSQLiteCleanupReopen(t, path)
+	})
+}
+
+func TestSQLitePeriodicCleanupRetriesFailedCommit(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newSQLitePeriodicCleanupFixture(t, true)
+		advanceSQLiteCleanupTick()
+		if _, ok := s.GetRun("remove-run"); ok {
+			t.Fatal("failed persist unexpectedly retained the run in memory")
+		}
+		if s.LastPersistError() == nil {
+			t.Fatal("periodic persist failure was not recorded")
+		}
+		if got := sqliteCleanupFixtureRows(t, s); got != 6 {
+			t.Fatalf("failed cleanup transaction left %d rows, want all six retained", got)
+		}
+		if _, err := s.db.Exec("DROP TRIGGER reject_periodic_cleanup"); err != nil {
+			t.Fatal(err)
+		}
+		// No business write or final Flush repairs the failed delete. The next
+		// actual background tick must retry it despite an empty cleanup result.
+		advanceSQLiteCleanupTick()
+		if err := s.LastPersistError(); err != nil {
+			t.Errorf("next cleanup tick did not clear recovered persist failure: %v", err)
+		}
+		if got := sqliteCleanupFixtureRows(t, s); got != 0 {
+			t.Errorf("recovered periodic cleanup left %d durable rows", got)
+		}
+		stopSQLiteWithoutFinalPersist(t, s)
+		assertSQLiteCleanupReopen(t, path)
+	})
+}
+
+func TestSQLiteCloseFlushesPendingCleanup(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newSQLitePeriodicCleanupFixture(t, true)
+		advanceSQLiteCleanupTick()
+		if s.LastPersistError() == nil {
+			t.Fatal("expected the injected periodic persist failure")
+		}
+		if _, err := s.db.Exec("DROP TRIGGER reject_periodic_cleanup"); err != nil {
+			t.Fatal(err)
+		}
+		// Do not advance to another tick: Close must flush pending memory after
+		// its workers have stopped, then close the database.
+		s.Close()
+		if err := s.LastPersistError(); err != nil {
+			t.Fatalf("final cleanup persist failed: %v", err)
+		}
+		assertSQLiteCleanupReopen(t, path)
+	})
+}
+
+func TestSQLiteCloseWaitsForBackgroundWork(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, err := NewSQLite(filepath.Join(t.TempDir(), "close.db"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(s.Close)
+		release := make(chan struct{})
+		s.backgroundWG.Go(func() { <-release })
+		closed := make(chan struct{})
+		go func() {
+			s.Close()
+			close(closed)
+		}()
+		synctest.Wait()
+		select {
+		case <-closed:
+			t.Error("Close returned before owned background work finished")
+		default:
+		}
+		if err := s.db.Ping(); err != nil {
+			t.Errorf("database closed before owned background work finished: %v", err)
+		}
+		close(release)
+		synctest.Wait()
+		select {
+		case <-closed:
+		default:
+			t.Error("Close did not finish after owned background work completed")
+		}
+		if err := s.db.Ping(); err == nil {
+			t.Error("database remains open after Close")
+		}
+	})
+}
+
+// Virtual time drives the constructor's real production-cadence loops. No
+// shortened production interval or wall-clock sleep is needed in these tests.
+func advanceSQLiteCleanupTick() {
+	synctest.Wait()
+	<-time.After(sqliteBackgroundLoopInterval)
+	synctest.Wait()
+}
+
+func newSQLitePeriodicCleanupFixture(t *testing.T, rejectDelete bool) (*SQLiteStore, string) {
+	t.Helper()
+	s, path := newSQLiteCheckpointDeltaFixture(t)
+	if _, err := s.UpsertRunDiffFile(RunDiffFile{RunID: "remove-run", Path: "before.txt", Diff: "old evidence", Status: "modified"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.UpsertArtifact(Artifact{ID: "expired-artifact", RunID: "remove-run", Path: "before.txt"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.UpsertPreview(Preview{ID: "expired-preview", RunID: "remove-run", Status: "ready"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CreateItem(Item{ID: "expired-item", ProjectID: "project", ThreadID: "remove-thread", RunID: "remove-run", Type: "event"}); err != nil {
+		t.Fatal(err)
+	}
+	s.store.mu.Lock()
+	run := s.store.runs["remove-run"]
+	run.Status = "finished"
+	run.FinishedAt = time.Now().Add(-sqliteCleanupTerminalTTL - time.Hour).UTC().Format(time.RFC3339)
+	s.store.runs[run.ID] = run
+	s.store.mu.Unlock()
+	if err := s.syncPersist(); err != nil {
+		t.Fatal(err)
+	}
+	if got := sqliteCleanupFixtureRows(t, s); got != 6 {
+		t.Fatalf("fixture durable rows = %d, want run and five evidence rows", got)
+	}
+	if rejectDelete {
+		_, err := s.db.Exec("CREATE TRIGGER reject_periodic_cleanup BEFORE DELETE ON agenthub_store_rows " +
+			"WHEN OLD.row_kind = 'run' AND OLD.row_id = 'remove-run' " +
+			"BEGIN SELECT RAISE(ABORT, 'injected periodic cleanup failure'); END")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	return s, path
+}
+
+func sqliteCleanupFixtureRows(t *testing.T, s *SQLiteStore) int {
+	t.Helper()
+	var count int
+	if err := s.db.QueryRow("SELECT COUNT(*) FROM agenthub_store_rows WHERE json_extract(payload, '$.runId') = ?", "remove-run").Scan(&count); err != nil {
+		t.Fatal(err)
+	}
+	return count
+}
+
+func stopSQLiteWithoutFinalPersist(t *testing.T, s *SQLiteStore) {
+	t.Helper()
+	s.closeOnce.Do(func() {
+		close(s.stopCheckpoint)
+		s.backgroundWG.Wait()
+		if err := s.db.Close(); err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+func assertSQLiteCleanupReopen(t *testing.T, path string) {
+	t.Helper()
+	restored, err := NewSQLite(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(restored.Close)
+	if got := sqliteCleanupFixtureRows(t, restored); got != 0 {
+		t.Errorf("reopen restored %d expired run/evidence rows", got)
+	}
+	if _, ok := restored.GetRun("remove-run"); ok {
+		t.Error("expired run reappeared after reopen")
+	}
+	if _, ok := restored.GetRunCheckpoint("remove-run"); ok {
+		t.Error("expired checkpoint reappeared after reopen")
+	}
+	if run, ok := restored.GetRun("keep-run"); !ok || run.Status != "queued" {
+		t.Error("unrelated queued run was removed or changed")
+	}
+	if cp, ok := restored.GetRunCheckpoint("keep-run"); !ok || len(cp.Files) != 1 || cp.Files[0].Content != "keep evidence" {
+		t.Error("unrelated queued checkpoint was removed or changed")
+	}
+}

edge-server/internal/store/sqlite_store.go

@@ -39,14 +39,15 @@ type SQLiteStore struct {
 	store        *Store
 	persistMu    sync.Mutex
 	closeOnce    sync.Once
+	backgroundWG sync.WaitGroup
 	lastErr      error
 	lastSnapshot fileSnapshot
 	// rowsSeeded marks agenthub_store_rows as the durable source of truth:
 	// once set, syncPersist skips the legacy full-store payload UPSERT
 	// (write amplification — the payload is only the pre-rows fallback and
 	// is kept fresh until the first commit that durably seeds the rows).
 	rowsSeeded     bool
-	stopCheckpoint chan struct{} // closed to stop periodic WAL checkpoint
+	stopCheckpoint chan struct{} // closed to stop both background loops
 }
 
 func NewSQLite(path string) (*SQLiteStore, error) {
@@ -77,11 +78,11 @@ func NewSQLite(path string) (*SQLiteStore, error) {
 	// in-memory page cache small. On Windows the Go runtime is reluctant to
 	// return freed memory to the OS, so keeping the WAL small is critical.
 	s.stopCheckpoint = make(chan struct{})
-	go s.checkpointLoop(sqliteBackgroundLoopInterval)
+	s.backgroundWG.Go(func() { s.checkpointLoop(sqliteBackgroundLoopInterval) })
 	// Periodically clean up old terminal runs to prevent unbounded Store map growth.
 	// Without this, Store maps only shrink when a new run is created,
 	// which may never happen on an idle server.
-	go s.cleanupLoop(sqliteBackgroundLoopInterval)
+	s.backgroundWG.Go(func() { s.cleanupLoop(sqliteBackgroundLoopInterval) })
 
 	return s, nil
 }
@@ -103,16 +104,27 @@ func (s *SQLiteStore) checkpointLoop(interval time.Duration) {
 	}
 }
 
-// cleanupLoop periodically removes old terminal runs to prevent unbounded
-// in-memory Store map growth. Without this, Store maps only shrink when a
-// new run is explicitly created — which may never happen on an idle server.
+// cleanupLoop periodically removes old terminal runs and persists the deletes.
+// Failed commits remain pending for the next tick, even if no new runs expire.
 func (s *SQLiteStore) cleanupLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
+	pendingPersist := false
 	for {
 		select {
 		case <-ticker.C:
-			s.store.CleanupRuns(sqlitePeriodicCleanupOptions())
+			result := s.store.CleanupRuns(sqlitePeriodicCleanupOptions())
+			// A failed commit already removed data from memory. Retry it even
+			// when the next tick has no additional runs to remove.
+			pendingPersist = pendingPersist || shouldSyncAfterCleanup(result)
+			if !pendingPersist {
+				continue
+			}
+			if err := s.syncPersist(); err != nil {
+				slog.Warn("sqlite store: periodic cleanup persist failed", "error", err)
+				continue
+			}
+			pendingPersist = false
 		case <-s.stopCheckpoint:
 			return
 		}
@@ -122,6 +134,7 @@ func (s *SQLiteStore) cleanupLoop(interval time.Duration) {
 func (s *SQLiteStore) Close() {
 	s.closeOnce.Do(func() {
 		close(s.stopCheckpoint)
+		s.backgroundWG.Wait()
 		// Final checkpoint to shrink the WAL before close. Failures are logged
 		// (not returned) to keep the shutdown signature; LastPersistError still
 		// reflects the final persist outcome.