mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-12 02:13:59 +00:00
The v3 pipeline compacts session history mid-loop (prune_stage + final-request guard) but only mutates the run's message buffer, never the session store. Each turn reloads full history and re-compacts from scratch: message_tokens climb 129k->156k across turns while every turn compacts back down to ~60k. The lossy compaction differs per run, degrading the agent. The same missing persistence stalls episodic memory: the cumulative compaction count never advances, so the episodic worker's idempotency key (sessionKey:count) is pinned and every cycle after the first is skipped. Observed on live traffic: 8 run.completed since deploy, 0 new episodic. Fixes, all reusing existing machinery (no new store methods, no migrations): - Bug A: emitSessionCompleted reads cumulative GetCompactionCount (matching the legacy v2 path) instead of the per-run counter that resets to 0. - Bug B/anti-loop: finalize passes state.Prune.MidLoopCompacted into maybeSummarize; under pressure it lowers the trigger to a unit-aligned threshold (compactionInputCap - overhead, same MaxRequestShare the guard uses) so the compaction is PERSISTED via the existing TruncateHistory + IncrementCompaction path. Defensive floor prevents over-compaction on pathological config; tool-result-only bloat still skips (history-only). - Bug C: SourceID embeds the count (sessionKey:count) so the eventbus dedup key advances per compaction cycle instead of swallowing rapid same-session turns within the 5m TTL. Tests: episodic compaction, maybe_summarize pressure, request budget. go build (PG + sqliteonly), go vet, go test -race all green.
203 lines
7.4 KiB
Go
203 lines
7.4 KiB
Go
package agent
|
||
|
||
import (
|
||
"context"
|
||
"strings"
|
||
"testing"
|
||
|
||
"github.com/nextlevelbuilder/goclaw/internal/providers"
|
||
"github.com/nextlevelbuilder/goclaw/internal/tokencount"
|
||
)
|
||
|
||
// buildVietnameseMsgs constructs n alternating user/assistant messages with
|
||
// Vietnamese UTF-8 content. Each message is ~viRunes runes to hit a realistic
|
||
// total token budget (~100k input tokens for 600 messages).
|
||
func buildVietnameseMsgs(n, viRunes int) []providers.Message {
|
||
// ~viRunes-rune Vietnamese segment (3-byte UTF-8 per diacritic char).
|
||
segment := strings.Repeat(
|
||
"Xin chào! Đây là nội dung kiểm tra với ký tự tiếng Việt đặc biệt: ắ ặ ầ ẩ ậ ề ể ệ ọ ộ. ",
|
||
(viRunes/80)+1,
|
||
)
|
||
runes := []rune(segment)
|
||
if len(runes) > viRunes {
|
||
segment = string(runes[:viRunes])
|
||
}
|
||
|
||
msgs := make([]providers.Message, n)
|
||
for i := range msgs {
|
||
role := "user"
|
||
if i%2 != 0 {
|
||
role = "assistant"
|
||
}
|
||
msgs[i] = providers.Message{Role: role, Content: segment}
|
||
}
|
||
return msgs
|
||
}
|
||
|
||
// TestLoopCompact_Integration_DynamicMaxTokens_VietnameseFixture verifies the
|
||
// end-to-end composition of Phase 03 (FallbackCounter) + Phase 04 (dynamicSummaryMax):
|
||
//
|
||
// 1. Loop with real FallbackCounter estimates ~100k input tokens from 600 Vietnamese messages.
|
||
// 2. compactMessagesInPlace passes max_tokens in [2000, 8192] to the provider.
|
||
// 3. The formula dynamicSummaryMax(in) = in/25 holds: for ~100k input → ~4000 output budget.
|
||
//
|
||
// Tolerance: FallbackCounter uses rune/3 heuristic so exact input count varies;
|
||
// we assert >= 2000 && <= 8192 rather than == 4000.
|
||
func TestLoopCompact_Integration_DynamicMaxTokens_VietnameseFixture(t *testing.T) {
|
||
cap := &capturingProvider{response: "Tóm tắt cuộc trò chuyện: Đã thảo luận về nhiều chủ đề."}
|
||
|
||
loop := &Loop{
|
||
provider: cap,
|
||
model: "claude-3-5-sonnet",
|
||
tokenCounter: tokencount.NewFallbackCounter(),
|
||
contextWindow: 200_000,
|
||
}
|
||
|
||
// 600 messages × ~500 runes each ≈ 300k runes ÷ 3 ≈ 100k tokens total.
|
||
// keepCount defaults to 4; splitIdx = 600-4 = 596 msgs to summarise.
|
||
// FallbackCounter on 596 msgs × ~500 runes ÷ 3 ≈ ~99k tokens → dynamicSummaryMax(99000) = 3960 (floor 1024).
|
||
msgs := buildVietnameseMsgs(600, 500)
|
||
|
||
result := loop.compactMessagesInPlace(context.Background(), msgs)
|
||
if result == nil {
|
||
t.Fatal("compactMessagesInPlace returned nil; expected compaction to succeed with 600 messages")
|
||
}
|
||
|
||
if len(cap.captured) != 1 {
|
||
t.Fatalf("provider.Chat called %d time(s), want 1", len(cap.captured))
|
||
}
|
||
|
||
req := cap.captured[0]
|
||
maxTokensRaw, ok := req.Options["max_tokens"]
|
||
if !ok {
|
||
t.Fatal("Options[\"max_tokens\"] not set in ChatRequest")
|
||
}
|
||
|
||
maxTokens, ok := maxTokensRaw.(int)
|
||
if !ok {
|
||
t.Fatalf("Options[\"max_tokens\"] type = %T, want int", maxTokensRaw)
|
||
}
|
||
|
||
// Tolerance: FallbackCounter rune/3 varies slightly by content.
|
||
// For ~100k token input: dynamicSummaryMax → ~4000 (formula in/25).
|
||
// Assert range [2000, 8192] to accommodate counter variance.
|
||
const minExpected = 2000
|
||
const maxExpected = 8192
|
||
if maxTokens < minExpected || maxTokens > maxExpected {
|
||
t.Errorf("max_tokens = %d, want in [%d, %d]; formula dynamicSummaryMax(estimatedInput)",
|
||
maxTokens, minExpected, maxExpected)
|
||
}
|
||
|
||
// Log actual observed value for diagnostics.
|
||
keepCount := 4
|
||
if minKeep := len(msgs) * 3 / 10; minKeep > keepCount {
|
||
keepCount = minKeep
|
||
}
|
||
splitIdx := len(msgs) - keepCount
|
||
estimatedIn := loop.estimateSummaryInputTokens(msgs[:splitIdx])
|
||
t.Logf("observed: msgs=%d splitIdx=%d estimatedIn=%d max_tokens=%d dynamicSummaryMax=%d",
|
||
len(msgs), splitIdx, estimatedIn, maxTokens, dynamicSummaryMax(estimatedIn))
|
||
}
|
||
|
||
func TestLoopCompact_ChunksEverySummaryRequestUnderAgentCap(t *testing.T) {
|
||
cap := &capturingProvider{response: "bounded summary"}
|
||
loop := &Loop{
|
||
provider: cap,
|
||
model: "claude-3-5-sonnet",
|
||
tokenCounter: tokencount.NewFallbackCounter(),
|
||
contextWindow: 20_000,
|
||
}
|
||
msgs := buildVietnameseMsgs(20, 4_000)
|
||
|
||
result := loop.compactMessagesInPlace(context.Background(), msgs)
|
||
if result == nil {
|
||
t.Fatal("compactMessagesInPlace returned nil")
|
||
}
|
||
if len(cap.captured) < 3 {
|
||
t.Fatalf("provider calls = %d, want multiple chunks plus merge", len(cap.captured))
|
||
}
|
||
inputCap := loop.compactionInputCap()
|
||
for i, req := range cap.captured {
|
||
got := loop.tokenCounter.CountMessages(loop.model, req.Messages)
|
||
if got > inputCap {
|
||
t.Fatalf("request %d input = %d, exceeds cap %d", i, got, inputCap)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestLoopCompact_InputCapReservesFixedMaxTokens locks the agent-only budget:
|
||
// compactionInputCap() reserves the CALLING agent's fixed effectiveMaxTokens()
|
||
// from the window, bounded by the request share. The dynamic-output term
|
||
// (dynamicSummaryMax) governs only the per-chunk output budget — it is NOT an
|
||
// authority over the input cap, so it does not appear here.
|
||
func TestLoopCompact_InputCapReservesFixedMaxTokens(t *testing.T) {
|
||
const (
|
||
window = 20_000
|
||
maxTokens = 8_192
|
||
)
|
||
loop := &Loop{
|
||
provider: &capturingProvider{},
|
||
model: "claude-3-5-sonnet",
|
||
contextWindow: window,
|
||
maxTokens: maxTokens,
|
||
}
|
||
inputCap := loop.compactionInputCap()
|
||
|
||
hardCap := window - maxTokens // 11808
|
||
softTarget := int(float64(window)*defaultCompactionShare) - maxTokens // 8808
|
||
want := min(hardCap, softTarget)
|
||
if inputCap != want {
|
||
t.Fatalf("compactionInputCap() = %d, want %d (min(window-maxTokens, share*window-maxTokens))", inputCap, want)
|
||
}
|
||
// Invariant: input cap plus the fixed reserve never exceeds the window.
|
||
if inputCap+maxTokens > window {
|
||
t.Fatalf("inputCap %d + reserve %d exceeds window %d", inputCap, maxTokens, window)
|
||
}
|
||
}
|
||
|
||
func TestLoopCompact_RejectsMoreThanSixteenChunksBeforeProvider(t *testing.T) {
|
||
cap := &capturingProvider{response: "unused"}
|
||
loop := &Loop{
|
||
provider: cap,
|
||
model: "claude-3-5-sonnet",
|
||
tokenCounter: tokencount.NewFallbackCounter(),
|
||
contextWindow: 200_000,
|
||
}
|
||
unit := strings.Repeat("bounded-word ", 1_500)
|
||
unitCap := loop.estimateCompactionRequestTokens(unit)
|
||
units := make([]string, maxCompactionChunks+1)
|
||
for i := range units {
|
||
units[i] = unit
|
||
}
|
||
|
||
if _, _, err := loop.summarizeCompactionUnits(context.Background(), units, unitCap, 1); err == nil {
|
||
t.Fatal("expected compaction chunk limit error")
|
||
}
|
||
if len(cap.captured) != 0 {
|
||
t.Fatalf("provider calls = %d, want 0 before chunk-limit rejection", len(cap.captured))
|
||
}
|
||
}
|
||
|
||
func TestBuildCompactionUnits_KeepsToolCycleTogether(t *testing.T) {
|
||
units := buildCompactionUnits([]providers.Message{
|
||
{Role: "assistant", Content: "calling tool", ToolCalls: []providers.ToolCall{{ID: "call-1", Name: "read_file", Arguments: map[string]any{"path": "a.go"}}}},
|
||
{Role: "tool", Content: "file body here", ToolCallID: "call-1"},
|
||
{Role: "user", Content: "continue"},
|
||
})
|
||
if len(units) != 2 {
|
||
t.Fatalf("units = %d, want 2 (tool cycle + following user)", len(units))
|
||
}
|
||
if !strings.Contains(units[0], "calling tool") || !strings.Contains(units[1], "continue") {
|
||
t.Fatalf("unexpected units: %#v", units)
|
||
}
|
||
// Regression: the tool RESULT payload and the tool CALL must survive into the
|
||
// compaction unit. Previously renderCompactionMessages dropped role=tool and
|
||
// assistant tool_calls, erasing technical data before summarization.
|
||
if !strings.Contains(units[0], "file body here") {
|
||
t.Fatalf("tool result dropped from compaction unit: %#v", units[0])
|
||
}
|
||
if !strings.Contains(units[0], "read_file") {
|
||
t.Fatalf("tool call dropped from compaction unit: %#v", units[0])
|
||
}
|
||
}
|