fix(store): prune run checkpoints and preserve SQLite delta baselines

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%）。
 

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
+}

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),

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"

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,