mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-13 02:14:39 +00:00
Clark Cant
5ca433b8ac
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.
57 lines
1.6 KiB
Go
57 lines
1.6 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"testing"
|
|
"time"
|
|
|
|
runtimelogs "github.com/nextlevelbuilder/goclaw/internal/logs"
|
|
)
|
|
|
|
func TestLogTeeAggregateIncludesWithAttrsAndGroupEntries(t *testing.T) {
|
|
tee := NewLogTee(slog.NewTextHandler(io.Discard, nil))
|
|
root := slog.New(tee)
|
|
root.Info("root", "source", "gateway")
|
|
root.With("source", "worker").Warn("child")
|
|
root.WithGroup("grouped").Error("grouped", "source", "jobs")
|
|
|
|
result := tee.AggregateRuntimeLogs(runtimelogs.RuntimeAggregateOpts{GroupBy: "source"})
|
|
if result.Capacity != 100 || result.Retention != "ring_buffer" {
|
|
t.Fatalf("result metadata = %+v", result)
|
|
}
|
|
got := map[string]int{}
|
|
for _, b := range result.Buckets {
|
|
got[b.Key] = b.Count
|
|
}
|
|
if got["gateway"] != 1 || got["worker"] != 1 || got["jobs"] != 1 {
|
|
t.Fatalf("buckets = %+v", result.Buckets)
|
|
}
|
|
}
|
|
|
|
func TestLogTeeAggregateRedactsSensitiveAttrs(t *testing.T) {
|
|
tee := NewLogTee(slog.NewTextHandler(io.Discard, nil))
|
|
rec := slog.NewRecord(time.Now(), slog.LevelInfo, "secret", 0)
|
|
rec.AddAttrs(
|
|
slog.String("api_token", "leak"),
|
|
slog.Int("total_tokens", 123),
|
|
slog.Int("last_prompt_tokens", 100),
|
|
slog.String("source", "test"),
|
|
)
|
|
if err := tee.Handle(context.Background(), rec); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
entries := tee.recentEntries()
|
|
attrs, _ := entries[0]["attrs"].(map[string]any)
|
|
if attrs["api_token"] != redactedValue {
|
|
t.Fatalf("attrs = %+v", attrs)
|
|
}
|
|
if attrs["total_tokens"] != "123" {
|
|
t.Fatalf("total_tokens = %#v, want 123", attrs["total_tokens"])
|
|
}
|
|
if attrs["last_prompt_tokens"] != "100" {
|
|
t.Fatalf("last_prompt_tokens = %#v, want 100", attrs["last_prompt_tokens"])
|
|
}
|
|
}
|