feat(pipeline): session compaction overflow recovery (#958)

- Add ZAI/GLM context overflow patterns to error_classify.go
- ThinkStage detects overflow, triggers emergency compaction + 1 retry
- Wire ReserveTokensFloor config to pipeline budget calculation
- Send user-friendly error on channel RunFailed events
- Add sessions.compact WebSocket method for manual truncation
This commit is contained in:
viettranx committed 2026-04-20 08:49:38 +07:00
1 parent 1b8627077a
commit a7962d182a
13 files changed
+406 -3

No files matched your search

+8
View File
@@ -70,6 +70,14 @@ func isContextOverflowError(lower string) bool {
"prompt is too long",
"exceeds model context window",
"request exceeds the maximum size",
// Issue 958: Additional patterns (sync with providers/error_classify.go)
"prompt exceeds max length", // ZAI/GLM-5
"input is too long", // DashScope
"token limit",
"too many tokens",
"请求输入过长", // Chinese generic
"超出最大长度限制", // Chinese Qwen
"上下文长度", // Chinese context length
) || (strings.Contains(lower, "context") &&
containsAny(lower, "overflow", "too large", "too long", "limit", "exceeded"))
}
+1
View File
@@ -58,6 +58,7 @@ func (l *Loop) buildPipelineDeps(req *RunRequest, bridgeRS *runState) pipeline.P
CheckpointInterval: 5,
ContextWindow: l.contextWindow,
MaxTokens: l.effectiveMaxTokens(),
ReserveTokens: l.resolveReserveTokens(),
Compaction: l.compactionCfg,
// V3 memory/retrieval flags removed — always true at runtime.
},
+9
View File
@@ -452,6 +452,15 @@ func (l *Loop) effectiveMaxTokens() int {
return defaultMaxTokens
}
// resolveReserveTokens returns the reserve token buffer from compaction config.
// Issue 958: Wire ReserveTokensFloor to prevent context overflow before compaction.
func (l *Loop) resolveReserveTokens() int {
if l.compactionCfg != nil && l.compactionCfg.ReserveTokensFloor > 0 {
return l.compactionCfg.ReserveTokensFloor
}
return 0
}
func NewLoop(cfg LoopConfig) *Loop {
if cfg.MaxIterations <= 0 {
cfg.MaxIterations = config.DefaultMaxIterations
+45
View File
@@ -0,0 +1,45 @@
package channels
import (
"strings"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
// FormatAgentError converts internal error to user-friendly message.
// Issue 958: Send user-friendly error on RunFailed instead of silent "...".
func FormatAgentError(errStr string) string {
if errStr == "" {
return ""
}
lower := strings.ToLower(errStr)
// Context overflow (highest priority — specific actionable message)
if providers.IsContextOverflowMessage(lower) {
return "⚠️ The conversation has grown too long. Please start a new chat or ask me to summarize."
}
// Rate limit
if strings.Contains(lower, "rate limit") || strings.Contains(lower, "too many requests") || strings.Contains(lower, "429") {
return "⏳ Too many requests. Please wait a moment and try again."
}
// Auth errors
if strings.Contains(lower, "unauthorized") || strings.Contains(lower, "invalid api key") || strings.Contains(lower, "401") || strings.Contains(lower, "403") {
return "🔑 Authentication error. Please check your API configuration."
}
// Timeout
if strings.Contains(lower, "timeout") || strings.Contains(lower, "deadline exceeded") {
return "⏱️ Request timed out. Please try again."
}
// Overloaded
if strings.Contains(lower, "overload") {
return "🔄 Service is busy. Please try again in a moment."
}
// Generic fallback (don't expose internal error details)
return "❌ Something went wrong. Please try again."
}
+65
View File
@@ -0,0 +1,65 @@
package channels
import (
"strings"
"testing"
)
func TestFormatAgentError_ContextOverflow(t *testing.T) {
t.Parallel()
testCases := []string{
"context length exceeded",
"Prompt exceeds max length",
"request_too_large: payload too big",
"Input is too long for this model",
"token limit exceeded",
"请求输入过长",
}
for _, tc := range testCases {
result := FormatAgentError(tc)
if !strings.Contains(result, "conversation has grown too long") {
t.Errorf("expected context overflow message for %q, got %q", tc, result)
}
}
}
func TestFormatAgentError_RateLimit(t *testing.T) {
t.Parallel()
result := FormatAgentError("rate limit exceeded")
if !strings.Contains(result, "Too many requests") {
t.Errorf("unexpected rate limit message: %s", result)
}
}
func TestFormatAgentError_Auth(t *testing.T) {
t.Parallel()
result := FormatAgentError("unauthorized access")
if !strings.Contains(result, "Authentication error") {
t.Errorf("unexpected auth message: %s", result)
}
}
func TestFormatAgentError_Timeout(t *testing.T) {
t.Parallel()
result := FormatAgentError("request timeout")
if !strings.Contains(result, "timed out") {
t.Errorf("unexpected timeout message: %s", result)
}
}
func TestFormatAgentError_Generic(t *testing.T) {
t.Parallel()
result := FormatAgentError("some unknown error")
if !strings.Contains(result, "Something went wrong") {
t.Errorf("unexpected generic message: %s", result)
}
}
func TestFormatAgentError_Empty(t *testing.T) {
t.Parallel()
result := FormatAgentError("")
if result != "" {
t.Errorf("expected empty string for empty error, got %q", result)
}
}
+21 -2
View File
@@ -234,8 +234,27 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
}
sc.FinalizeStream(ctx, rc.ChatID, currentStream)
}
case protocol.AgentEventRunFailed, protocol.AgentEventRunCancelled:
// Clean up streaming state on failure or cancellation
case protocol.AgentEventRunFailed:
// Clean up streaming state on failure
rc.mu.Lock()
currentStream := rc.stream
rc.stream = nil
rc.mu.Unlock()
if currentStream != nil {
_ = currentStream.Stop(ctx)
}
// Issue 958: Send user-friendly error message instead of silent "..."
errStr := extractPayloadString(payload, "error")
if friendlyMsg := FormatAgentError(errStr); friendlyMsg != "" {
m.bus.PublishOutbound(bus.OutboundMessage{
Channel: rc.ChannelName,
ChatID: rc.ChatID,
Content: friendlyMsg,
TenantID: rc.TenantID,
})
}
case protocol.AgentEventRunCancelled:
// Clean up streaming state on cancellation
rc.mu.Lock()
currentStream := rc.stream
rc.stream = nil
+63
View File
@@ -30,6 +30,7 @@ func (m *SessionsMethods) Register(router *gateway.MethodRouter) {
router.Register(protocol.MethodSessionsPatch, m.handlePatch)
router.Register(protocol.MethodSessionsDelete, m.handleDelete)
router.Register(protocol.MethodSessionsReset, m.handleReset)
router.Register(protocol.MethodSessionsCompact, m.handleCompact)
}
type sessionsListParams struct {
@@ -230,3 +231,65 @@ func (m *SessionsMethods) handleReset(ctx context.Context, client *gateway.Clien
}))
emitAudit(m.eventBus, client, "session.reset", "session", params.Key)
}
type sessionCompactParams struct {
Key string `json:"key"`
KeepLast int `json:"keepLast,omitempty"` // default 4
}
// handleCompact truncates session history to the last N messages.
// Issue 958: Manual session compaction API (truncate-only, no LLM summarization).
func (m *SessionsMethods) handleCompact(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) {
locale := store.LocaleFromContext(ctx)
var params sessionCompactParams
if err := json.Unmarshal(req.Params, &params); err != nil {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON)))
return
}
if params.Key == "" {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "key is required"))
return
}
keepLast := params.KeepLast
if keepLast <= 0 {
keepLast = 4 // default: keep last 2 exchanges
}
// Auth check
sess := m.sessions.Get(ctx, params.Key)
if sess == nil {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgNotFound, "session", params.Key)))
return
}
if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) {
if sess.UserID != client.UserID() {
client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "session")))
return
}
}
history := m.sessions.GetHistory(ctx, params.Key)
originalLen := len(history)
if originalLen < 6 {
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
"ok": true,
"message": "session too short to compact",
"kept": originalLen,
}))
return
}
// Truncate history to last N messages
m.sessions.TruncateHistory(ctx, params.Key, keepLast)
m.sessions.IncrementCompaction(ctx, params.Key)
m.sessions.Save(ctx, params.Key)
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
"ok": true,
"original": originalLen,
"kept": keepLast,
}))
emitAudit(m.eventBus, client, "session.compacted", "session", params.Key)
}
+112
View File
@@ -336,6 +336,118 @@ func TestThinkStage_LLMError_Propagates(t *testing.T) {
}
}
// Issue 958: Context overflow triggers emergency compaction + retry
func TestThinkStage_ContextOverflow_TriggersCompaction(t *testing.T) {
t.Parallel()
callCount := 0
compacted := false
deps := &PipelineDeps{
Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000},
CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) {
callCount++
if callCount == 1 {
return nil, &providers.HTTPError{Status: 400, Body: "Prompt exceeds max length"}
}
return &providers.ChatResponse{Content: "success after compact", FinishReason: "stop"}, nil
},
CompactMessages: func(_ context.Context, msgs []providers.Message, _ string) ([]providers.Message, error) {
compacted = true
return []providers.Message{{Role: "user", Content: "[Summary]"}}, nil
},
}
stage := NewThinkStage(deps)
state := defaultState()
state.Messages.SetHistory([]providers.Message{{Role: "user", Content: "test"}})
// First call: overflow → compact → retry
err := stage.Execute(context.Background(), state)
if err != nil {
t.Fatalf("first Execute() should trigger retry, got error: %v", err)
}
if !compacted {
t.Error("expected compaction to be triggered")
}
if state.Think.OverflowRetries != 1 {
t.Errorf("expected OverflowRetries=1, got %d", state.Think.OverflowRetries)
}
// Stage returns Continue (nil error) to signal retry this iteration
if stage.Result() != Continue {
t.Errorf("Result() = %v after compaction, want Continue", stage.Result())
}
}
func TestThinkStage_ContextOverflow_FailsAfterOneRetry(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{
Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000},
CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) {
return nil, &providers.HTTPError{Status: 400, Body: "Prompt exceeds max length"}
},
CompactMessages: func(_ context.Context, _ []providers.Message, _ string) ([]providers.Message, error) {
return []providers.Message{{Role: "user", Content: "[Summary]"}}, nil
},
}
stage := NewThinkStage(deps)
state := defaultState()
state.Think.OverflowRetries = 1 // Already retried once
err := stage.Execute(context.Background(), state)
if err == nil {
t.Error("expected error after second overflow")
}
if !strings.Contains(err.Error(), "context overflow after compaction") {
t.Errorf("expected 'context overflow after compaction' message, got %v", err)
}
}
func TestThinkStage_ContextOverflow_NoCompactCallback_FailsGracefully(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{
Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000},
CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) {
return nil, &providers.HTTPError{Status: 400, Body: "Prompt exceeds max length"}
},
CompactMessages: nil, // No compaction available
}
stage := NewThinkStage(deps)
state := defaultState()
err := stage.Execute(context.Background(), state)
if err == nil {
t.Error("expected error when no compaction available")
}
}
func TestThinkStage_ContextOverflow_CompactionFails_ReturnsOriginalError(t *testing.T) {
t.Parallel()
deps := &PipelineDeps{
Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000},
CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) {
return nil, &providers.HTTPError{Status: 400, Body: "Prompt exceeds max length"}
},
CompactMessages: func(_ context.Context, _ []providers.Message, _ string) ([]providers.Message, error) {
return nil, errors.New("compaction failed")
},
}
stage := NewThinkStage(deps)
state := defaultState()
err := stage.Execute(context.Background(), state)
if err == nil {
t.Error("expected error when compaction fails")
}
// Should return LLM error wrapped, not compaction error
if !strings.Contains(err.Error(), "llm call") {
t.Errorf("expected 'llm call' in error message, got %v", err)
}
}
// --- PruneStage tests ---
func TestPruneStage_UnderBudget_NoOp(t *testing.T) {
+1
View File
@@ -32,6 +32,7 @@ type ThinkState struct {
LastResponse *providers.ChatResponse
TotalUsage providers.Usage
TruncRetries int // consecutive truncation retries (max 3)
OverflowRetries int // context overflow compact+retry attempts (max 1)
StreamingActive bool // true during active stream
}
+36 -1
View File
@@ -3,6 +3,8 @@ package pipeline
import (
"context"
"fmt"
"log/slog"
"strings"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
@@ -57,6 +59,28 @@ func (s *ThinkStage) Execute(ctx context.Context, state *RunState) error {
}
resp, err := s.deps.CallLLM(ctx, state, req)
if err != nil {
// Issue 958: Check for context overflow — attempt emergency compaction + retry
if isContextOverflowErr(err) {
if state.Think.OverflowRetries > 0 {
return fmt.Errorf("context overflow after compaction: %w", err)
}
state.Think.OverflowRetries++
// Attempt emergency compaction
if s.deps.CompactMessages != nil {
originalLen := len(state.Messages.History())
compacted, compactErr := s.deps.CompactMessages(ctx, state.Messages.History(), state.Model)
if compactErr == nil {
state.Messages.ReplaceHistory(compacted)
slog.Info("emergency_compaction_triggered",
"run_id", state.RunID,
"original_msgs", originalLen,
"compacted_msgs", len(compacted),
)
return nil // Retry this iteration (Continue result)
}
slog.Warn("emergency_compaction_failed", "error", compactErr)
}
}
return fmt.Errorf("llm call: %w", err)
}
state.Think.LastResponse = resp
@@ -92,7 +116,8 @@ func (s *ThinkStage) Execute(ctx context.Context, state *RunState) error {
state.Messages.AppendPending(providers.Message{Role: "user", Content: hint})
return nil // Continue to next iteration for retry
}
state.Think.TruncRetries = 0 // reset on success
state.Think.TruncRetries = 0 // reset on success
state.Think.OverflowRetries = 0 // reset on success
// 7. Uniquify tool call IDs (OpenAI returns 400 on duplicates across iterations).
// Skip if raw content present (Anthropic thinking passback) to avoid desync.
@@ -192,3 +217,13 @@ func toolCallsHaveMissingRequiredArgs(calls []providers.ToolCall) bool {
}
return false
}
// isContextOverflowErr checks if an error indicates context window overflow.
// Uses the exported helper from providers package for pattern matching.
func isContextOverflowErr(err error) bool {
if err == nil {
return false
}
lower := strings.ToLower(err.Error())
return providers.IsContextOverflowMessage(lower)
}
+10
View File
@@ -155,9 +155,19 @@ func isContextOverflow(lower string) bool {
// Chinese patterns (Qwen/DashScope)
"超出最大长度限制",
"上下文长度",
// Issue 958: Additional patterns
"prompt exceeds max length", // ZAI/GLM-5
"request_too_large", // Generic
"input is too long", // DashScope
"请求输入过长", // Chinese generic
)
}
// IsContextOverflowMessage exports overflow detection for use by pipeline.
func IsContextOverflowMessage(lower string) bool {
return isContextOverflow(lower)
}
// isNetworkError checks if an error is a network-level failure.
func isNetworkError(err error) bool {
if err == nil {
+34
View File
@@ -324,3 +324,37 @@ func TestClassifyUnknownError(t *testing.T) {
t.Errorf("expected FailoverUnknown, got %s", result.Reason)
}
}
// Issue 958: New context overflow patterns for ZAI/GLM, DashScope, generic
func TestClassifyPromptExceedsMaxLength(t *testing.T) {
classifier := NewDefaultClassifier()
result := classifier.Classify(nil, 400, `{"error":{"code":"1261","message":"Prompt exceeds max length"}}`)
if result.Kind != "context_overflow" {
t.Errorf("expected context_overflow, got %s (reason: %s)", result.Kind, result.Reason)
}
}
func TestClassifyInputTooLong(t *testing.T) {
classifier := NewDefaultClassifier()
result := classifier.Classify(nil, 400, "Input is too long for this model")
if result.Kind != "context_overflow" {
t.Errorf("expected context_overflow, got %s (reason: %s)", result.Kind, result.Reason)
}
}
func TestClassifyRequestTooLarge(t *testing.T) {
classifier := NewDefaultClassifier()
result := classifier.Classify(nil, 400, "request_too_large: payload exceeds limit")
if result.Kind != "context_overflow" {
t.Errorf("expected context_overflow, got %s (reason: %s)", result.Kind, result.Reason)
}
}
func TestClassifyChineseInputTooLong(t *testing.T) {
classifier := NewDefaultClassifier()
result := classifier.Classify(nil, 400, "请求输入过长")
if result.Kind != "context_overflow" {
t.Errorf("expected context_overflow, got %s (reason: %s)", result.Kind, result.Reason)
}
}
+1
View File
@@ -39,6 +39,7 @@ const (
MethodSessionsPatch = "sessions.patch"
MethodSessionsDelete = "sessions.delete"
MethodSessionsReset = "sessions.reset"
MethodSessionsCompact = "sessions.compact"
// System
MethodConnect = "connect"