diff --git a/deps/agent-native b/deps/agent-native index 0a52d9f..fb2b3b7 160000 --- a/deps/agent-native +++ b/deps/agent-native @@ -1 +1 @@ -Subproject commit 0a52d9f9763372ef1b4d061934d60f15fd8bcbea +Subproject commit fb2b3b796c57246306cd7384b60fb671ce371143 diff --git a/deps/agent-squad b/deps/agent-squad index beea97e..a3b49e7 160000 --- a/deps/agent-squad +++ b/deps/agent-squad @@ -1 +1 @@ -Subproject commit beea97e8cf8f509289545827bc3034a940433ca4 +Subproject commit a3b49e7b1f563b34661cec026f58eec3f135fdbd diff --git a/deps/agent-system-core b/deps/agent-system-core index cf7df54..5dc190e 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit cf7df54874c20f8ae9cb4e48b2c75f9ef6968f14 +Subproject commit 5dc190ee63174be71208609309f0b36a5fe070e0 diff --git a/internal/app/app.go b/internal/app/app.go index 47561d9..436a32c 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -23,9 +23,11 @@ import ( "logging-system/logsys" "logging-system/sink" + "agent-native/runtime/checkpoint" agentllm "agent-native/runtime/llm" agentmcp "agent-native/runtime/mcp" "agent-native/runtime/policy" + "agent-native/runtime/telemetry" agenttools "agent-native/runtime/tools" nativeadp "agent-squad/adapters/native" "agent-system-backend/internal/bus" @@ -69,6 +71,7 @@ func Run(opt Options) error { Sinks: []logsys.Sink{&sink.Console{Out: os.Stderr, Format: "json"}}, }) logger.Info("agent-system starting", "addr", opt.Addr, "data", opt.DataDir) + traceFn := telemetry.StartFunc(logger) var st *store.Store var err error if opt.StorageDSN != "" { @@ -188,6 +191,17 @@ func Run(opt Options) error { if mgr := mcpManagerFor(rawMCP); mgr != nil { cfg.MCP = mgr } + cfg.Trace = traceFn + if host, ok := rc.(interface { + OpenCheckpoints() (*checkpoint.Log, error) + CheckpointAnchor() string + }); ok { + if lg, err := host.OpenCheckpoints(); err == nil { + cfg.Checkpoints = lg + anchor := host.CheckpointAnchor() + cfg.CheckpointAnchor = func() string { return anchor } + } + } return cfg, true } profSvc := profiles.New(st) @@ -200,8 +214,8 @@ func Run(opt Options) error { } } - convRunner := &runtime.LLMRunner{Prompts: promptSvc, SkillsDir: skillsRoot} - subRunner := &runtime.SubLLMRunner{Prompts: promptSvc} + convRunner := &runtime.LLMRunner{Prompts: promptSvc, SkillsDir: skillsRoot, Trace: traceFn} + subRunner := &runtime.SubLLMRunner{Prompts: promptSvc, Trace: traceFn} mockRunner := &runtime.MockRunner{} mgr := runmgr.New(st, b, reg, profSvc, setSvc, convRunner, subRunner, mockRunner, &runtime.MockSubagent{}) instanceID := opt.InstanceID diff --git a/internal/runmgr/rewind.go b/internal/runmgr/rewind.go new file mode 100644 index 0000000..5db16c2 --- /dev/null +++ b/internal/runmgr/rewind.go @@ -0,0 +1,137 @@ +package runmgr + +import ( + "fmt" + "path/filepath" + "strings" + + "agent-native/runtime/checkpoint" + core "agent-system-core" +) + +func (m *Manager) checkpointFile(owner, convID string) (string, error) { + if m == nil || m.workspaceRoot == "" || owner == "" || convID == "" { + return "", fmt.Errorf("未配置工作区，不能记录文件检查点") + } + return filepath.Join(m.workspaceRoot, owner, convID, ".agentx", "checkpoints.jsonl"), nil +} + +// OpenCheckpoints 打开这个会话的文件检查点日志。 +func (m *Manager) OpenCheckpoints(owner, convID string) (*checkpoint.Log, error) { + path, err := m.checkpointFile(owner, convID) + if err != nil { + return nil, err + } + return checkpoint.Open(path) +} + +func (c *runCtx) OpenCheckpoints() (*checkpoint.Log, error) { + if c == nil || c.mgr == nil { + return nil, fmt.Errorf("未配置工作区，不能记录文件检查点") + } + return c.mgr.OpenCheckpoints(c.run.Owner, c.run.ConversationID) +} + +func (c *runCtx) CheckpointAnchor() string { + if c == nil { + return "" + } + return c.userMessage.ID +} + +// NoteFileCheckpoint 记录一次工作区写入的改前内容。 +func (c *runCtx) NoteFileCheckpoint(rel string, before []byte, existed bool, after []byte) error { + log, err := c.OpenCheckpoints() + if err != nil { + return err + } + root := c.ConversationSandbox() + return log.Record(c.CheckpointAnchor(), root, rel, before, existed, after) +} + +func (t *toolCtx) NoteFileCheckpoint(rel string, before []byte, existed bool, after []byte) error { + if t == nil || t.rc == nil { + return nil + } + return t.rc.NoteFileCheckpoint(rel, before, existed, after) +} + +// PlanRewind 列出从这条用户消息起会被回滚的文件，不改磁盘。 +func (m *Manager) PlanRewind(owner, convID, messageID string) (checkpoint.Result, error) { + ordered, err := m.userMessageIDs(convID, messageID) + if err != nil { + return checkpoint.Result{}, err + } + log, err := m.OpenCheckpoints(owner, convID) + if err != nil { + return checkpoint.Result{}, err + } + return log.Plan(messageID, ordered) +} + +// RewindFiles 把文件退回到这条用户消息之前，并追加一条系统说明。 +func (m *Manager) RewindFiles(owner, convID, messageID string) (checkpoint.Result, error) { + ordered, err := m.userMessageIDs(convID, messageID) + if err != nil { + return checkpoint.Result{}, err + } + log, err := m.OpenCheckpoints(owner, convID) + if err != nil { + return checkpoint.Result{}, err + } + res, err := log.Rewind(messageID, ordered) + if err != nil { + return checkpoint.Result{}, err + } + note := rewindNote(res) + msg := core.Message{ + ConversationID: convID, Role: core.RoleSystem, + Parts: []core.Part{{Type: core.PartText, Text: note}}, + } + if err := m.st.AppendMessage(&msg); err != nil { + return res, err + } + return res, nil +} + +func (m *Manager) userMessageIDs(convID, messageID string) ([]string, error) { + msgs, err := m.st.ListMessages(convID) + if err != nil { + return nil, err + } + var ordered []string + found := false + for _, msg := range msgs { + if msg.Role != core.RoleUser { + continue + } + ordered = append(ordered, msg.ID) + if msg.ID == messageID { + found = true + } + } + if !found { + return nil, fmt.Errorf("找不到用户消息 %s", messageID) + } + return ordered, nil +} + +func rewindNote(res checkpoint.Result) string { + var b strings.Builder + b.WriteString("已把文件回滚到这条消息之前。") + if len(res.Restored) == 0 && len(res.Skipped) == 0 { + b.WriteString("没有需要回滚的文件。") + return b.String() + } + if len(res.Restored) > 0 { + b.WriteString("已恢复：") + b.WriteString(strings.Join(res.Restored, "、")) + b.WriteString("。") + } + if len(res.Skipped) > 0 { + b.WriteString("已跳过：") + b.WriteString(strings.Join(res.Skipped, "、")) + b.WriteString("（文件在追踪之外被改过）。") + } + return b.String() +} diff --git a/internal/runmgr/rewind_test.go b/internal/runmgr/rewind_test.go new file mode 100644 index 0000000..45bbde8 --- /dev/null +++ b/internal/runmgr/rewind_test.go @@ -0,0 +1,77 @@ +package runmgr + +import ( + "os" + "path/filepath" + "testing" + + "agent-system-backend/internal/tools" + core "agent-system-core" +) + +func TestRewindRestoresTrackedWriteAndLeavesShellFile(t *testing.T) { + c, st := newExecEnv(t) + root := t.TempDir() + c.mgr.SetWorkspaceRoot(root) + c.eff = core.EffectiveSettings{"runtime": map[string]any{"fsTools": true}} + if err := tools.RegisterFSTools(c.mgr.registry); err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: c.run.ConversationID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "go"}}} + if err := st.AppendMessage(&msg); err != nil { + t.Fatal(err) + } + c.userMessage = msg + res := c.ExecuteTool(core.ToolCallRequest{ + CallID: "w1", Name: "fs_write_file", + Arguments: []byte(`{"path":"note.txt","content":"new"}`), + }) + if res.Status != "ok" { + t.Fatalf("write %s %s", res.Status, res.Output) + } + sandbox := c.ConversationSandbox() + if err := os.WriteFile(filepath.Join(sandbox, "shell.txt"), []byte("from-shell"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sandbox, "note.txt"), []byte("external"), 0o644); err != nil { + t.Fatal(err) + } + // 外部改动应被跳过。再写一个只被追踪、没有被外改的文件。 + res = c.ExecuteTool(core.ToolCallRequest{ + CallID: "w2", Name: "fs_write_file", + Arguments: []byte(`{"path":"kept.txt","content":"yes"}`), + }) + if res.Status != "ok" { + t.Fatalf("kept %s %s", res.Status, res.Output) + } + got, err := c.mgr.RewindFiles(c.run.Owner, c.run.ConversationID, msg.ID) + if err != nil { + t.Fatal(err) + } + if len(got.Restored) != 1 || got.Restored[0] != "kept.txt" { + t.Fatalf("restored %#v skipped %#v", got.Restored, got.Skipped) + } + if len(got.Skipped) != 1 || got.Skipped[0] != "note.txt" { + t.Fatalf("skipped %#v", got.Skipped) + } + if _, err := os.Stat(filepath.Join(sandbox, "kept.txt")); !os.IsNotExist(err) { + t.Fatal("tracked create was not removed") + } + side, err := os.ReadFile(filepath.Join(sandbox, "shell.txt")) + if err != nil || string(side) != "from-shell" { + t.Fatalf("shell file %q %v", side, err) + } + msgs, err := st.ListMessages(c.run.ConversationID) + if err != nil { + t.Fatal(err) + } + var note string + for _, m := range msgs { + if m.Role == core.RoleSystem && len(m.Parts) > 0 { + note = m.Parts[0].Text + } + } + if note == "" { + t.Fatal("missing system note") + } +} diff --git a/internal/runtime/llm.go b/internal/runtime/llm.go index 50b4f65..e7ed564 100644 --- a/internal/runtime/llm.go +++ b/internal/runtime/llm.go @@ -31,6 +31,8 @@ type LLMRunner struct { Now func() time.Time // VerifyStream 是终轮验证器的注入点（nil = 按轻量模型构造客户端）。 VerifyStream agentllm.StreamFunc + // Trace 记录 invoke_agent、chat、execute_tool。nil 不采集。 + Trace func(name string, attrs map[string]string) func() } func (l *LLMRunner) Name() string { return "llm" } @@ -382,6 +384,8 @@ type SubLLMRunner struct { Prompts *prompts.Service // Stream 提供方边界注入点（nil = 按 profile 的模型档构造客户端）。 Stream agentllm.StreamFunc + // Trace 记录子代理循环的 span。nil 不采集。 + Trace func(name string, attrs map[string]string) func() } func currentUserText(rc RunCtx) string { diff --git a/internal/runtime/llm_direct.go b/internal/runtime/llm_direct.go index 6aa6c4e..c0f4384 100644 --- a/internal/runtime/llm_direct.go +++ b/internal/runtime/llm_direct.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "sync" "time" nruntime "agent-native/runtime" @@ -52,6 +53,9 @@ type directRun struct { checklistReminders int checks []core.CheckPart notes *reminders.Bus + servedModel string + switches []string + switchMu sync.Mutex } type toolEvidence struct { @@ -95,8 +99,13 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin } st.pinSubagentArtifacts(text) } + st.servedModel = model + stream = agentllm.WithFallback(stream, ChatFallbacks(eff), st.noteModelSwitch) defer func() { if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok { + st.switchMu.Lock() + st.usageTotal.Model = st.servedModel + st.switchMu.Unlock() sink.AddUsage(st.usageTotal) } }() @@ -173,6 +182,7 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin OnEvent: st.onEvent, Hooks: bus, Reminders: st.reminders, + Trace: l.Trace, UnknownTool: func(call agentllm.ToolCall) agenttools.Result { return st.runTool(call.Name, call.Arguments) }, @@ -190,6 +200,16 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin for _, ref := range st.deliverables { st.finalParts = append(st.finalParts, core.Part{Type: core.PartArtifactRef, Artifact: ref}) } + st.switchMu.Lock() + notes := append([]string(nil), st.switches...) + st.switchMu.Unlock() + if len(notes) > 0 { + prefix := make([]core.Part, 0, len(notes)) + for _, note := range notes { + prefix = append(prefix, core.Part{Type: core.PartText, Text: note, Process: true}) + } + st.finalParts = append(prefix, st.finalParts...) + } _ = rc.SetAssistantParts(st.finalParts) if err != nil { return err @@ -208,6 +228,13 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin // noteToolDelta 把模型正在编写的工具参数立刻转成会话事件。 // 执行开始前的这段时间原先没有任何工具事件，界面只剩光标。 +func (st *directRun) noteModelSwitch(from, to string) { + st.switchMu.Lock() + defer st.switchMu.Unlock() + st.servedModel = to + st.switches = append(st.switches, fmt.Sprintf("模型 %s 在输出前失败，已改用 %s", from, to)) +} + func (st *directRun) noteToolDelta(ev agentllm.Event) { if !st.acting { _ = st.rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseActing}) diff --git a/internal/runtime/model.go b/internal/runtime/model.go index f97ebd7..7e615f1 100644 --- a/internal/runtime/model.go +++ b/internal/runtime/model.go @@ -1,6 +1,8 @@ package runtime import ( + "strings" + core "agent-system-core" ) @@ -27,6 +29,25 @@ func ResolveChatModel(prof core.AgentProfile, eff core.EffectiveSettings) Resolv Key: eff.GetString("models.chat.apiKey", "")} } +// ChatFallbacks 解析 models.chat.fallbacks。默认空，不改变现有模型选择。 +func ChatFallbacks(eff core.EffectiveSettings) []string { + raw := strings.TrimSpace(eff.GetString("models.chat.fallbacks", "")) + if raw == "" { + return nil + } + var out []string + seen := map[string]bool{} + for _, part := range strings.Split(raw, ",") { + name := strings.TrimSpace(part) + if name == "" || seen[name] { + continue + } + seen[name] = true + out = append(out, name) + } + return out +} + // ResolveLightweightModel 解析轻量模型：画像 route > 设置层轻量档 > 回落主档。 func ResolveLightweightModel(prof core.AgentProfile, eff core.EffectiveSettings) ResolvedModel { m := eff.GetString("models.lightweight.model", "") diff --git a/internal/runtime/model_test.go b/internal/runtime/model_test.go index 97d6a12..bac984b 100644 --- a/internal/runtime/model_test.go +++ b/internal/runtime/model_test.go @@ -45,3 +45,16 @@ func TestResolveLightweightFallback(t *testing.T) { t.Fatalf("profile lightweight + chat endpoint: %+v", got) } } + +func TestChatFallbacksDefaultEmpty(t *testing.T) { + if got := ChatFallbacks(effOf(nil)); got != nil { + t.Fatalf("default %#v", got) + } + eff := effOf(map[string]any{"models": map[string]any{"chat": map[string]any{ + "fallbacks": " backup , primary ,backup", + }}}) + got := ChatFallbacks(eff) + if len(got) != 2 || got[0] != "backup" || got[1] != "primary" { + t.Fatalf("%#v", got) + } +} diff --git a/internal/runtime/sub_loop.go b/internal/runtime/sub_loop.go index d5b65d2..9206315 100644 --- a/internal/runtime/sub_loop.go +++ b/internal/runtime/sub_loop.go @@ -57,6 +57,9 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error { client := openLLMClient(route.Base, route.Key) stream = client.Stream } + if spec.Model == "chat" { + stream = agentllm.WithFallback(stream, ChatFallbacks(rc.Settings()), nil) + } decls := subagentDecls(rc, spec) st := &subRun{rc: rc} reg := agenttools.NewRegistry("") @@ -105,6 +108,7 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error { OutputReserve: reserve, ContextBuffer: buffer, OnEvent: st.onEvent, + Trace: l.Trace, }, user) if len(st.parts) == 0 { st.parts = append(st.parts, core.Part{Type: core.PartText, Text: "（子代理未返回内容）"}) diff --git a/internal/server/rewind.go b/internal/server/rewind.go new file mode 100644 index 0000000..80d7748 --- /dev/null +++ b/internal/server/rewind.go @@ -0,0 +1,56 @@ +package server + +import ( + "encoding/json" + "errors" + "net/http" + + "agent-system-backend/internal/runmgr" +) + +func (s *Server) listCheckpoints(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + owner := s.owner(r) + if _, err := s.st.GetConversation(owner, id); err != nil { + writeStoreErr(w, err) + return + } + messageID := r.URL.Query().Get("messageId") + plan, err := s.mgr.PlanRewind(owner, id, messageID) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusOK, map[string]any{"restored": plan.Restored, "skipped": plan.Skipped}) +} + +func (s *Server) rewindFiles(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + owner := s.owner(r) + if _, err := s.st.GetConversation(owner, id); err != nil { + writeStoreErr(w, err) + return + } + var req struct { + MessageID string `json:"messageId"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil || req.MessageID == "" { + http.Error(w, "messageId required", http.StatusBadRequest) + return + } + if err := s.mgr.ClaimConversation(id); err != nil { + if errors.Is(err, runmgr.ErrConversationBusy) { + http.Error(w, "conversation busy", http.StatusConflict) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + defer s.mgr.ReleaseConversation(id) + res, err := s.mgr.RewindFiles(owner, id, req.MessageID) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusOK, map[string]any{"restored": res.Restored, "skipped": res.Skipped}) +} diff --git a/internal/server/server.go b/internal/server/server.go index 1f4dc16..a06fb36 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -165,6 +165,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("PATCH /api/conversations/{id}", s.patchConversation) mux.HandleFunc("DELETE /api/conversations/{id}", s.deleteConversation) mux.HandleFunc("POST /api/conversations/{id}/messages", s.postMessage) + mux.HandleFunc("GET /api/conversations/{id}/checkpoints", s.listCheckpoints) + mux.HandleFunc("POST /api/conversations/{id}/rewind", s.rewindFiles) mux.HandleFunc("POST /api/runs/{id}/cancel", s.cancelRun) mux.HandleFunc("GET /api/runs/{id}", s.getRun) diff --git a/internal/settings/settings.go b/internal/settings/settings.go index 2b6eaa6..79d89f1 100644 --- a/internal/settings/settings.go +++ b/internal/settings/settings.go @@ -18,9 +18,10 @@ func SystemDefaults() map[string]any { return map[string]any{ "models": map[string]any{ "chat": map[string]any{ - "baseUrl": "", // 例：https://api.openai.com/v1 （OpenAI 兼容） - "apiKey": "", - "model": "", + "baseUrl": "", // 例：https://api.openai.com/v1 （OpenAI 兼容） + "apiKey": "", + "model": "", + "fallbacks": "", // 逗号分隔；空 = 不回退 }, "lightweight": map[string]any{"baseUrl": "", "apiKey": "", "model": ""}, "embedding": map[string]any{"baseUrl": "", "apiKey": "", "model": ""}, @@ -65,6 +66,7 @@ func Schema() []core.SettingDesc { {Key: "models.chat.baseUrl", Group: "models", Label: "主模型 Base URL", Type: core.SettingString, Default: "", Effect: core.EffectHot, Description: "OpenAI 兼容端点，如 https://api.openai.com/v1"}, {Key: "models.chat.apiKey", Group: "models", Label: "主模型 API Key", Type: core.SettingSecret, Default: "", Effect: core.EffectHot}, {Key: "models.chat.model", Group: "models", Label: "主模型名称", Type: core.SettingString, Default: "", Effect: core.EffectNewRun}, + {Key: "models.chat.fallbacks", Group: "models", Label: "主模型回退链", Type: core.SettingString, Default: "", Effect: core.EffectNewRun, Description: "逗号分隔的模型名。只在尚未输出时，前一个模型返回 5xx 才改用下一个。空表示不回退"}, {Key: "models.chat.maxOutputTokens", Group: "models", Label: "主模型输出上限", Type: core.SettingNumber, Default: 0, Effect: core.EffectNewRun, Description: "单次回复的 max_tokens。0 = 不限制，由模型服务端决定"}, {Key: "models.lightweight.baseUrl", Group: "models", Label: "轻量模型 Base URL", Type: core.SettingString, Default: "", Effect: core.EffectHot, Description: "标题生成/澄清等轻任务；空则回落主模型"}, {Key: "models.lightweight.apiKey", Group: "models", Label: "轻量模型 API Key", Type: core.SettingSecret, Default: "", Effect: core.EffectHot}, diff --git a/internal/tools/fs.go b/internal/tools/fs.go index 9b0ef9a..913c93a 100644 --- a/internal/tools/fs.go +++ b/internal/tools/fs.go @@ -151,9 +151,15 @@ func execFSWrite(rc ToolExecCtx, args json.RawMessage) (string, error) { } } } + before, existed := fsBefore(abs) if err := os.WriteFile(abs, []byte(a.Content), 0o644); err != nil { return "", err } + if cp, ok := rc.(interface { + NoteFileCheckpoint(rel string, before []byte, existed bool, after []byte) error + }); ok { + _ = cp.NoteFileCheckpoint(a.Path, before, existed, []byte(a.Content)) + } // 不变式：工作区文件 = 工件面板可见（内容同步入对象库，工件自包含可下载）。 base := filepath.Base(abs) art, err := rc.WriteArtifact(base, "file", mimeOfExt(base), []byte(a.Content)) @@ -163,6 +169,14 @@ func execFSWrite(rc ToolExecCtx, args json.RawMessage) (string, error) { return fmt.Sprintf("已写入 %s（%d 字节，工件 %s）", a.Path, len(a.Content), art.ID), nil } +func fsBefore(abs string) ([]byte, bool) { + b, err := os.ReadFile(abs) + if err != nil { + return nil, false + } + return b, true +} + func execFSList(rc ToolExecCtx, args json.RawMessage) (string, error) { var a struct { Path string `json:"path"`