feat(dingtalk): add proactive messaging support via robot API

Enable DingTalk bot to send messages proactively without requiring
user to initiate conversation first. This allows cron jobs and other
automated tasks to send notifications.

- Add token management with auto-refresh (token.go)
- Add robot API for group/direct messaging (api.go)
- Modify Send() to try session_webhook first, fallback to robot API
- Store complete session info including staffId and conversationId
- Auto-detect message type based on chatID format (cid prefix = group)

Configuration: set proactive_send: true in dingtalk channel config.
For direct messages, use staffId in the "to" field of cron jobs.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

pkg/channels/dingtalk/api.go

@@ -0,0 +1,151 @@
+// PicoClaw - Ultra-lightweight personal AI agent
+// DingTalk robot API for proactive messaging
+
+package dingtalk
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/sipeed/picoclaw/pkg/channels"
+	"github.com/sipeed/picoclaw/pkg/logger"
+)
+
+// sendGroupMessage sends a message to a group chat via robot API
+func (c *DingTalkChannel) sendGroupMessage(ctx context.Context, accessToken, openConversationId, content string) error {
+	url := fmt.Sprintf("%s/v1.0/robot/groupMessages/send", dingtalkAPIBase)
+
+	// Build msgParam for markdown message
+	msgParam, _ := json.Marshal(map[string]string{
+		"title": "PicoClaw",
+		"text":  content,
+	})
+
+	body := map[string]any{
+		"openConversationId": openConversationId,
+		"robotCode":          c.clientID,
+		"msgKey":             "sampleMarkdown",
+		"msgParam":           string(msgParam),
+	}
+
+	return c.sendRobotAPIRequest(ctx, accessToken, url, body, "group")
+}
+
+// sendOToMessage sends a message to a user via robot API (one-to-one)
+func (c *DingTalkChannel) sendOToMessage(ctx context.Context, accessToken, userID, content string) error {
+	url := fmt.Sprintf("%s/v1.0/robot/oToMessages/batchSend", dingtalkAPIBase)
+
+	// Build msgParam for markdown message
+	msgParam, _ := json.Marshal(map[string]string{
+		"title": "PicoClaw",
+		"text":  content,
+	})
+
+	body := map[string]any{
+		"robotCode": c.clientID,
+		"userIds":   []string{userID},
+		"msgKey":    "sampleMarkdown",
+		"msgParam":  string(msgParam),
+	}
+
+	return c.sendRobotAPIRequest(ctx, accessToken, url, body, "direct")
+}
+
+// sendRobotAPIRequest sends a request to DingTalk robot API
+func (c *DingTalkChannel) sendRobotAPIRequest(
+	ctx context.Context,
+	accessToken, url string,
+	body any,
+	msgType string,
+) error {
+	jsonData, err := json.Marshal(body)
+	if err != nil {
+		return fmt.Errorf("marshal request body failed: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
+	if err != nil {
+		return fmt.Errorf("create request failed: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("x-acs-dingtalk-access-token", accessToken)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return channels.ClassifyNetError(err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("read response failed: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		// Parse error response for better error message
+		var errResp struct {
+			Code string `json:"code"`
+			Msg  string `json:"message"`
+		}
+		if json.Unmarshal(respBody, &errResp) == nil && errResp.Msg != "" {
+			return fmt.Errorf("dingtalk robot API error: %s (%s)", errResp.Msg, errResp.Code)
+		}
+		return channels.ClassifySendError(resp.StatusCode,
+			fmt.Errorf("dingtalk robot API error (status %d): %s", resp.StatusCode, string(respBody)))
+	}
+
+	logger.InfoCF("dingtalk", "Robot API message sent successfully", map[string]any{
+		"type":   msgType,
+		"status": resp.StatusCode,
+	})
+
+	return nil
+}
+
+// sendViaRobotAPI sends a message via the robot API
+// For proactive messaging: chatID is used directly as staffId (direct) or openConversationId (group)
+// Group IDs start with "cid", everything else is treated as direct message staffId
+func (c *DingTalkChannel) sendViaRobotAPI(ctx context.Context, chatID, content string) error {
+	accessToken, err := c.ensureValidToken(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get access token: %w", err)
+	}
+
+	// First, check if we have session info (from prior conversation)
+	session := c.getSession(chatID)
+	if session != nil {
+		if session.ConversationType == "1" {
+			// Direct message with session info
+			if session.SenderStaffId != "" {
+				return c.sendOToMessage(ctx, accessToken, session.SenderStaffId, content)
+			}
+		} else {
+			// Group message with session info
+			if session.OpenConversationId != "" {
+				return c.sendGroupMessage(ctx, accessToken, session.OpenConversationId, content)
+			}
+		}
+	}
+
+	// No session info - detect type based on chatID format
+	// Group IDs start with "cid" (openConversationId format)
+	isGroup := len(chatID) > 3 && chatID[:3] == "cid"
+
+	if isGroup {
+		logger.DebugCF("dingtalk", "Sending proactive group message", map[string]any{
+			"open_conversation_id": chatID,
+		})
+		return c.sendGroupMessage(ctx, accessToken, chatID, content)
+	}
+
+	// Direct message - chatID is the staffId
+	logger.DebugCF("dingtalk", "Sending proactive direct message", map[string]any{
+		"staff_id": chatID,
+	})
+	return c.sendOToMessage(ctx, accessToken, chatID, content)
+}

pkg/channels/dingtalk/dingtalk.go

@@ -6,6 +6,7 @@ package dingtalk
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"sync"
 	"time"
 
@@ -28,6 +29,16 @@ const (
 	recoveryRetryDelay  = 30 * time.Second // Delay before retrying failed recovery
 )
 
+// DingTalkSession stores complete session information for proactive messaging
+type DingTalkSession struct {
+	SessionWebhook            string    `json:"session_webhook"`
+	SessionWebhookExpiredTime int64     `json:"session_webhook_expired_time"` // milliseconds timestamp
+	OpenConversationId        string    `json:"open_conversation_id"`         // Group chat ID
+	SenderStaffId             string    `json:"sender_staff_id"`              // User ID for direct chat
+	ConversationType          string    `json:"conversation_type"`            // "1" = direct, else = group
+	LastUpdated               time.Time `json:"last_updated"`
+}
+
 // DingTalkChannel implements the Channel interface for DingTalk (钉钉)
 // It uses WebSocket for receiving messages via stream mode and API for sending
 type DingTalkChannel struct {
@@ -38,8 +49,13 @@ type DingTalkChannel struct {
 	streamClient *client.StreamClient
 	ctx          context.Context
 	cancel       context.CancelFunc
-	// Map to store session webhooks for each chat
+	// Map to store session info for each chat
-	sessionWebhooks sync.Map // chatID -> sessionWebhook
+	sessions sync.Map // chatID -> *DingTalkSession
+	// Token management for proactive messaging via robot API
+	accessToken string
+	tokenExpiry time.Time
+	tokenMu     sync.RWMutex
+	httpClient  *http.Client
 	// Health monitoring
 	lastMessageTime time.Time
 	mu              sync.RWMutex
@@ -72,6 +88,19 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
 	c.ctx, c.cancel = context.WithCancel(ctx)
 	c.lastMessageTime = time.Now() // Initialize on start
 
+	// Initialize HTTP client for robot API
+	c.httpClient = &http.Client{Timeout: 30 * time.Second}
+
+	// If proactive send is enabled, get initial token and start refresh loop
+	if c.config.ProactiveSend {
+		if err := c.refreshAccessToken(); err != nil {
+			logger.WarnCF("dingtalk", "Failed to get initial access token, proactive send may not work", map[string]any{
+				"error": err.Error(),
+			})
+		}
+		go c.tokenRefreshLoop()
+	}
+
 	// Start the stream client
 	if err := c.startStreamClient(); err != nil {
 		return err
@@ -81,7 +110,9 @@ func (c *DingTalkChannel) Start(ctx context.Context) error {
 	go c.healthMonitor()
 
 	c.SetRunning(true)
-	logger.InfoC("dingtalk", "DingTalk channel started (Stream Mode)")
+	logger.InfoCF("dingtalk", "DingTalk channel started", map[string]any{
+		"proactive_send": c.config.ProactiveSend,
+	})
 	return nil
 }
 
@@ -199,30 +230,60 @@ func (c *DingTalkChannel) Stop(ctx context.Context) error {
 	return nil
 }
 
-// Send sends a message to DingTalk via the chatbot reply API
+// Send sends a message to DingTalk
+// Priority: 1) session_webhook (if valid), 2) robot API (if proactive_send enabled)
 func (c *DingTalkChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
 	if !c.IsRunning() {
 		return channels.ErrNotRunning
 	}
 
-	// Get session webhook from storage
-	sessionWebhookRaw, ok := c.sessionWebhooks.Load(msg.ChatID)
-	if !ok {
-		return fmt.Errorf("no session_webhook found for chat %s, cannot send message", msg.ChatID)
-	}
-
-	sessionWebhook, ok := sessionWebhookRaw.(string)
-	if !ok {
-		return fmt.Errorf("invalid session_webhook type for chat %s", msg.ChatID)
-	}
-
 	logger.DebugCF("dingtalk", "Sending message", map[string]any{
 		"chat_id": msg.ChatID,
 		"preview": utils.Truncate(msg.Content, 100),
 	})
 
-	// Use the session webhook to send the reply
+	// 1. Try session webhook first
-	return c.SendDirectReply(ctx, sessionWebhook, msg.Content)
+	session := c.getSession(msg.ChatID)
+	if session != nil && c.isSessionWebhookValid(session) {
+		err := c.SendDirectReply(ctx, session.SessionWebhook, msg.Content)
+		if err == nil {
+			return nil
+		}
+		// Session webhook failed, log and try fallback
+		logger.WarnCF("dingtalk", "Session webhook send failed, trying fallback", map[string]any{
+			"error":   err.Error(),
+			"chat_id": msg.ChatID,
+		})
+	}
+
+	// 2. Fallback to robot API if proactive send is enabled
+	if !c.config.ProactiveSend {
+		return fmt.Errorf("no valid session_webhook for chat %s and proactive_send is disabled", msg.ChatID)
+	}
+
+	return c.sendViaRobotAPI(ctx, msg.ChatID, msg.Content)
+}
+
+// getSession retrieves session info for a chat
+func (c *DingTalkChannel) getSession(chatID string) *DingTalkSession {
+	if v, ok := c.sessions.Load(chatID); ok {
+		if session, ok := v.(*DingTalkSession); ok {
+			return session
+		}
+	}
+	return nil
+}
+
+// isSessionWebhookValid checks if the session webhook is still valid
+func (c *DingTalkChannel) isSessionWebhookValid(session *DingTalkSession) bool {
+	if session.SessionWebhook == "" {
+		return false
+	}
+	// Check expiry if available (SessionWebhookExpiredTime is in milliseconds)
+	if session.SessionWebhookExpiredTime > 0 {
+		return time.Now().UnixMilli() < session.SessionWebhookExpiredTime
+	}
+	return true // No expiry info, assume valid
 }
 
 // onChatBotMessageReceived implements the IChatBotMessageHandler function signature
@@ -258,8 +319,16 @@ func (c *DingTalkChannel) onChatBotMessageReceived(
 		chatID = data.ConversationId
 	}
 
-	// Store the session webhook for this chat so we can reply later
+	// Store complete session info for proactive messaging
-	c.sessionWebhooks.Store(chatID, data.SessionWebhook)
+	session := &DingTalkSession{
+		SessionWebhook:            data.SessionWebhook,
+		SessionWebhookExpiredTime: data.SessionWebhookExpiredTime,
+		OpenConversationId:        data.ConversationId,
+		SenderStaffId:             data.SenderStaffId,
+		ConversationType:          data.ConversationType,
+		LastUpdated:               time.Now(),
+	}
+	c.sessions.Store(chatID, session)
 
 	metadata := map[string]string{
 		"sender_name":       senderNick,

pkg/channels/dingtalk/token.go

@@ -0,0 +1,139 @@
+// PicoClaw - Ultra-lightweight personal AI agent
+// DingTalk access token management
+
+package dingtalk
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/sipeed/picoclaw/pkg/logger"
+)
+
+const (
+	dingtalkAPIBase      = "https://api.dingtalk.com"
+	tokenRefreshInterval = 5 * time.Minute
+	tokenExpireBuffer    = 5 * time.Minute // Refresh token 5 minutes before expiry
+)
+
+// accessTokenResponse 钉钉 access_token 响应
+type accessTokenResponse struct {
+	AccessToken string `json:"accessToken"`
+	ExpireIn    int64  `json:"expireIn"` // seconds
+}
+
+// refreshAccessToken refreshes the access token from DingTalk API
+func (c *DingTalkChannel) refreshAccessToken() error {
+	url := fmt.Sprintf("%s/v1.0/oauth2/accessToken", dingtalkAPIBase)
+
+	body := map[string]string{
+		"appKey":    c.clientID,
+		"appSecret": c.clientSecret,
+	}
+	jsonData, err := json.Marshal(body)
+	if err != nil {
+		return fmt.Errorf("marshal request body failed: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(c.ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
+	if err != nil {
+		return fmt.Errorf("create request failed: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("request failed: %w", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("read response failed: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("dingtalk API error (status %d): %s", resp.StatusCode, string(respBody))
+	}
+
+	var tokenResp accessTokenResponse
+	if err := json.Unmarshal(respBody, &tokenResp); err != nil {
+		return fmt.Errorf("parse response failed: %w", err)
+	}
+
+	if tokenResp.AccessToken == "" {
+		return fmt.Errorf("empty access token in response")
+	}
+
+	c.tokenMu.Lock()
+	c.accessToken = tokenResp.AccessToken
+	c.tokenExpiry = time.Now().Add(time.Duration(tokenResp.ExpireIn) * time.Second)
+	c.tokenMu.Unlock()
+
+	logger.DebugCF("dingtalk", "Access token refreshed successfully", map[string]any{
+		"expires_in": tokenResp.ExpireIn,
+	})
+	return nil
+}
+
+// getAccessToken returns the current valid access token
+// Returns empty string if token is expired or about to expire
+func (c *DingTalkChannel) getAccessToken() string {
+	c.tokenMu.RLock()
+	defer c.tokenMu.RUnlock()
+
+	if c.accessToken == "" {
+		return ""
+	}
+
+	// Check if token is about to expire (within buffer time)
+	if time.Now().After(c.tokenExpiry.Add(-tokenExpireBuffer)) {
+		return "" // Token expired or about to expire
+	}
+
+	return c.accessToken
+}
+
+// ensureValidToken ensures we have a valid access token, refreshing if necessary
+func (c *DingTalkChannel) ensureValidToken(ctx context.Context) (string, error) {
+	token := c.getAccessToken()
+	if token != "" {
+		return token, nil
+	}
+
+	// Need to refresh
+	if err := c.refreshAccessToken(); err != nil {
+		return "", fmt.Errorf("failed to refresh access token: %w", err)
+	}
+
+	return c.getAccessToken(), nil
+}
+
+// tokenRefreshLoop runs periodically to refresh the access token
+func (c *DingTalkChannel) tokenRefreshLoop() {
+	ticker := time.NewTicker(tokenRefreshInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			if c.config.ProactiveSend {
+				// Check if token needs refresh
+				if c.getAccessToken() == "" {
+					if err := c.refreshAccessToken(); err != nil {
+						logger.ErrorCF("dingtalk", "Failed to refresh access token", map[string]any{
+							"error": err.Error(),
+						})
+					}
+				}
+			}
+		}
+	}
+}

pkg/config/config.go

@@ -322,6 +322,7 @@ type DingTalkConfig struct {
 	AllowFrom          FlexibleStringSlice `json:"allow_from"              env:"PICOCLAW_CHANNELS_DINGTALK_ALLOW_FROM"`
 	GroupTrigger       GroupTriggerConfig  `json:"group_trigger,omitempty"`
 	ReasoningChannelID string              `json:"reasoning_channel_id"    env:"PICOCLAW_CHANNELS_DINGTALK_REASONING_CHANNEL_ID"`
+	ProactiveSend      bool                `json:"proactive_send"          env:"PICOCLAW_CHANNELS_DINGTALK_PROACTIVE_SEND"`
 }
 
 type SlackConfig struct {