fix(agent): configure and instrument compaction timeout (#1259)

Co-authored-by: nguyenha935 <208228297+nguyenha935@users.noreply.github.com>
This commit is contained in:
nguyenha935andnguyenha935 authored and GitHub committed 2026-06-23 01:23:43 +07:00
1 parent 42a20c0f22
commit 9ec542251d
14 files changed
+320 -23

No files matched your search

+19 -3
View File
@@ -37,6 +37,15 @@ Conversation to summarize:
`
const defaultCompactionTimeout = 120 * time.Second
func (l *Loop) compactionTimeout() time.Duration {
if l.compactionCfg != nil && l.compactionCfg.TimeoutSeconds > 0 {
return time.Duration(l.compactionCfg.TimeoutSeconds) * time.Second
}
return defaultCompactionTimeout
}
// compactMessagesInPlace summarizes the first ~70% of messages into a condensed
// summary, keeping the last ~30% intact. Operates purely on the local messages
// slice — no session state touched, no locks needed.
@@ -84,11 +93,18 @@ func (l *Loop) compactMessagesInPlace(ctx context.Context, messages []providers.
}
}
sctx, cancel := context.WithTimeout(ctx, 30*time.Second)
timeout := l.compactionTimeout()
sctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
inTokens := l.estimateSummaryInputTokens(toSummarize)
slog.Info("compact_budget", "agent", l.id, "in_tokens", inTokens, "out_tokens", dynamicSummaryMax(inTokens))
slog.Info("compact_budget",
"path", "mid-loop",
"agent", l.id,
"in_tokens", inTokens,
"out_tokens", dynamicSummaryMax(inTokens),
"timeout_seconds", int(timeout/time.Second),
)
chatReq := providers.ChatRequest{
Messages: []providers.Message{{
Role: "user",
@@ -99,7 +115,7 @@ func (l *Loop) compactMessagesInPlace(ctx context.Context, messages []providers.
}
resp, err := l.callInternalLLMWithUsage(sctx, chatReq, "mid-loop-compaction")
if err != nil {
slog.Warn("mid_loop_compaction_failed", "agent", l.id, "error", err)
slog.Warn("mid_loop_compaction_failed", "agent", l.id, "timeout_seconds", int(timeout/time.Second), "error", err)
return nil
}
+108
View File
@@ -0,0 +1,108 @@
package agent
import (
"context"
"testing"
"time"
"github.com/nextlevelbuilder/goclaw/internal/config"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
type deadlineCapturingProvider struct {
capturingProvider
deadline time.Time
hasDeadline bool
}
func (d *deadlineCapturingProvider) Chat(ctx context.Context, req providers.ChatRequest) (*providers.ChatResponse, error) {
d.deadline, d.hasDeadline = ctx.Deadline()
return d.capturingProvider.Chat(ctx, req)
}
func TestCompactMessagesInPlace_UsesDefaultTimeout(t *testing.T) {
provider := &deadlineCapturingProvider{
capturingProvider: capturingProvider{response: "Summary of conversation."},
}
loop := &Loop{
provider: provider,
model: "claude-3-5-sonnet",
}
start := time.Now()
result := loop.compactMessagesInPlace(context.Background(), compactableMessages())
if result == nil {
t.Fatal("compactMessagesInPlace returned nil; expected compaction to succeed")
}
assertDeadlineWithin(t, provider, start, 120*time.Second)
}
func TestCompactMessagesInPlace_UsesConfiguredTimeout(t *testing.T) {
provider := &deadlineCapturingProvider{
capturingProvider: capturingProvider{response: "Summary of conversation."},
}
loop := &Loop{
provider: provider,
model: "claude-3-5-sonnet",
compactionCfg: &config.CompactionConfig{
TimeoutSeconds: 45,
},
}
start := time.Now()
result := loop.compactMessagesInPlace(context.Background(), compactableMessages())
if result == nil {
t.Fatal("compactMessagesInPlace returned nil; expected compaction to succeed")
}
assertDeadlineWithin(t, provider, start, 45*time.Second)
}
func TestCompactMessagesInPlace_NonPositiveTimeoutFallsBackToDefault(t *testing.T) {
provider := &deadlineCapturingProvider{
capturingProvider: capturingProvider{response: "Summary of conversation."},
}
loop := &Loop{
provider: provider,
model: "claude-3-5-sonnet",
compactionCfg: &config.CompactionConfig{
TimeoutSeconds: -1,
},
}
start := time.Now()
result := loop.compactMessagesInPlace(context.Background(), compactableMessages())
if result == nil {
t.Fatal("compactMessagesInPlace returned nil; expected compaction to succeed")
}
assertDeadlineWithin(t, provider, start, 120*time.Second)
}
func compactableMessages() []providers.Message {
msgs := make([]providers.Message, 10)
for i := range msgs {
if i%2 == 0 {
msgs[i] = providers.Message{Role: "user", Content: "user message"}
} else {
msgs[i] = providers.Message{Role: "assistant", Content: "assistant reply"}
}
}
return msgs
}
func assertDeadlineWithin(t *testing.T, provider *deadlineCapturingProvider, start time.Time, want time.Duration) {
t.Helper()
if !provider.hasDeadline {
t.Fatal("provider context has no deadline")
}
got := provider.deadline.Sub(start)
lower := want - time.Second
upper := want + time.Second
if got < lower || got > upper {
t.Fatalf("deadline duration = %s, want within [%s, %s]", got, lower, upper)
}
}
+43 -3
View File
@@ -189,7 +189,8 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
// lastPromptTokens includes everything (system prompt, tools, context files, history).
// We subtract estimated overhead so the threshold comparison is history-only.
lastPT, lastMC := l.sessions.GetLastPromptTokens(ctx, sessionKey)
adjustedLastPT := max(lastPT-l.estimateOverhead(history, lastPT, lastMC), 0)
overheadEstimate := l.estimateOverhead(history, lastPT, lastMC)
adjustedLastPT := max(lastPT-overheadEstimate, 0)
tokenEstimate := EstimateTokensWithCalibration(history, adjustedLastPT, lastMC)
// Resolve compaction threshold from config: token-only (no message count guard).
@@ -201,6 +202,7 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
threshold := int(float64(l.contextWindow) * historyShare)
if tokenEstimate <= threshold {
l.logCompactionDecision(sessionKey, "skip", "under_threshold", tokenEstimate, threshold, historyShare, lastPT, lastMC, adjustedLastPT, overheadEstimate)
return
}
@@ -210,10 +212,13 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
muI, _ := l.summarizeMu.LoadOrStore(sessionKey, &sync.Mutex{})
sessionMu := muI.(*sync.Mutex)
if !sessionMu.TryLock() {
l.logCompactionDecision(sessionKey, "skip", "already_in_progress", tokenEstimate, threshold, historyShare, lastPT, lastMC, adjustedLastPT, overheadEstimate)
slog.Debug("summarization already in progress, skipping", "session", sessionKey)
return
}
l.logCompactionDecision(sessionKey, "trigger", "", tokenEstimate, threshold, historyShare, lastPT, lastMC, adjustedLastPT, overheadEstimate)
// Memory flush runs synchronously INSIDE the guard
// (so concurrent runs don't both trigger flush for the same compaction cycle).
flushSettings := ResolveMemoryFlushSettings(l.compactionCfg)
@@ -234,7 +239,8 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
// Re-check: history may have been truncated by a concurrent summarize
// that finished between our threshold check and acquiring the lock.
sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 120*time.Second)
timeout := l.compactionTimeout()
sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout)
defer cancel()
history := l.sessions.GetHistory(sctx, sessionKey)
@@ -283,7 +289,19 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
prompt.WriteString(sb.String())
inTokens := l.estimateSummaryInputTokens(toSummarize)
slog.Info("compact_budget", "agent", l.id, "in_tokens", inTokens, "out_tokens", dynamicSummaryMax(inTokens))
slog.Info("compact_budget",
"path", "post-turn",
"agent", l.id,
"session", sessionKey,
"in_tokens", inTokens,
"out_tokens", dynamicSummaryMax(inTokens),
"context_window", l.contextWindow,
"threshold", threshold,
"token_estimate", tokenEstimate,
"max_history_share", historyShare,
"reserve_tokens_floor", l.resolveReserveTokens(),
"timeout_seconds", int(timeout/time.Second),
)
chatReq := providers.ChatRequest{
Messages: []providers.Message{{Role: "user", Content: prompt.String()}},
Model: l.model,
@@ -332,6 +350,28 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) {
}()
}
func (l *Loop) logCompactionDecision(sessionKey, decision, skipReason string, tokenEstimate, threshold int, historyShare float64, lastPromptTokens, lastMessageCount, adjustedLastPromptTokens, overheadEstimate int) {
args := []any{
"path", "post-turn",
"agent", l.id,
"session", sessionKey,
"decision", decision,
"context_window", l.contextWindow,
"threshold", threshold,
"token_estimate", tokenEstimate,
"max_history_share", historyShare,
"reserve_tokens_floor", l.resolveReserveTokens(),
"last_prompt_tokens", lastPromptTokens,
"last_message_count", lastMessageCount,
"adjusted_last_prompt_tokens", adjustedLastPromptTokens,
"overhead_estimate", overheadEstimate,
}
if skipReason != "" {
args = append(args, "skip_reason", skipReason)
}
slog.Info("compaction_decision", args...)
}
// estimateOverhead derives the non-history token overhead (system prompt + tool definitions +
// context files) from calibration data. Used by maybeSummarize to compare history-only tokens
// against the compaction threshold.
@@ -1,7 +1,10 @@
package agent
import (
"bytes"
"context"
"log/slog"
"strings"
"testing"
"time"
@@ -24,29 +27,29 @@ type nopSessionStore struct {
func (n *nopSessionStore) GetOrCreate(_ context.Context, _ string) *store.SessionData {
return &store.SessionData{}
}
func (n *nopSessionStore) Get(_ context.Context, _ string) *store.SessionData { return nil }
func (n *nopSessionStore) Get(_ context.Context, _ string) *store.SessionData { return nil }
func (n *nopSessionStore) AddMessage(_ context.Context, _ string, _ providers.Message) {}
func (n *nopSessionStore) GetHistory(_ context.Context, _ string) []providers.Message {
return n.history
}
func (n *nopSessionStore) GetSummary(_ context.Context, _ string) string { return "" }
func (n *nopSessionStore) SetSummary(_ context.Context, _, _ string) {}
func (n *nopSessionStore) GetLabel(_ context.Context, _ string) string { return "" }
func (n *nopSessionStore) SetLabel(_ context.Context, _, _ string) {}
func (n *nopSessionStore) GetSummary(_ context.Context, _ string) string { return "" }
func (n *nopSessionStore) SetSummary(_ context.Context, _, _ string) {}
func (n *nopSessionStore) GetLabel(_ context.Context, _ string) string { return "" }
func (n *nopSessionStore) SetLabel(_ context.Context, _, _ string) {}
func (n *nopSessionStore) SetAgentInfo(_ context.Context, _ string, _ uuid.UUID, _ string) {}
func (n *nopSessionStore) TruncateHistory(_ context.Context, _ string, _ int) {}
func (n *nopSessionStore) SetHistory(_ context.Context, _ string, _ []providers.Message) {}
func (n *nopSessionStore) Reset(_ context.Context, _ string) {}
func (n *nopSessionStore) Delete(_ context.Context, _ string) error { return nil }
func (n *nopSessionStore) Save(_ context.Context, _ string) error { return nil }
func (n *nopSessionStore) TruncateHistory(_ context.Context, _ string, _ int) {}
func (n *nopSessionStore) SetHistory(_ context.Context, _ string, _ []providers.Message) {}
func (n *nopSessionStore) Reset(_ context.Context, _ string) {}
func (n *nopSessionStore) Delete(_ context.Context, _ string) error { return nil }
func (n *nopSessionStore) Save(_ context.Context, _ string) error { return nil }
// SessionMetadataStore methods
func (n *nopSessionStore) UpdateMetadata(_ context.Context, _, _, _, _ string) {}
func (n *nopSessionStore) AccumulateTokens(_ context.Context, _ string, _, _ int64) {}
func (n *nopSessionStore) IncrementCompaction(_ context.Context, _ string) {}
func (n *nopSessionStore) GetCompactionCount(_ context.Context, _ string) int { return 0 }
func (n *nopSessionStore) UpdateMetadata(_ context.Context, _, _, _, _ string) {}
func (n *nopSessionStore) AccumulateTokens(_ context.Context, _ string, _, _ int64) {}
func (n *nopSessionStore) IncrementCompaction(_ context.Context, _ string) {}
func (n *nopSessionStore) GetCompactionCount(_ context.Context, _ string) int { return 0 }
func (n *nopSessionStore) GetMemoryFlushCompactionCount(_ context.Context, _ string) int { return 0 }
func (n *nopSessionStore) SetMemoryFlushDone(_ context.Context, _ string) {}
func (n *nopSessionStore) SetMemoryFlushDone(_ context.Context, _ string) {}
func (n *nopSessionStore) GetSessionMetadata(_ context.Context, _ string) map[string]string {
return nil
}
@@ -123,7 +126,7 @@ func TestMaybeSummarize_MaxTokensDynamic(t *testing.T) {
contextWindow: contextWindow,
sessions: sessions,
// hasMemory = false → shouldRunMemoryFlush returns false (skip memory flush)
hasMemory: false,
hasMemory: false,
// compactionCfg nil → uses DefaultHistoryShare (0.85), keepLast=4
compactionCfg: nil,
// tokenCounter nil → estimateSummaryInputTokens uses rune/3 fallback
@@ -164,6 +167,91 @@ func TestMaybeSummarize_MaxTokensDynamic(t *testing.T) {
}
}
func TestMaybeSummarize_LogsSkipDecisionUnderThreshold(t *testing.T) {
sessions := &nopSessionStore{
history: []providers.Message{
{Role: "user", Content: "short request"},
{Role: "assistant", Content: "short response"},
},
lastPromptTokens: 0,
lastMsgCount: 0,
}
loop := &Loop{
id: "test-agent",
provider: &capturingProvider{response: "unused"},
model: "claude-3-5-sonnet",
contextWindow: 10000,
sessions: sessions,
hasMemory: false,
}
logs := captureSlog(t, func() {
loop.maybeSummarize(context.Background(), "test-session-key")
})
assertLogContains(t, logs,
"compaction_decision",
`"path":"post-turn"`,
`"decision":"skip"`,
`"skip_reason":"under_threshold"`,
`"agent":"test-agent"`,
`"session":"test-session-key"`,
`"context_window":10000`,
)
}
func TestMaybeSummarize_LogsTriggerDecisionOverThreshold(t *testing.T) {
const contextWindow = 10000
longContent := makeLongString(9000)
history := make([]providers.Message, 10)
for i := range history {
if i%2 == 0 {
history[i] = providers.Message{Role: "user", Content: longContent}
} else {
history[i] = providers.Message{Role: "assistant", Content: longContent}
}
}
done := make(chan struct{}, 1)
provider := &signallingProvider{
capturingProvider: capturingProvider{response: "compaction summary"},
done: done,
}
sessions := &nopSessionStore{
history: history,
lastPromptTokens: 0,
lastMsgCount: 0,
}
loop := &Loop{
id: "test-agent",
provider: provider,
model: "claude-3-5-sonnet",
contextWindow: contextWindow,
sessions: sessions,
hasMemory: false,
}
logs := captureSlog(t, func() {
loop.maybeSummarize(context.Background(), "test-session-key")
})
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("timed out waiting for maybeSummarize to call provider.Chat")
}
assertLogContains(t, logs,
"compaction_decision",
`"path":"post-turn"`,
`"decision":"trigger"`,
`"agent":"test-agent"`,
`"session":"test-session-key"`,
`"context_window":10000`,
)
}
// makeLongString returns a string of n ASCII characters ('a').
func makeLongString(n int) string {
b := make([]byte, n)
@@ -172,3 +260,25 @@ func makeLongString(n int) string {
}
return string(b)
}
func captureSlog(t *testing.T, fn func()) string {
t.Helper()
var buf bytes.Buffer
previous := slog.Default()
slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
defer slog.SetDefault(previous)
fn()
return buf.String()
}
func assertLogContains(t *testing.T, logs string, fragments ...string) {
t.Helper()
for _, fragment := range fragments {
if !strings.Contains(logs, fragment) {
t.Fatalf("logs do not contain %q\nlogs:\n%s", fragment, logs)
}
}
}
+1
View File
@@ -265,6 +265,7 @@ type CompactionConfig struct {
ReserveTokensFloor int `json:"reserveTokensFloor,omitempty"` // min reserve tokens (default 20000)
MaxHistoryShare float64 `json:"maxHistoryShare,omitempty"` // max share of context for history (default 0.85)
KeepLastMessages int `json:"keepLastMessages,omitempty"` // messages to keep after compaction (default 4)
TimeoutSeconds int `json:"timeoutSeconds,omitempty"` // summarization timeout in seconds (default 120)
MemoryFlush *MemoryFlushConfig `json:"memoryFlush,omitempty"` // pre-compaction flush
}
+2
View File
@@ -812,6 +812,8 @@
"maxHistoryShareTip": "Maximum fraction of context window for conversation history (e.g. 0.85 = 85%). Compaction triggers when exceeded.",
"keepLastMessages": "Keep Last Messages",
"keepLastMessagesTip": "Recent messages kept after compaction. Older messages are replaced by a summary.",
"timeoutSeconds": "Timeout Seconds",
"timeoutSecondsTip": "Maximum time to wait for a compaction summary before keeping the original history.",
"memoryFlush": "Memory Flush",
"memoryFlushTip": "Before compaction, the agent gets a turn to save important context to memory files. Also triggers Knowledge Graph extraction."
},
+2
View File
@@ -98,6 +98,8 @@
"agents.compaction.reserveTokensFloorTip": "Minimum token buffer to reserve for new responses before compacting.",
"agents.compaction.maxHistoryShare": "Max History Share (0-1)",
"agents.compaction.maxHistoryShareTip": "Fraction of the context window that conversation history can occupy.",
"agents.compaction.timeoutSeconds": "Timeout Seconds",
"agents.compaction.timeoutSecondsTip": "Maximum time to wait for compaction before keeping the original history.",
"agents.pruning.title": "Context Pruning",
"agents.pruning.desc": "Trim old tool results to free context space",
+2
View File
@@ -797,6 +797,8 @@
"maxHistoryShareTip": "Tỷ lệ tối đa của cửa sổ ngữ cảnh dành cho lịch sử hội thoại (vd: 0.85 = 85%). Nén kích hoạt khi vượt quá.",
"keepLastMessages": "Giữ tin nhắn cuối",
"keepLastMessagesTip": "Số tin nhắn gần nhất giữ lại sau khi nén. Các tin nhắn cũ hơn được thay thế bằng bản tóm tắt.",
"timeoutSeconds": "Thời gian chờ (giây)",
"timeoutSecondsTip": "Thời gian tối đa chờ bản tóm tắt nén ngữ cảnh trước khi giữ nguyên lịch sử cũ.",
"memoryFlush": "Ghi nhớ trước nén",
"memoryFlushTip": "Trước khi nén, agent được một lượt để lưu ngữ cảnh quan trọng vào file bộ nhớ. Cũng kích hoạt trích xuất Knowledge Graph."
},
+2
View File
@@ -98,6 +98,8 @@
"agents.compaction.reserveTokensFloorTip": "Bộ đệm token tối thiểu cần dự trữ cho phản hồi mới trước khi nén.",
"agents.compaction.maxHistoryShare": "Phần lịch sử tối đa (0-1)",
"agents.compaction.maxHistoryShareTip": "Tỷ lệ cửa sổ ngữ cảnh mà lịch sử cuộc trò chuyện có thể chiếm.",
"agents.compaction.timeoutSeconds": "Thời gian chờ (giây)",
"agents.compaction.timeoutSecondsTip": "Thời gian tối đa chờ nén ngữ cảnh trước khi giữ nguyên lịch sử cũ.",
"agents.pruning.title": "Cắt bớt ngữ cảnh",
"agents.pruning.desc": "Cắt bỏ kết quả công cụ cũ để giải phóng không gian ngữ cảnh",
+2
View File
@@ -797,6 +797,8 @@
"maxHistoryShareTip": "上下文窗口用于对话历史的最大比例(如0.85 = 85%)。超过时触发压缩。",
"keepLastMessages": "保留最后消息数",
"keepLastMessagesTip": "压缩后保留的最近消息数。旧消息被摘要替换。",
"timeoutSeconds": "超时秒数",
"timeoutSecondsTip": "等待上下文压缩摘要的最长时间,超时后保留原始历史。",
"memoryFlush": "压缩前记忆",
"memoryFlushTip": "压缩前,Agent可以将重要上下文保存到记忆文件。同时触发知识图谱提取。"
},
+2
View File
@@ -98,6 +98,8 @@
"agents.compaction.reserveTokensFloorTip": "压缩前为新响应保留的最小 Token 缓冲区。",
"agents.compaction.maxHistoryShare": "最大历史占比(0-1)",
"agents.compaction.maxHistoryShareTip": "对话历史可占用的上下文窗口比例。",
"agents.compaction.timeoutSeconds": "超时秒数",
"agents.compaction.timeoutSecondsTip": "等待上下文压缩的最长时间,超时后保留原始历史。",
"agents.pruning.title": "上下文裁剪",
"agents.pruning.desc": "裁剪旧工具结果以释放上下文空间",
@@ -39,6 +39,15 @@ export function CompactionSection({ value, onChange }: CompactionSectionProps) {
onChange={(e) => onChange({ ...value, keepLastMessages: numOrUndef(e.target.value) })}
/>
</div>
<div className="space-y-2">
<InfoLabel tip={t(`${s}.timeoutSecondsTip`)}>{t(`${s}.timeoutSeconds`)}</InfoLabel>
<Input
type="number"
placeholder="120"
value={value.timeoutSeconds ?? ""}
onChange={(e) => onChange({ ...value, timeoutSeconds: numOrUndef(e.target.value) })}
/>
</div>
</div>
<div className="flex items-center gap-2">
<Switch
@@ -196,6 +196,7 @@ export function AiDefaultsSection({ data, onSave, saving }: Props) {
<div className="grid grid-cols-1 gap-4 sm:grid-cols-2">
<Field label={t("agents.compaction.reserveTokensFloor")} tip={t("agents.compaction.reserveTokensFloorTip")} type="number" value={compaction.reserveTokensFloor} onChange={(v) => updateNested("compaction", { reserveTokensFloor: Number(v) })} placeholder="20000" />
<Field label={t("agents.compaction.maxHistoryShare")} tip={t("agents.compaction.maxHistoryShareTip")} type="number" step="0.05" value={compaction.maxHistoryShare} onChange={(v) => updateNested("compaction", { maxHistoryShare: Number(v) })} placeholder="0.75" />
<Field label={t("agents.compaction.timeoutSeconds")} tip={t("agents.compaction.timeoutSecondsTip")} type="number" value={compaction.timeoutSeconds} onChange={(v) => updateNested("compaction", { timeoutSeconds: Number(v) })} placeholder="120" />
</div>
</SubSection>
@@ -263,4 +264,3 @@ export function AiDefaultsSection({ data, onSave, saving }: Props) {
</Card>
);
}
+1
View File
@@ -28,6 +28,7 @@ export interface CompactionConfig {
reserveTokensFloor?: number;
maxHistoryShare?: number;
keepLastMessages?: number;
timeoutSeconds?: number;
memoryFlush?: {
enabled?: boolean;
softThresholdTokens?: number;