Add session compaction, metadata, and crash-safe transcript loading
Why: long multi-turn executions grow the JSONL transcript until they hit provider context limits (parity-matrix #11), and a crashed append left a partial line that made the whole session unloadable (parity-matrix #12 resume semantics). Evidence: go vet ./... clean; go test ./... 13 packages ok. Session cases: crash-residue tail skipped while mid-file corruption still errors, Stats line/byte counts, meta sidecar roundtrip (missing meta nil), Compact rewriting system+summary+keepLastN atomically with no tmp residue, no-op on short transcripts, summarizer error leaving the file untouched. Adapter case: threshold-triggered compaction runs before the loop and the summary marker reaches both the transcript and the model's context. Impact: session gains Stats/SaveMeta/ReadMeta/Compact; native.Options gains Compact *CompactConfig (default summarizer is one tool-less stream call; compaction failure logs and continues - optimization, never a silent failure); session meta sidecar records model/scope/ exec id per execution. Refs: docs/parity-matrix.md #11/#12 (Batch3 scope).
This commit is contained in:
@@ -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{
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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 把会话元数据写进 <path>.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
|
||||
}
|
||||
@@ -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) }
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user