diff --git a/pkg/channels/qq/qq.go b/pkg/channels/qq/qq.go index dfea85ba3..68a84ba81 100644 --- a/pkg/channels/qq/qq.go +++ b/pkg/channels/qq/qq.go @@ -35,12 +35,15 @@ import ( ) const ( - dedupTTL = 5 * time.Minute - dedupInterval = 60 * time.Second - dedupMaxSize = 10000 // hard cap on dedup map entries - typingResend = 8 * time.Second - typingSeconds = 10 - bytesPerMiB = 1024 * 1024 + dedupTTL = 5 * time.Minute + dedupInterval = 60 * time.Second + dedupMaxSize = 10000 // hard cap on dedup map entries + typingResend = 8 * time.Second + typingSeconds = 10 + bytesPerMiB = 1024 * 1024 + reconnectInitial = 5 * time.Second + reconnectMax = 5 * time.Minute + reconnectMultiplier = 2.0 ) type qqAPI interface { @@ -127,6 +130,28 @@ func (c *QQChannel) Start(ctx context.Context) error { // initialize OpenAPI client c.api = botgo.NewOpenAPI(c.config.AppID, c.tokenSource).WithTimeout(5 * time.Second) + // start dedup janitor goroutine + go c.dedupJanitor() + + // Pre-register reasoning_channel_id as group chat if configured, + // so outbound-only destinations are routed correctly. + if c.config.ReasoningChannelID != "" { + c.chatType.Store(c.config.ReasoningChannelID, "group") + } + + // Start the reconnect loop. It handles both the initial connection and + // subsequent reconnection attempts with exponential backoff. + go c.reconnectLoop() + + c.SetRunning(true) + logger.InfoC("qq", "QQ bot started successfully") + + return nil +} + +// startSession fetches a fresh WebSocket endpoint and starts the session manager. +// It blocks until the session goroutine exits (i.e. the WebSocket connection drops). +func (c *QQChannel) startSession() error { // register event handlers intent := event.RegisterHandlers( c.handleC2CMessage(), @@ -146,29 +171,59 @@ func (c *QQChannel) Start(ctx context.Context) error { // create and save sessionManager c.sessionManager = botgo.NewSessionManager() - // start WebSocket connection in goroutine to avoid blocking - go func() { - if err := c.sessionManager.Start(wsInfo, c.tokenSource, &intent); err != nil { - logger.ErrorCF("qq", "WebSocket session error", map[string]any{ - "error": err.Error(), - }) - c.SetRunning(false) + return c.sessionManager.Start(wsInfo, c.tokenSource, &intent) +} + +// reconnectLoop manages the QQ WebSocket session lifecycle with exponential backoff. +// On initial start or after a disconnect, it re-acquires the WebSocket endpoint +// and re-initializes the session manager. +func (c *QQChannel) reconnectLoop() { + backoff := reconnectInitial + + for { + select { + case <-c.ctx.Done(): + return + case <-c.done: + return + default: } - }() - // start dedup janitor goroutine - go c.dedupJanitor() + // Check if channel was explicitly stopped. + if !c.IsRunning() { + return + } - // Pre-register reasoning_channel_id as group chat if configured, - // so outbound-only destinations are routed correctly. - if c.config.ReasoningChannelID != "" { - c.chatType.Store(c.config.ReasoningChannelID, "group") + logger.InfoCF("qq", "Starting QQ session", map[string]any{ + "backoff": backoff.String(), + }) + + err := c.startSession() + if err == nil { + // Session exited cleanly (e.g. via Stop/cancel). + return + } + + // Session failed or disconnected -- log and schedule retry. + logger.WarnCF("qq", "QQ session ended, reconnecting", map[string]any{ + "error": err.Error(), + }) + + select { + case <-c.ctx.Done(): + return + case <-c.done: + return + case <-time.After(backoff): + if backoff < reconnectMax { + next := time.Duration(float64(backoff) * reconnectMultiplier) + if next > reconnectMax { + next = reconnectMax + } + backoff = next + } + } } - - c.SetRunning(true) - logger.InfoC("qq", "QQ bot started successfully") - - return nil } func (c *QQChannel) Stop(ctx context.Context) error { diff --git a/pkg/providers/error_classifier_test.go b/pkg/providers/error_classifier_test.go index 46b180835..503efb313 100644 --- a/pkg/providers/error_classifier_test.go +++ b/pkg/providers/error_classifier_test.go @@ -264,6 +264,97 @@ func TestClassifyError_UnknownError(t *testing.T) { } } +func TestClassifyError_ModelNotFoundPatterns(t *testing.T) { + tests := []struct { + name string + msg string + want FailoverReason + wantOK bool + }{ + // Exact pattern: model_not_found (zhipu actual format) + {name: "model_not_found", msg: "model_not_found", want: FailoverFormat, wantOK: true}, + // Space-separated + {name: "model not found", msg: "model not found", want: FailoverFormat, wantOK: true}, + // invalid model + {name: "invalid model", msg: "invalid model", want: FailoverFormat, wantOK: true}, + // model does not exist + {name: "model does not exist", msg: "model does not exist", want: FailoverFormat, wantOK: true}, + // Case insensitive + {name: "Model_Not_Found upper", msg: "Model_Not_Found", want: FailoverFormat, wantOK: true}, + // model not supported + {name: "model not supported", msg: "model not supported", want: FailoverFormat, wantOK: true}, + // model not available + {name: "model not available", msg: "model not available", want: FailoverFormat, wantOK: true}, + // unknown model + {name: "unknown model", msg: "unknown model", want: FailoverFormat, wantOK: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := errors.New(tt.msg) + result := ClassifyError(err, "zhipu", "glm-4") + if tt.wantOK { + if result == nil { + t.Fatalf("expected non-nil for %q", tt.msg) + } + if result.Reason != tt.want { + t.Errorf("reason = %q, want %q", result.Reason, tt.want) + } + } else { + if result != nil { + t.Errorf("expected nil for %q, got %+v", tt.msg, result) + } + } + }) + } +} + +func TestClassifyError_TransientStatusOverride(t *testing.T) { + // When a transient 5xx status (e.g. 503) is accompanied by a + // model_not_found message body, the message pattern should win over + // the status code classification. This verifies the isTransientStatus + // override logic in ClassifyError. + tests := []struct { + name string + errMsg string + want FailoverReason + }{ + { + name: "503 with model_not_found should be format not timeout", + errMsg: "API error: status: 503 model_not_found", + want: FailoverFormat, + }, + { + name: "500 with model not found should be format not timeout", + errMsg: "API error: status: 500 model not found", + want: FailoverFormat, + }, + { + name: "502 with invalid model should be format not timeout", + errMsg: "status 502 - invalid model", + want: FailoverFormat, + }, + { + name: "503 without model pattern should remain timeout", + errMsg: "API error: status: 503 service unavailable", + want: FailoverTimeout, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := errors.New(tt.errMsg) + result := ClassifyError(err, "zhipu", "glm-4") + if result == nil { + t.Fatalf("expected non-nil for %q", tt.errMsg) + } + if result.Reason != tt.want { + t.Errorf("reason = %q, want %q", result.Reason, tt.want) + } + }) + } +} + func TestClassifyError_ProviderModelPropagation(t *testing.T) { err := errors.New("rate limit exceeded") result := ClassifyError(err, "my-provider", "my-model")