From 350ee4fba3bc938e0fe472fe059218724fcbed40 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Sun, 6 Sep 2026 14:57:51 +0800 Subject: [PATCH] fix(store): prune run checkpoints and preserve SQLite delta baselines Fix checkpoint retention after run cleanup or thread deletion and stop rewriting unchanged checkpoint rows. Closes #2335. Co-authored-by: Codex --- docs/architecture/02-edge-server.md | 2 +- .../store/sqlite_checkpoint_delta_test.go | 148 ++++++++++++++++++ .../internal/store/sqlite_store_query.go | 1 + .../internal/store/sqlite_store_query_test.go | 12 ++ edge-server/internal/store/store_projects.go | 1 + 5 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 edge-server/internal/store/sqlite_checkpoint_delta_test.go diff --git a/docs/architecture/02-edge-server.md b/docs/architecture/02-edge-server.md index 7062670f9..27410398f 100644 --- a/docs/architecture/02-edge-server.md +++ b/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（completed/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items。 +终端状态 runs（finished/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items/checkpoints。Checkpoint 在 run 完成时保留，在 run 清理或所属 thread 删除时随 run 移除；这不删除工作区文件。 `EventBus`（`internal/events/bus.go`）是基于 channel 的发布/订阅模型：4 worker 并发 observer、子 channel 缓冲（256）、gap detection（`system.gap` 事件）；通过 `PersistFn` 钩子先持久化再广播。`EventLog` 是 append-only JSON-lines 事件日志（默认 50 MiB 上限，超限截断保留尾部 75%）。 diff --git a/edge-server/internal/store/sqlite_checkpoint_delta_test.go b/edge-server/internal/store/sqlite_checkpoint_delta_test.go new file mode 100644 index 000000000..597ceeb5b --- /dev/null +++ b/edge-server/internal/store/sqlite_checkpoint_delta_test.go @@ -0,0 +1,148 @@ +package store + +import ( + "path/filepath" + "reflect" + "testing" + "time" +) + +func TestSQLiteCheckpointDeltaSkipsUnchangedRows(t *testing.T) { + s, path := newSQLiteCheckpointDeltaFixture(t) + // Observe real row updates, not an implementation helper or mock. A + // legitimate checkpoint edit below verifies the audit trigger's sensitivity. + if _, err := s.db.Exec( + "CREATE TABLE checkpoint_updates (run_id TEXT NOT NULL); " + + "CREATE TRIGGER audit_checkpoint_update AFTER UPDATE ON agenthub_store_rows " + + "WHEN NEW.row_kind = 'checkpoint' BEGIN " + + "INSERT INTO checkpoint_updates(run_id) VALUES (NEW.row_id); END;"); err != nil { + t.Fatal(err) + } + updates := func() int { + t.Helper() + var count int + if err := s.db.QueryRow("SELECT COUNT(*) FROM checkpoint_updates").Scan(&count); err != nil { + t.Fatal(err) + } + return count + } + if _, ok := s.SetRunRetryCount("remove-run", 1); !ok { + t.Fatalf("unrelated run mutation failed: %v", s.LastPersistError()) + } + if got := updates(); got != 0 { + t.Errorf("unrelated run mutation rewrote %d unchanged checkpoints", got) + } + + cp, ok := s.GetRunCheckpoint("keep-run") + if !ok { + t.Fatal("keep checkpoint missing") + } + cp.Files[0].Content = "revised evidence" + before := updates() + if _, err := s.UpsertRunCheckpoint(cp); err != nil { + t.Fatal(err) + } + if got := updates() - before; got != 1 { + t.Errorf("legitimate checkpoint edit updated %d rows, want only its own row", got) + } + // Read back without a preceding Close/Flush that could repair a missed write. + restored, err := NewSQLite(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(restored.Close) + got, ok := restored.GetRunCheckpoint(cp.RunID) + if !ok || !reflect.DeepEqual(got, cp) { + t.Errorf("checkpoint edit was not durable: got=%#v ok=%v", got, ok) + } +} + +func TestSQLiteCheckpointDeltaRemovesRunEvidence(t *testing.T) { + for _, removal := range []string{"terminal-cleanup", "delete-thread"} { + t.Run(removal, func(t *testing.T) { + s, path := newSQLiteCheckpointDeltaFixture(t) + keep, ok := s.GetRunCheckpoint("keep-run") + if !ok { + t.Fatal("retained checkpoint missing before removal") + } + switch removal { + case "terminal-cleanup": + if _, ok := s.SetRunStatus("remove-run", "finished"); !ok { + t.Fatal("complete run") + } + result := s.CleanupRuns(RunCleanupOptions{ + Now: time.Now().Add(2 * time.Hour), TerminalTTL: time.Hour, + }) + if result.RemovedRuns != 1 { + t.Fatalf("removed runs = %d, want 1", result.RemovedRuns) + } + case "delete-thread": + if !s.DeleteThread("remove-thread") { + t.Fatal("delete thread") + } + } + if err := s.LastPersistError(); err != nil { + t.Fatal(err) + } + if _, ok := s.GetRun("remove-run"); ok { + t.Fatal("run remains after removal") + } + if _, ok := s.GetRunCheckpoint("remove-run"); ok { + t.Error("removed run's checkpoint remains in memory") + } + var count int + if err := s.db.QueryRow("SELECT COUNT(*) FROM agenthub_store_rows WHERE row_kind = ? AND row_id = ?", + sqliteRowKindCheckpoint, "remove-run").Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Errorf("removed run still has %d durable checkpoint rows", count) + } + // A new handle uses the actual load path before the original Close. + restored, err := NewSQLite(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(restored.Close) + if _, ok := restored.GetRunCheckpoint("remove-run"); ok { + t.Error("removed run's checkpoint reappears after reopen") + } + if _, ok := restored.GetRun("keep-run"); !ok { + t.Error("unrelated queued run was removed") + } + got, ok := restored.GetRunCheckpoint("keep-run") + if !ok || !reflect.DeepEqual(got, keep) { + t.Errorf("unrelated checkpoint changed: got=%#v ok=%v", got, ok) + } + }) + } +} + +func newSQLiteCheckpointDeltaFixture(t *testing.T) (*SQLiteStore, string) { + t.Helper() + path := filepath.Join(t.TempDir(), "checkpoint-delta.db") + s, err := NewSQLite(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + if _, err := s.CreateProject("project", "Checkpoints", ""); err != nil { + t.Fatal(err) + } + for _, name := range []string{"keep", "remove"} { + threadID, runID := name+"-thread", name+"-run" + if _, err := s.CreateThread(threadID, "project", "Checkpoint", "", "", ""); err != nil { + t.Fatal(err) + } + if _, err := s.CreateRun(runID, "project", threadID); err != nil { + t.Fatal(err) + } + if _, err := s.UpsertRunCheckpoint(RunCheckpoint{ + ID: name + "-checkpoint", RunID: runID, WorkDir: "fixture", FileCount: 1, + Files: []CheckpointFile{{Path: "input.txt", Content: name + " evidence"}}, + }); err != nil { + t.Fatal(err) + } + } + return s, path +} diff --git a/edge-server/internal/store/sqlite_store_query.go b/edge-server/internal/store/sqlite_store_query.go index 52f6b6086..84e7ee968 100644 --- a/edge-server/internal/store/sqlite_store_query.go +++ b/edge-server/internal/store/sqlite_store_query.go @@ -156,6 +156,7 @@ func cloneFileSnapshot(snapshot fileSnapshot) fileSnapshot { Pins: copyMap(snapshot.Pins), Diffs: copyMap(snapshot.Diffs), Artifacts: cloneArtifactMap(snapshot.Artifacts), + Checkpoints: cloneCheckpointMap(snapshot.Checkpoints), Previews: copyMap(snapshot.Previews), UserProfiles: copyMap(snapshot.UserProfiles), AgentProfiles: copyMap(snapshot.AgentProfiles), diff --git a/edge-server/internal/store/sqlite_store_query_test.go b/edge-server/internal/store/sqlite_store_query_test.go index 97a4a543f..f688edcbe 100644 --- a/edge-server/internal/store/sqlite_store_query_test.go +++ b/edge-server/internal/store/sqlite_store_query_test.go @@ -262,6 +262,9 @@ func TestCloneFileSnapshot(t *testing.T) { Artifacts: map[string]Artifact{ "a1": {ID: "a1", RunID: "r1", ContentSource: source}, }, + Checkpoints: map[string]RunCheckpoint{ + "r1": {RunID: "r1", Files: []CheckpointFile{{Path: "input.txt", Content: "original"}}}, + }, ProjectOrder: []string{"p1"}, ArtifactOrder: []string{"a1"}, Settings: map[string]string{"theme": "dark"}, @@ -278,6 +281,15 @@ func TestCloneFileSnapshot(t *testing.T) { if cloned.Artifacts["a1"].ContentSource == original.Artifacts["a1"].ContentSource { t.Fatal("artifact content source pointer should be deep-cloned") } + checkpoint, ok := cloned.Checkpoints["r1"] + if !ok || !reflect.DeepEqual(checkpoint, original.Checkpoints["r1"]) { + t.Fatal("checkpoint missing or changed in cloned snapshot") + } + checkpoint.Files[0].Content = "mutated" + delete(cloned.Checkpoints, "r1") + if got := original.Checkpoints["r1"].Files[0].Content; got != "original" { + t.Fatalf("checkpoint clone shares nested files: %q", got) + } cloned.Projects["p1"] = Project{ID: "p1", Name: "mutated"} cloned.ProjectOrder[0] = "mutated" cloned.Settings["theme"] = "light" diff --git a/edge-server/internal/store/store_projects.go b/edge-server/internal/store/store_projects.go index fe67ea5df..804a0ab3a 100644 --- a/edge-server/internal/store/store_projects.go +++ b/edge-server/internal/store/store_projects.go @@ -75,6 +75,7 @@ func (s *Store) removePins(match func(ThreadPin) bool) { } func (s *Store) removeRunEvidence(runID string) { + delete(s.checkpoints, runID) s.diffOrder, s.artifactOrder, s.previewOrder = pruneRunEvidence( s.diffs, s.artifacts, s.previews, s.diffOrder, s.artifactOrder, s.previewOrder,