Merge 7be27f2d92 into b6927c9a7a
This commit is contained in:
commit
b6d62aa7f5
2 changed files with 38 additions and 7 deletions
|
|
@ -270,6 +270,15 @@ func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, cha
|
|||
}
|
||||
|
||||
func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) {
|
||||
// Filter out empty or whitespace-only messages
|
||||
if strings.TrimSpace(msg.Content) == "" {
|
||||
logger.DebugCF("agent", "Dropped empty message", map[string]any{
|
||||
"channel": msg.Channel,
|
||||
"sender_id": msg.SenderID,
|
||||
})
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// Add message preview to log (show full content for error messages)
|
||||
var logContent string
|
||||
if strings.Contains(msg.Content, "Error:") || strings.Contains(msg.Content, "error") {
|
||||
|
|
@ -458,8 +467,7 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt
|
|||
}
|
||||
|
||||
// 9. Log response
|
||||
responsePreview := utils.Truncate(finalContent, 120)
|
||||
logger.InfoCF("agent", fmt.Sprintf("Response: %s", responsePreview),
|
||||
logger.InfoCF("agent", fmt.Sprintf("Response: [%s]", finalContent),
|
||||
map[string]any{
|
||||
"agent_id": agent.ID,
|
||||
"session_key": opts.SessionKey,
|
||||
|
|
|
|||
|
|
@ -215,7 +215,15 @@ func (c *WeComAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
|
|||
|
||||
accessToken := c.getAccessToken()
|
||||
if accessToken == "" {
|
||||
return fmt.Errorf("no valid access token available")
|
||||
// Token expired or not yet acquired — attempt an on-demand refresh
|
||||
logger.WarnC("wecom_app", "Access token missing or expired, attempting on-demand refresh")
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
return fmt.Errorf("access token unavailable and refresh failed: %w", err)
|
||||
}
|
||||
accessToken = c.getAccessToken()
|
||||
if accessToken == "" {
|
||||
return fmt.Errorf("no valid access token available after refresh")
|
||||
}
|
||||
}
|
||||
|
||||
logger.DebugCF("wecom_app", "Sending message", map[string]any{
|
||||
|
|
@ -223,7 +231,7 @@ func (c *WeComAppChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
|
|||
"preview": utils.Truncate(msg.Content, 100),
|
||||
})
|
||||
|
||||
return c.sendTextMessage(ctx, accessToken, msg.ChatID, msg.Content)
|
||||
return c.sendMarkdownMessage(ctx, accessToken, msg.ChatID, msg.Content)
|
||||
}
|
||||
|
||||
// handleWebhook handles incoming webhook requests from WeCom
|
||||
|
|
@ -453,14 +461,29 @@ func (c *WeComAppChannel) processMessage(ctx context.Context, msg WeComXMLMessag
|
|||
|
||||
// tokenRefreshLoop periodically refreshes the access token
|
||||
func (c *WeComAppChannel) tokenRefreshLoop() {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
const fallbackInterval = 30 * time.Minute
|
||||
const earlyRefresh = 5 * time.Minute
|
||||
|
||||
for {
|
||||
// Calculate sleep duration based on current token expiry
|
||||
c.tokenMu.RLock()
|
||||
expiry := c.tokenExpiry
|
||||
c.tokenMu.RUnlock()
|
||||
|
||||
var sleepDur time.Duration
|
||||
if expiry.IsZero() {
|
||||
// Token never successfully acquired
|
||||
sleepDur = fallbackInterval
|
||||
} else {
|
||||
sleepDur = max(time.Until(expiry.Add(-earlyRefresh)),
|
||||
// minimum 1 minute to avoid tight loop
|
||||
time.Minute)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
case <-time.After(sleepDur):
|
||||
if err := c.refreshAccessToken(); err != nil {
|
||||
logger.ErrorCF("wecom_app", "Failed to refresh access token", map[string]any{
|
||||
"error": err.Error(),
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue