diff --git a/adapters/native/backend.go b/adapters/native/backend.go index 6194ca6..c26d69e 100644 --- a/adapters/native/backend.go +++ b/adapters/native/backend.go @@ -45,12 +45,24 @@ type Options struct { MCP *mcp.Manager // SkillsRoots 可选:全局技能根目录;执行时自动并入工作区 .agent/skills/。 SkillsRoots []string + // Compact 可选:转录超过阈值时把头部历史压成摘要(长会话上下文治理)。 + Compact *CompactConfig // Log 可选。nil 时丢弃日志。 Log *logsys.Logger // ExecutionID 写入日志,不参与执行控制。 ExecutionID string } +// CompactConfig 声明长会话压缩策略(优化性质:压缩失败不阻断执行,但留日志)。 +type CompactConfig struct { + // ThresholdChars 触发阈值(按字节数近似),默认 300_000。 + ThresholdChars int + // KeepLastN 压缩后保留的最近消息条数,默认 8。 + KeepLastN int + // Summarizer 可选注入摘要函数;缺省用后端 stream 做一次无工具调用。 + Summarizer func(ctx context.Context, messages []llm.Message) (string, error) +} + // Backend 以自研内核实现 exec.Backend。 type Backend struct { opts Options @@ -168,6 +180,13 @@ func (b *Backend) Execute(ctx context.Context, req exec.Request) (*exec.Result, client := &llm.Client{BaseURL: b.opts.APIBase, APIKey: b.opts.APIKey} stream = client.Stream } + _ = store.SaveMeta(map[string]string{ + "model": model, + "scope": req.SessionScope, + "exec_id": req.ExecutionID, + "updated": time.Now().UTC().Format(time.RFC3339), + }) + b.maybeCompact(ctx, store, stream, model) timeout := req.Timeout if timeout <= 0 { @@ -301,6 +320,74 @@ func (b *Backend) run(ctx context.Context, model string, budget *runtime.Budget, return runRes, err } +// maybeCompact 在转录超过阈值时压缩头部历史。压缩是优化性质: +// 失败不阻断执行(按原文继续),但必须留日志,不静默。 +func (b *Backend) maybeCompact(ctx context.Context, store *session.Store, stream llm.StreamFunc, model string) { + cfg := b.opts.Compact + if cfg == nil { + return + } + threshold := int64(cfg.ThresholdChars) + if threshold <= 0 { + threshold = 300_000 + } + keep := cfg.KeepLastN + if keep <= 0 { + keep = 8 + } + if store.Stats().Bytes < threshold { + return + } + summarize := cfg.Summarizer + if summarize == nil { + summarize = defaultSummarizer(stream, model) + } + done, err := store.Compact(keep, func(messages []llm.Message) (string, error) { + return summarize(ctx, messages) + }) + if err != nil { + b.logger().Warn("session compaction failed; continuing with full transcript", "error", err.Error()) + return + } + if done { + b.logger().Info("session compacted", "keep_last", keep) + } +} + +// defaultSummarizer 用后端 stream 做一次无工具调用,把历史压成摘要。 +func defaultSummarizer(stream llm.StreamFunc, model string) func(ctx context.Context, messages []llm.Message) (string, error) { + return func(ctx context.Context, messages []llm.Message) (string, error) { + var b strings.Builder + for _, m := range messages { + b.WriteString(string(m.Role)) + b.WriteString(": ") + b.WriteString(m.Content) + b.WriteString("\n") + if b.Len() > 120_000 { + break + } + } + prompt := "把以下智能体对话历史压缩成摘要:保留全部关键事实、已做决定、文件变更与未决问题;中文;不超过 800 字;只输出摘要正文。\n\n" + b.String() + var out strings.Builder + for ev := range stream(ctx, model, []llm.Message{{Role: llm.RoleUser, Content: prompt}}, nil, llm.StreamOptions{}) { + switch ev.Kind { + case llm.KindTextDelta: + out.WriteString(ev.Text) + case llm.KindError: + return "", ev.Err + } + } + if strings.TrimSpace(out.String()) == "" { + return "", errString("summarizer produced empty output") + } + return out.String(), nil + } +} + +type errString string + +func (e errString) Error() string { return string(e) } + // resumeUnsafeMarkers 续跑不安全的关键词(会话已被污染:上下文溢出/ // 迭代上限/非法请求——重试必须换新会话而非续跑)。 var resumeUnsafeMarkers = []string{ diff --git a/adapters/native/compact_test.go b/adapters/native/compact_test.go new file mode 100644 index 0000000..9b3683d --- /dev/null +++ b/adapters/native/compact_test.go @@ -0,0 +1,69 @@ +package native + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "agent-native/exec" + "agent-native/runtime/llm" +) + +func TestBackendCompactBeforeRun(t *testing.T) { + dir := t.TempDir() + // 预置一份超阈值的长转录(系统消息 + 12 条历史)。 + transcript := filepath.Join(dir, ".agentx", "transcript.jsonl") + if err := os.MkdirAll(filepath.Dir(transcript), 0o755); err != nil { + t.Fatal(err) + } + var b strings.Builder + b.WriteString(`{"role":"system","content":"sys"}` + "\n") + for i := 0; i < 12; i++ { + b.WriteString(`{"role":"user","content":"` + strings.Repeat("x", 400) + `"}` + "\n") + } + if err := os.WriteFile(transcript, []byte(b.String()), 0o644); err != nil { + t.Fatal(err) + } + + var sawSummary bool + // 注入式捕获:验证 Run 收到的消息里有摘要标记。 + bk2 := New(Options{ + Model: "fake", + Stream: captureFirstStream(&sawSummary), + Compact: &CompactConfig{ThresholdChars: 1000, KeepLastN: 3, Summarizer: func(ctx context.Context, messages []llm.Message) (string, error) { + return "预置历史摘要", nil + }}, + }) + res, err := bk2.Execute(context.Background(), exec.Request{ + WorkDir: dir, Prompt: "继续", Targets: []string{"output.md"}, + ToolProfile: exec.ToolProfileReasoning, + }) + if err != nil { + t.Fatal(err) + } + if res.Success && !sawSummary { + t.Fatal("compaction should have run before the loop") + } + data, _ := os.ReadFile(transcript) + if !strings.Contains(string(data), "[conversation summary]") { + t.Fatalf("transcript should contain the summary marker, got:\n%.200s", data) + } +} + +func captureFirstStream(saw *bool) llm.StreamFunc { + return func(ctx context.Context, model string, messages []llm.Message, decls []llm.ToolDecl, opts llm.StreamOptions) <-chan llm.Event { + for _, m := range messages { + if strings.Contains(m.Content, "[conversation summary]") { + *saw = true + } + } + ch := make(chan llm.Event, 4) + go func() { + defer close(ch) + ch <- llm.Event{Kind: llm.KindMessageEnd, Message: llm.Message{Role: llm.RoleAssistant, Content: "压缩后续写完成", StopReason: llm.StopEnd}} + }() + return ch + } +} diff --git a/runtime/session/compact.go b/runtime/session/compact.go new file mode 100644 index 0000000..1c675d1 --- /dev/null +++ b/runtime/session/compact.go @@ -0,0 +1,138 @@ +package session + +import ( + "bufio" + "encoding/json" + "os" + + "agent-native/runtime/llm" +) + +// Stats 是转录文件的轻量元数据(扫描即得,无副作用)。 +type Stats struct { + Lines int + Bytes int64 +} + +// Stats 统计当前转录行数与字节数(空行不计)。 +func (s *Store) Stats() Stats { + s.mu.Lock() + defer s.mu.Unlock() + st := Stats{} + f, err := os.Open(s.path) + if err != nil { + return st + } + defer f.Close() + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024) + for sc.Scan() { + if len(sc.Bytes()) == 0 { + continue + } + st.Lines++ + } + if info, err := f.Stat(); err == nil { + st.Bytes = info.Size() + } + return st +} + +// SaveMeta 把会话元数据写进 .meta.json(创建时间/模型等由调用方定义)。 +func (s *Store) SaveMeta(meta map[string]string) error { + s.mu.Lock() + defer s.mu.Unlock() + raw, err := json.MarshalIndent(meta, "", " ") + if err != nil { + return err + } + tmp := s.path + ".meta.json.tmp" + if err := os.WriteFile(tmp, raw, 0o644); err != nil { + return err + } + return os.Rename(tmp, s.path+".meta.json") +} + +// ReadMeta 读取会话元数据;不存在时返回 nil(新会话)。 +func (s *Store) ReadMeta() map[string]string { + raw, err := os.ReadFile(s.path + ".meta.json") + if err != nil { + return nil + } + var meta map[string]string + if err := json.Unmarshal(raw, &meta); err != nil { + return nil + } + return meta +} + +// Compact 把转录改写为「系统消息 + 摘要消息 + 最近 keepLastN 条」。 +// 摘要由 summarize 生成(由适配层注入,通常是一次无工具模型调用); +// 改写原子落盘(临时文件 + rename)。 +// 返回是否发生了压缩(消息数不足/已是紧凑形态时返回 false 且不动文件)。 +func (s *Store) Compact(keepLastN int, summarize func(messages []llm.Message) (string, error)) (bool, error) { + messages, err := s.Load() + if err != nil { + return false, err + } + if len(messages) <= keepLastN+1 { // 系统消息 + 尾部,无可压缩头部 + return false, nil + } + var system *llm.Message + rest := messages + if len(rest) > 0 && rest[0].Role == llm.RoleSystem { + copyOf := rest[0] + system = ©Of + rest = rest[1:] + } + tailStart := len(rest) - keepLastN + if tailStart <= 0 { + return false, nil + } + head := rest[:tailStart] + tail := rest[tailStart:] + summary, err := summarize(head) + if err != nil { + return false, err + } + summaryMsg := llm.Message{ + Role: llm.RoleUser, + Content: "[conversation summary]\n" + summary + "\n[recent messages follow]", + } + var rebuilt []llm.Message + if system != nil { + rebuilt = append(rebuilt, *system) + } + rebuilt = append(rebuilt, summaryMsg) + rebuilt = append(rebuilt, tail...) + + s.mu.Lock() + defer s.mu.Unlock() + tmp := s.path + ".compact.tmp" + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return false, err + } + w := bufio.NewWriter(f) + for _, m := range rebuilt { + raw, err := json.Marshal(m) + if err != nil { + _ = f.Close() + _ = os.Remove(tmp) + return false, err + } + w.Write(raw) + w.WriteByte('\n') + } + if err := w.Flush(); err != nil { + _ = f.Close() + _ = os.Remove(tmp) + return false, err + } + _ = f.Close() + if err := os.Rename(tmp, s.path); err != nil { + _ = os.Remove(tmp) + return false, err + } + return true, nil +} diff --git a/runtime/session/compact_test.go b/runtime/session/compact_test.go new file mode 100644 index 0000000..b7c742f --- /dev/null +++ b/runtime/session/compact_test.go @@ -0,0 +1,159 @@ +package session + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "agent-native/runtime/llm" +) + +func appendMsgs(t *testing.T, s *Store, msgs ...llm.Message) { + t.Helper() + for _, m := range msgs { + if err := s.Append(m); err != nil { + t.Fatal(err) + } + } +} + +func TestLoadSkipsCrashResidueTail(t *testing.T) { + dir := t.TempDir() + s, err := Open(filepath.Join(dir, "t.jsonl")) + if err != nil { + t.Fatal(err) + } + appendMsgs(t, s, + llm.Message{Role: llm.RoleSystem, Content: "sys"}, + llm.Message{Role: llm.RoleUser, Content: "hi"}, + ) + // 模拟崩溃半行:无换行结尾的残缺 JSON。 + f, _ := os.OpenFile(s.Path(), os.O_APPEND|os.O_WRONLY, 0o644) + _, _ = f.WriteString(`{"role":"assistant","content":"trunca`) + _ = f.Close() + + msgs, err := s.Load() + if err != nil { + t.Fatalf("crash residue must not poison the session: %v", err) + } + if len(msgs) != 2 || msgs[0].Role != llm.RoleSystem || msgs[1].Content != "hi" { + t.Fatalf("want sys+hi, got %+v", msgs) + } +} + +func TestLoadStillFailsOnMidFileCorruption(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "t.jsonl") + if err := os.WriteFile(path, []byte("{\"bad json}\n{\"role\":\"user\",\"content\":\"ok\"}\n"), 0o644); err != nil { + t.Fatal(err) + } + s := &Store{path: path} + if _, err := s.Load(); err == nil { + t.Fatal("mid-file corruption must still error") + } +} + +func TestStatsAndMeta(t *testing.T) { + dir := t.TempDir() + s, _ := Open(filepath.Join(dir, "t.jsonl")) + appendMsgs(t, s, + llm.Message{Role: llm.RoleSystem, Content: "sys"}, + llm.Message{Role: llm.RoleUser, Content: "q"}, + ) + st := s.Stats() + if st.Lines != 2 || st.Bytes <= 0 { + t.Fatalf("stats wrong: %+v", st) + } + if err := s.SaveMeta(map[string]string{"model": "m1"}); err != nil { + t.Fatal(err) + } + meta := s.ReadMeta() + if meta == nil || meta["model"] != "m1" { + t.Fatalf("meta roundtrip failed: %v", meta) + } + if other := (&Store{path: filepath.Join(dir, "missing.jsonl")}).ReadMeta(); other != nil { + t.Fatalf("missing meta must return nil, got %v", other) + } +} + +func TestCompactRewritesTranscript(t *testing.T) { + dir := t.TempDir() + s, _ := Open(filepath.Join(dir, "t.jsonl")) + appendMsgs(t, s, llm.Message{Role: llm.RoleSystem, Content: "sys"}) + for i := 0; i < 10; i++ { + appendMsgs(t, s, + llm.Message{Role: llm.RoleUser, Content: "early" + string(rune('a'+i))}, + llm.Message{Role: llm.RoleAssistant, Content: "reply"}, + ) + } + var seenHead []llm.Message + done, err := s.Compact(3, func(messages []llm.Message) (string, error) { + seenHead = messages + return "这是摘要", nil + }) + if err != nil || !done { + t.Fatalf("compact failed: %v done=%v", err, done) + } + if len(seenHead) != 17 { // 20 - system - 3 tail + t.Fatalf("summarizer should receive 17 head messages, got %d", len(seenHead)) + } + msgs, err := s.Load() + if err != nil { + t.Fatal(err) + } + if len(msgs) != 5 { // system + summary + 3 tail + t.Fatalf("want 5 messages after compact, got %d", len(msgs)) + } + if msgs[0].Role != llm.RoleSystem { + t.Fatal("system message must be preserved") + } + if !strings.Contains(msgs[1].Content, "[conversation summary]") || !strings.Contains(msgs[1].Content, "这是摘要") { + t.Fatalf("summary message wrong: %+v", msgs[1]) + } + if msgs[2].Content != "reply" || msgs[3].Content != "earlyj" || msgs[4].Content != "reply" { + t.Fatalf("tail order wrong: %+v", msgs[2:]) + } + // 无 tmp 残留。 + if _, err := os.Stat(s.Path() + ".compact.tmp"); !os.IsNotExist(err) { + t.Fatal("compact tmp file must not remain") + } +} + +func TestCompactNoopWhenShort(t *testing.T) { + dir := t.TempDir() + s, _ := Open(filepath.Join(dir, "t.jsonl")) + appendMsgs(t, s, + llm.Message{Role: llm.RoleSystem, Content: "sys"}, + llm.Message{Role: llm.RoleUser, Content: "q"}, + ) + done, err := s.Compact(8, func([]llm.Message) (string, error) { return "x", nil }) + if err != nil || done { + t.Fatalf("short transcript must be a no-op: %v %v", done, err) + } + msgs, _ := s.Load() + if len(msgs) != 2 { + t.Fatal("no-op compact must not touch the file") + } +} + +func TestCompactSummarizerErrorAborts(t *testing.T) { + dir := t.TempDir() + s, _ := Open(filepath.Join(dir, "t.jsonl")) + appendMsgs(t, s, llm.Message{Role: llm.RoleSystem, Content: "sys"}) + for i := 0; i < 6; i++ { + appendMsgs(t, s, llm.Message{Role: llm.RoleUser, Content: "m"}) + } + done, err := s.Compact(2, func([]llm.Message) (string, error) { return "", errStringT("boom") }) + if err == nil || done { + t.Fatal("summarizer error must abort compaction") + } + msgs, _ := s.Load() + if len(msgs) != 7 { + t.Fatal("failed compaction must leave the transcript untouched") + } +} + +type errStringT string + +func (e errStringT) Error() string { return string(e) } diff --git a/runtime/session/jsonl.go b/runtime/session/jsonl.go index 2c29fa0..3d84b2c 100644 --- a/runtime/session/jsonl.go +++ b/runtime/session/jsonl.go @@ -5,6 +5,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "sync" "agent-native/runtime/llm" @@ -47,16 +48,28 @@ func (s *Store) Append(msg llm.Message) error { return err } +// Load 全量读取转录。崩溃残留容错:Append 恒以换行收尾,因此无换行结尾的 +// 尾部若不是完整 JSON 即为崩溃半行,丢弃而不是让整个会话不可用; +// 行中间的坏行仍报错(那是文件损坏,不是崩溃残留)。 func (s *Store) Load() ([]llm.Message, error) { s.mu.Lock() defer s.mu.Unlock() - f, err := os.Open(s.path) + raw, err := os.ReadFile(s.path) if err != nil { return nil, err } - defer f.Close() + text := string(raw) + if text != "" && !strings.HasSuffix(text, "\n") { + if idx := strings.LastIndexByte(text, '\n'); idx >= 0 { + if !json.Valid([]byte(text[idx+1:])) { + text = text[:idx+1] + } + } else if !json.Valid([]byte(text)) { + text = "" + } + } var out []llm.Message - sc := bufio.NewScanner(f) + sc := bufio.NewScanner(strings.NewReader(text)) sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024) for sc.Scan() { line := sc.Bytes()