diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 82f39939..426e3585 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -38,6 +38,13 @@ var ( setConsoleInputCodePage = platformSetConsoleInputCodePage buildToolManagerFunc = buildToolManager newTUIWithMemo = tui.NewWithMemo + cleanupExpiredSessions = func( + ctx context.Context, + store agentsession.Store, + maxAge time.Duration, + ) (int, error) { + return store.CleanupExpiredSessions(ctx, maxAge) + } ) // BootstrapOptions 描述应用启动时可注入的运行时选项。 @@ -150,6 +157,11 @@ func BuildRuntime(ctx context.Context, opts BootstrapOptions) (RuntimeBundle, er // 这意味着所有会话都归属到启动时指定的项目目录下，运行时不会因配置变更而迁移存储位置。 sessionStore := agentsession.NewStore(loader.BaseDir(), cfg.Workdir) + // 启动时自动清理过期会话，避免数据库无限膨胀。 + if _, err := cleanupExpiredSessions(ctx, sessionStore, agentsession.DefaultSessionMaxAge); err != nil { + log.Printf("session cleanup warning: %v", err) + } + // 注册内置工具的内容摘要器，使 micro-compact 在清理旧工具结果时保留关键上下文。 tools.RegisterBuiltinSummarizers(toolRegistry) diff --git a/internal/app/bootstrap_test.go b/internal/app/bootstrap_test.go index efc03ef1..8dd6b489 100644 --- a/internal/app/bootstrap_test.go +++ b/internal/app/bootstrap_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "log" "os" "path/filepath" "reflect" @@ -721,6 +722,11 @@ func TestBuildRuntimeUsesWorkdirOverride(t *testing.T) { if bundle.ConfigManager == nil || bundle.Runtime == nil || bundle.ProviderSelection == nil { t.Fatalf("expected runtime bundle dependencies, got %+v", bundle) } + if bundle.Close != nil { + t.Cleanup(func() { + _ = bundle.Close() + }) + } } func TestBuildRuntimeSucceedsWhenSkillsRootMissing(t *testing.T) { @@ -987,6 +993,43 @@ func TestBuildRuntimeCleansResourcesWhenToolManagerBuildFails(t *testing.T) { } } +func TestBuildRuntimeLogsSessionCleanupWarningAndContinues(t *testing.T) { + disableBuiltinProviderAPIKeys(t) + + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("USERPROFILE", home) + + originalCleanupExpiredSessions := cleanupExpiredSessions + t.Cleanup(func() { cleanupExpiredSessions = originalCleanupExpiredSessions }) + cleanupExpiredSessions = func( + ctx context.Context, + store agentsession.Store, + maxAge time.Duration, + ) (int, error) { + return 0, errors.New("cleanup failed") + } + + var logBuffer bytes.Buffer + originalLogWriter := log.Writer() + log.SetOutput(&logBuffer) + t.Cleanup(func() { log.SetOutput(originalLogWriter) }) + + bundle, err := BuildRuntime(context.Background(), BootstrapOptions{}) + if err != nil { + t.Fatalf("BuildRuntime() error = %v", err) + } + if bundle.Close != nil { + defer bundle.Close() + } + if bundle.Runtime == nil { + t.Fatalf("expected runtime bundle to be created") + } + if !strings.Contains(logBuffer.String(), "session cleanup warning: cleanup failed") { + t.Fatalf("expected cleanup warning in logs, got %q", logBuffer.String()) + } +} + func TestNewProgramCleansResourcesWhenTUIBuildFails(t *testing.T) { disableBuiltinProviderAPIKeys(t) diff --git a/internal/context/microcompact.go b/internal/context/microcompact.go index 37993eb2..4e82adef 100644 --- a/internal/context/microcompact.go +++ b/internal/context/microcompact.go @@ -23,58 +23,86 @@ func microCompactMessages(messages []providertypes.Message) []providertypes.Mess } // microCompactMessagesWithPolicies 按工具策略对裁剪后的消息做只读投影式微压缩。 +// 仅对需要压缩的工具消息做深拷贝，其余消息共享原始引用以减少内存分配。 func microCompactMessagesWithPolicies(messages []providertypes.Message, policies MicroCompactPolicySource, retainedToolSpans int, summarizers MicroCompactSummarizerSource) []providertypes.Message { if retainedToolSpans <= 0 { retainedToolSpans = defaultMicroCompactRetainedToolSpans } - cloned := cloneContextMessages(messages) - if len(cloned) == 0 { - return cloned + if len(messages) == 0 { + return nil } - spans := internalcompact.BuildMessageSpans(cloned) + spans := internalcompact.BuildMessageSpans(messages) protectedStart, hasProtectedTail := internalcompact.ProtectedTailStart(spans) retainedCompactableSpans := 0 + modifiedIndices := make(map[int]struct{}) + var pendingCompactions []compactionPending + for spanIndex := len(spans) - 1; spanIndex >= 0; spanIndex-- { span := spans[spanIndex] if hasProtectedTail && span.Start >= protectedStart { continue } - if !isToolCallSpan(cloned, span) { + if !isToolCallSpan(messages, span) { continue } - compactableIDs, toolNames := compactableToolCallIDs(cloned[span.Start].ToolCalls, policies) + compactableIDs, toolNames := compactableToolCallIDs(messages[span.Start].ToolCalls, policies) if len(compactableIDs) == 0 { continue } if retainedCompactableSpans < retainedToolSpans { - if hasCompactableToolMessage(cloned, span, compactableIDs) { + if hasCompactableToolMessage(messages, span, compactableIDs) { retainedCompactableSpans++ } continue } - compactableContents := compactableToolMessageContents(cloned, span, compactableIDs) + compactableContents := compactableToolMessageContents(messages, span, compactableIDs) if len(compactableContents) == 0 { continue } - for messageIndex := span.Start + 1; messageIndex < span.End; messageIndex++ { - content, ok := compactableContents[messageIndex] - if !ok { - continue - } - summary := summarizeOrClear(cloned[messageIndex], content, toolNames, summarizers) - cloned[messageIndex].Parts = []providertypes.ContentPart{providertypes.NewTextPart(summary)} + for messageIndex, content := range compactableContents { + modifiedIndices[messageIndex] = struct{}{} + pendingCompactions = append(pendingCompactions, compactionPending{ + index: messageIndex, + content: content, + toolNames: toolNames, + }) + } + } + + if len(modifiedIndices) == 0 { + return append([]providertypes.Message(nil), messages...) + } + + cloned := make([]providertypes.Message, len(messages)) + for i, msg := range messages { + if _, needsClone := modifiedIndices[i]; needsClone { + cloned[i] = cloneSingleMessage(msg) + } else { + cloned[i] = msg } } + for _, pending := range pendingCompactions { + summary := summarizeOrClear(cloned[pending.index], pending.content, pending.toolNames, summarizers) + cloned[pending.index].Parts = []providertypes.ContentPart{providertypes.NewTextPart(summary)} + } + return cloned } +// compactionPending 记录待压缩的消息索引和所需上下文。 +type compactionPending struct { + index int + content string + toolNames map[string]string +} + // cloneContextMessages 深拷贝消息切片，避免读时投影污染 runtime 持有的原始会话消息。 func cloneContextMessages(messages []providertypes.Message) []providertypes.Message { if len(messages) == 0 { @@ -83,19 +111,24 @@ func cloneContextMessages(messages []providertypes.Message) []providertypes.Mess cloned := make([]providertypes.Message, 0, len(messages)) for _, message := range messages { - next := message - next.ToolCalls = append([]providertypes.ToolCall(nil), message.ToolCalls...) - if len(message.ToolMetadata) > 0 { - next.ToolMetadata = make(map[string]string, len(message.ToolMetadata)) - for key, value := range message.ToolMetadata { - next.ToolMetadata[key] = value - } - } - cloned = append(cloned, next) + cloned = append(cloned, cloneSingleMessage(message)) } return cloned } +// cloneSingleMessage 深拷贝单条消息，隔离 ToolCalls 和 ToolMetadata 的底层引用。 +func cloneSingleMessage(msg providertypes.Message) providertypes.Message { + next := msg + next.ToolCalls = append([]providertypes.ToolCall(nil), msg.ToolCalls...) + if len(msg.ToolMetadata) > 0 { + next.ToolMetadata = make(map[string]string, len(msg.ToolMetadata)) + for key, value := range msg.ToolMetadata { + next.ToolMetadata[key] = value + } + } + return next +} + // isToolCallSpan 判断当前 span 是否是由 assistant tool call 起始的原子工具块。 func isToolCallSpan(messages []providertypes.Message, span internalcompact.MessageSpan) bool { if span.Start < 0 || span.Start >= len(messages) { diff --git a/internal/context/source_system.go b/internal/context/source_system.go index d78bdda7..e5fa4eae 100644 --- a/internal/context/source_system.go +++ b/internal/context/source_system.go @@ -6,8 +6,12 @@ import ( "fmt" "os/exec" "strings" + "time" ) +// gitCommandTimeout 定义 git 命令的最大等待时间，避免网络挂载或损坏仓库阻塞上下文构建。 +const gitCommandTimeout = 5 * time.Second + type gitCommandRunner func(ctx context.Context, workdir string, args ...string) (string, error) // collectSystemState 汇总运行时上下文，并通过一次 git status 调用获取分支与脏状态。 @@ -104,8 +108,11 @@ func renderSystemStateSection(state SystemState) promptSection { } } +// runGitCommand 执行 git 命令并在超时后自动取消，避免阻塞上下文构建主链路。 func runGitCommand(ctx context.Context, workdir string, args ...string) (string, error) { - command := exec.CommandContext(ctx, "git", append([]string{"-C", workdir}, args...)...) + timeoutCtx, cancel := context.WithTimeout(ctx, gitCommandTimeout) + defer cancel() + command := exec.CommandContext(timeoutCtx, "git", append([]string{"-C", workdir}, args...)...) output, err := command.Output() if err != nil { return "", err diff --git a/internal/memo/auto_extractor.go b/internal/memo/auto_extractor.go index 666342af..7c172c78 100644 --- a/internal/memo/auto_extractor.go +++ b/internal/memo/auto_extractor.go @@ -2,6 +2,8 @@ package memo import ( "context" + "encoding/json" + "hash/fnv" "log" "strings" "sync" @@ -29,13 +31,14 @@ type AutoExtractor struct { } type autoExtractState struct { - mu sync.Mutex - pending *autoExtractRequest - running bool - timer *time.Timer - idleTimer *time.Timer - scheduleSeq uint64 - idleSeq uint64 + mu sync.Mutex + pending *autoExtractRequest + running bool + timer *time.Timer + idleTimer *time.Timer + scheduleSeq uint64 + idleSeq uint64 + lastFingerprint uint64 } type autoExtractRequest struct { @@ -62,16 +65,11 @@ func NewAutoExtractor(extractor Extractor, svc *Service, extractTimeout time.Dur // Schedule 按会话维度安排一次后台自动提取。 func (a *AutoExtractor) Schedule(sessionID string, messages []providertypes.Message) { - a.schedule(sessionID, messages, a.extractor) + a.ScheduleWithExtractor(sessionID, messages, a.extractor) } // ScheduleWithExtractor 允许调用方在调度时绑定本次请求专用的提取器快照。 func (a *AutoExtractor) ScheduleWithExtractor(sessionID string, messages []providertypes.Message, extractor Extractor) { - a.schedule(sessionID, messages, extractor) -} - -// schedule 统一处理会话级别的自动提取调度。 -func (a *AutoExtractor) schedule(sessionID string, messages []providertypes.Message, extractor Extractor) { if a == nil || extractor == nil || a.svc == nil { return } @@ -134,7 +132,7 @@ func (a *AutoExtractor) armDebounceTimerLocked( }) } -// handleDebounce 在防抖窗口结束后启动一次后台提取。 +// handleDebounce 在防抖窗口结束后启动一次后台提取，若消息未变化则跳过以避免重复 LLM 调用。 func (a *AutoExtractor) handleDebounce(sessionID string, state *autoExtractState, seq uint64) { state.mu.Lock() defer state.mu.Unlock() @@ -149,11 +147,23 @@ func (a *AutoExtractor) handleDebounce(sessionID string, state *autoExtractState req := *state.pending state.pending = nil + + // 增量检测：消息未变化时跳过提取 + fp := computeMessageFingerprint(req.messages) + if fp == state.lastFingerprint { + a.armIdleTimerLocked(sessionID, state) + return + } + state.running = true state.timer = nil go func() { - a.extractAndStore(req.extractor, req.messages) + if a.extractAndStore(req.extractor, req.messages) { + state.mu.Lock() + state.lastFingerprint = fp + state.mu.Unlock() + } a.handleRunDone(sessionID, state) }() } @@ -209,20 +219,22 @@ func isIdleStateLocked(state *autoExtractState, seq uint64) bool { } // extractAndStore 执行提取，并在写入前做本地批次去重和持久化级别的原子去重。 -func (a *AutoExtractor) extractAndStore(extractor Extractor, messages []providertypes.Message) { +// 返回值表示本次提取和写入流程是否成功完成，可用于更新增量指纹。 +func (a *AutoExtractor) extractAndStore(extractor Extractor, messages []providertypes.Message) bool { ctx, cancel := context.WithTimeout(context.Background(), a.extractTimeout) defer cancel() entries, err := extractor.Extract(ctx, messages) if err != nil { a.logError("memo: auto extract failed: %v", err) - return + return false } if len(entries) == 0 { - return + return true } seen := make(map[string]struct{}, len(entries)) + succeeded := true for _, entry := range entries { entry.Source = SourceAutoExtract key := autoExtractDedupKey(entry) @@ -236,6 +248,7 @@ func (a *AutoExtractor) extractAndStore(extractor Extractor, messages []provider added, err := a.svc.addAutoExtractIfAbsent(ctx, entry) if err != nil { a.logError("memo: auto extract add failed: %v", err) + succeeded = false continue } @@ -244,6 +257,7 @@ func (a *AutoExtractor) extractAndStore(extractor Extractor, messages []provider continue } } + return succeeded } // autoExtractDedupKey 生成自动提取条目的精确去重键。 @@ -307,3 +321,26 @@ func (a *AutoExtractor) logError(format string, args ...any) { a.logf(format, args...) } } + +// computeMessageFingerprint 使用 FNV-1a 64bit 哈希计算消息窗口的内容指纹，用于增量提取检测。 +func computeMessageFingerprint(messages []providertypes.Message) uint64 { + h := fnv.New64a() + payload, err := json.Marshal(messages) + if err == nil { + _, _ = h.Write(payload) + return h.Sum64() + } + + // 回退路径仅在意外序列化失败时使用，至少保留文本消息的增量检测能力。 + for _, msg := range messages { + _, _ = h.Write([]byte(msg.Role)) + _, _ = h.Write([]byte{0}) + for _, part := range msg.Parts { + _, _ = h.Write([]byte(part.Kind)) + _, _ = h.Write([]byte{0}) + _, _ = h.Write([]byte(part.Text)) + _, _ = h.Write([]byte{0}) + } + } + return h.Sum64() +} diff --git a/internal/memo/auto_extractor_test.go b/internal/memo/auto_extractor_test.go index 502903f4..d9f83763 100644 --- a/internal/memo/auto_extractor_test.go +++ b/internal/memo/auto_extractor_test.go @@ -263,6 +263,212 @@ func TestAutoExtractorLoadsDedupIndexOutsideCurrentProcessState(t *testing.T) { } } +func TestAutoExtractorFailedRunDoesNotAdvanceFingerprint(t *testing.T) { + svc := newAutoExtractorTestService(t) + var attemptMu sync.Mutex + attempt := 0 + extractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + attemptMu.Lock() + defer attemptMu.Unlock() + attempt++ + if attempt == 1 { + return nil, errors.New("boom") + } + return []Entry{{Type: TypeProject, Title: "retry", Content: "retry", Source: SourceAutoExtract}}, nil + }, + } + auto := NewAutoExtractor(extractor, svc, time.Second) + auto.debounce = 5 * time.Millisecond + auto.logf = func(string, ...any) {} + registerAutoExtractorCleanup(t, auto) + + messages := []providertypes.Message{{Role: providertypes.RoleUser, Parts: []providertypes.ContentPart{providertypes.NewTextPart("same payload")}}} + auto.Schedule("session-1", messages) + waitFor(t, time.Second, func() bool { return extractor.Calls() == 1 }) + + auto.Schedule("session-1", messages) + waitFor(t, time.Second, func() bool { return extractor.Calls() == 2 }) + waitFor(t, time.Second, func() bool { + entries, err := svc.List(context.Background(), ScopeProject) + return err == nil && len(entries) == 1 + }) +} + +func TestAutoExtractorHandleDebounceSkipsDuplicateFingerprint(t *testing.T) { + svc := newAutoExtractorTestService(t) + extractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + t.Fatal("duplicate fingerprint should skip extractor invocation") + return nil, nil + }, + } + auto := NewAutoExtractor(extractor, svc, time.Second) + auto.logf = func(string, ...any) {} + registerAutoExtractorCleanup(t, auto) + + messages := []providertypes.Message{ + {Role: providertypes.RoleUser, Parts: []providertypes.ContentPart{providertypes.NewTextPart("same payload")}}, + } + state := auto.ensureState("session-1") + state.mu.Lock() + state.pending = &autoExtractRequest{ + messages: cloneProviderMessages(messages), + dueAt: time.Now().Add(-time.Millisecond), + extractor: extractor, + } + state.scheduleSeq = 1 + state.lastFingerprint = computeMessageFingerprint(messages) + state.mu.Unlock() + + auto.handleDebounce("session-1", state, 1) + + waitFor(t, time.Second, func() bool { + state.mu.Lock() + defer state.mu.Unlock() + return !state.running && state.pending == nil && state.idleTimer != nil + }) + if extractor.Calls() != 0 { + t.Fatalf("extractor call count = %d, want 0", extractor.Calls()) + } +} + +func TestAutoExtractorScheduleWithExtractorUsesCallScopedExtractorAndSkipsSuccessfulDuplicates(t *testing.T) { + svc := newAutoExtractorTestService(t) + defaultExtractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + t.Fatal("default extractor should not be used for call-scoped scheduling") + return nil, nil + }, + } + firstExtractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + return []Entry{{Type: TypeProject, Title: "deduped", Content: "deduped", Source: SourceAutoExtract}}, nil + }, + } + secondExtractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + t.Fatal("duplicate fingerprint should short-circuit before second extractor runs") + return nil, nil + }, + } + + auto := NewAutoExtractor(defaultExtractor, svc, time.Second) + auto.debounce = 5 * time.Millisecond + auto.logf = func(string, ...any) {} + registerAutoExtractorCleanup(t, auto) + + messages := []providertypes.Message{ + {Role: providertypes.RoleUser, Parts: []providertypes.ContentPart{providertypes.NewTextPart("same payload")}}, + } + fingerprint := computeMessageFingerprint(messages) + + auto.ScheduleWithExtractor("session-1", messages, firstExtractor) + waitFor(t, time.Second, func() bool { return firstExtractor.Calls() == 1 }) + waitFor(t, time.Second, func() bool { + auto.mu.Lock() + state := auto.states["session-1"] + auto.mu.Unlock() + if state == nil { + return false + } + + state.mu.Lock() + defer state.mu.Unlock() + return !state.running && state.pending == nil && state.lastFingerprint == fingerprint + }) + + auto.ScheduleWithExtractor("session-1", messages, secondExtractor) + waitFor(t, time.Second, func() bool { + auto.mu.Lock() + state := auto.states["session-1"] + auto.mu.Unlock() + if state == nil { + return false + } + + state.mu.Lock() + defer state.mu.Unlock() + return !state.running && state.pending == nil && state.lastFingerprint == fingerprint && state.idleTimer != nil + }) + + if defaultExtractor.Calls() != 0 { + t.Fatalf("default extractor call count = %d, want 0", defaultExtractor.Calls()) + } + if secondExtractor.Calls() != 0 { + t.Fatalf("second extractor call count = %d, want 0", secondExtractor.Calls()) + } + + entries, err := svc.List(context.Background(), ScopeProject) + if err != nil { + t.Fatalf("List() error = %v", err) + } + if len(entries) != 1 { + t.Fatalf("len(entries) = %d, want 1", len(entries)) + } +} + +func TestAutoExtractorFingerprintIncludesNonTextParts(t *testing.T) { + svc := newAutoExtractorTestService(t) + firstExtractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + return []Entry{{Type: TypeProject, Title: "first-image", Content: "first-image", Source: SourceAutoExtract}}, nil + }, + } + secondExtractor := &stubMemoExtractor{ + extractFn: func(ctx context.Context, messages []providertypes.Message) ([]Entry, error) { + return []Entry{{Type: TypeProject, Title: "second-image", Content: "second-image", Source: SourceAutoExtract}}, nil + }, + } + + auto := NewAutoExtractor(nil, svc, time.Second) + auto.debounce = 5 * time.Millisecond + auto.logf = func(string, ...any) {} + registerAutoExtractorCleanup(t, auto) + + firstMessages := []providertypes.Message{{ + Role: providertypes.RoleUser, + Parts: []providertypes.ContentPart{ + providertypes.NewTextPart("same text"), + providertypes.NewRemoteImagePart("https://example.com/first.png"), + }, + }} + secondMessages := []providertypes.Message{{ + Role: providertypes.RoleUser, + Parts: []providertypes.ContentPart{ + providertypes.NewTextPart("same text"), + providertypes.NewRemoteImagePart("https://example.com/second.png"), + }, + }} + + auto.ScheduleWithExtractor("session-1", firstMessages, firstExtractor) + waitFor(t, time.Second, func() bool { return firstExtractor.Calls() == 1 }) + + auto.ScheduleWithExtractor("session-1", secondMessages, secondExtractor) + waitFor(t, time.Second, func() bool { return secondExtractor.Calls() == 1 }) + + entries, err := svc.List(context.Background(), ScopeProject) + if err != nil { + t.Fatalf("List() error = %v", err) + } + if len(entries) != 2 { + t.Fatalf("len(entries) = %d, want 2", len(entries)) + } +} + +func TestStopTimerDrainsExpiredTimer(t *testing.T) { + timer := time.NewTimer(5 * time.Millisecond) + time.Sleep(20 * time.Millisecond) + + stopTimer(timer) + + select { + case <-timer.C: + t.Fatal("expected stopTimer to drain expired timer channel") + default: + } +} + func waitFor(t *testing.T, timeout time.Duration, fn func() bool) { t.Helper() deadline := time.Now().Add(timeout) diff --git a/internal/memo/service.go b/internal/memo/service.go index d983efef..c7415c50 100644 --- a/internal/memo/service.go +++ b/internal/memo/service.go @@ -474,19 +474,31 @@ func cloneIndex(index *Index) *Index { Entries: make([]Entry, len(index.Entries)), UpdatedAt: index.UpdatedAt, } - copy(cloned.Entries, index.Entries) + for i, entry := range index.Entries { + cloned.Entries[i] = cloneEntry(entry) + } return cloned } -// trimIndexEntries 先按条目数、再按索引字节数裁剪最旧条目，并返回被删除的记录。 +// cloneEntry 深拷贝单条记忆条目，避免 Keywords 等切片字段共享底层数组。 +func cloneEntry(entry Entry) Entry { + cloned := entry + if len(entry.Keywords) > 0 { + cloned.Keywords = append([]string(nil), entry.Keywords...) + } + return cloned +} + +// trimIndexEntries 按优先级和条目数裁剪索引，优先移除低优先级条目，并返回被删除的记录。 func trimIndexEntries(index *Index, maxEntries int, maxIndexBytes int) []Entry { if index == nil { return nil } removed := make([]Entry, 0) for maxEntries > 0 && len(index.Entries) > maxEntries { - removed = append(removed, index.Entries[0]) - index.Entries = index.Entries[1:] + victimIdx := findEvictionVictim(index.Entries) + removed = append(removed, index.Entries[victimIdx]) + index.Entries = append(index.Entries[:victimIdx], index.Entries[victimIdx+1:]...) } if maxIndexBytes > 0 && len(index.Entries) > 0 { removed = append(removed, trimIndexEntriesByBytes(index, maxIndexBytes)...) @@ -494,7 +506,24 @@ func trimIndexEntries(index *Index, maxEntries int, maxIndexBytes int) []Entry { return removed } -// trimIndexEntriesByBytes 在索引超过字节阈值时，通过二分定位最小移除数量并返回被删除条目。 +// findEvictionVictim 找到优先级最低的条目索引；同优先级时优先移除最旧（靠前）的条目。 +func findEvictionVictim(entries []Entry) int { + if len(entries) == 0 { + return -1 + } + minPriority := entries[0].evictionPriority() + victimIdx := 0 + for i, entry := range entries { + p := entry.evictionPriority() + if p < minPriority { + minPriority = p + victimIdx = i + } + } + return victimIdx +} + +// trimIndexEntriesByBytes 在索引超过字节阈值时，按统一淘汰优先级循环移除条目直到回到字节阈值内。 func trimIndexEntriesByBytes(index *Index, maxIndexBytes int) []Entry { if index == nil || len(index.Entries) == 0 || maxIndexBytes <= 0 { return nil @@ -503,20 +532,15 @@ func trimIndexEntriesByBytes(index *Index, maxIndexBytes int) []Entry { return nil } - entries := index.Entries - lo, hi := 0, len(entries) - for lo < hi { - mid := lo + (hi-lo)/2 - candidate := &Index{Entries: entries[mid:], UpdatedAt: index.UpdatedAt} - if len(RenderIndex(candidate)) > maxIndexBytes { - lo = mid + 1 - continue + removed := make([]Entry, 0) + for len(index.Entries) > 0 && len(RenderIndex(index)) > maxIndexBytes { + victimIdx := findEvictionVictim(index.Entries) + if victimIdx < 0 { + break } - hi = mid + removed = append(removed, index.Entries[victimIdx]) + index.Entries = append(index.Entries[:victimIdx], index.Entries[victimIdx+1:]...) } - - removed := append([]Entry(nil), entries[:lo]...) - index.Entries = entries[lo:] return removed } diff --git a/internal/memo/service_test.go b/internal/memo/service_test.go index 21988dba..988b71ff 100644 --- a/internal/memo/service_test.go +++ b/internal/memo/service_test.go @@ -395,3 +395,68 @@ func TestTrimIndexEntriesByBytesRemovesMinimalPrefix(t *testing.T) { t.Fatalf("remaining entries = %#v", index.Entries) } } + +func TestTrimIndexEntriesPrefersLowestPriorityEntries(t *testing.T) { + index := &Index{ + Entries: []Entry{ + {Type: TypeUser, Title: "user", Source: SourceAutoExtract}, + {Type: TypeReference, Title: "reference", Source: SourceAutoExtract}, + {Type: TypeProject, Title: "project", Source: SourceUserManual}, + }, + } + + removed := trimIndexEntries(index, 2, 0) + if len(removed) != 1 || removed[0].Title != "reference" { + t.Fatalf("removed = %#v, want reference entry", removed) + } + if len(index.Entries) != 2 { + t.Fatalf("len(index.Entries) = %d, want 2", len(index.Entries)) + } +} + +func TestTrimIndexEntriesByBytesPrefersLowestPriorityEntries(t *testing.T) { + index := &Index{ + Entries: []Entry{ + {Type: TypeUser, Title: "keep-user", Content: "u", Source: SourceUserManual}, + {Type: TypeReference, Title: strings.Repeat("ref", 32), Content: strings.Repeat("payload", 64), Source: SourceAutoExtract}, + {Type: TypeProject, Title: "keep-project", Content: "p", Source: SourceAutoExtract}, + }, + } + + target := &Index{ + Entries: []Entry{ + index.Entries[0], + index.Entries[2], + }, + } + maxIndexBytes := len(RenderIndex(target)) + removed := trimIndexEntries(index, 10, maxIndexBytes) + + if len(removed) != 1 || removed[0].Type != TypeReference { + t.Fatalf("removed = %#v, want only low-priority reference entry", removed) + } + if len(index.Entries) != 2 { + t.Fatalf("len(index.Entries) = %d, want 2", len(index.Entries)) + } + if index.Entries[0].Title != "keep-user" || index.Entries[1].Title != "keep-project" { + t.Fatalf("remaining entries = %#v, want keep-user and keep-project", index.Entries) + } + if got := len(RenderIndex(index)); got > maxIndexBytes { + t.Fatalf("rendered index bytes = %d, want <= %d", got, maxIndexBytes) + } +} + +func TestFindEvictionVictimBranches(t *testing.T) { + if got := findEvictionVictim(nil); got != -1 { + t.Fatalf("findEvictionVictim(nil) = %d, want -1", got) + } + + entries := []Entry{ + {Type: TypeReference, Title: "first", Source: SourceAutoExtract}, + {Type: TypeReference, Title: "second", Source: SourceAutoExtract}, + {Type: TypeProject, Title: "third", Source: SourceAutoExtract}, + } + if got := findEvictionVictim(entries); got != 0 { + t.Fatalf("findEvictionVictim() = %d, want 0 for first lowest-priority entry", got) + } +} diff --git a/internal/memo/store.go b/internal/memo/store.go index befbb527..d7ee2036 100644 --- a/internal/memo/store.go +++ b/internal/memo/store.go @@ -8,6 +8,7 @@ import ( "path/filepath" "sort" "sync" + "time" agentsession "neo-code/internal/session" ) @@ -18,11 +19,20 @@ const ( memoFileName = "MEMO.md" ) +// cachedIndex 缓存已解析的索引及其文件修改时间，用于减少高频读取时的磁盘 I/O。 +type cachedIndex struct { + content *Index + modTime time.Time +} + // FileStore 基于文件系统实现 Store 接口，采用工作区隔离的双层目录布局。 type FileStore struct { mu sync.RWMutex baseDir string workspaceRoot string + + cacheMu sync.RWMutex + indexCache map[Scope]*cachedIndex } // NewFileStore 创建 FileStore 实例，目录基于 baseDir 和 workspaceRoot 计算工作区隔离路径。 @@ -30,10 +40,11 @@ func NewFileStore(baseDir string, workspaceRoot string) *FileStore { return &FileStore{ baseDir: baseDir, workspaceRoot: workspaceRoot, + indexCache: make(map[Scope]*cachedIndex), } } -// LoadIndex 加载指定分层下的 MEMO.md 索引文件并解析为 Index 结构。 +// LoadIndex 加载指定分层下的 MEMO.md 索引文件并解析为 Index 结构，优先复用内存缓存。 func (s *FileStore) LoadIndex(ctx context.Context, scope Scope) (*Index, error) { if err := ctx.Err(); err != nil { return nil, err @@ -42,6 +53,19 @@ func (s *FileStore) LoadIndex(ctx context.Context, scope Scope) (*Index, error) return nil, err } + indexPath := filepath.Join(s.scopeDir(scope), memoFileName) + var cachedModTime time.Time + if info, statErr := os.Stat(indexPath); statErr == nil { + cachedModTime = info.ModTime() + s.cacheMu.RLock() + if cached, ok := s.indexCache[scope]; ok && cachedModTime.Equal(cached.modTime) { + result := cloneIndex(cached.content) + s.cacheMu.RUnlock() + return result, nil + } + s.cacheMu.RUnlock() + } + s.mu.RLock() defer s.mu.RUnlock() @@ -52,7 +76,29 @@ func (s *FileStore) LoadIndex(ctx context.Context, scope Scope) (*Index, error) if err != nil { return nil, fmt.Errorf("memo: read index: %w", err) } - return ParseIndex(string(data)) + index, err := ParseIndex(string(data)) + if err != nil { + return nil, err + } + cachedContent := cloneIndex(index) + + if !cachedModTime.IsZero() { + s.cacheMu.Lock() + s.indexCache[scope] = &cachedIndex{ + content: cachedContent, + modTime: cachedModTime, + } + s.cacheMu.Unlock() + } else if info, statErr := os.Stat(indexPath); statErr == nil { + s.cacheMu.Lock() + s.indexCache[scope] = &cachedIndex{ + content: cachedContent, + modTime: info.ModTime(), + } + s.cacheMu.Unlock() + } + + return cloneIndex(cachedContent), nil } // SaveIndex 将索引写入指定分层下的 MEMO.md 文件，采用临时文件 + 原子替换策略。 @@ -92,9 +138,25 @@ func (s *FileStore) SaveIndex(ctx context.Context, scope Scope, index *Index) er if err := os.Rename(temp, target); err != nil { return fmt.Errorf("memo: commit index: %w", err) } + + s.cacheMu.Lock() + s.indexCache[scope] = &cachedIndex{ + content: cloneIndex(index), + modTime: cacheModTimeAfterSave(target), + } + s.cacheMu.Unlock() + return nil } +// cacheModTimeAfterSave 在原子写入后读取文件的真实 mtime，确保缓存与后续 LoadIndex 的 stat 比较一致。 +func cacheModTimeAfterSave(target string) time.Time { + if info, err := os.Stat(target); err == nil { + return info.ModTime() + } + return time.Now() +} + // LoadTopic 读取指定分层下的 topic 文件完整内容。 func (s *FileStore) LoadTopic(ctx context.Context, scope Scope, filename string) (string, error) { if err := ctx.Err(); err != nil { diff --git a/internal/memo/store_test.go b/internal/memo/store_test.go index 1a917dc0..de58dbf5 100644 --- a/internal/memo/store_test.go +++ b/internal/memo/store_test.go @@ -55,6 +55,92 @@ func TestFileStoreSaveAndLoadIndexByScope(t *testing.T) { } } +func TestFileStoreLoadIndexReturnsClonedCacheContent(t *testing.T) { + store := NewFileStore(t.TempDir(), "/workspace/project") + if err := store.SaveIndex(context.Background(), ScopeUser, &Index{ + Entries: []Entry{{Type: TypeUser, Title: "original", Content: "body", Keywords: []string{"go", "tabs"}}}, + }); err != nil { + t.Fatalf("SaveIndex() error = %v", err) + } + + first, err := store.LoadIndex(context.Background(), ScopeUser) + if err != nil { + t.Fatalf("LoadIndex(first) error = %v", err) + } + first.Entries[0].Title = "mutated in memory" + first.Entries[0].Keywords[0] = "mutated-keyword" + + second, err := store.LoadIndex(context.Background(), ScopeUser) + if err != nil { + t.Fatalf("LoadIndex(second) error = %v", err) + } + if got := second.Entries[0].Title; got != "original" { + t.Fatalf("cached title = %q, want %q", got, "original") + } + if got := second.Entries[0].Keywords[0]; got != "go" { + t.Fatalf("cached keyword = %q, want %q", got, "go") + } +} + +func TestFileStoreSaveIndexCachesCloneOfInput(t *testing.T) { + store := NewFileStore(t.TempDir(), "/workspace/project") + index := &Index{ + Entries: []Entry{{Type: TypeUser, Title: "saved", Content: "body", Keywords: []string{"persisted"}}}, + } + if err := store.SaveIndex(context.Background(), ScopeUser, index); err != nil { + t.Fatalf("SaveIndex() error = %v", err) + } + + index.Entries[0].Title = "changed after save" + index.Entries[0].Keywords[0] = "changed-keyword" + + loaded, err := store.LoadIndex(context.Background(), ScopeUser) + if err != nil { + t.Fatalf("LoadIndex() error = %v", err) + } + if got := loaded.Entries[0].Title; got != "saved" { + t.Fatalf("loaded title = %q, want %q", got, "saved") + } + if got := loaded.Entries[0].Keywords[0]; got != "persisted" { + t.Fatalf("loaded keyword = %q, want %q", got, "persisted") + } +} + +func TestFileStoreLoadIndexRefreshesCacheWhenMemoFileMtimeChanges(t *testing.T) { + store := NewFileStore(t.TempDir(), "/workspace/project") + if err := store.SaveIndex(context.Background(), ScopeUser, &Index{ + Entries: []Entry{{Type: TypeUser, Title: "first", Content: "body"}}, + }); err != nil { + t.Fatalf("SaveIndex() error = %v", err) + } + + first, err := store.LoadIndex(context.Background(), ScopeUser) + if err != nil { + t.Fatalf("LoadIndex(first) error = %v", err) + } + if len(first.Entries) != 1 || first.Entries[0].Title != "first" { + t.Fatalf("first load entries = %#v", first.Entries) + } + + indexPath := filepath.Join(store.scopeDir(ScopeUser), memoFileName) + updated := &Index{Entries: []Entry{{Type: TypeUser, Title: "second", Content: "updated"}}} + if err := os.WriteFile(indexPath, []byte(RenderIndex(updated)), 0o644); err != nil { + t.Fatalf("WriteFile(updated index) error = %v", err) + } + newModTime := time.Now().Add(2 * time.Second) + if err := os.Chtimes(indexPath, newModTime, newModTime); err != nil { + t.Fatalf("Chtimes() error = %v", err) + } + + second, err := store.LoadIndex(context.Background(), ScopeUser) + if err != nil { + t.Fatalf("LoadIndex(second) error = %v", err) + } + if len(second.Entries) != 1 || second.Entries[0].Title != "second" { + t.Fatalf("second load entries = %#v, want refreshed disk content", second.Entries) + } +} + func TestFileStoreSaveAndLoadTopicByScope(t *testing.T) { store := NewFileStore(t.TempDir(), "/workspace/project") content := "---\ntype: user\n---\n\nbody\n" diff --git a/internal/memo/types.go b/internal/memo/types.go index 563cde52..8572e0e4 100644 --- a/internal/memo/types.go +++ b/internal/memo/types.go @@ -44,6 +44,26 @@ type Entry struct { UpdatedAt time.Time } +// evictionPriority 返回条目在裁剪时的优先级权重，值越大越不容易被裁剪。 +// user > feedback > project > reference，且手动创建优先于自动提取。 +func (e Entry) evictionPriority() int { + base := 0 + switch e.Type { + case TypeUser: + base = 40 + case TypeFeedback: + base = 30 + case TypeProject: + base = 20 + case TypeReference: + base = 10 + } + if e.Source == SourceUserManual || e.Source == SourceToolInitiated { + base += 50 + } + return base +} + // Scope 表示 memo 的逻辑分层范围。 type Scope string diff --git a/internal/memo/types_test.go b/internal/memo/types_test.go index 5ef69a01..ac2a0c66 100644 --- a/internal/memo/types_test.go +++ b/internal/memo/types_test.go @@ -93,3 +93,52 @@ func TestEntryFields(t *testing.T) { t.Errorf("len(Entry.Keywords) = %d, want 2", len(e.Keywords)) } } + +func TestEntryEvictionPriority(t *testing.T) { + tests := []struct { + name string + entry Entry + want int + }{ + { + name: "reference auto extract", + entry: Entry{Type: TypeReference, Source: SourceAutoExtract}, + want: 10, + }, + { + name: "project auto extract", + entry: Entry{Type: TypeProject, Source: SourceAutoExtract}, + want: 20, + }, + { + name: "feedback auto extract", + entry: Entry{Type: TypeFeedback, Source: SourceAutoExtract}, + want: 30, + }, + { + name: "user auto extract", + entry: Entry{Type: TypeUser, Source: SourceAutoExtract}, + want: 40, + }, + { + name: "manual reference", + entry: Entry{Type: TypeReference, Source: SourceUserManual}, + want: 60, + }, + { + name: "tool initiated feedback", + entry: Entry{Type: TypeFeedback, Source: SourceToolInitiated}, + want: 80, + }, + { + name: "unknown type", + entry: Entry{Type: Type("unknown"), Source: SourceAutoExtract}, + want: 0, + }, + } + for _, tt := range tests { + if got := tt.entry.evictionPriority(); got != tt.want { + t.Fatalf("%s: evictionPriority() = %d, want %d", tt.name, got, tt.want) + } + } +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index d686d039..01d1b2b8 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -134,9 +134,9 @@ type Service struct { permissionAskLocks map[string]*permissionAskLockEntry } -// sessionLockEntry 维护单个会话锁及其当前引用计数，用于在无引用时回收 map 项。 +// sessionLockEntry 维护单个会话读写锁及其当前引用计数，用于在无引用时回收 map 项。 type sessionLockEntry struct { - mu sync.Mutex + mu sync.RWMutex refs int } diff --git a/internal/runtime/runtime_internal_helpers_test.go b/internal/runtime/runtime_internal_helpers_test.go index 8e21a062..4fd36d7b 100644 --- a/internal/runtime/runtime_internal_helpers_test.go +++ b/internal/runtime/runtime_internal_helpers_test.go @@ -57,6 +57,10 @@ func (s *lockProbeStore) ReplaceTranscript(ctx context.Context, input agentsessi return errors.New("not implemented") } +func (s *lockProbeStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + return 0, nil +} + func (s *stubMemoExtractor) Schedule(_ string, messages []providertypes.Message) { s.mu.Lock() s.calls++ diff --git a/internal/runtime/runtime_remaining_branches_test.go b/internal/runtime/runtime_remaining_branches_test.go index 398a4add..6e10b221 100644 --- a/internal/runtime/runtime_remaining_branches_test.go +++ b/internal/runtime/runtime_remaining_branches_test.go @@ -152,6 +152,10 @@ func (s *postSaveHookStore) ReplaceTranscript(ctx context.Context, input agentse return s.afterSave(s.base.ReplaceTranscript(ctx, input)) } +func (s *postSaveHookStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + return 0, nil +} + func TestResolveCompactProviderSelectionResolveErrorBranch(t *testing.T) { t.Parallel() diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index e771ff01..16700e57 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -239,6 +239,10 @@ func (s *memoryStore) ReplaceTranscript(ctx context.Context, input agentsession. return nil } +func (s *memoryStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + return 0, nil +} + // CreateSession 转发到底层 Store，并按旧 save 计数规则注入失败。 func (s *failingStore) CreateSession(ctx context.Context, input agentsession.CreateSessionInput) (agentsession.Session, error) { if err := s.nextSaveError(ctx); err != nil { @@ -498,6 +502,10 @@ func (s *blockingLoadStore) ListSummaries(ctx context.Context) ([]agentsession.S return summaries, nil } +func (s *blockingLoadStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + return 0, nil +} + type scriptedProvider struct { name string streams [][]providertypes.StreamEvent diff --git a/internal/runtime/session_scheduler.go b/internal/runtime/session_scheduler.go index 9cb2b25f..a4928d7d 100644 --- a/internal/runtime/session_scheduler.go +++ b/internal/runtime/session_scheduler.go @@ -107,9 +107,20 @@ func (s *Service) finishRun(token uint64) { } } -// acquireSessionLock 获取指定会话锁并返回释放引用的函数。 +// acquireSessionLock 获取指定会话写锁并返回释放引用的函数。 // 调用方在完成会话级串行操作后，必须调用 release 以允许锁条目回收。 -func (s *Service) acquireSessionLock(sessionID string) (*sync.Mutex, func()) { +func (s *Service) acquireSessionLock(sessionID string) (*sync.RWMutex, func()) { + return s.acquireSessionLockEntry(sessionID) +} + +// acquireSessionRLock 获取指定会话读锁并返回释放引用的函数。 +// 读锁允许同一会话的多个并发读取，适用于 LoadSession 等只读操作。 +func (s *Service) acquireSessionRLock(sessionID string) (*sync.RWMutex, func()) { + return s.acquireSessionLockEntry(sessionID) +} + +// acquireSessionLockEntry 获取指定会话的锁条目并增加引用计数。 +func (s *Service) acquireSessionLockEntry(sessionID string) (*sync.RWMutex, func()) { s.sessionMu.Lock() if s.sessionLocks == nil { s.sessionLocks = make(map[string]*sessionLockEntry) diff --git a/internal/runtime/system_tool.go b/internal/runtime/system_tool.go index fa70de49..d0fb4dae 100644 --- a/internal/runtime/system_tool.go +++ b/internal/runtime/system_tool.go @@ -42,12 +42,12 @@ func (s *Service) ExecuteSystemTool(ctx context.Context, input SystemToolInput) loaded agentsession.Session ) if sessionID != "" { - sessionMu, releaseLockRef := s.acquireSessionLock(sessionID) - sessionMu.Lock() + sessionMu, releaseLockRef := s.acquireSessionRLock(sessionID) + sessionMu.RLock() session, err := s.sessionStore.LoadSession(ctx, sessionID) if err != nil { - sessionMu.Unlock() + sessionMu.RUnlock() releaseLockRef() return tools.ToolResult{}, err } @@ -57,7 +57,7 @@ func (s *Service) ExecuteSystemTool(ctx context.Context, input SystemToolInput) } runStateValue := newRunState(runID, session) state = &runStateValue - sessionMu.Unlock() + sessionMu.RUnlock() releaseLockRef() } diff --git a/internal/runtime/todo_mutator_test.go b/internal/runtime/todo_mutator_test.go index 7c9b9d8c..9377af46 100644 --- a/internal/runtime/todo_mutator_test.go +++ b/internal/runtime/todo_mutator_test.go @@ -5,6 +5,7 @@ import ( "errors" "slices" "testing" + "time" agentsession "neo-code/internal/session" ) @@ -131,6 +132,10 @@ func (s *mutatorStore) ReplaceTranscript(ctx context.Context, input agentsession return nil } +func (s *mutatorStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + return 0, nil +} + func TestRuntimeSessionMutatorMutateAndSave(t *testing.T) { t.Parallel() diff --git a/internal/session/sqlite_store.go b/internal/session/sqlite_store.go index b8253185..0da93f71 100644 --- a/internal/session/sqlite_store.go +++ b/internal/session/sqlite_store.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "log" "os" "path/filepath" "strings" @@ -42,6 +43,8 @@ type sqliteMessageRow struct { ToolMetadataJSON string } +const maxSessionDeleteBatchSize = 900 + // SQLiteStore 使用单个工作区级 SQLite 数据库持久化会话。 type SQLiteStore struct { projectDir string @@ -83,6 +86,52 @@ func (s *SQLiteStore) Close() error { return s.db.Close() } +// CleanupExpiredSessions 删除超过指定时长未更新的会话及其附件，返回删除数量。 +func (s *SQLiteStore) CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + if maxAge <= 0 { + return 0, nil + } + db, err := s.ensureDB(ctx) + if err != nil { + return 0, err + } + + cutoffMS := toUnixMillis(time.Now().Add(-maxAge)) + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("session: begin cleanup tx: %w", err) + } + defer rollbackTx(tx) + + expiredIDs, err := listExpiredSessionIDs(ctx, tx, cutoffMS) + if err != nil { + return 0, err + } + + if len(expiredIDs) == 0 { + return 0, nil + } + + // 在同一事务内按固定 ID 集合删除记录，避免查询结果与实际删除集合不一致。 + affected, err := deleteSessionsByIDSet(ctx, tx, expiredIDs) + if err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("session: commit cleanup tx: %w", err) + } + + // 清理附件目录 + if err := s.cleanupExpiredSessionAssets(ctx, expiredIDs); err != nil { + return int(affected), err + } + + return int(affected), nil +} + // CreateSession 创建并持久化一个新的空会话头。 func (s *SQLiteStore) CreateSession(ctx context.Context, input CreateSessionInput) (Session, error) { if err := ctx.Err(); err != nil { @@ -229,6 +278,7 @@ func (s *SQLiteStore) AppendMessages(ctx context.Context, input AppendMessagesIn if err != nil { return err } + normalizedMessages = trimMessagesToSessionLimit(normalizedMessages) updatedAt := resolveUpdatedAt(input.UpdatedAt) tx, err := db.BeginTx(ctx, nil) @@ -242,6 +292,11 @@ func (s *SQLiteStore) AppendMessages(ctx context.Context, input AppendMessagesIn return err } + // 自动裁剪超限的旧消息，保持会话消息数不超过 MaxSessionMessages。 + if err := trimOverflowMessages(ctx, tx, input.SessionID, len(normalizedMessages)); err != nil { + return fmt.Errorf("session: trim overflow messages %s: %w", input.SessionID, err) + } + for _, message := range normalizedMessages { lastSeq++ if err := insertMessage(ctx, tx, input.SessionID, lastSeq, updatedAt, message); err != nil { @@ -610,14 +665,7 @@ func (s *SQLiteStore) DeleteSession(ctx context.Context, sessionID string) error return fmt.Errorf("session: delete session %s: %w", sessionID, err) } - assetDir := filepath.Join(s.assetsDir, sessionID) - if err := ensurePathWithinBase(s.projectDir, assetDir); err != nil { - return fmt.Errorf("session: resolve assets dir path: %w", err) - } - if err := os.RemoveAll(assetDir); err != nil { - return fmt.Errorf("session: delete assets dir %s: %w", sessionID, err) - } - return nil + return s.removeSessionAssetsDir(sessionID) } // ensureDB 懒加载数据库并执行 schema 初始化。 @@ -635,12 +683,14 @@ func (s *SQLiteStore) ensureDB(ctx context.Context) (*sql.DB, error) { // initialize 打开数据库、设置 PRAGMA 并初始化 schema。 func (s *SQLiteStore) initialize(ctx context.Context) error { - if err := os.MkdirAll(s.projectDir, 0o755); err != nil { + if err := os.MkdirAll(s.projectDir, 0o700); err != nil { return fmt.Errorf("session: create project dir: %w", err) } - if err := os.MkdirAll(s.assetsDir, 0o755); err != nil { + _ = os.Chmod(s.projectDir, 0o700) + if err := os.MkdirAll(s.assetsDir, 0o700); err != nil { return fmt.Errorf("session: create assets dir: %w", err) } + _ = os.Chmod(s.assetsDir, 0o700) db, err := sql.Open("sqlite", s.dbPath) if err != nil { @@ -658,6 +708,9 @@ func (s *SQLiteStore) initialize(ctx context.Context) error { return err } + // 收紧数据库文件权限，仅 owner 可读写 + _ = os.Chmod(s.dbPath, 0o600) + s.db = db return nil } @@ -1050,6 +1103,141 @@ func currentLastSeq(ctx context.Context, tx *sql.Tx, sessionID string) (int, err return lastSeq, nil } +// trimOverflowMessages 在事务内裁剪超限的旧消息，确保追加新消息后会话消息数不超过 MaxSessionMessages。 +func trimOverflowMessages(ctx context.Context, tx *sql.Tx, sessionID string, newCount int) error { + if MaxSessionMessages <= 0 || newCount <= 0 { + return nil + } + var currentCount int + if err := tx.QueryRowContext(ctx, + `SELECT COUNT(*) FROM messages WHERE session_id = ?`, sessionID, + ).Scan(¤tCount); err != nil { + return fmt.Errorf("session: query message rows: %w", err) + } + overflow := (currentCount + newCount) - MaxSessionMessages + if overflow <= 0 { + return nil + } + if _, err := tx.ExecContext(ctx, + `DELETE FROM messages WHERE session_id = ? AND seq IN (SELECT seq FROM messages WHERE session_id = ? ORDER BY seq ASC LIMIT ?)`, + sessionID, sessionID, overflow, + ); err != nil { + return fmt.Errorf("session: delete overflow messages: %w", err) + } + if _, err := tx.ExecContext(ctx, + `UPDATE sessions SET message_count = message_count - ? WHERE id = ?`, + overflow, sessionID, + ); err != nil { + return fmt.Errorf("session: update message_count after trim: %w", err) + } + return nil +} + +// trimMessagesToSessionLimit 保留最新一批消息，使单次追加的消息数不会超过会话上限。 +func trimMessagesToSessionLimit(messages []providertypes.Message) []providertypes.Message { + if MaxSessionMessages <= 0 || len(messages) <= MaxSessionMessages { + return messages + } + start := len(messages) - MaxSessionMessages + return append([]providertypes.Message(nil), messages[start:]...) +} + +// listExpiredSessionIDs 在事务内查询过期会话 ID，供后续按固定集合删除。 +func listExpiredSessionIDs(ctx context.Context, tx *sql.Tx, cutoffMS int64) ([]string, error) { + rows, err := tx.QueryContext(ctx, `SELECT id FROM sessions WHERE updated_at_ms < ?`, cutoffMS) + if err != nil { + return nil, fmt.Errorf("session: query expired sessions: %w", err) + } + defer rows.Close() + + var expiredIDs []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("session: scan expired session id: %w", err) + } + expiredIDs = append(expiredIDs, id) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("session: iterate expired sessions: %w", err) + } + return expiredIDs, nil +} + +// deleteSessionsByIDSet 在事务内按固定 ID 集合删除会话，避免删除范围漂移。 +func deleteSessionsByIDSet(ctx context.Context, tx *sql.Tx, sessionIDs []string) (int, error) { + return deleteSessionsByIDSetWithBatchSize(ctx, tx, sessionIDs, maxSessionDeleteBatchSize) +} + +// deleteSessionsByIDSetWithBatchSize 按批次删除固定 ID 集，避免 SQL 参数数量超过 SQLite 限制。 +func deleteSessionsByIDSetWithBatchSize( + ctx context.Context, + tx *sql.Tx, + sessionIDs []string, + batchSize int, +) (int, error) { + if len(sessionIDs) == 0 { + return 0, nil + } + if batchSize <= 0 { + batchSize = maxSessionDeleteBatchSize + } + + totalAffected := 0 + for start := 0; start < len(sessionIDs); start += batchSize { + end := start + batchSize + if end > len(sessionIDs) { + end = len(sessionIDs) + } + batch := sessionIDs[start:end] + args := make([]any, 0, len(batch)) + placeholders := make([]string, 0, len(batch)) + for _, id := range batch { + args = append(args, id) + placeholders = append(placeholders, "?") + } + query := fmt.Sprintf(`DELETE FROM sessions WHERE id IN (%s)`, strings.Join(placeholders, ", ")) + result, err := tx.ExecContext(ctx, query, args...) + if err != nil { + return 0, fmt.Errorf("session: cleanup expired sessions: %w", err) + } + affected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("session: inspect expired cleanup rows: %w", err) + } + totalAffected += int(affected) + } + return totalAffected, nil +} + +// removeSessionAssetsDir 删除单个会话的附件目录，并校验目标路径始终位于 assets 根目录内。 +func (s *SQLiteStore) removeSessionAssetsDir(sessionID string) error { + if err := validateStorageID("session id", sessionID); err != nil { + return fmt.Errorf("session: %w", err) + } + assetDir := filepath.Join(s.assetsDir, sessionID) + if err := ensurePathWithinBase(s.assetsDir, assetDir); err != nil { + return fmt.Errorf("session: resolve assets dir path: %w", err) + } + if err := os.RemoveAll(assetDir); err != nil { + return fmt.Errorf("session: delete assets dir %s: %w", sessionID, err) + } + return nil +} + +// cleanupExpiredSessionAssets 清理过期会话的附件目录；若上下文已取消则尽快中止，其余路径错误仅记录告警并继续。 +func (s *SQLiteStore) cleanupExpiredSessionAssets(ctx context.Context, sessionIDs []string) error { + for _, id := range sessionIDs { + if err := ctx.Err(); err != nil { + return err + } + if err := s.removeSessionAssetsDir(id); err != nil { + log.Printf("session cleanup warning: skip asset cleanup for %q: %v", id, err) + } + } + return nil +} + // insertMessage 在事务内插入单条消息记录。 func insertMessage( ctx context.Context, diff --git a/internal/session/sqlite_store_additional_test.go b/internal/session/sqlite_store_additional_test.go index 627bf6df..4d967756 100644 --- a/internal/session/sqlite_store_additional_test.go +++ b/internal/session/sqlite_store_additional_test.go @@ -6,6 +6,7 @@ import ( "errors" "os" "path/filepath" + "runtime" "strings" "testing" "time" @@ -234,3 +235,419 @@ func TestResolveUpdatedAtReturnsProvidedValue(t *testing.T) { t.Fatalf("resolveUpdatedAt should keep non-zero value, got %v want %v", got, provided) } } + +func TestSQLiteStoreCleanupExpiredSessionsNoopBranches(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + session, err := store.CreateSession(ctx, CreateSessionInput{ + ID: "cleanup_noop", + Title: "cleanup noop", + CreatedAt: time.Now().UTC().Add(-time.Hour), + UpdatedAt: time.Now().UTC().Add(-time.Hour), + }) + if err != nil { + t.Fatalf("CreateSession() error = %v", err) + } + + removed, err := store.CleanupExpiredSessions(ctx, 0) + if err != nil { + t.Fatalf("CleanupExpiredSessions(0) error = %v", err) + } + if removed != 0 { + t.Fatalf("CleanupExpiredSessions(0) removed = %d, want 0", removed) + } + + removed, err = store.CleanupExpiredSessions(ctx, DefaultSessionMaxAge) + if err != nil { + t.Fatalf("CleanupExpiredSessions(DefaultSessionMaxAge) error = %v", err) + } + if removed != 0 { + t.Fatalf("CleanupExpiredSessions(DefaultSessionMaxAge) removed = %d, want 0", removed) + } + + if _, err := store.LoadSession(ctx, session.ID); err != nil { + t.Fatalf("LoadSession() error = %v", err) + } +} + +func TestSQLiteStoreCleanupExpiredSessionsRemovesExpiredSessionsAndAssets(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + expiredAt := time.Now().UTC().Add(-DefaultSessionMaxAge - time.Hour).Truncate(time.Millisecond) + freshAt := time.Now().UTC().Add(-time.Hour).Truncate(time.Millisecond) + + expired, err := store.CreateSession(ctx, CreateSessionInput{ + ID: "cleanup_expired", + Title: "expired", + CreatedAt: expiredAt, + UpdatedAt: expiredAt, + }) + if err != nil { + t.Fatalf("CreateSession(expired) error = %v", err) + } + fresh, err := store.CreateSession(ctx, CreateSessionInput{ + ID: "cleanup_fresh", + Title: "fresh", + CreatedAt: freshAt, + UpdatedAt: freshAt, + }) + if err != nil { + t.Fatalf("CreateSession(fresh) error = %v", err) + } + + expiredAssetDir := filepath.Join(store.assetsDir, expired.ID) + freshAssetDir := filepath.Join(store.assetsDir, fresh.ID) + for _, dir := range []string{expiredAssetDir, freshAssetDir} { + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("MkdirAll(%q) error = %v", dir, err) + } + if err := os.WriteFile(filepath.Join(dir, "note.txt"), []byte("asset"), 0o644); err != nil { + t.Fatalf("WriteFile(%q) error = %v", dir, err) + } + } + + removed, err := store.CleanupExpiredSessions(ctx, DefaultSessionMaxAge) + if err != nil { + t.Fatalf("CleanupExpiredSessions() error = %v", err) + } + if removed != 1 { + t.Fatalf("CleanupExpiredSessions() removed = %d, want 1", removed) + } + + if _, err := store.LoadSession(ctx, expired.ID); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected expired session to be removed, got %v", err) + } + if _, err := store.LoadSession(ctx, fresh.ID); err != nil { + t.Fatalf("expected fresh session to remain, got %v", err) + } + if _, err := os.Stat(expiredAssetDir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected expired asset dir to be removed, got %v", err) + } + if _, err := os.Stat(freshAssetDir); err != nil { + t.Fatalf("expected fresh asset dir to remain, got %v", err) + } +} + +func TestSQLiteStoreCleanupExpiredSessionsSkipsInvalidAssetPath(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + expiredAt := time.Now().UTC().Add(-DefaultSessionMaxAge - time.Hour).Truncate(time.Millisecond) + session, err := store.CreateSession(ctx, CreateSessionInput{ + ID: "cleanup_escape_seed", + Title: "escape seed", + CreatedAt: expiredAt, + UpdatedAt: expiredAt, + }) + if err != nil { + t.Fatalf("CreateSession() error = %v", err) + } + + db, err := store.ensureDB(ctx) + if err != nil { + t.Fatalf("ensureDB() error = %v", err) + } + maliciousID := filepath.Join("..", "escaped-cleanup-target") + if _, err := db.ExecContext(ctx, `UPDATE sessions SET id = ? WHERE id = ?`, maliciousID, session.ID); err != nil { + t.Fatalf("UPDATE session id error = %v", err) + } + + victimDir := filepath.Join(store.assetsDir, maliciousID) + if err := os.MkdirAll(victimDir, 0o755); err != nil { + t.Fatalf("MkdirAll(victim) error = %v", err) + } + victimFile := filepath.Join(victimDir, "note.txt") + if err := os.WriteFile(victimFile, []byte("keep me"), 0o644); err != nil { + t.Fatalf("WriteFile(victim) error = %v", err) + } + + removed, err := store.CleanupExpiredSessions(ctx, DefaultSessionMaxAge) + if err != nil { + t.Fatalf("CleanupExpiredSessions() error = %v", err) + } + if removed != 1 { + t.Fatalf("CleanupExpiredSessions() removed = %d, want 1", removed) + } + if _, err := os.Stat(victimFile); err != nil { + t.Fatalf("expected invalid-path victim file to remain, got %v", err) + } +} + +func TestSQLiteStoreInitializeTightensExistingDirectoryPermissions(t *testing.T) { + t.Parallel() + if runtime.GOOS == "windows" { + t.Skip("directory mode assertions are not reliable on Windows") + } + + baseDir := t.TempDir() + workspaceRoot := t.TempDir() + store := NewStore(baseDir, workspaceRoot) + t.Cleanup(func() { _ = store.Close() }) + + for _, dir := range []string{store.projectDir, store.assetsDir} { + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatalf("MkdirAll(%q) error = %v", dir, err) + } + if err := os.Chmod(dir, 0o755); err != nil { + t.Fatalf("Chmod(%q, 0755) error = %v", dir, err) + } + } + + if _, err := store.ensureDB(context.Background()); err != nil { + t.Fatalf("ensureDB() error = %v", err) + } + + for _, dir := range []string{store.projectDir, store.assetsDir} { + info, err := os.Stat(dir) + if err != nil { + t.Fatalf("Stat(%q) error = %v", dir, err) + } + if got := info.Mode().Perm(); got != 0o700 { + t.Fatalf("%s mode = %o, want 700", dir, got) + } + } +} + +func TestSQLiteStoreAppendMessagesCapsBatchAndSessionCount(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + session, err := store.CreateSession(ctx, CreateSessionInput{ID: "append_cap", Title: "append cap"}) + if err != nil { + t.Fatalf("CreateSession() error = %v", err) + } + + if err := store.AppendMessages(ctx, AppendMessagesInput{ + SessionID: session.ID, + Messages: []providertypes.Message{ + {Role: providertypes.RoleUser, Parts: []providertypes.ContentPart{providertypes.NewTextPart("seed-0")}}, + {Role: providertypes.RoleAssistant, Parts: []providertypes.ContentPart{providertypes.NewTextPart("seed-1")}}, + }, + }); err != nil { + t.Fatalf("AppendMessages(seed) error = %v", err) + } + + batch := make([]providertypes.Message, 0, MaxSessionMessages+2) + for i := 0; i < MaxSessionMessages+2; i++ { + batch = append(batch, providertypes.Message{ + Role: providertypes.RoleUser, + Parts: []providertypes.ContentPart{providertypes.NewTextPart(buildIndexedSuffix(i))}, + }) + } + if err := store.AppendMessages(ctx, AppendMessagesInput{ + SessionID: session.ID, + Messages: batch, + }); err != nil { + t.Fatalf("AppendMessages(large batch) error = %v", err) + } + + loaded, err := store.LoadSession(ctx, session.ID) + if err != nil { + t.Fatalf("LoadSession() error = %v", err) + } + if len(loaded.Messages) != MaxSessionMessages { + t.Fatalf("len(Messages) = %d, want %d", len(loaded.Messages), MaxSessionMessages) + } + if got := renderSessionMessageParts(loaded.Messages[0]); got != buildIndexedSuffix(2) { + t.Fatalf("first kept message = %q, want %q", got, buildIndexedSuffix(2)) + } + if got := renderSessionMessageParts(loaded.Messages[len(loaded.Messages)-1]); got != buildIndexedSuffix(MaxSessionMessages+1) { + t.Fatalf("last kept message = %q, want %q", got, buildIndexedSuffix(MaxSessionMessages+1)) + } + + db, err := store.ensureDB(ctx) + if err != nil { + t.Fatalf("ensureDB() error = %v", err) + } + var headerCount int + if err := db.QueryRowContext(ctx, `SELECT message_count FROM sessions WHERE id = ?`, session.ID).Scan(&headerCount); err != nil { + t.Fatalf("query message_count error = %v", err) + } + if headerCount != MaxSessionMessages { + t.Fatalf("message_count = %d, want %d", headerCount, MaxSessionMessages) + } + + var rowCount int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM messages WHERE session_id = ?`, session.ID).Scan(&rowCount); err != nil { + t.Fatalf("query row count error = %v", err) + } + if rowCount != MaxSessionMessages { + t.Fatalf("message rows = %d, want %d", rowCount, MaxSessionMessages) + } +} + +func TestDeleteSessionsByIDSetWithBatchSizeDeletesAllBatches(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + expiredAt := time.Now().UTC().Add(-DefaultSessionMaxAge - time.Hour).Truncate(time.Millisecond) + sessionIDs := []string{"cleanup_batch_01", "cleanup_batch_02", "cleanup_batch_03", "cleanup_batch_04", "cleanup_batch_05"} + for _, id := range sessionIDs { + if _, err := store.CreateSession(ctx, CreateSessionInput{ + ID: id, + Title: id, + CreatedAt: expiredAt, + UpdatedAt: expiredAt, + }); err != nil { + t.Fatalf("CreateSession(%q) error = %v", id, err) + } + } + + db, err := store.ensureDB(ctx) + if err != nil { + t.Fatalf("ensureDB() error = %v", err) + } + tx, err := db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx() error = %v", err) + } + defer rollbackTx(tx) + + affected, err := deleteSessionsByIDSetWithBatchSize(ctx, tx, sessionIDs, 2) + if err != nil { + t.Fatalf("deleteSessionsByIDSetWithBatchSize() error = %v", err) + } + if affected != len(sessionIDs) { + t.Fatalf("affected = %d, want %d", affected, len(sessionIDs)) + } + if err := tx.Commit(); err != nil { + t.Fatalf("Commit() error = %v", err) + } + + for _, id := range sessionIDs { + if _, err := store.LoadSession(ctx, id); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected %q to be deleted, got %v", id, err) + } + } +} + +func TestDeleteSessionsByIDSetHelpersCoverEmptyAndDefaultBatchSize(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + expiredAt := time.Now().UTC().Add(-DefaultSessionMaxAge - time.Hour).Truncate(time.Millisecond) + session, err := store.CreateSession(ctx, CreateSessionInput{ + ID: "cleanup_batch_default", + Title: "cleanup batch default", + CreatedAt: expiredAt, + UpdatedAt: expiredAt, + }) + if err != nil { + t.Fatalf("CreateSession() error = %v", err) + } + + db, err := store.ensureDB(ctx) + if err != nil { + t.Fatalf("ensureDB() error = %v", err) + } + tx, err := db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("BeginTx() error = %v", err) + } + defer rollbackTx(tx) + + affected, err := deleteSessionsByIDSetWithBatchSize(ctx, tx, nil, 2) + if err != nil { + t.Fatalf("deleteSessionsByIDSetWithBatchSize(nil) error = %v", err) + } + if affected != 0 { + t.Fatalf("deleteSessionsByIDSetWithBatchSize(nil) affected = %d, want 0", affected) + } + + affected, err = deleteSessionsByIDSet(ctx, tx, []string{session.ID}) + if err != nil { + t.Fatalf("deleteSessionsByIDSet() error = %v", err) + } + if affected != 1 { + t.Fatalf("deleteSessionsByIDSet() affected = %d, want 1", affected) + } +} + +func TestTrimMessagesToSessionLimitBranches(t *testing.T) { + t.Parallel() + + within := []providertypes.Message{ + {Role: providertypes.RoleUser, Parts: []providertypes.ContentPart{providertypes.NewTextPart("a")}}, + {Role: providertypes.RoleAssistant, Parts: []providertypes.ContentPart{providertypes.NewTextPart("b")}}, + } + if got := trimMessagesToSessionLimit(within); len(got) != len(within) { + t.Fatalf("trimMessagesToSessionLimit(within) len = %d, want %d", len(got), len(within)) + } + + overflow := make([]providertypes.Message, 0, MaxSessionMessages+1) + for i := 0; i < MaxSessionMessages+1; i++ { + overflow = append(overflow, providertypes.Message{ + Role: providertypes.RoleUser, + Parts: []providertypes.ContentPart{providertypes.NewTextPart(buildIndexedSuffix(i))}, + }) + } + got := trimMessagesToSessionLimit(overflow) + if len(got) != MaxSessionMessages { + t.Fatalf("trimMessagesToSessionLimit(overflow) len = %d, want %d", len(got), MaxSessionMessages) + } + if renderSessionMessageParts(got[0]) != buildIndexedSuffix(1) { + t.Fatalf("first kept trimmed message = %q, want %q", renderSessionMessageParts(got[0]), buildIndexedSuffix(1)) + } +} + +func TestRemoveSessionAssetsDirBranches(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newTestStore(t) + if _, err := store.CreateSession(ctx, CreateSessionInput{ID: "remove_assets_ok", Title: "remove assets"}); err != nil { + t.Fatalf("CreateSession() error = %v", err) + } + + assetDir := filepath.Join(store.assetsDir, "remove_assets_ok") + if err := os.MkdirAll(assetDir, 0o755); err != nil { + t.Fatalf("MkdirAll(assetDir) error = %v", err) + } + if err := store.removeSessionAssetsDir("remove_assets_ok"); err != nil { + t.Fatalf("removeSessionAssetsDir(valid) error = %v", err) + } + if _, err := os.Stat(assetDir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected asset dir removed, got %v", err) + } + + if err := store.removeSessionAssetsDir("../bad-id"); err == nil { + t.Fatalf("expected invalid session id error") + } +} + +func TestCleanupExpiredSessionAssetsStopsOnCanceledContext(t *testing.T) { + t.Parallel() + + store := newTestStore(t) + sessionIDs := []string{"cleanup_ctx_01", "cleanup_ctx_02"} + for _, id := range sessionIDs { + assetDir := filepath.Join(store.assetsDir, id) + if err := os.MkdirAll(assetDir, 0o755); err != nil { + t.Fatalf("MkdirAll(%q) error = %v", assetDir, err) + } + if err := os.WriteFile(filepath.Join(assetDir, "note.txt"), []byte("keep"), 0o644); err != nil { + t.Fatalf("WriteFile(%q) error = %v", assetDir, err) + } + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := store.cleanupExpiredSessionAssets(ctx, sessionIDs) + if !errors.Is(err, context.Canceled) { + t.Fatalf("cleanupExpiredSessionAssets() error = %v, want context.Canceled", err) + } + for _, id := range sessionIDs { + if _, err := os.Stat(filepath.Join(store.assetsDir, id)); err != nil { + t.Fatalf("expected asset dir %q to remain after canceled cleanup, got %v", id, err) + } + } +} diff --git a/internal/session/store.go b/internal/session/store.go index 6a2228c3..6fcd7a25 100644 --- a/internal/session/store.go +++ b/internal/session/store.go @@ -14,6 +14,12 @@ const ( sessionDatabaseFileName = "session.db" assetsDirName = "assets" sqliteSchemaVersion = 1 + + // MaxSessionMessages 定义单个会话允许持久化的最大消息数，超出时自动裁剪最旧消息。 + MaxSessionMessages = 8192 + + // DefaultSessionMaxAge 定义默认的会话过期时间，30 天未更新的会话将被自动清理。 + DefaultSessionMaxAge = 30 * 24 * time.Hour ) var storageIDPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_-]{0,127}$`) @@ -118,6 +124,8 @@ type Store interface { UpdateSessionWorkdir(ctx context.Context, input UpdateSessionWorkdirInput) error UpdateSessionState(ctx context.Context, input UpdateSessionStateInput) error ReplaceTranscript(ctx context.Context, input ReplaceTranscriptInput) error + // CleanupExpiredSessions 删除超过指定时长未更新的会话及其关联数据，返回删除数量。 + CleanupExpiredSessions(ctx context.Context, maxAge time.Duration) (int, error) } // NewSQLiteStore 创建基于 SQLite 的会话存储实现。