docs(dingtalk): document vision and escalation setup

Add configurable PostgreSQL and database/sql pool limits, validate invalid environment values, clamp idle connections, and cover the parsing/application logic with unit tests.

.env.example

@@ -210,6 +210,15 @@ DB_PASSWORD=postgres123!@#
 # 数据库名称
 DB_NAME=WeKnora
 
+# PostgreSQL 全局最大连接数（Compose 内置 PostgreSQL 服务）
+POSTGRES_MAX_CONNECTIONS=300
+
+# WeKnora database/sql 连接池
+DB_MAX_OPEN_CONNS=120
+DB_MAX_IDLE_CONNS=20
+DB_CONN_MAX_IDLE_TIME=5m
+DB_CONN_MAX_LIFETIME=10m
+
 # 如果使用 redis 作为流处理后端，需要配置以下参数
 # Redis用户名，Redis 6.0+ ACL 功能支持（可选）
 # REDIS_USERNAME=

docker-compose.yml

@@ -79,6 +79,10 @@ services:
       - DB_USER=${DB_USER:-}
       - DB_PASSWORD=${DB_PASSWORD:-}
       - DB_NAME=${DB_NAME:-}
+      - DB_MAX_OPEN_CONNS=${DB_MAX_OPEN_CONNS:-120}
+      - DB_MAX_IDLE_CONNS=${DB_MAX_IDLE_CONNS:-20}
+      - DB_CONN_MAX_IDLE_TIME=${DB_CONN_MAX_IDLE_TIME:-5m}
+      - DB_CONN_MAX_LIFETIME=${DB_CONN_MAX_LIFETIME:-10m}
       - TZ=${TZ:-Asia/Shanghai}
       - WEKNORA_LANGUAGE=${WEKNORA_LANGUAGE:-zh-CN}
       # ========== Langfuse (optional observability) ==========
@@ -291,6 +295,7 @@ services:
   postgres:
     image: paradedb/paradedb:v0.22.2-pg17
     container_name: WeKnora-postgres
+    command: ["postgres", "-c", "max_connections=${POSTGRES_MAX_CONNECTIONS:-300}"]
     environment:
       - POSTGRES_USER=${DB_USER}
       - POSTGRES_PASSWORD=${DB_PASSWORD}

docs/IM集成开发文档.md

@@ -419,7 +419,9 @@ curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
 **第二步：配置机器人**
 
 1. 在应用详情页 → **机器人** → 配置消息接收模式为 **Stream 模式**
-2.（可选）如需流式 AI 卡片效果，在 **互动卡片** 中创建一个 **AI 卡片模板**，记录 **Card Template ID**
+2. 如需识别用户发送的图片，在钉钉控制台为应用授予机器人消息文件下载所需权限，并在 WeKnora Agent 配置中启用**图片上传/多模态**
+3.（可选）如需流式 AI 卡片效果，在 **互动卡片** 中创建一个 **AI 卡片模板**，记录 **Card Template ID**
+4. 如需点赞/点踩和转人工，在卡片模板中配置互动按钮，按钮参数使用 `action=like`、`action=dislike`、`action=unresolved` 或 `action=handoff`；点踩原因可附加 `reason` 参数
 
 **第三步：在 WeKnora 中添加 IM 渠道**
 
@@ -431,6 +433,7 @@ curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
    - **Client ID**：填入 AppKey
    - **Client Secret**：填入 AppSecret
    - **Card Template ID**（可选）：填入 AI 卡片模板 ID，启用后流式回复将以 AI 卡片形式展示
+   - **技术支持群 openConversationId**（可选）：配置后，转人工工单会自动推送到该群；可让机器人先加入支持群并从该群消息回调的 `conversationId` 获取
 3. 保存
 
 启动后日志出现以下内容表示连接成功：
@@ -439,7 +442,7 @@ curl -X POST "https://api.telegram.org/bot<YOUR_BOT_TOKEN>/setWebhook" \
 [IM] DingTalk Stream connecting...
 ```
 
-> **关于 AI 卡片**：配置 Card Template ID 后，流式回复将通过钉钉 AI 卡片实时展示打字效果；未配置时流式内容将在结束后一次性发送。
+> **关于 AI 卡片**：配置 Card Template ID 后，流式回复将通过钉钉 AI 卡片实时展示打字效果；未配置时流式内容将在结束后一次性发送。点赞/点踩/转人工回调使用钉钉 Stream 卡片回调，需保持渠道为 WebSocket/Stream 模式。
 
 ---
 

frontend/src/components/IMChannelPanel.vue

@@ -401,6 +401,12 @@
                   placeholder="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.schema" />
                 <p class="form-desc">{{ $t('agentEditor.im.dingtalkCardTemplateIdHint') }}</p>
               </div>
+              <div class="form-item">
+                <label class="form-label">{{ $t('agentEditor.im.dingtalkSupportGroupId') }}</label>
+                <t-input v-model="formData.credentials.support_group_open_conversation_id"
+                  placeholder="cidxxxxxxxxxxxxxxxx" />
+                <p class="form-desc">{{ $t('agentEditor.im.dingtalkSupportGroupIdHint') }}</p>
+              </div>
             </template>
 
             <!-- QQBot credentials -->

frontend/src/i18n/locales/en-US.ts

@@ -5375,6 +5375,8 @@ export default {
       dingtalkConsole: 'DingTalk Open Platform',
       dingtalkCardTemplateId: 'Card Template ID (optional)',
       dingtalkCardTemplateIdHint: 'Create an AI Card template at open-dev.dingtalk.com/fe/card to enable streaming output with typewriter effect',
+      dingtalkSupportGroupId: 'Support group openConversationId (optional)',
+      dingtalkSupportGroupIdHint: 'Escalated tickets are pushed to this DingTalk group. Obtain the ID from a callback received in the support group.',
       mattermostConsole: 'Mattermost integrations',
       qqbotConsole: 'QQ Open Platform',
       qqbotAPIBaseURLHint: 'Optional. Leave empty to use the default production API endpoint.',

frontend/src/i18n/locales/ko-KR.ts

@@ -5316,6 +5316,8 @@ export default {
       dingtalkConsole: "DingTalk 개발 플랫폼",
       dingtalkCardTemplateId: "카드 템플릿 ID (선택 사항)",
       dingtalkCardTemplateIdHint: "open-dev.dingtalk.com/fe/card에서 AI 카드 템플릿을 만들면 타자기 효과 스트리밍 출력이 활성화됩니다",
+      dingtalkSupportGroupId: "지원 그룹 openConversationId (선택 사항)",
+      dingtalkSupportGroupIdHint: "에스컬레이션된 티켓을 이 DingTalk 그룹으로 전송합니다",
       mattermostConsole: "Mattermost 통합",
       qqbotConsole: "QQ Open Platform",
       qqbotAPIBaseURLHint: "선택 사항. 비워두면 기본 프로덕션 API 엔드포인트를 사용합니다.",

frontend/src/i18n/locales/ru-RU.ts

@@ -4886,6 +4886,8 @@ export default {
       dingtalkConsole: 'Платформа DingTalk',
       dingtalkCardTemplateId: 'ID шаблона карточки (необязательно)',
       dingtalkCardTemplateIdHint: 'Создайте шаблон AI-карточки на open-dev.dingtalk.com/fe/card для потоковой передачи с эффектом печатной машинки',
+      dingtalkSupportGroupId: 'openConversationId группы поддержки (необязательно)',
+      dingtalkSupportGroupIdHint: 'Эскалированные заявки отправляются в эту группу DingTalk',
       mattermostConsole: 'Интеграции Mattermost',
       qqbotConsole: 'QQ Open Platform',
       qqbotAPIBaseURLHint: 'Необязательно. Оставьте пустым, чтобы использовать стандартный production API endpoint.',

frontend/src/i18n/locales/zh-CN.ts

@@ -5318,6 +5318,8 @@ export default {
       dingtalkConsole: "钉钉开放平台",
       dingtalkCardTemplateId: "卡片模板 ID（可选）",
       dingtalkCardTemplateIdHint: "在 open-dev.dingtalk.com/fe/card 创建 AI 卡片模板，启用后支持打字机效果的流式输出",
+      dingtalkSupportGroupId: "技术支持群 openConversationId（可选）",
+      dingtalkSupportGroupIdHint: "转人工工单会自动推送到该钉钉群，可从支持群内机器人回调中获取此 ID",
       mattermostConsole: "Mattermost 集成说明",
       qqbotConsole: "QQ 开放平台",
       qqbotAPIBaseURLHint: "可选，留空则使用默认正式环境 API 地址。",

internal/agent/engine.go

@@ -48,6 +48,7 @@ type AgentEngine struct {
 	tokenEstimator       *agenttoken.Estimator     // Token estimator for context window management
 	memoryConsolidator   *agentmemory.Consolidator // Memory consolidator for LLM-powered summarization (optional)
 	lastUsage            types.TokenUsage          // Token usage from the most recent LLM call
+	totalUsage           types.TokenUsage          // Aggregate token usage across all LLM rounds in this execution
 	lastSentMsgCount     int                       // Number of messages sent in the most recent LLM call
 }
 
@@ -183,6 +184,17 @@ func (e *AgentEngine) estimateCurrentTokens(messages []chat.Message) int {
 	return e.tokenEstimator.EstimateMessages(messages)
 }
 
+func (e *AgentEngine) recordTokenUsage(usage types.TokenUsage) {
+	if usage.TotalTokens <= 0 {
+		return
+	}
+	e.lastUsage = usage
+	e.totalUsage.PromptTokens += usage.PromptTokens
+	e.totalUsage.CompletionTokens += usage.CompletionTokens
+	e.totalUsage.TotalTokens += usage.TotalTokens
+	e.totalUsage.CachedTokens += usage.CachedTokens
+}
+
 // Execute executes the agent with conversation history and streaming output
 // All events are emitted to EventBus and handled by subscribers (like Handler layer)
 func (e *AgentEngine) Execute(
@@ -191,6 +203,9 @@ func (e *AgentEngine) Execute(
 	llmContext []chat.Message,
 	imageURLs ...[]string,
 ) (*types.AgentState, error) {
+	e.lastUsage = types.TokenUsage{}
+	e.totalUsage = types.TokenUsage{}
+	e.lastSentMsgCount = 0
 	logger.Infof(ctx, "[Agent] Starting execution: session=%s, message=%s, query_len=%d, context_msgs=%d",
 		sessionID, messageID, len(query), len(llmContext))
 	// Ensure tools are cleaned up after execution
@@ -540,7 +555,7 @@ func (e *AgentEngine) runReActIteration(
 	}
 	response = resp
 	if response.Usage.TotalTokens > 0 {
-		e.lastUsage = response.Usage
+		e.recordTokenUsage(response.Usage)
 		logger.Debugf(ctx, "[Agent][Round-%d] Usage: prompt=%d, completion=%d, total=%d",
 			round, response.Usage.PromptTokens,
 			response.Usage.CompletionTokens, response.Usage.TotalTokens)

internal/agent/finalize.go

@@ -169,6 +169,11 @@ func (e *AgentEngine) emitCompletionEvent(
 	for _, ref := range state.KnowledgeRefs {
 		knowledgeRefsInterface = append(knowledgeRefsInterface, ref)
 	}
+	var tokenUsage *types.TokenUsage
+	if e.totalUsage.TotalTokens > 0 {
+		usage := e.totalUsage
+		tokenUsage = &usage
+	}
 
 	e.eventBus.Emit(ctx, event.Event{
 		ID:        generateEventID("complete"),
@@ -181,6 +186,7 @@ func (e *AgentEngine) emitCompletionEvent(
 			TotalSteps:      len(state.RoundSteps),
 			TotalDurationMs: time.Since(startTime).Milliseconds(),
 			MessageID:       messageID, // Include message ID for proper message update
+			TokenUsage:      tokenUsage,
 		},
 	})
 

internal/agent/usage_audit_test.go

@@ -0,0 +1,30 @@
+package agent
+
+import (
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+func TestRecordTokenUsageAggregatesAgentRounds(t *testing.T) {
+	engine := &AgentEngine{}
+	engine.recordTokenUsage(types.TokenUsage{PromptTokens: 100, CompletionTokens: 20, TotalTokens: 120, CachedTokens: 40})
+	engine.recordTokenUsage(types.TokenUsage{PromptTokens: 140, CompletionTokens: 30, TotalTokens: 170, CachedTokens: 60})
+
+	wantTotal := types.TokenUsage{PromptTokens: 240, CompletionTokens: 50, TotalTokens: 290, CachedTokens: 100}
+	if engine.totalUsage != wantTotal {
+		t.Fatalf("totalUsage = %#v, want %#v", engine.totalUsage, wantTotal)
+	}
+	wantLast := types.TokenUsage{PromptTokens: 140, CompletionTokens: 30, TotalTokens: 170, CachedTokens: 60}
+	if engine.lastUsage != wantLast {
+		t.Fatalf("lastUsage = %#v, want %#v", engine.lastUsage, wantLast)
+	}
+}
+
+func TestRecordTokenUsageIgnoresEmptyUsage(t *testing.T) {
+	engine := &AgentEngine{}
+	engine.recordTokenUsage(types.TokenUsage{})
+	if engine.totalUsage != (types.TokenUsage{}) || engine.lastUsage != (types.TokenUsage{}) {
+		t.Fatalf("empty usage changed state: total=%#v last=%#v", engine.totalUsage, engine.lastUsage)
+	}
+}

internal/application/service/chat_pipeline/chat_completion_stream.go

@@ -186,8 +186,9 @@ func (p *PluginChatCompletionStream) OnEvent(ctx context.Context,
 						Type:      types.EventType(event.EventAgentFinalAnswer),
 						SessionID: chatManage.SessionID,
 						Data: event.AgentFinalAnswerData{
-							Content: response.Content,
+							Content:    response.Content,
-							Done:    response.Done,
+							Done:       response.Done,
+							TokenUsage: response.Usage,
 						},
 					})
 				}

internal/container/container.go

@@ -90,6 +90,62 @@ import (
 	wgrpc "github.com/weaviate/weaviate-go-client/v5/weaviate/grpc"
 )
 
+func nonNegativeEnvInt(name string, fallback int) int {
+	raw := strings.TrimSpace(os.Getenv(name))
+	if raw == "" {
+		return fallback
+	}
+
+	value, err := strconv.Atoi(raw)
+	if err != nil || value < 0 {
+		return fallback
+	}
+	return value
+}
+
+func nonNegativeEnvDuration(name string, fallback time.Duration) time.Duration {
+	raw := strings.TrimSpace(os.Getenv(name))
+	if raw == "" {
+		return fallback
+	}
+	if raw == "0" {
+		return 0
+	}
+
+	value, err := time.ParseDuration(raw)
+	if err != nil || value < 0 {
+		return fallback
+	}
+	return value
+}
+
+type databasePoolConfig struct {
+	maxOpen     int
+	maxIdle     int
+	maxIdleTime time.Duration
+	maxLifetime time.Duration
+}
+
+func databasePoolConfigFromEnv() databasePoolConfig {
+	cfg := databasePoolConfig{
+		maxOpen:     nonNegativeEnvInt("DB_MAX_OPEN_CONNS", 0),
+		maxIdle:     nonNegativeEnvInt("DB_MAX_IDLE_CONNS", 10),
+		maxIdleTime: nonNegativeEnvDuration("DB_CONN_MAX_IDLE_TIME", 0),
+		maxLifetime: nonNegativeEnvDuration("DB_CONN_MAX_LIFETIME", 10*time.Minute),
+	}
+	if cfg.maxOpen > 0 && cfg.maxIdle > cfg.maxOpen {
+		cfg.maxIdle = cfg.maxOpen
+	}
+	return cfg
+}
+
+func applyDatabasePoolConfig(db *sql.DB, cfg databasePoolConfig) {
+	db.SetMaxOpenConns(cfg.maxOpen)
+	db.SetMaxIdleConns(cfg.maxIdle)
+	db.SetConnMaxIdleTime(cfg.maxIdleTime)
+	db.SetConnMaxLifetime(cfg.maxLifetime)
+}
+
 // BuildContainer constructs the dependency injection container
 // Registers all components, services, repositories and handlers needed by the application
 // Creates a fully configured application container with proper dependency resolution
@@ -608,10 +664,19 @@ func initDatabase(cfg *config.Config) (*gorm.DB, error) {
 		// Limiting to a single open connection serialises all DB access and
 		// prevents "database is locked" errors from concurrent goroutines.
 		sqlDB.SetMaxOpenConns(1)
+		sqlDB.SetConnMaxLifetime(10 * time.Minute)
 	} else {
-		sqlDB.SetMaxIdleConns(10)
+		poolCfg := databasePoolConfigFromEnv()
+		applyDatabasePoolConfig(sqlDB, poolCfg)
+		logger.Infof(
+			context.Background(),
+			"Database pool configured: max_open=%d max_idle=%d max_idle_time=%s max_lifetime=%s",
+			poolCfg.maxOpen,
+			poolCfg.maxIdle,
+			poolCfg.maxIdleTime,
+			poolCfg.maxLifetime,
+		)
 	}
-	sqlDB.SetConnMaxLifetime(time.Duration(10) * time.Minute)
 
 	return db, nil
 }

internal/container/db_pool_config_test.go

@@ -0,0 +1,105 @@
+package container
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+)
+
+func TestNonNegativeEnvInt(t *testing.T) {
+	tests := []struct {
+		name     string
+		raw      string
+		fallback int
+		want     int
+	}{
+		{name: "uses fallback when unset", raw: "", fallback: 10, want: 10},
+		{name: "accepts positive value", raw: "120", fallback: 10, want: 120},
+		{name: "accepts zero", raw: "0", fallback: 10, want: 0},
+		{name: "rejects negative value", raw: "-1", fallback: 10, want: 10},
+		{name: "rejects invalid value", raw: "many", fallback: 10, want: 10},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv("TEST_DB_POOL_INT", tt.raw)
+			if got := nonNegativeEnvInt("TEST_DB_POOL_INT", tt.fallback); got != tt.want {
+				t.Fatalf("nonNegativeEnvInt() = %d, want %d", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestNonNegativeEnvDuration(t *testing.T) {
+	tests := []struct {
+		name     string
+		raw      string
+		fallback time.Duration
+		want     time.Duration
+	}{
+		{name: "uses fallback when unset", raw: "", fallback: 10 * time.Minute, want: 10 * time.Minute},
+		{name: "accepts duration", raw: "5m", fallback: 10 * time.Minute, want: 5 * time.Minute},
+		{name: "accepts zero", raw: "0", fallback: 10 * time.Minute, want: 0},
+		{name: "rejects negative duration", raw: "-1m", fallback: 10 * time.Minute, want: 10 * time.Minute},
+		{name: "rejects invalid duration", raw: "later", fallback: 10 * time.Minute, want: 10 * time.Minute},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv("TEST_DB_POOL_DURATION", tt.raw)
+			if got := nonNegativeEnvDuration("TEST_DB_POOL_DURATION", tt.fallback); got != tt.want {
+				t.Fatalf("nonNegativeEnvDuration() = %s, want %s", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDatabasePoolConfigFromEnv(t *testing.T) {
+	t.Setenv("DB_MAX_OPEN_CONNS", "120")
+	t.Setenv("DB_MAX_IDLE_CONNS", "20")
+	t.Setenv("DB_CONN_MAX_IDLE_TIME", "5m")
+	t.Setenv("DB_CONN_MAX_LIFETIME", "10m")
+
+	got := databasePoolConfigFromEnv()
+	if got.maxOpen != 120 {
+		t.Fatalf("maxOpen = %d, want 120", got.maxOpen)
+	}
+	if got.maxIdle != 20 {
+		t.Fatalf("maxIdle = %d, want 20", got.maxIdle)
+	}
+	if got.maxIdleTime != 5*time.Minute {
+		t.Fatalf("maxIdleTime = %s, want 5m", got.maxIdleTime)
+	}
+	if got.maxLifetime != 10*time.Minute {
+		t.Fatalf("maxLifetime = %s, want 10m", got.maxLifetime)
+	}
+}
+
+func TestDatabasePoolConfigClampsIdleToOpen(t *testing.T) {
+	t.Setenv("DB_MAX_OPEN_CONNS", "5")
+	t.Setenv("DB_MAX_IDLE_CONNS", "20")
+
+	got := databasePoolConfigFromEnv()
+	if got.maxIdle != 5 {
+		t.Fatalf("maxIdle = %d, want 5", got.maxIdle)
+	}
+}
+
+func TestApplyDatabasePoolConfig(t *testing.T) {
+	db, err := sql.Open("sqlite3", ":memory:")
+	if err != nil {
+		t.Fatalf("sql.Open() error = %v", err)
+	}
+	defer db.Close()
+
+	applyDatabasePoolConfig(db, databasePoolConfig{
+		maxOpen:     120,
+		maxIdle:     20,
+		maxIdleTime: 5 * time.Minute,
+		maxLifetime: 10 * time.Minute,
+	})
+
+	if got := db.Stats().MaxOpenConnections; got != 120 {
+		t.Fatalf("MaxOpenConnections = %d, want 120", got)
+	}
+}

internal/event/event_data.go

@@ -1,5 +1,7 @@
 package event
 
+import "github.com/Tencent/WeKnora/internal/types"
+
 // EventData contains common event data structures for different stages
 
 // QueryData represents query-related event data
@@ -142,6 +144,7 @@ type AgentCompleteData struct {
 	TotalDurationMs int64                  `json:"total_duration_ms"`
 	MessageID       string                 `json:"message_id,omitempty"` // Assistant message ID
 	RequestID       string                 `json:"request_id,omitempty"`
+	TokenUsage      *types.TokenUsage      `json:"token_usage,omitempty"`
 	Extra           map[string]interface{} `json:"extra,omitempty"`
 }
 
@@ -184,9 +187,10 @@ type AgentReferencesData struct {
 
 // AgentFinalAnswerData represents final answer streaming data
 type AgentFinalAnswerData struct {
-	Content    string `json:"content"`
+	Content    string            `json:"content"`
-	Done       bool   `json:"done"`
+	Done       bool              `json:"done"`
-	IsFallback bool   `json:"is_fallback,omitempty"` // True when response is a fallback (no knowledge base match)
+	IsFallback bool              `json:"is_fallback,omitempty"` // True when response is a fallback (no knowledge base match)
+	TokenUsage *types.TokenUsage `json:"token_usage,omitempty"`
 }
 
 // AgentReflectionData represents agent reflection data

internal/handler/session/agent_stream_handler.go

@@ -475,6 +475,10 @@ func (h *AgentStreamHandler) handleFinalAnswer(ctx context.Context, evt event.Ev
 	if data.IsFallback {
 		h.assistantMessage.IsFallback = true
 	}
+	if data.TokenUsage != nil {
+		usage := *data.TokenUsage
+		h.assistantMessage.TokenUsage = &usage
+	}
 
 	// Calculate duration if done
 	var metadata map[string]interface{}
@@ -604,6 +608,10 @@ func (h *AgentStreamHandler) handleComplete(ctx context.Context, evt event.Event
 		// h.assistantMessage.Content = data.FinalAnswer
 		h.assistantMessage.IsCompleted = true
 		h.assistantMessage.AgentDurationMs = data.TotalDurationMs
+		if data.TokenUsage != nil {
+			usage := *data.TokenUsage
+			h.assistantMessage.TokenUsage = &usage
+		}
 
 		// Update knowledge references if provided
 		if len(data.KnowledgeRefs) > 0 {

internal/handler/session/qa.go

@@ -718,6 +718,10 @@ func (h *Handler) executeQA(reqCtx *qaRequestContext, mode qaMode, generateTitle
 				return nil
 			}
 			streamCtx.assistantMessage.Content += data.Content
+			if data.TokenUsage != nil {
+				usage := *data.TokenUsage
+				streamCtx.assistantMessage.TokenUsage = &usage
+			}
 			if data.IsFallback {
 				streamCtx.assistantMessage.IsFallback = true
 			}

internal/im/adapter.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 
+	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/gin-gonic/gin"
 )
 
@@ -64,6 +65,10 @@ type IncomingMessage struct {
 	FileName string
 	// FileSize is the file size in bytes (for file messages, optional).
 	FileSize int64
+	// ImageURLs carries transient model-ready image inputs (usually data URIs). Never log these values.
+	ImageURLs []string
+	// ImageReferences carries durable, non-secret source references persisted on the user message.
+	ImageReferences types.MessageImages
 	// ThreadID is the platform-specific thread identifier.
 	// - Slack: thread_ts (top-level message uses its own timestamp)
 	// - Mattermost: root_id, or post_id if top-level

internal/im/dingtalk/adapter.go

@@ -9,7 +9,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"mime"
+	"net"
 	"net/http"
+	"net/url"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -33,15 +37,18 @@ const dingtalkConvTypeGroup = "2"
 
 // Compile-time checks.
 var (
-	_ im.Adapter      = (*Adapter)(nil)
+	_ im.Adapter        = (*Adapter)(nil)
-	_ im.StreamSender = (*Adapter)(nil)
+	_ im.StreamSender   = (*Adapter)(nil)
+	_ im.FileDownloader = (*Adapter)(nil)
 )
 
 // Adapter implements im.Adapter for DingTalk.
 type Adapter struct {
-	clientID       string
+	clientID               string
-	clientSecret   string
+	clientSecret           string
-	cardTemplateID string // optional: enables AI card streaming when set
+	cardTemplateID         string // optional: enables AI card streaming when set
+	messageFileDownloadURL string
+	allowInsecureDownload  bool // tests only; production requires HTTPS
 
 	// accessToken cache
 	tokenMu    sync.RWMutex
@@ -114,19 +121,20 @@ func (a *Adapter) VerifyCallback(c *gin.Context) error {
 
 // DingTalk callback message structure.
 type callbackMessage struct {
-	ConversationID   string       `json:"conversationId"`
+	ConversationID   string          `json:"conversationId"`
-	ConversationType string       `json:"conversationType"`
+	ConversationType string          `json:"conversationType"`
-	MsgID            string       `json:"msgId"`
+	MsgID            string          `json:"msgId"`
-	Msgtype          string       `json:"msgtype"`
+	Msgtype          string          `json:"msgtype"`
-	Text             *textContent `json:"text"`
+	Text             *textContent    `json:"text"`
-	SenderNick       string       `json:"senderNick"`
+	Content          json.RawMessage `json:"content"`
-	SenderStaffId    string       `json:"senderStaffId"`
+	SenderNick       string          `json:"senderNick"`
-	SenderID         string       `json:"senderId"`
+	SenderStaffId    string          `json:"senderStaffId"`
-	SessionWebhook   string       `json:"sessionWebhook"`
+	SenderID         string          `json:"senderId"`
-	RobotCode        string       `json:"robotCode"`
+	SessionWebhook   string          `json:"sessionWebhook"`
-	AtUsers          []atUser     `json:"atUsers"`
+	RobotCode        string          `json:"robotCode"`
-	IsInAtList       bool         `json:"isInAtList"`
+	AtUsers          []atUser        `json:"atUsers"`
-	ChatbotCorpId    string       `json:"chatbotCorpId"`
+	IsInAtList       bool            `json:"isInAtList"`
+	ChatbotCorpId    string          `json:"chatbotCorpId"`
 }
 
 type textContent struct {
@@ -170,6 +178,20 @@ func parseCallbackMessage(msg *callbackMessage) *im.IncomingMessage {
 	if msg.Text != nil {
 		content = strings.TrimSpace(msg.Text.Content)
 	}
+	messageType := im.MessageTypeText
+	fileKey := ""
+	fileName := ""
+	if strings.EqualFold(msg.Msgtype, "picture") {
+		var picture struct {
+			DownloadCode string `json:"downloadCode"`
+		}
+		if json.Unmarshal(msg.Content, &picture) == nil && strings.TrimSpace(picture.DownloadCode) != "" {
+			messageType = im.MessageTypeImage
+			fileKey = strings.TrimSpace(picture.DownloadCode)
+			fileName = msg.MsgID + ".png"
+			content = ""
+		}
+	}
 
 	return &im.IncomingMessage{
 		Platform:    im.PlatformDingtalk,
@@ -178,14 +200,124 @@ func parseCallbackMessage(msg *callbackMessage) *im.IncomingMessage {
 		ChatID:      chatID,
 		ChatType:    chatType,
 		MessageID:   msg.MsgID,
-		MessageType: im.MessageTypeText,
+		MessageType: messageType,
 		Content:     content,
+		FileKey:     fileKey,
+		FileName:    fileName,
 		Extra: map[string]string{
 			"session_webhook": msg.SessionWebhook,
 		},
 	}
 }
 
+const dingtalkMessageFileDownloadURL = "https://api.dingtalk.com/v1.0/robot/messageFiles/download"
+
+func (a *Adapter) DownloadFile(ctx context.Context, msg *im.IncomingMessage) (io.ReadCloser, string, error) {
+	if msg == nil || strings.TrimSpace(msg.FileKey) == "" {
+		return nil, "", fmt.Errorf("download code is required")
+	}
+	token, err := a.getAccessToken(ctx)
+	if err != nil {
+		return nil, "", fmt.Errorf("get access token: %w", err)
+	}
+	body, err := json.Marshal(map[string]string{
+		"downloadCode": msg.FileKey,
+		"robotCode":    a.clientID,
+	})
+	if err != nil {
+		return nil, "", fmt.Errorf("marshal message file request: %w", err)
+	}
+	apiURL := a.messageFileDownloadURL
+	if apiURL == "" {
+		apiURL = dingtalkMessageFileDownloadURL
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(body))
+	if err != nil {
+		return nil, "", fmt.Errorf("create message file request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("x-acs-dingtalk-access-token", token)
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return nil, "", fmt.Errorf("request message file URL: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		return nil, "", fmt.Errorf("dingtalk message file API returned %d", resp.StatusCode)
+	}
+	var result struct {
+		DownloadURL string `json:"downloadUrl"`
+	}
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&result); err != nil {
+		return nil, "", fmt.Errorf("decode message file response: %w", err)
+	}
+	if err := validateDingTalkDownloadURL(result.DownloadURL, a.allowInsecureDownload); err != nil {
+		return nil, "", err
+	}
+
+	downloadReq, err := http.NewRequestWithContext(ctx, http.MethodGet, result.DownloadURL, nil)
+	if err != nil {
+		return nil, "", fmt.Errorf("create image download request: %w", err)
+	}
+	downloadClient := *httpClient
+	downloadClient.CheckRedirect = func(req *http.Request, _ []*http.Request) error {
+		return validateDingTalkDownloadURL(req.URL.String(), a.allowInsecureDownload)
+	}
+	downloadResp, err := downloadClient.Do(downloadReq)
+	if err != nil {
+		return nil, "", fmt.Errorf("download dingtalk image: %w", err)
+	}
+	if downloadResp.StatusCode < http.StatusOK || downloadResp.StatusCode >= http.StatusMultipleChoices {
+		downloadResp.Body.Close()
+		return nil, "", fmt.Errorf("dingtalk image download returned %d", downloadResp.StatusCode)
+	}
+	contentType := strings.ToLower(strings.TrimSpace(strings.Split(downloadResp.Header.Get("Content-Type"), ";")[0]))
+	if msg.MessageType == im.MessageTypeImage && !strings.HasPrefix(contentType, "image/") {
+		downloadResp.Body.Close()
+		return nil, "", fmt.Errorf("dingtalk picture response is not an image")
+	}
+	return downloadResp.Body, dingtalkDownloadedFileName(downloadResp, msg, contentType), nil
+}
+
+func validateDingTalkDownloadURL(rawURL string, allowInsecure bool) error {
+	u, err := url.Parse(strings.TrimSpace(rawURL))
+	if err != nil || u.Hostname() == "" || u.User != nil {
+		return fmt.Errorf("invalid dingtalk download URL")
+	}
+	if allowInsecure {
+		if u.Scheme != "http" && u.Scheme != "https" {
+			return fmt.Errorf("invalid dingtalk download URL scheme")
+		}
+		return nil
+	}
+	if u.Scheme != "https" {
+		return fmt.Errorf("dingtalk download URL must use HTTPS")
+	}
+	if ip := net.ParseIP(u.Hostname()); ip != nil && (ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsUnspecified()) {
+		return fmt.Errorf("dingtalk download URL points to a private address")
+	}
+	return nil
+}
+
+func dingtalkDownloadedFileName(resp *http.Response, msg *im.IncomingMessage, contentType string) string {
+	if _, params, err := mime.ParseMediaType(resp.Header.Get("Content-Disposition")); err == nil {
+		if name := filepath.Base(strings.TrimSpace(params["filename"])); name != "" && name != "." && name != ".." {
+			return name
+		}
+	}
+	if msg != nil && strings.TrimSpace(msg.FileName) != "" {
+		return filepath.Base(msg.FileName)
+	}
+	ext := ".bin"
+	if extensions, _ := mime.ExtensionsByType(contentType); len(extensions) > 0 {
+		ext = extensions[0]
+	}
+	if msg != nil && msg.MessageID != "" {
+		return msg.MessageID + ext
+	}
+	return "dingtalk-file" + ext
+}
+
 // ── Send reply ──
 
 func (a *Adapter) SendReply(ctx context.Context, incoming *im.IncomingMessage, reply *im.ReplyMessage) error {
@@ -509,6 +641,10 @@ func (a *Adapter) StartStream(ctx context.Context, incoming *im.IncomingMessage)
 			logger.Warnf(ctx, "[DingTalk] Failed to create AI card, falling back to sessionWebhook: %v", err)
 		} else {
 			state.outTrackID = outTrackID
+			if incoming.Extra == nil {
+				incoming.Extra = make(map[string]string)
+			}
+			incoming.Extra["dingtalk_out_track_id"] = outTrackID
 		}
 	}
 

internal/im/dingtalk/adapter_test.go

@@ -0,0 +1,93 @@
+package dingtalk
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Tencent/WeKnora/internal/im"
+)
+
+func TestParseCallbackMessagePicture(t *testing.T) {
+	msg := &callbackMessage{
+		ConversationID:   "cid-group",
+		ConversationType: dingtalkConvTypeGroup,
+		MsgID:            "msg-picture-1",
+		Msgtype:          "picture",
+		Content:          []byte(`{"downloadCode":"download-code-1"}`),
+		SenderStaffId:    "staff-1",
+		SenderNick:       "Alice",
+		SessionWebhook:   "https://example.invalid/session",
+	}
+
+	got := parseCallbackMessage(msg)
+	if got.MessageType != im.MessageTypeImage {
+		t.Fatalf("MessageType = %q, want image", got.MessageType)
+	}
+	if got.FileKey != "download-code-1" {
+		t.Fatalf("FileKey = %q, want download code", got.FileKey)
+	}
+	if got.FileName != "msg-picture-1.png" {
+		t.Fatalf("FileName = %q", got.FileName)
+	}
+	if got.Content != "" {
+		t.Fatalf("Content = %q, want empty", got.Content)
+	}
+}
+
+func TestAdapterDownloadFileUsesMessageFilesAPI(t *testing.T) {
+	var sawToken bool
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/v1.0/robot/messageFiles/download":
+			sawToken = r.Header.Get("x-acs-dingtalk-access-token") == "cached-access-token"
+			body, _ := io.ReadAll(r.Body)
+			if !strings.Contains(string(body), `"downloadCode":"download-code-1"`) || !strings.Contains(string(body), `"robotCode":"robot-code-1"`) {
+				t.Errorf("unexpected request body: %s", body)
+			}
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"downloadUrl":"` + serverURLForRequest(r) + `/media/picture"}`))
+		case "/media/picture":
+			w.Header().Set("Content-Type", "image/png")
+			w.Header().Set("Content-Disposition", `attachment; filename="question.png"`)
+			_, _ = w.Write([]byte("PNGDATA"))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer server.Close()
+
+	adapter := NewAdapter("robot-code-1", "secret", "")
+	adapter.token = "cached-access-token"
+	adapter.tokenExpAt = time.Now().Add(time.Hour)
+	adapter.messageFileDownloadURL = server.URL + "/v1.0/robot/messageFiles/download"
+	adapter.allowInsecureDownload = true
+
+	reader, fileName, err := adapter.DownloadFile(context.Background(), &im.IncomingMessage{
+		MessageType: im.MessageTypeImage,
+		MessageID:   "msg-picture-1",
+		FileKey:     "download-code-1",
+	})
+	if err != nil {
+		t.Fatalf("DownloadFile: %v", err)
+	}
+	defer reader.Close()
+	body, _ := io.ReadAll(reader)
+	if string(body) != "PNGDATA" {
+		t.Fatalf("body = %q", body)
+	}
+	if fileName != "question.png" {
+		t.Fatalf("fileName = %q", fileName)
+	}
+	if !sawToken {
+		t.Fatal("missing DingTalk access token header")
+	}
+}
+
+func serverURLForRequest(r *http.Request) string {
+	return "http://" + r.Host
+}

internal/im/dingtalk/longconn.go

@@ -2,6 +2,7 @@ package dingtalk
 
 import (
 	"context"
+	"encoding/json"
 	"strings"
 
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
@@ -71,7 +72,11 @@ func (c *LongConnClient) onChatBotMessage(ctx context.Context, data *chatbot.Bot
 		userID = data.SenderId
 	}
 
-	content := strings.TrimSpace(data.Text.Content)
+	messageType, content, fileKey := parseDingTalkStreamPayload(data.Msgtype, data.Content, data.Text.Content)
+	fileName := ""
+	if messageType == im.MessageTypeImage {
+		fileName = data.MsgId + ".png"
+	}
 
 	incoming := &im.IncomingMessage{
 		Platform:    im.PlatformDingtalk,
@@ -80,8 +85,10 @@ func (c *LongConnClient) onChatBotMessage(ctx context.Context, data *chatbot.Bot
 		ChatID:      chatID,
 		ChatType:    chatType,
 		MessageID:   data.MsgId,
-		MessageType: im.MessageTypeText,
+		MessageType: messageType,
 		Content:     content,
+		FileKey:     fileKey,
+		FileName:    fileName,
 		Extra: map[string]string{
 			"session_webhook": data.SessionWebhook,
 		},
@@ -93,3 +100,31 @@ func (c *LongConnClient) onChatBotMessage(ctx context.Context, data *chatbot.Bot
 
 	return nil, nil
 }
+
+func parseDingTalkStreamPayload(msgType string, rawContent interface{}, text string) (im.MessageType, string, string) {
+	if !strings.EqualFold(strings.TrimSpace(msgType), "picture") {
+		return im.MessageTypeText, strings.TrimSpace(text), ""
+	}
+	var payload []byte
+	switch value := rawContent.(type) {
+	case string:
+		payload = []byte(value)
+	case []byte:
+		payload = value
+	case json.RawMessage:
+		payload = value
+	default:
+		var err error
+		payload, err = json.Marshal(rawContent)
+		if err != nil {
+			return im.MessageTypeImage, "", ""
+		}
+	}
+	var picture struct {
+		DownloadCode string `json:"downloadCode"`
+	}
+	if json.Unmarshal(payload, &picture) != nil {
+		return im.MessageTypeImage, "", ""
+	}
+	return im.MessageTypeImage, "", strings.TrimSpace(picture.DownloadCode)
+}

internal/im/dingtalk/longconn_test.go

@@ -0,0 +1,37 @@
+package dingtalk
+
+import (
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/im"
+)
+
+func TestParseDingTalkStreamPayloadPicture(t *testing.T) {
+	messageType, content, fileKey := parseDingTalkStreamPayload("picture", map[string]interface{}{
+		"downloadCode": "stream-download-code",
+	}, "ignored text")
+
+	if messageType != im.MessageTypeImage {
+		t.Fatalf("MessageType = %q, want image", messageType)
+	}
+	if content != "" {
+		t.Fatalf("content = %q, want empty", content)
+	}
+	if fileKey != "stream-download-code" {
+		t.Fatalf("fileKey = %q", fileKey)
+	}
+}
+
+func TestParseDingTalkStreamPayloadText(t *testing.T) {
+	messageType, content, fileKey := parseDingTalkStreamPayload("text", nil, "  hello  ")
+	if messageType != im.MessageTypeText || content != "hello" || fileKey != "" {
+		t.Fatalf("got type=%q content=%q fileKey=%q", messageType, content, fileKey)
+	}
+}
+
+func TestParseDingTalkStreamPayloadPictureJSONString(t *testing.T) {
+	messageType, _, fileKey := parseDingTalkStreamPayload("picture", `{"downloadCode":"string-code"}`, "")
+	if messageType != im.MessageTypeImage || fileKey != "string-code" {
+		t.Fatalf("got type=%q fileKey=%q", messageType, fileKey)
+	}
+}

internal/im/image_qa_test.go

@@ -0,0 +1,83 @@
+package im
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+type staticImageDownloader struct {
+	body     string
+	fileName string
+}
+
+func (d staticImageDownloader) DownloadFile(context.Context, *IncomingMessage) (io.ReadCloser, string, error) {
+	return io.NopCloser(strings.NewReader(d.body)), d.fileName, nil
+}
+
+func TestPrepareIMImageForQAAndBuildRequest(t *testing.T) {
+	png := string([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1a, '\n'}) + strings.Repeat("x", 600)
+	msg := &IncomingMessage{
+		Platform:    PlatformDingtalk,
+		MessageType: MessageTypeImage,
+		MessageID:   "picture-message-1",
+		FileKey:     "temporary-download-code",
+	}
+
+	if err := prepareIMImageForQA(context.Background(), msg, staticImageDownloader{body: png, fileName: "question.png"}); err != nil {
+		t.Fatalf("prepareIMImageForQA: %v", err)
+	}
+	if len(msg.ImageURLs) != 1 || !strings.HasPrefix(msg.ImageURLs[0], "data:image/png;base64,") {
+		t.Fatalf("ImageURLs = %#v", msg.ImageURLs)
+	}
+	if len(msg.ImageReferences) != 1 {
+		t.Fatalf("ImageReferences = %#v", msg.ImageReferences)
+	}
+	if got := msg.ImageReferences[0]; got.URL != "dingtalk://message/picture-message-1" || got.Caption != "question.png" {
+		t.Fatalf("image reference = %#v", got)
+	}
+	if msg.Content == "" {
+		t.Fatal("image-only message should receive a default visual question")
+	}
+
+	req := buildIMQARequest(&types.Session{ID: "session-1"}, msg.Content, "assistant-1", "user-1", nil, nil, nil, msg.ImageURLs)
+	if len(req.ImageURLs) != 1 || req.ImageURLs[0] != msg.ImageURLs[0] {
+		t.Fatalf("request ImageURLs = %#v", req.ImageURLs)
+	}
+
+	audit := newIMMessageAuditContext(msg, "channel-1", nil)
+	userMessage := createIMUserMessagePayload("session-1", msg.Content, "request-1", audit)
+	assistantMessage := createIMAssistantMessagePayload("session-1", "request-1", audit)
+	if len(userMessage.Images) != 1 || userMessage.Images[0].URL != "dingtalk://message/picture-message-1" {
+		t.Fatalf("user message Images = %#v", userMessage.Images)
+	}
+	if len(assistantMessage.Images) != 0 {
+		t.Fatalf("assistant message should not duplicate input image: %#v", assistantMessage.Images)
+	}
+}
+
+func TestPrepareIMImageForQARejectsInvalidOrOversizedPayload(t *testing.T) {
+	tests := []struct {
+		name string
+		body string
+		want string
+	}{
+		{name: "not an image", body: strings.Repeat("plain text", 80), want: "unsupported image content type"},
+		{name: "oversized", body: strings.Repeat("x", maxIMImageBytes+1), want: "exceeds 10 MiB limit"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			msg := &IncomingMessage{Platform: PlatformDingtalk, MessageType: MessageTypeImage, MessageID: "image-1"}
+			err := prepareIMImageForQA(context.Background(), msg, staticImageDownloader{body: tt.body, fileName: "payload.bin"})
+			if err == nil || !strings.Contains(err.Error(), tt.want) {
+				t.Fatalf("error = %v, want containing %q", err, tt.want)
+			}
+			if len(msg.ImageURLs) != 0 {
+				t.Fatal("rejected image must not reach the model")
+			}
+		})
+	}
+}

internal/im/proactive_reply_test.go

@@ -0,0 +1,93 @@
+package im
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/gin-gonic/gin"
+)
+
+type recordingReplyAdapter struct {
+	incoming *IncomingMessage
+	reply    *ReplyMessage
+}
+
+func (a *recordingReplyAdapter) Platform() Platform                { return PlatformDingtalk }
+func (a *recordingReplyAdapter) VerifyCallback(*gin.Context) error { return nil }
+func (a *recordingReplyAdapter) ParseCallback(*gin.Context) (*IncomingMessage, error) {
+	return nil, nil
+}
+func (a *recordingReplyAdapter) SendReply(_ context.Context, incoming *IncomingMessage, reply *ReplyMessage) error {
+	a.incoming = incoming
+	a.reply = reply
+	return nil
+}
+func (a *recordingReplyAdapter) HandleURLVerification(*gin.Context) bool { return false }
+
+func TestSendReplyToSource(t *testing.T) {
+	adapter := &recordingReplyAdapter{}
+	service := &Service{channels: map[string]*channelState{
+		"channel-1": {
+			Adapter: adapter,
+			Channel: &IMChannel{ID: "channel-1", Platform: "dingtalk", Enabled: true},
+		},
+	}}
+
+	err := service.SendReplyToSource(context.Background(), ReplyTarget{
+		ChannelID: "channel-1",
+		UserID:    "staff-1",
+		ChatID:    "conversation-1",
+		ChatType:  ChatTypeGroup,
+	}, "技术支持回复内容")
+	if err != nil {
+		t.Fatalf("SendReplyToSource: %v", err)
+	}
+	if adapter.incoming == nil || adapter.incoming.UserID != "staff-1" || adapter.incoming.ChatID != "conversation-1" || adapter.incoming.ChatType != ChatTypeGroup {
+		t.Fatalf("incoming = %#v", adapter.incoming)
+	}
+	if adapter.reply == nil || adapter.reply.Content != "技术支持回复内容" || !adapter.reply.IsFinal {
+		t.Fatalf("reply = %#v", adapter.reply)
+	}
+}
+
+func TestSendReplyToSourceRejectsUnavailableChannel(t *testing.T) {
+	service := &Service{channels: map[string]*channelState{}}
+	if err := service.SendReplyToSource(context.Background(), ReplyTarget{ChannelID: "missing"}, "reply"); err == nil {
+		t.Fatal("expected missing channel error")
+	}
+}
+
+func TestSendSupportNotificationUsesConfiguredGroup(t *testing.T) {
+	adapter := &recordingReplyAdapter{}
+	service := &Service{channels: map[string]*channelState{
+		"channel-1": {
+			Adapter: adapter,
+			Channel: &IMChannel{
+				ID:          "channel-1",
+				Platform:    "dingtalk",
+				Enabled:     true,
+				Credentials: types.JSON(`{"support_group_open_conversation_id":"support-group-1"}`),
+			},
+		},
+	}}
+
+	if err := service.SendSupportNotification(context.Background(), "channel-1", "新工单：客户未解决"); err != nil {
+		t.Fatalf("SendSupportNotification: %v", err)
+	}
+	if adapter.incoming == nil || adapter.incoming.ChatType != ChatTypeGroup || adapter.incoming.ChatID != "support-group-1" {
+		t.Fatalf("support target = %#v", adapter.incoming)
+	}
+	if adapter.reply == nil || adapter.reply.Content != "新工单：客户未解决" {
+		t.Fatalf("support reply = %#v", adapter.reply)
+	}
+}
+
+func TestSendSupportNotificationRequiresConfiguredGroup(t *testing.T) {
+	service := &Service{channels: map[string]*channelState{
+		"channel-1": {Adapter: &recordingReplyAdapter{}, Channel: &IMChannel{ID: "channel-1", Platform: "dingtalk", Enabled: true}},
+	}}
+	if err := service.SendSupportNotification(context.Background(), "channel-1", "ticket"); err == nil {
+		t.Fatal("expected missing support group configuration error")
+	}
+}

internal/im/service.go

@@ -3,11 +3,14 @@ package im
 import (
 	"bytes"
 	"context"
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"io"
 	"mime/multipart"
+	"net/http"
 	"net/textproto"
+	"net/url"
 	"os"
 	"regexp"
 	"strings"
@@ -483,12 +486,17 @@ func buildIMQARequest(
 	customAgent *types.CustomAgent,
 	kbIDs []string,
 	quote *QuotedMessage,
+	imageURLs ...[]string,
 ) *types.QARequest {
 	// WebSearchEnabled: the web handler passes this per-request from the
 	// frontend toggle; for IM channels the user has no per-message toggle,
 	// so we derive it from the agent config (the single source of truth).
 	webSearchEnabled := customAgent != nil && customAgent.Config.WebSearchEnabled
 	quotedContext := formatQuotedContext(quote)
+	var images []string
+	if len(imageURLs) > 0 {
+		images = append([]string(nil), imageURLs[0]...)
+	}
 	return &types.QARequest{
 		Session:            session,
 		Query:              query,
@@ -498,9 +506,62 @@ func buildIMQARequest(
 		UserMessageID:      userMessageID,
 		WebSearchEnabled:   webSearchEnabled,
 		QuotedContext:      quotedContext,
+		ImageURLs:          images,
 	}
 }
 
+const maxIMImageBytes = 10 << 20
+
+func prepareIMImageForQA(ctx context.Context, msg *IncomingMessage, downloader FileDownloader) error {
+	if msg == nil || downloader == nil {
+		return fmt.Errorf("image message and downloader are required")
+	}
+	reader, fileName, err := downloader.DownloadFile(ctx, msg)
+	if err != nil {
+		return fmt.Errorf("download image: %w", err)
+	}
+	defer reader.Close()
+
+	data, err := io.ReadAll(io.LimitReader(reader, maxIMImageBytes+1))
+	if err != nil {
+		return fmt.Errorf("read image: %w", err)
+	}
+	if len(data) == 0 {
+		return fmt.Errorf("downloaded image is empty")
+	}
+	if len(data) > maxIMImageBytes {
+		return fmt.Errorf("image exceeds %d MiB limit", maxIMImageBytes>>20)
+	}
+	contentType := strings.ToLower(http.DetectContentType(data))
+	switch contentType {
+	case "image/jpeg", "image/png", "image/gif", "image/webp":
+	default:
+		return fmt.Errorf("unsupported image content type %q", contentType)
+	}
+
+	msg.FileName = fileName
+	msg.ImageURLs = []string{"data:" + contentType + ";base64," + base64.StdEncoding.EncodeToString(data)}
+	referenceID := strings.TrimSpace(msg.MessageID)
+	if referenceID == "" {
+		referenceID = "unknown"
+	}
+	msg.ImageReferences = types.MessageImages{{
+		URL:     string(msg.Platform) + "://message/" + url.PathEscape(referenceID),
+		Caption: fileName,
+	}}
+	if strings.TrimSpace(msg.Content) == "" {
+		msg.Content = "请识别这张图片的内容，并回答图片中的问题。"
+	}
+	return nil
+}
+
+func auditImageURLs(audit *imMessageAuditContext) []string {
+	if audit == nil {
+		return nil
+	}
+	return append([]string(nil), audit.imageURLs...)
+}
+
 func buildIMLastRequestState(agentID string, customAgent *types.CustomAgent, kbIDs []string) *types.SessionLastRequestState {
 	state := &types.SessionLastRequestState{
 		AgentID:          agentID,
@@ -521,8 +582,56 @@ func buildIMLastRequestState(agentID string, customAgent *types.CustomAgent, kbI
 	return state
 }
 
-func createIMUserMessagePayload(sessionID, content, requestID string) *types.Message {
+type imMessageAuditContext struct {
-	return &types.Message{
+	platform         string
+	channelID        string
+	messageID        string
+	deliveryID       string
+	userID           string
+	chatID           string
+	threadID         string
+	knowledgeBaseIDs types.MessageKnowledgeBaseIDs
+	imageReferences  types.MessageImages
+	imageURLs        []string
+}
+
+func newIMMessageAuditContext(msg *IncomingMessage, channelID string, kbIDs []string) *imMessageAuditContext {
+	audit := &imMessageAuditContext{
+		channelID:        channelID,
+		knowledgeBaseIDs: append(types.MessageKnowledgeBaseIDs(nil), kbIDs...),
+	}
+	if msg == nil {
+		return audit
+	}
+	audit.platform = string(msg.Platform)
+	audit.messageID = msg.MessageID
+	if msg.Extra != nil {
+		audit.deliveryID = msg.Extra["dingtalk_out_track_id"]
+	}
+	audit.userID = msg.UserID
+	audit.chatID = msg.ChatID
+	audit.threadID = msg.ThreadID
+	audit.imageReferences = append(types.MessageImages(nil), msg.ImageReferences...)
+	audit.imageURLs = append([]string(nil), msg.ImageURLs...)
+	return audit
+}
+
+func applyIMMessageAudit(message *types.Message, audit *imMessageAuditContext) {
+	if message == nil || audit == nil {
+		return
+	}
+	message.SourcePlatform = audit.platform
+	message.SourceChannelID = audit.channelID
+	message.SourceMessageID = audit.messageID
+	message.SourceDeliveryID = audit.deliveryID
+	message.SourceUserID = audit.userID
+	message.SourceChatID = audit.chatID
+	message.SourceThreadID = audit.threadID
+	message.SearchedKnowledgeBaseIDs = append(types.MessageKnowledgeBaseIDs(nil), audit.knowledgeBaseIDs...)
+}
+
+func createIMUserMessagePayload(sessionID, content, requestID string, audit *imMessageAuditContext) *types.Message {
+	message := &types.Message{
 		SessionID:   sessionID,
 		Role:        "user",
 		Content:     content,
@@ -531,10 +640,15 @@ func createIMUserMessagePayload(sessionID, content, requestID string) *types.Mes
 		IsCompleted: true,
 		Channel:     "im",
 	}
+	applyIMMessageAudit(message, audit)
+	if audit != nil {
+		message.Images = append(types.MessageImages(nil), audit.imageReferences...)
+	}
+	return message
 }
 
-func createIMAssistantMessagePayload(sessionID, requestID string) *types.Message {
+func createIMAssistantMessagePayload(sessionID, requestID string, audit *imMessageAuditContext) *types.Message {
-	return &types.Message{
+	message := &types.Message{
 		SessionID:   sessionID,
 		Role:        "assistant",
 		RequestID:   requestID,
@@ -542,6 +656,8 @@ func createIMAssistantMessagePayload(sessionID, requestID string) *types.Message
 		IsCompleted: false,
 		Channel:     "im",
 	}
+	applyIMMessageAudit(message, audit)
+	return message
 }
 
 func collectIMKnowledgeReferences(dst *[]*types.SearchResult, refs interface{}) {
@@ -577,6 +693,10 @@ func applyIMCompleteDataToMessage(msg *types.Message, data event.AgentCompleteDa
 	}
 	msg.IsCompleted = true
 	msg.AgentDurationMs = data.TotalDurationMs
+	if data.TokenUsage != nil {
+		usage := *data.TokenUsage
+		msg.TokenUsage = &usage
+	}
 	if len(data.KnowledgeRefs) > 0 {
 		refs := make([]*types.SearchResult, 0, len(data.KnowledgeRefs))
 		collectIMKnowledgeReferences(&refs, data.KnowledgeRefs)
@@ -589,6 +709,14 @@ func applyIMCompleteDataToMessage(msg *types.Message, data event.AgentCompleteDa
 	}
 }
 
+func applyIMFinalAnswerDataToMessage(msg *types.Message, data event.AgentFinalAnswerData) {
+	if msg == nil || data.TokenUsage == nil {
+		return
+	}
+	usage := *data.TokenUsage
+	msg.TokenUsage = &usage
+}
+
 // waitForIMAgentComplete blocks until EventAgentComplete, ctx cancellation, or timeout.
 func waitForIMAgentComplete(ctx context.Context, completeDone <-chan struct{}, sessionID string) {
 	timer := time.NewTimer(agentCompleteWaitTimeout)
@@ -1125,6 +1253,70 @@ func (s *Service) GetChannelAdapter(channelID string) (Adapter, *IMChannel, bool
 	return cs.Adapter, cs.Channel, true
 }
 
+// ReplyTarget identifies the original customer conversation for a proactive reply.
+type ReplyTarget struct {
+	ChannelID string
+	UserID    string
+	ChatID    string
+	ChatType  ChatType
+	ThreadID  string
+}
+
+// SendReplyToSource delivers a support reply through the configured platform adapter.
+// It intentionally does not depend on a short-lived session webhook; adapters can use
+// their authenticated proactive-message API when Extra is empty.
+func (s *Service) SendReplyToSource(ctx context.Context, target ReplyTarget, content string) error {
+	if strings.TrimSpace(target.ChannelID) == "" {
+		return fmt.Errorf("channel ID is required")
+	}
+	content = strings.TrimSpace(content)
+	if content == "" {
+		return fmt.Errorf("reply content is required")
+	}
+	adapter, channel, ok := s.GetChannelAdapter(target.ChannelID)
+	if !ok || adapter == nil || channel == nil {
+		return fmt.Errorf("IM channel %s is not running", target.ChannelID)
+	}
+	if !channel.Enabled {
+		return fmt.Errorf("IM channel %s is disabled", target.ChannelID)
+	}
+	chatType := target.ChatType
+	if chatType == "" {
+		chatType = ChatTypeDirect
+	}
+	incoming := &IncomingMessage{
+		Platform:  Platform(channel.Platform),
+		UserID:    strings.TrimSpace(target.UserID),
+		ChatID:    strings.TrimSpace(target.ChatID),
+		ChatType:  chatType,
+		ThreadID:  strings.TrimSpace(target.ThreadID),
+		MessageID: "support:" + uuid.New().String(),
+	}
+	return adapter.SendReply(ctx, incoming, &ReplyMessage{Content: content, IsFinal: true})
+}
+
+// SendSupportNotification pushes a newly escalated ticket to the support group
+// configured in the source channel credentials. The group identifier is never logged.
+func (s *Service) SendSupportNotification(ctx context.Context, channelID, content string) error {
+	_, channel, ok := s.GetChannelAdapter(channelID)
+	if !ok || channel == nil {
+		return fmt.Errorf("IM channel %s is not running", channelID)
+	}
+	credentials, err := ParseCredentials(channel.Credentials)
+	if err != nil {
+		return fmt.Errorf("parse support group configuration: %w", err)
+	}
+	supportGroupID := strings.TrimSpace(GetString(credentials, "support_group_open_conversation_id"))
+	if supportGroupID == "" {
+		return fmt.Errorf("support group is not configured for IM channel %s", channelID)
+	}
+	return s.SendReplyToSource(ctx, ReplyTarget{
+		ChannelID: channelID,
+		ChatID:    supportGroupID,
+		ChatType:  ChatTypeGroup,
+	}, content)
+}
+
 // GetChannelByID loads a channel from the database.
 func (s *Service) GetChannelByID(channelID string) (*IMChannel, error) {
 	var ch IMChannel
@@ -1238,7 +1430,7 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann
 	// ── File/Image message shortcut ──
 	// If the message is a file or image and the channel has a knowledge_base_id configured,
 	// handle it separately without entering the QA pipeline.
-	if (msg.MessageType == MessageTypeFile || msg.MessageType == MessageTypeImage) && channel.KnowledgeBaseID != "" {
+	if msg.MessageType == MessageTypeFile && channel.KnowledgeBaseID != "" {
 		return s.handleFileMessage(ctx, msg, adapter, channel)
 	}
 
@@ -1246,7 +1438,7 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann
 	// If the message is an image/file/video but has no text content, the QA pipeline
 	// cannot do anything useful (no vision support in IM yet). Sending an empty query
 	// to KB retrieval would return irrelevant results and cause hallucination.
-	if msg.Content == "" && (msg.MessageType == MessageTypeImage || msg.MessageType == MessageTypeFile) {
+	if msg.Content == "" && msg.MessageType == MessageTypeFile {
 		logger.Infof(ctx, "[IM] Skipping QA for non-text message without content: type=%s", msg.MessageType)
 		if err := adapter.SendReply(ctx, msg, &ReplyMessage{
 			Content: "当前渠道未配置文件知识库，无法处理图片/文件消息。请在渠道设置中配置文件知识库后再发送，或直接用文字描述您的问题。",
@@ -1282,6 +1474,26 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann
 		}
 	}
 
+	if msg.MessageType == MessageTypeImage {
+		if customAgent == nil || !customAgent.Config.ImageUploadEnabled {
+			if channel.KnowledgeBaseID != "" {
+				return s.handleFileMessage(ctx, msg, adapter, channel)
+			}
+			return adapter.SendReply(ctx, msg, &ReplyMessage{
+				Content: "当前机器人未启用图片识别，请在智能体设置中开启图片上传并配置视觉模型。",
+				IsFinal: true,
+			})
+		}
+		downloader, ok := adapter.(FileDownloader)
+		if !ok {
+			return adapter.SendReply(ctx, msg, &ReplyMessage{Content: "当前钉钉渠道暂不支持图片下载。", IsFinal: true})
+		}
+		if err := prepareIMImageForQA(sessionCtx, msg, downloader); err != nil {
+			logger.Warnf(ctx, "[IM] Failed to prepare image for QA: %v", err)
+			return adapter.SendReply(ctx, msg, &ReplyMessage{Content: "图片下载或格式校验失败，请发送10MB以内的 JPG、PNG、GIF 或 WebP 图片。", IsFinal: true})
+		}
+	}
+
 	// ── Slash-command dispatch ──
 	// Commands are handled before the QA pipeline so they respond instantly.
 	if cmd, args, ok := s.cmdRegistry.Parse(msg.Content); ok {
@@ -1417,6 +1629,11 @@ func (s *Service) executeQARequest(req *qaRequest) {
 
 	// kbIDs is left empty so the QA pipeline resolves them from the agent config.
 	var kbIDs []string
+	auditedKBIDs := kbIDs
+	if len(auditedKBIDs) == 0 && req.agent != nil {
+		auditedKBIDs = req.agent.Config.KnowledgeBases
+	}
+	audit := newIMMessageAuditContext(req.msg, req.channelID, auditedKBIDs)
 
 	// Determine output mode from channel config.
 	streamDisabled := req.channel.OutputMode == "full"
@@ -1424,7 +1641,7 @@ func (s *Service) executeQARequest(req *qaRequest) {
 	// If the adapter supports streaming and output is not "full", use streaming.
 	if !streamDisabled {
 		if streamer, ok := req.adapter.(StreamSender); ok {
-			if err := s.handleMessageStream(ctx, req.msg, req.session, req.agent, kbIDs, streamer, req.adapter, req.userKey, req.tenant); err != nil {
+			if err := s.handleMessageStream(ctx, req.msg, req.session, req.agent, kbIDs, streamer, req.adapter, req.userKey, req.tenant, audit); err != nil {
 				logger.Errorf(ctx, "[IM] Stream QA failed: %v", err)
 			}
 			return
@@ -1432,7 +1649,7 @@ func (s *Service) executeQARequest(req *qaRequest) {
 	}
 
 	// Non-streaming fallback: collect full answer then send.
-	answer, err := s.runQA(ctx, req.session, req.msg.Content, req.agent, kbIDs, req.userKey, req.msg.Quote)
+	answer, err := s.runQA(ctx, req.session, req.msg.Content, req.agent, kbIDs, req.userKey, req.msg.Quote, audit)
 	if err != nil {
 		logger.Errorf(ctx, "[IM] QA failed: %v, sending fallback reply", err)
 		answer = "抱歉，处理您的问题时出现了异常，请稍后再试。"
@@ -1855,12 +2072,12 @@ func briefToolSummary(output string) string {
 // handleMessageStream runs the QA pipeline and streams answer chunks to the IM platform
 // in real-time via the StreamSender interface. Chunks are batched at streamFlushInterval
 // to avoid API rate-limiting.
-func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, session *types.Session, customAgent *types.CustomAgent, kbIDs []string, streamer StreamSender, adapter Adapter, userKey string, tenant *types.Tenant) error {
+func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage, session *types.Session, customAgent *types.CustomAgent, kbIDs []string, streamer StreamSender, adapter Adapter, userKey string, tenant *types.Tenant, audit *imMessageAuditContext) error {
 	// Start the stream on the IM platform (e.g., create Feishu streaming card)
 	streamID, err := streamer.StartStream(ctx, msg)
 	if err != nil {
 		logger.Warnf(ctx, "[IM] StartStream failed, falling back to non-streaming: %v", err)
-		return s.fallbackNonStream(ctx, msg, session, customAgent, kbIDs, adapter, userKey, tenant)
+		return s.fallbackNonStream(ctx, msg, session, customAgent, kbIDs, adapter, userKey, tenant, audit)
 	}
 
 	// Prepare the QA pipeline
@@ -1951,6 +2168,7 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 		}
 
 		bufMu.Lock()
+		applyIMFinalAnswerDataToMessage(assistantMsg, data)
 		if useAgent && !agentDone {
 			if data.Content != "" {
 				agentLiveAnswer.WriteString(data.Content)
@@ -2104,13 +2322,13 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 	requestID := uuid.New().String()
 
 	// Create user message
-	userMsg, err := s.messageService.CreateMessage(qaCtx, createIMUserMessagePayload(session.ID, msg.Content, requestID))
+	userMsg, err := s.messageService.CreateMessage(qaCtx, createIMUserMessagePayload(session.ID, msg.Content, requestID, audit))
 	if err != nil {
 		return fmt.Errorf("create user message: %w", err)
 	}
 
 	// Create placeholder assistant message
-	assistantMsg, err = s.messageService.CreateMessage(qaCtx, createIMAssistantMessagePayload(session.ID, requestID))
+	assistantMsg, err = s.messageService.CreateMessage(qaCtx, createIMAssistantMessagePayload(session.ID, requestID, audit))
 	if err != nil {
 		return fmt.Errorf("create assistant message: %w", err)
 	}
@@ -2132,7 +2350,7 @@ func (s *Service) handleMessageStream(ctx context.Context, msg *IncomingMessage,
 	// Run QA async
 	go func() {
 		var err error
-		req := buildIMQARequest(session, msg.Content, assistantMsg.ID, userMsg.ID, customAgent, kbIDs, msg.Quote)
+		req := buildIMQARequest(session, msg.Content, assistantMsg.ID, userMsg.ID, customAgent, kbIDs, msg.Quote, msg.ImageURLs)
 		if req.QuotedContext != "" {
 			logger.Debugf(qaCtx, "[IM] QuotedContext set: length=%d", len(req.QuotedContext))
 		}
@@ -2246,8 +2464,8 @@ loop:
 }
 
 // fallbackNonStream is used when streaming initialization fails.
-func (s *Service) fallbackNonStream(ctx context.Context, msg *IncomingMessage, session *types.Session, customAgent *types.CustomAgent, kbIDs []string, adapter Adapter, userKey string, tenant *types.Tenant) error {
+func (s *Service) fallbackNonStream(ctx context.Context, msg *IncomingMessage, session *types.Session, customAgent *types.CustomAgent, kbIDs []string, adapter Adapter, userKey string, tenant *types.Tenant, audit *imMessageAuditContext) error {
-	answer, err := s.runQA(ctx, session, msg.Content, customAgent, kbIDs, userKey, msg.Quote)
+	answer, err := s.runQA(ctx, session, msg.Content, customAgent, kbIDs, userKey, msg.Quote, audit)
 	if err != nil {
 		logger.Errorf(ctx, "[IM] QA fallback failed: %v", err)
 		answer = "抱歉，处理您的问题时出现了异常，请稍后再试。"
@@ -2257,7 +2475,7 @@ func (s *Service) fallbackNonStream(ctx context.Context, msg *IncomingMessage, s
 }
 
 // runQA executes the WeKnora QA pipeline and returns the full answer text.
-func (s *Service) runQA(ctx context.Context, session *types.Session, query string, customAgent *types.CustomAgent, kbIDs []string, userKey string, quote *QuotedMessage) (string, error) {
+func (s *Service) runQA(ctx context.Context, session *types.Session, query string, customAgent *types.CustomAgent, kbIDs []string, userKey string, quote *QuotedMessage, audit *imMessageAuditContext) (string, error) {
 	// Cancellable context (no hard deadline): each agent round has its own
 	// LLMCallTimeout. The context can still be cancelled by /stop.
 	ctx, cancel := context.WithCancel(ctx)
@@ -2268,6 +2486,7 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 	// Thread-safe answer collection
 	var answerMu sync.Mutex
 	var answerBuilder strings.Builder
+	var assistantMsg *types.Message
 	var qaErr error
 	done := make(chan struct{})
 	completeDone := make(chan struct{})
@@ -2282,6 +2501,7 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 			return nil
 		}
 		answerMu.Lock()
+		applyIMFinalAnswerDataToMessage(assistantMsg, data)
 		answerBuilder.WriteString(data.Content)
 		answerMu.Unlock()
 		if data.Done {
@@ -2311,13 +2531,13 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 	requestID := uuid.New().String()
 
 	// Create user message so it appears in conversation history
-	userMsg, err := s.messageService.CreateMessage(ctx, createIMUserMessagePayload(session.ID, query, requestID))
+	userMsg, err := s.messageService.CreateMessage(ctx, createIMUserMessagePayload(session.ID, query, requestID, audit))
 	if err != nil {
 		return "", fmt.Errorf("create user message: %w", err)
 	}
 
 	// Create a placeholder assistant message
-	assistantMsg, err := s.messageService.CreateMessage(ctx, createIMAssistantMessagePayload(session.ID, requestID))
+	assistantMsg, err = s.messageService.CreateMessage(ctx, createIMAssistantMessagePayload(session.ID, requestID, audit))
 	if err != nil {
 		return "", fmt.Errorf("create assistant message: %w", err)
 	}
@@ -2365,7 +2585,7 @@ func (s *Service) runQA(ctx context.Context, session *types.Session, query strin
 	// Run QA async
 	go func() {
 		var err error
-		req := buildIMQARequest(session, query, assistantMsg.ID, userMsg.ID, customAgent, kbIDs, quote)
+		req := buildIMQARequest(session, query, assistantMsg.ID, userMsg.ID, customAgent, kbIDs, quote, auditImageURLs(audit))
 		if req.QuotedContext != "" {
 			logger.Debugf(ctx, "[IM] QuotedContext set: length=%d", len(req.QuotedContext))
 		}

internal/im/session_test.go

@@ -202,8 +202,17 @@ func TestBuildIMLastRequestStateKeepsExplicitKBs(t *testing.T) {
 }
 
 func TestCreateIMMessagePayloadsShareRequestShape(t *testing.T) {
-	userMsg := createIMUserMessagePayload("session-1", "hello", "request-1")
+	incoming := &IncomingMessage{
-	assistantMsg := createIMAssistantMessagePayload("session-1", "request-1")
+		Platform:  PlatformDingtalk,
+		UserID:    "platform-user-1",
+		ChatID:    "platform-chat-1",
+		ThreadID:  "platform-thread-1",
+		MessageID: "platform-message-1",
+		Extra:     map[string]string{"dingtalk_out_track_id": "card-track-1"},
+	}
+	audit := newIMMessageAuditContext(incoming, "channel-1", []string{"kb-1", "kb-2"})
+	userMsg := createIMUserMessagePayload("session-1", "hello", "request-1", audit)
+	assistantMsg := createIMAssistantMessagePayload("session-1", "request-1", audit)
 
 	if userMsg.SessionID != "session-1" || assistantMsg.SessionID != "session-1" {
 		t.Fatalf("SessionID mismatch: user=%q assistant=%q", userMsg.SessionID, assistantMsg.SessionID)
@@ -229,6 +238,20 @@ func TestCreateIMMessagePayloadsShareRequestShape(t *testing.T) {
 	if assistantMsg.Content != "" {
 		t.Fatalf("assistant placeholder content = %q, want empty", assistantMsg.Content)
 	}
+	for _, message := range []*types.Message{userMsg, assistantMsg} {
+		if message.SourcePlatform != "dingtalk" || message.SourceChannelID != "channel-1" {
+			t.Fatalf("source channel metadata = platform %q channel %q", message.SourcePlatform, message.SourceChannelID)
+		}
+		if message.SourceMessageID != "platform-message-1" || message.SourceUserID != "platform-user-1" || message.SourceDeliveryID != "card-track-1" {
+			t.Fatalf("source actor metadata = message %q user %q", message.SourceMessageID, message.SourceUserID)
+		}
+		if message.SourceChatID != "platform-chat-1" || message.SourceThreadID != "platform-thread-1" {
+			t.Fatalf("source conversation metadata = chat %q thread %q", message.SourceChatID, message.SourceThreadID)
+		}
+		if !reflect.DeepEqual(message.SearchedKnowledgeBaseIDs, types.MessageKnowledgeBaseIDs{"kb-1", "kb-2"}) {
+			t.Fatalf("searched knowledge bases = %#v", message.SearchedKnowledgeBaseIDs)
+		}
+	}
 }
 
 func TestApplyIMCompleteDataToMessage(t *testing.T) {
@@ -239,11 +262,13 @@ func TestApplyIMCompleteDataToMessage(t *testing.T) {
 		ReasoningContent: "thinking",
 	}}
 
+	usage := &types.TokenUsage{PromptTokens: 100, CompletionTokens: 20, TotalTokens: 120, CachedTokens: 40}
 	applyIMCompleteDataToMessage(msg, event.AgentCompleteData{
 		MessageID:       "assistant-1",
 		TotalDurationMs: 1234,
 		KnowledgeRefs:   []interface{}{ref},
 		AgentSteps:      steps,
+		TokenUsage:      usage,
 	})
 
 	if !msg.IsCompleted {
@@ -258,6 +283,20 @@ func TestApplyIMCompleteDataToMessage(t *testing.T) {
 	if len(msg.AgentSteps) != 1 || msg.AgentSteps[0].ReasoningContent != "thinking" {
 		t.Fatalf("AgentSteps = %#v, want one thinking step", msg.AgentSteps)
 	}
+	if !reflect.DeepEqual(msg.TokenUsage, usage) {
+		t.Fatalf("TokenUsage = %#v, want %#v", msg.TokenUsage, usage)
+	}
+}
+
+func TestApplyIMFinalAnswerUsageToMessage(t *testing.T) {
+	msg := &types.Message{ID: "assistant-1", Role: "assistant"}
+	usage := &types.TokenUsage{PromptTokens: 80, CompletionTokens: 16, TotalTokens: 96}
+
+	applyIMFinalAnswerDataToMessage(msg, event.AgentFinalAnswerData{Done: true, TokenUsage: usage})
+
+	if !reflect.DeepEqual(msg.TokenUsage, usage) {
+		t.Fatalf("TokenUsage = %#v, want %#v", msg.TokenUsage, usage)
+	}
 }
 
 func TestPickIMStoredAnswerPrefersFirstNonEmpty(t *testing.T) {

internal/types/message.go

@@ -150,6 +150,10 @@ func (m *MessageAttachments) Scan(value interface{}) error {
 // MentionedItems is a slice of MentionedItem for database storage
 type MentionedItems []MentionedItem
 
+// MessageKnowledgeBaseIDs stores the knowledge-base scope searched for a message.
+// It is kept even when retrieval returns no references so the request remains auditable.
+type MessageKnowledgeBaseIDs []string
+
 // Value implements the driver.Valuer interface for database serialization
 func (m MentionedItems) Value() (driver.Value, error) {
 	if m == nil {
@@ -216,6 +220,24 @@ type Message struct {
 	RenderedContent string `json:"-" gorm:"type:text;column:rendered_content;default:''"`
 	// Channel indicates the source channel of this message (e.g., "web", "api", "im")
 	Channel string `json:"channel,omitempty" gorm:"type:varchar(50);default:''"`
+	// SourcePlatform is the concrete external platform for IM messages (for example, dingtalk).
+	SourcePlatform string `json:"source_platform,omitempty" gorm:"type:varchar(20);default:''"`
+	// SourceChannelID identifies the configured WeKnora IM channel that received the message.
+	SourceChannelID string `json:"source_channel_id,omitempty" gorm:"type:varchar(36);default:''"`
+	// SourceMessageID is the platform message identifier used for end-to-end correlation.
+	SourceMessageID string `json:"source_message_id,omitempty" gorm:"type:varchar(128);default:''"`
+	// SourceDeliveryID correlates outbound cards/messages with later platform callbacks.
+	SourceDeliveryID string `json:"source_delivery_id,omitempty" gorm:"type:varchar(128);default:''"`
+	// SourceUserID records the platform user for this individual message.
+	SourceUserID string `json:"source_user_id,omitempty" gorm:"type:varchar(128);default:''"`
+	// SourceChatID records the platform conversation or group identifier.
+	SourceChatID string `json:"source_chat_id,omitempty" gorm:"type:varchar(128);default:''"`
+	// SourceThreadID records the platform thread identifier when present.
+	SourceThreadID string `json:"source_thread_id,omitempty" gorm:"type:varchar(128);default:''"`
+	// SearchedKnowledgeBaseIDs records the request scope even when retrieval has zero hits.
+	SearchedKnowledgeBaseIDs MessageKnowledgeBaseIDs `json:"searched_knowledge_base_ids,omitempty" gorm:"serializer:json;type:jsonb;column:searched_knowledge_base_ids"`
+	// TokenUsage records answer-generation model usage for the assistant message.
+	TokenUsage *TokenUsage `json:"token_usage,omitempty" gorm:"serializer:json;type:jsonb;column:token_usage"`
 	// KnowledgeID links this message to a Knowledge entry in the chat history knowledge base
 	// Used for vector search indexing: when set, the message content has been indexed as a Knowledge passage
 	KnowledgeID string `json:"knowledge_id,omitempty" gorm:"type:varchar(36);index"`

internal/types/message_audit_migration_test.go

@@ -0,0 +1,58 @@
+package types
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestMessageAuditMigrationCoversModelFields(t *testing.T) {
+	upPath := "../../migrations/versioned/000064_message_audit.up.sql"
+	downPath := "../../migrations/versioned/000064_message_audit.down.sql"
+
+	up, err := os.ReadFile(upPath)
+	if err != nil {
+		t.Fatalf("read up migration: %v", err)
+	}
+	down, err := os.ReadFile(downPath)
+	if err != nil {
+		t.Fatalf("read down migration: %v", err)
+	}
+
+	columns := []string{
+		"source_platform",
+		"source_channel_id",
+		"source_message_id",
+		"source_delivery_id",
+		"source_user_id",
+		"source_chat_id",
+		"source_thread_id",
+		"searched_knowledge_base_ids",
+		"token_usage",
+	}
+	upSQL := strings.ToLower(string(up))
+	downSQL := strings.ToLower(string(down))
+	for _, column := range columns {
+		if !strings.Contains(upSQL, column) {
+			t.Errorf("up migration missing column %q", column)
+		}
+		if !strings.Contains(downSQL, column) {
+			t.Errorf("down migration missing column %q", column)
+		}
+	}
+
+	indexes := []string{
+		"idx_messages_request_role",
+		"idx_messages_source_channel",
+		"idx_messages_source_message",
+		"idx_messages_source_user",
+	}
+	for _, index := range indexes {
+		if !strings.Contains(upSQL, index) {
+			t.Errorf("up migration missing index %q", index)
+		}
+		if !strings.Contains(downSQL, index) {
+			t.Errorf("down migration missing index %q", index)
+		}
+	}
+}

internal/types/message_audit_test.go

@@ -0,0 +1,53 @@
+package types
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+)
+
+func TestMessageCarriesAuditMetadata(t *testing.T) {
+	original := Message{
+		ID:                       "message-1",
+		SourcePlatform:           "dingtalk",
+		SourceChannelID:          "channel-1",
+		SourceMessageID:          "platform-message-1",
+		SourceDeliveryID:         "card-track-1",
+		SourceUserID:             "user-1",
+		SourceChatID:             "chat-1",
+		SourceThreadID:           "thread-1",
+		SearchedKnowledgeBaseIDs: MessageKnowledgeBaseIDs{"kb-1", "kb-2"},
+		TokenUsage: &TokenUsage{
+			PromptTokens:     100,
+			CompletionTokens: 20,
+			TotalTokens:      120,
+			CachedTokens:     40,
+		},
+	}
+
+	payload, err := json.Marshal(original)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+
+	var decoded Message
+	if err := json.Unmarshal(payload, &decoded); err != nil {
+		t.Fatalf("json.Unmarshal() error = %v", err)
+	}
+
+	if decoded.SourcePlatform != original.SourcePlatform ||
+		decoded.SourceChannelID != original.SourceChannelID ||
+		decoded.SourceMessageID != original.SourceMessageID ||
+		decoded.SourceDeliveryID != original.SourceDeliveryID ||
+		decoded.SourceUserID != original.SourceUserID ||
+		decoded.SourceChatID != original.SourceChatID ||
+		decoded.SourceThreadID != original.SourceThreadID {
+		t.Fatalf("source metadata was not preserved: got %+v", decoded)
+	}
+	if !reflect.DeepEqual(decoded.SearchedKnowledgeBaseIDs, original.SearchedKnowledgeBaseIDs) {
+		t.Fatalf("searched knowledge bases = %#v, want %#v", decoded.SearchedKnowledgeBaseIDs, original.SearchedKnowledgeBaseIDs)
+	}
+	if !reflect.DeepEqual(decoded.TokenUsage, original.TokenUsage) {
+		t.Fatalf("token usage = %#v, want %#v", decoded.TokenUsage, original.TokenUsage)
+	}
+}

migrations/sqlite/000000_init.up.sql

@@ -168,6 +168,15 @@ CREATE TABLE IF NOT EXISTS messages (
     is_completed BOOLEAN NOT NULL DEFAULT 0,
     is_fallback BOOLEAN NOT NULL DEFAULT 0,
     channel VARCHAR(50) NOT NULL DEFAULT '',
+    source_platform VARCHAR(20) NOT NULL DEFAULT '',
+    source_channel_id VARCHAR(36) NOT NULL DEFAULT '',
+    source_message_id VARCHAR(128) NOT NULL DEFAULT '',
+    source_delivery_id VARCHAR(128) NOT NULL DEFAULT '',
+    source_user_id VARCHAR(128) NOT NULL DEFAULT '',
+    source_chat_id VARCHAR(128) NOT NULL DEFAULT '',
+    source_thread_id VARCHAR(128) NOT NULL DEFAULT '',
+    searched_knowledge_base_ids TEXT NOT NULL DEFAULT '[]',
+    token_usage TEXT,
     agent_duration_ms INTEGER DEFAULT 0,
     knowledge_id VARCHAR(36),
     created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
@@ -177,6 +186,11 @@ CREATE TABLE IF NOT EXISTS messages (
 
 CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id);
 CREATE INDEX IF NOT EXISTS idx_messages_knowledge_id ON messages(knowledge_id);
+CREATE INDEX IF NOT EXISTS idx_messages_request_role ON messages(request_id, role);
+CREATE INDEX IF NOT EXISTS idx_messages_source_channel ON messages(source_platform, source_channel_id);
+CREATE INDEX IF NOT EXISTS idx_messages_source_message ON messages(source_platform, source_message_id);
+CREATE INDEX IF NOT EXISTS idx_messages_source_delivery ON messages(source_platform, source_delivery_id);
+CREATE INDEX IF NOT EXISTS idx_messages_source_user ON messages(source_platform, source_user_id);
 
 CREATE TABLE IF NOT EXISTS chunks (
     id VARCHAR(36) PRIMARY KEY,

migrations/versioned/000064_message_audit.down.sql

@@ -0,0 +1,20 @@
+-- Migration: 000064_message_audit (rollback)
+DO $$ BEGIN RAISE NOTICE '[Migration 000064 rollback] Removing message audit metadata'; END $$;
+
+DROP INDEX IF EXISTS idx_messages_source_user;
+DROP INDEX IF EXISTS idx_messages_source_delivery;
+DROP INDEX IF EXISTS idx_messages_source_message;
+DROP INDEX IF EXISTS idx_messages_source_channel;
+DROP INDEX IF EXISTS idx_messages_request_role;
+
+ALTER TABLE messages DROP COLUMN IF EXISTS token_usage;
+ALTER TABLE messages DROP COLUMN IF EXISTS searched_knowledge_base_ids;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_thread_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_chat_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_user_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_delivery_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_message_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_channel_id;
+ALTER TABLE messages DROP COLUMN IF EXISTS source_platform;
+
+DO $$ BEGIN RAISE NOTICE '[Migration 000064 rollback] Message audit metadata removed'; END $$;

migrations/versioned/000064_message_audit.up.sql

@@ -0,0 +1,40 @@
+-- Migration: 000064_message_audit
+-- Description: Persist per-message source metadata, delivery correlation, retrieval scope, and model token usage.
+DO $$ BEGIN RAISE NOTICE '[Migration 000064] Adding message audit metadata'; END $$;
+
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_platform VARCHAR(20) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_channel_id VARCHAR(36) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_message_id VARCHAR(128) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_delivery_id VARCHAR(128) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_user_id VARCHAR(128) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_chat_id VARCHAR(128) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS source_thread_id VARCHAR(128) NOT NULL DEFAULT '';
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS searched_knowledge_base_ids JSONB NOT NULL DEFAULT '[]'::jsonb;
+ALTER TABLE messages ADD COLUMN IF NOT EXISTS token_usage JSONB;
+
+CREATE INDEX IF NOT EXISTS idx_messages_request_role
+    ON messages (request_id, role);
+CREATE INDEX IF NOT EXISTS idx_messages_source_channel
+    ON messages (source_platform, source_channel_id)
+    WHERE source_platform != '';
+CREATE INDEX IF NOT EXISTS idx_messages_source_message
+    ON messages (source_platform, source_message_id)
+    WHERE source_message_id != '';
+CREATE INDEX IF NOT EXISTS idx_messages_source_delivery
+    ON messages (source_platform, source_delivery_id)
+    WHERE source_delivery_id != '';
+CREATE INDEX IF NOT EXISTS idx_messages_source_user
+    ON messages (source_platform, source_user_id)
+    WHERE source_user_id != '';
+
+COMMENT ON COLUMN messages.source_platform IS 'Concrete source platform such as dingtalk; channel remains im for compatibility';
+COMMENT ON COLUMN messages.source_channel_id IS 'WeKnora im_channels.id that received the message';
+COMMENT ON COLUMN messages.source_message_id IS 'External platform message identifier';
+COMMENT ON COLUMN messages.source_delivery_id IS 'External delivery or card identifier used to correlate callbacks';
+COMMENT ON COLUMN messages.source_user_id IS 'External platform user identifier for this individual message';
+COMMENT ON COLUMN messages.source_chat_id IS 'External platform conversation or group identifier';
+COMMENT ON COLUMN messages.source_thread_id IS 'External platform thread identifier when available';
+COMMENT ON COLUMN messages.searched_knowledge_base_ids IS 'Knowledge-base search scope, retained even when retrieval has zero hits';
+COMMENT ON COLUMN messages.token_usage IS 'Answer-generation token usage: prompt, completion, total, and cached tokens';
+
+DO $$ BEGIN RAISE NOTICE '[Migration 000064] Message audit metadata ready'; END $$;