From f862892544f5a076d9e65a1763dae97328298eb7 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 15:59:14 +0000 Subject: [PATCH] Take the checkpoint closure: retention on rewind, HTTP tests, eligibility pin (N3 WP3) Why: the library now pins eligibility, guards symlinks, reports partial restores, and bounds the log (Agent-Native#33); the product wires retention after successful rewinds and needed the HTTP-layer and eligibility evidence. Evidence: go test -count=1 ./... green; TestRewindFilesHTTPHappy and TestRewindFilesHTTPBusy409 (HTTP happy + conflict), TestBackendCheckpointEligibilityIsWriteFileOnly (the product surface checkpoints exactly {fs_write_file}). Impact: checkpoints.jsonl is compacted to the latest 200 anchors after each successful rewind. Refs: Agent-Native#33, Agent-Squad#37, N3 plan WP3 --- deps/agent-native | 2 +- deps/agent-squad | 2 +- internal/runmgr/rewind.go | 9 +- internal/server/rewind_http_test.go | 140 +++++++++++++++++++++++++++ internal/tools/fs_checkpoint_test.go | 40 ++++++++ 5 files changed, 189 insertions(+), 4 deletions(-) create mode 100644 internal/server/rewind_http_test.go create mode 100644 internal/tools/fs_checkpoint_test.go diff --git a/deps/agent-native b/deps/agent-native index fe0c4a9..f04c87a 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit fe0c4a9f82d8a5e160997f3608f2eb96594fe0fd +Subproject commit f04c87ac317efc7281b614827a6a2ec879fa5b69 diff --git a/deps/agent-squad b/deps/agent-squad index f38cf7c..ff36b23 160000 --- a/deps/agent-squad +++ b/deps/agent-squad @@ -1 +1 @@ -Subproject commit f38cf7c441d19a1bb06ef0dfbe65b2b5d87f5675 +Subproject commit ff36b231916531f16b166d3a9142d54715f0c7e9 diff --git a/internal/runmgr/rewind.go b/internal/runmgr/rewind.go index 5db16c2..1e1763e 100644 --- a/internal/runmgr/rewind.go +++ b/internal/runmgr/rewind.go @@ -2,6 +2,7 @@ package runmgr import ( "fmt" + "log" "path/filepath" "strings" @@ -75,14 +76,18 @@ func (m *Manager) RewindFiles(owner, convID, messageID string) (checkpoint.Resul if err != nil { return checkpoint.Result{}, err } - log, err := m.OpenCheckpoints(owner, convID) + cpLog, err := m.OpenCheckpoints(owner, convID) if err != nil { return checkpoint.Result{}, err } - res, err := log.Rewind(messageID, ordered) + res, err := cpLog.Rewind(messageID, ordered) if err != nil { return checkpoint.Result{}, err } + // 保留策略（N3 WP3）：成功的回滚后按 anchor 数裁到 200，日志有界。 + if cerr := cpLog.Compact(200); cerr != nil { + log.Printf("[runmgr] checkpoint compact %s: %v", convID, cerr) + } note := rewindNote(res) msg := core.Message{ ConversationID: convID, Role: core.RoleSystem, diff --git a/internal/server/rewind_http_test.go b/internal/server/rewind_http_test.go new file mode 100644 index 0000000..6032350 --- /dev/null +++ b/internal/server/rewind_http_test.go @@ -0,0 +1,140 @@ +package server + +import ( + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/bus" + "agent-system-backend/internal/cmdbus" + "agent-system-backend/internal/plugins" + "agent-system-backend/internal/profiles" + "agent-system-backend/internal/runmgr" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/tools" + core "agent-system-core" +) + +// ---- N3 WP3：rewind HTTP 层（happy + busy 409）---- + +// happy：经 API 回滚被追踪写入，restored/skipped 上抛，文件内容退回。 +func TestRewindFilesHTTPHappy(t *testing.T) { + dir := t.TempDir() + st, err := store.OpenSQLite(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + _ = st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": true}}) + reg := tools.NewRegistry() + reg.RegisterBuiltin() + setSvc := settings.NewService(st) + profSvc := profiles.New(st) + profSvc.EnsureDefault() + mgr := runmgr.New(st, bus.New(), reg, profSvc, setSvc, &runtime.MockRunner{}, &runtime.MockSubagent{}, &runtime.MockRunner{}, &runtime.MockSubagent{}) + ws := t.TempDir() + mgr.SetWorkspaceRoot(ws) + ph := plugins.New(st, reg, setSvc, filepath.Join(dir, "plugins")) + srv := New(st, mgr, reg, profSvc, setSvc, ph) + srv.Bus = cmdbus.New(st, nil, nil) + h := srv.Handler() + _, convID := setupConversation(t, h) + runID := postMessage(t, h, convID, "写个文件") + // 等 run 终态（mock run 是异步的；不回等会撞到会话占用 409）。 + deadline := time.Now().Add(2 * time.Second) + for { + run, err := st.GetRunByID(runID) + if err == nil && (run.Status == core.RunCompleted || run.Status == core.RunFailed || run.Status == core.RunCancelled) { + break + } + if time.Now().After(deadline) { + t.Fatalf("run 未终态：%+v", run) + } + time.Sleep(10 * time.Millisecond) + } + + // 找到用户消息（anchor）并造一条被追踪写入。 + var userMsgID string + msgs, err := st.ListMessages(convID) + if err != nil { + t.Fatal(err) + } + for _, m := range msgs { + if m.Role == core.RoleUser { + userMsgID = m.ID + } + } + if userMsgID == "" { + t.Fatal("no user message") + } + sandbox := filepath.Join(ws, store.OwnerDefault, convID) + if err := os.MkdirAll(sandbox, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sandbox, "tracked.txt"), []byte("写入后"), 0o644); err != nil { + t.Fatal(err) + } + cpLog, err := mgr.OpenCheckpoints(store.OwnerDefault, convID) + if err != nil { + t.Fatal(err) + } + if err := cpLog.Record(userMsgID, sandbox, "tracked.txt", []byte("写入前"), true, []byte("写入后")); err != nil { + t.Fatal(err) + } + + code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/rewind", map[string]any{"messageId": userMsgID}) + if code != http.StatusOK { + t.Fatalf("rewind status %d: %v", code, out) + } + restored, _ := out["restored"].([]any) + if len(restored) != 1 || restored[0] != "tracked.txt" { + t.Fatalf("restored: %v", out) + } + body, err := os.ReadFile(filepath.Join(sandbox, "tracked.txt")) + if err != nil || string(body) != "写入前" { + t.Fatalf("文件未退回：%v %q", err, body) + } +} + +// busy：会话被占时回滚返回 409。 +func TestRewindFilesHTTPBusy409(t *testing.T) { + dir := t.TempDir() + st, err := store.OpenSQLite(filepath.Join(dir, "test.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + _ = st.MergeSettingsScope(store.OwnerDefault, settings.ScopeUser, map[string]any{"runtime": map[string]any{"mock": false}}) + reg := tools.NewRegistry() + reg.RegisterBuiltin() + setSvc := settings.NewService(st) + profSvc := profiles.New(st) + profSvc.EnsureDefault() + release := make(chan struct{}) + blocking := blockingRunner{block: make(chan struct{}, 1), release: release} + defer close(release) + mgr := runmgr.New(st, bus.New(), reg, profSvc, setSvc, blocking, blocking, &runtime.MockRunner{}, &runtime.MockSubagent{}) + t.Cleanup(mgr.Shutdown) + ph := plugins.New(st, reg, setSvc, filepath.Join(dir, "plugins")) + srv := New(st, mgr, reg, profSvc, setSvc, ph) + srv.Bus = cmdbus.New(st, nil, nil) + h := srv.Handler() + _, convID := setupConversation(t, h) + postMessage(t, h, convID, "先占住") + + var userMsgID string + msgs, _ := st.ListMessages(convID) + for _, m := range msgs { + if m.Role == core.RoleUser { + userMsgID = m.ID + } + } + code, _ := doJSON(t, h, "POST", "/api/conversations/"+convID+"/rewind", map[string]any{"messageId": userMsgID}) + if code != http.StatusConflict { + t.Fatalf("busy rewind 应 409，得到 %d", code) + } +} diff --git a/internal/tools/fs_checkpoint_test.go b/internal/tools/fs_checkpoint_test.go new file mode 100644 index 0000000..35bdbc3 --- /dev/null +++ b/internal/tools/fs_checkpoint_test.go @@ -0,0 +1,40 @@ +package tools + +import ( + "encoding/json" + "testing" + + core "agent-system-core" +) + +// ---- N3 WP3：backend 自家工具面的回滚资格钉住（验收② 产品侧）---- + +type checkpointNoteCtx struct { + *fakeFSCtx + noted []string +} + +func (c *checkpointNoteCtx) NoteFileCheckpoint(rel string, before []byte, existed bool, after []byte) error { + c.noted = append(c.noted, rel) + return nil +} + +// backend 工具面里只有 fs_write_file 写检查点（产品面不暴露编辑工具是刻意的 +// 收窄）；fs_read_file / fs_list_dir 永不记录。 +func TestBackendCheckpointEligibilityIsWriteFileOnly(t *testing.T) { + r := newFSRegistry(t) + ctx := &checkpointNoteCtx{fakeFSCtx: &fakeFSCtx{root: t.TempDir(), owner: "o1", conv: "c1", fsToolsOn: true}} + exec := func(name, args string) { + t.Helper() + r.Execute(ctx, core.ToolCallRequest{CallID: "c-" + name, Name: name, Arguments: json.RawMessage(args)}) + } + exec("fs_write_file", `{"path":"a.txt","content":"x"}`) + if len(ctx.noted) != 1 || ctx.noted[0] != "a.txt" { + t.Fatalf("fs_write_file 必须写检查点：%v", ctx.noted) + } + exec("fs_read_file", `{"path":"a.txt"}`) + exec("fs_list_dir", `{"path":"."}`) + if len(ctx.noted) != 1 { + t.Fatalf("读类工具不得写检查点：%v", ctx.noted) + } +}