fix(store): prevent FileStore persistence starvation during continuous writes

docs/architecture/02-edge-server.md

@@ -64,7 +64,9 @@ ProcessExecutor 配置 `RunTimeout`（默认 30 分钟）、`ShutdownGracePeriod
 
 `Store`（`internal/store/store.go`）是核心内存数据结构，管理 Project、Thread、Run、Item、Pin、Diff、Artifact、Preview、UserProfile、AgentProfile、Settings 的全量 CRUD。`FileStore` 与 `SQLiteStore` 的常规 `Get*` / `List*` 委托这个内存 Store，不是直接 SQL 查询；业务读取主要经过内存锁，而不是 SQLite 连接池。
 
-`FileStore`（`internal/store/file_store.go`）对写入信号做 debounce，再保存全量 JSON 快照；显式 `Flush` 同步执行快照、编码和文件 Sync/rename。`SQLiteStore`（`internal/store/sqlite_store.go`）在写入后同步持久化快照差分到 SQLite（WAL 模式，定期 checkpoint），支持崩溃恢复。SQL 连接初始化与持久化串行化独立于普通业务读面，不能用额外 SQL 读者的争用直接代替业务读取测量。
+`FileStore`（`internal/store/file_store.go`）由首条待保存写入启动 50 ms 合批窗口，后续写入只合批、不推迟已有截止点；窗口到期后保存全量 JSON 快照，空闲时不轮询写盘。50 ms 是调度窗口，不是包含编码与文件 I/O 的完成时限；显式 `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 移除；这不删除工作区文件。 SQLite 后台清理有删除时同步提交；失败通过 `LastPersistError` 和日志留痕，下一周期即使没有新删除也会重试。关闭时先停止并等待后台清理/checkpoint 任务，再做最终持久化与数据库关闭。
 

edge-server/internal/store/file_store.go

@@ -48,7 +48,8 @@ type fileSnapshot struct {
 }
 
 // FileStore wraps the in-memory store with a JSON snapshot saved asynchronously after writes.
-// Writes are debounced — rapid mutations batch into a single disk write.
+// Writes within a fixed window batch into one disk write; continuous traffic
+// cannot postpone persistence by repeatedly restarting that window.
 type FileStore struct {
 	path string
 
@@ -144,14 +145,16 @@ func (f *FileStore) schedulePersist() {
 	}
 }
 
-// persistLoop runs in the background, debouncing persist calls.
+// persistLoop coalesces writes until the first pending signal's deadline.
 func (f *FileStore) persistLoop() {
 	defer close(f.done)
 
 	timer := time.NewTimer(0)
 	if !timer.Stop() {
 		<-timer.C
 	}
+	defer timer.Stop()
+	pending := false
 
 	for {
 		select {
@@ -161,8 +164,12 @@ func (f *FileStore) persistLoop() {
 				_ = f.syncPersist()
 				return
 			}
-			timer.Reset(debounceInterval)
+			if !pending {
+				timer.Reset(debounceInterval)
+				pending = true
+			}
 		case <-timer.C:
+			pending = false
 			// Failure is surfaced via LastPersistError, matching Flush() semantics.
 			_ = f.syncPersist()
 		}

edge-server/internal/store/file_store_persist_loop_test.go

@@ -0,0 +1,164 @@
+package store
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+func TestFileStorePersistsDuringContinuousWrites(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newFilePersistLoopFixture(t)
+		addFilePersistLoopItem(t, s, "first-message")
+		synctest.Wait()
+
+		// Drive the real constructor-owned loop with virtual time. A fresh
+		// write before each quiet-period deadline must not starve persistence.
+		for i := range 40 {
+			<-time.After(debounceInterval / 2)
+			addFilePersistLoopItem(t, s, fmt.Sprintf("busy-%d", i))
+			synctest.Wait()
+		}
+		if err := s.LastPersistError(); err != nil {
+			t.Fatal(err)
+		}
+		// Use real disk recovery before Flush/Close can repair the snapshot.
+		// A late item requires repeated progress, not only the first flush.
+		restored := readFilePersistLoopDisk(t, path)
+		for _, id := range []string{"first-message", "busy-36"} {
+			if _, ok := restored.GetItem(id); !ok {
+				t.Errorf("continuous writes starved FileStore persistence for 20 batch intervals: missing %s, memory items=%d, durable items=%d", id, len(s.ListThreadItems("thread")), len(restored.ListThreadItems("thread")))
+			}
+		}
+	})
+}
+
+func TestFileStoreCoalescesWithoutPostponingBatchDeadline(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newFilePersistLoopFixture(t)
+		addFilePersistLoopItem(t, s, "first-message")
+		synctest.Wait()
+		<-time.After(debounceInterval / 2)
+		addFilePersistLoopItem(t, s, "second-message")
+		synctest.Wait()
+		if got := len(readFilePersistLoopDisk(t, path).ListThreadItems("thread")); got != 0 {
+			t.Fatalf("batch persisted %d items before its first-write deadline", got)
+		}
+		<-time.After(debounceInterval / 2)
+		synctest.Wait()
+		restored := readFilePersistLoopDisk(t, path)
+		for _, id := range []string{"first-message", "second-message"} {
+			if _, ok := restored.GetItem(id); !ok {
+				t.Errorf("later write postponed the first batch: missing %s", id)
+			}
+		}
+
+		addFilePersistLoopItem(t, s, "next-batch")
+		synctest.Wait()
+		<-time.After(debounceInterval / 2)
+		synctest.Wait()
+		if _, ok := readFilePersistLoopDisk(t, path).GetItem("next-batch"); ok {
+			t.Error("next batch was not coalesced")
+		}
+		<-time.After(debounceInterval / 2)
+		synctest.Wait()
+		if _, ok := readFilePersistLoopDisk(t, path).GetItem("next-batch"); !ok {
+			t.Error("next batch did not re-arm persistence")
+		}
+	})
+}
+
+func TestFileStoreIdleDoesNotPersistAndNewBatchReportsFailure(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newFilePersistLoopFixture(t)
+		addFilePersistLoopItem(t, s, "first-message")
+		synctest.Wait()
+		<-time.After(debounceInterval)
+		synctest.Wait()
+		if _, ok := readFilePersistLoopDisk(t, path).GetItem("first-message"); !ok {
+			t.Fatal("initial batch did not persist")
+		}
+		// An actual filesystem failure makes an unwanted idle persist visible.
+		// The next real write below checks that the fault is effective.
+		if err := os.Remove(path); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Mkdir(path, 0o755); err != nil {
+			t.Fatal(err)
+		}
+		<-time.After(4 * debounceInterval)
+		synctest.Wait()
+		if err := s.LastPersistError(); err != nil {
+			t.Errorf("idle store attempted persistence without a write: %v", err)
+		}
+		addFilePersistLoopItem(t, s, "after-idle")
+		synctest.Wait()
+		<-time.After(debounceInterval)
+		synctest.Wait()
+		if err := s.LastPersistError(); err == nil {
+			t.Error("new batch did not surface the real snapshot write failure")
+		}
+		if err := os.Remove(path); err != nil {
+			t.Fatal(err)
+		}
+		s.Flush()
+		if err := s.LastPersistError(); err != nil {
+			t.Fatalf("explicit Flush did not recover: %v", err)
+		}
+		if _, ok := readFilePersistLoopDisk(t, path).GetItem("after-idle"); !ok {
+			t.Error("failed batch was not retained for explicit Flush")
+		}
+	})
+}
+
+func TestFileStoreCloseFlushesPendingBatch(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		s, path := newFilePersistLoopFixture(t)
+		addFilePersistLoopItem(t, s, "pending-at-close")
+		synctest.Wait()
+		// No timer advance or explicit Flush: shutdown itself must save it.
+		s.Close()
+		if err := s.LastPersistError(); err != nil {
+			t.Fatal(err)
+		}
+		if _, ok := readFilePersistLoopDisk(t, path).GetItem("pending-at-close"); !ok {
+			t.Error("Close did not persist the pending batch")
+		}
+	})
+}
+
+func newFilePersistLoopFixture(t *testing.T) (*FileStore, string) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "store.json")
+	s, err := NewFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(s.Close)
+	if _, err := s.CreateProject("project", "Project", ""); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CreateThread("thread", "project", "Thread", "", "", ""); err != nil {
+		t.Fatal(err)
+	}
+	return s, path
+}
+
+func addFilePersistLoopItem(t *testing.T, s *FileStore, id string) {
+	t.Helper()
+	if _, err := s.CreateItem(Item{ID: id, ProjectID: "project", ThreadID: "thread", Type: "event"}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func readFilePersistLoopDisk(t *testing.T, path string) *Store {
+	t.Helper()
+	restored := New()
+	if err := loadFileSnapshot(path, restored); err != nil {
+		t.Fatal(err)
+	}
+	return restored
+}