feat: make chat-channel and implement WhatsApp bot (#17518)

2026-07-30 12:39:27 +08:00 · 2026-07-29 18:55:22 +08:00
parent d5604f8947
commit c7b1b3a4d2
21 changed files with 1899 additions and 60 deletions
--- a/api/channels/whatsapp/gateway-node/README.md
+++ b/api/channels/whatsapp/gateway-node/README.md
@@ -20,6 +20,7 @@ npm start

 ## API

+- `GET /health`
 - `POST /whatsapp/:sessionKey/start`
 - `GET /whatsapp/:sessionKey/status`
 - `GET /whatsapp/:sessionKey/events/ws?after=<seq>` (WebSocket)
--- a/api/channels/whatsapp/gateway-node/package-lock.json
+++ b/api/channels/whatsapp/gateway-node/package-lock.json
@@ -199,9 +199,6 @@
      "cpu": [
        "arm"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -219,9 +216,6 @@
      "cpu": [
        "arm64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -239,9 +233,6 @@
      "cpu": [
        "ppc64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -259,9 +250,6 @@
      "cpu": [
        "riscv64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -279,9 +267,6 @@
      "cpu": [
        "s390x"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -299,9 +284,6 @@
      "cpu": [
        "x64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -319,9 +301,6 @@
      "cpu": [
        "arm64"
      ],
-      "libc": [
-        "musl"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -339,9 +318,6 @@
      "cpu": [
        "x64"
      ],
-      "libc": [
-        "musl"
-      ],
      "license": "LGPL-3.0-or-later",
      "optional": true,
      "os": [
@@ -359,9 +335,6 @@
      "cpu": [
        "arm"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -385,9 +358,6 @@
      "cpu": [
        "arm64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -411,9 +381,6 @@
      "cpu": [
        "ppc64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -437,9 +404,6 @@
      "cpu": [
        "riscv64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -463,9 +427,6 @@
      "cpu": [
        "s390x"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -489,9 +450,6 @@
      "cpu": [
        "x64"
      ],
-      "libc": [
-        "glibc"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -515,9 +473,6 @@
      "cpu": [
        "arm64"
      ],
-      "libc": [
-        "musl"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
@@ -541,9 +496,6 @@
      "cpu": [
        "x64"
      ],
-      "libc": [
-        "musl"
-      ],
      "license": "Apache-2.0",
      "optional": true,
      "os": [
--- a/cmd/ragflow_server.go
+++ b/cmd/ragflow_server.go
@@ -27,6 +27,7 @@ import (
 	"ragflow/internal/agent/audio"
 	"ragflow/internal/agent/canvas"
 	agenttool "ragflow/internal/agent/tool"
+	"ragflow/internal/channels"
 	"ragflow/internal/handler"
 	ingestion "ragflow/internal/ingestion/service"
 	"ragflow/internal/mcp"
@@ -846,6 +847,8 @@ func startServer(ctx context.Context, config *server.Config) {
 	// Setup routes
 	r.Setup(ginEngine)

+	channels.Start(ctx)
+
 	// Create HTTP server with timeouts to prevent slow clients from blocking shutdown
 	addr := fmt.Sprintf(":%d", config.APIServer.Port)
 	srv := &http.Server{
--- a/internal/channels/bootstrap.go
+++ b/internal/channels/bootstrap.go
@@ -0,0 +1,325 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"ragflow/internal/channels/core"
+	"ragflow/internal/dao"
+	"ragflow/internal/entity"
+	"ragflow/internal/service"
+)
+
+const (
+	reconcileInterval      = 10 * time.Second
+	initialStartRetryDelay = 10 * time.Second
+	maxStartRetryDelay     = 5 * time.Minute
+)
+
+type runningChannel struct {
+	channel core.Channel
+	fp      string
+}
+
+type failedChannel struct {
+	fp          string
+	attempts    int
+	nextRetryAt time.Time
+}
+
+type Runtime struct {
+	mu      sync.Mutex
+	running map[string]runningChannel
+	failed  map[string]failedChannel
+}
+
+// NewRuntime creates an empty chat-channel runtime reconciler.
+func NewRuntime() *Runtime {
+	return &Runtime{
+		running: map[string]runningChannel{},
+		failed:  map[string]failedChannel{},
+	}
+}
+
+// Start launches the chat-channel runtime reconciler in the background.
+func Start(ctx context.Context) *Runtime {
+	rt := NewRuntime()
+	service.SetChatChannelRuntimeProvider(whatsappRuntimeSnapshotMap)
+	go rt.Run(ctx)
+	return rt
+}
+
+// Run reconciles configured chat channels until the context is cancelled.
+func (r *Runtime) Run(ctx context.Context) {
+	ticker := time.NewTicker(reconcileInterval)
+	defer ticker.Stop()
+	defer r.stopAll(context.Background())
+
+	for {
+		if err := r.Reconcile(ctx); err != nil {
+			log.Printf("chat channel reconcile failed: %v", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+// Reconcile starts, stops, and restarts channel instances to match the database state.
+func (r *Runtime) Reconcile(ctx context.Context) error {
+	desired, err := desiredChannels(ctx)
+	if err != nil {
+		return err
+	}
+
+	var toStop []core.Channel
+	r.mu.Lock()
+	for accountID, entry := range r.running {
+		wanted, ok := desired[accountID]
+		if !ok || wanted.fp != entry.fp {
+			delete(r.running, accountID)
+			toStop = append(toStop, entry.channel)
+		}
+	}
+	for accountID, failure := range r.failed {
+		wanted, ok := desired[accountID]
+		if !ok || wanted.fp != failure.fp {
+			delete(r.failed, accountID)
+		}
+	}
+	r.mu.Unlock()
+
+	for _, ch := range toStop {
+		stopChannel(context.Background(), ch)
+	}
+
+	activeWhatsApp := false
+	for _, wanted := range desired {
+		if wanted.channel == "whatsapp" {
+			activeWhatsApp = true
+			break
+		}
+	}
+	if err := syncWhatsAppGateway(ctx, activeWhatsApp); err != nil && activeWhatsApp {
+		log.Printf("failed to sync WhatsApp gateway: %v", err)
+	}
+
+	now := time.Now()
+	for accountID, wanted := range desired {
+		r.mu.Lock()
+		_, isRunning := r.running[accountID]
+		failure, failed := r.failed[accountID]
+		retryPending := failed && failure.fp == wanted.fp && now.Before(failure.nextRetryAt)
+		r.mu.Unlock()
+		if isRunning || retryPending {
+			continue
+		}
+		if err := r.startChannel(ctx, accountID, wanted); err != nil {
+			log.Printf("failed to start chat channel %s (%s): %v", accountID, wanted.channel, err)
+			r.recordStartFailure(accountID, wanted.fp, now)
+			continue
+		}
+		r.clearStartFailure(accountID)
+	}
+	return nil
+}
+
+// recordStartFailure saves the next retry window for a failed channel start.
+func (r *Runtime) recordStartFailure(accountID string, fp string, now time.Time) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	attempts := 1
+	if failure, ok := r.failed[accountID]; ok && failure.fp == fp {
+		attempts = failure.attempts + 1
+	}
+	r.failed[accountID] = failedChannel{
+		fp:          fp,
+		attempts:    attempts,
+		nextRetryAt: now.Add(startRetryDelay(attempts)),
+	}
+}
+
+// clearStartFailure removes a stale failed-start record after a successful start.
+func (r *Runtime) clearStartFailure(accountID string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	delete(r.failed, accountID)
+}
+
+// startRetryDelay returns the bounded exponential backoff for a start attempt.
+func startRetryDelay(attempts int) time.Duration {
+	if attempts < 1 {
+		attempts = 1
+	}
+	delay := initialStartRetryDelay
+	for i := 1; i < attempts; i++ {
+		delay *= 2
+		if delay >= maxStartRetryDelay {
+			return maxStartRetryDelay
+		}
+	}
+	return delay
+}
+
+type desiredChannel struct {
+	channel    string
+	credential map[string]any
+	fp         string
+}
+
+// desiredChannels loads enabled chat-channel rows and reduces them to runtime configuration.
+func desiredChannels(ctx context.Context) (map[string]desiredChannel, error) {
+	rows, err := dao.NewChatChannel().ListActive(ctx, dao.DB)
+	if err != nil {
+		return nil, err
+	}
+	out := make(map[string]desiredChannel, len(rows))
+	for _, row := range rows {
+		credential := credentialFromConfig(row.Config)
+		out[row.ID] = desiredChannel{
+			channel:    row.Channel,
+			credential: credential,
+			fp:         fingerprint(row.Channel, credential),
+		}
+	}
+	return out, nil
+}
+
+// credentialFromConfig extracts the platform credential block from chat_channel.config.
+func credentialFromConfig(config entity.JSONMap) map[string]any {
+	if config == nil {
+		return map[string]any{}
+	}
+	if raw, ok := config["credential"].(map[string]any); ok {
+		return raw
+	}
+	if raw, ok := config["credential"].(entity.JSONMap); ok {
+		return raw
+	}
+	return map[string]any{}
+}
+
+// fingerprint returns a stable hash for the configuration that requires a channel restart.
+func fingerprint(channel string, credential map[string]any) string {
+	payload, _ := json.Marshal(map[string]any{
+		"channel":    channel,
+		"credential": credential,
+	})
+	sum := sha256.Sum256(payload)
+	return hex.EncodeToString(sum[:])
+}
+
+// startChannel builds a platform channel, attaches the RAG bridge, and starts it.
+func (r *Runtime) startChannel(ctx context.Context, accountID string, wanted desiredChannel) error {
+	ch, err := buildChannel(accountID, wanted)
+	if err != nil {
+		return err
+	}
+	if ch == nil {
+		return nil
+	}
+
+	channelService := service.NewChatChannelService()
+	ch.SetMessageHandler(func(ctx context.Context, msg core.IncomingMessage) error {
+		answer, err := channelService.HandleIncomingMessage(ctx, service.ChatChannelIncomingMessage{
+			Channel:   msg.Channel,
+			AccountID: msg.AccountID,
+			ChatID:    msg.ChatID,
+			ChatType:  msg.ChatType,
+			MessageID: msg.MessageID,
+			SenderID:  msg.SenderID,
+			Text:      msg.Text,
+		})
+		if err != nil || answer == "" {
+			return err
+		}
+		return ch.Send(ctx, core.OutgoingMessage{
+			ChatID:           msg.ChatID,
+			Text:             answer,
+			ReplyToMessageID: msg.MessageID,
+		})
+	})
+	if err = ch.Start(ctx); err != nil {
+		return err
+	}
+	r.mu.Lock()
+	r.running[accountID] = runningChannel{channel: ch, fp: wanted.fp}
+	r.mu.Unlock()
+	log.Printf("started chat channel %s:%s", ch.ChannelID(), accountID)
+	return nil
+}
+
+// buildChannel constructs the platform-specific channel implementation for one chat_channel row.
+func buildChannel(accountID string, wanted desiredChannel) (core.Channel, error) {
+	switch wanted.channel {
+	case "whatsapp":
+		return newWhatsAppChannelFromConfig(accountID, wanted.credential)
+	default:
+		return nil, fmt.Errorf("unknown channel: %s", wanted.channel)
+	}
+}
+
+// whatsappRuntimeSnapshotMap returns the API payload for a live WhatsApp runtime snapshot.
+func whatsappRuntimeSnapshotMap(accountID string) (map[string]any, bool) {
+	snapshot, ok := getWhatsAppRuntimeSnapshot(accountID)
+	if !ok {
+		return nil, false
+	}
+	return map[string]any{
+		"account_id":       snapshot.AccountID,
+		"session_key":      snapshot.SessionKey,
+		"status":           snapshot.Status,
+		"connected_at":     snapshot.ConnectedAt,
+		"qr_updated_at":    snapshot.QRUpdatedAt,
+		"qr_data_url":      snapshot.QRDataURL,
+		"last_error":       snapshot.LastError,
+		"session_id":       snapshot.SessionID,
+		"last_snapshot_at": snapshot.LastSnapshotAt,
+		"gateway_base_url": snapshot.GatewayBaseURL,
+		"event_cursor":     snapshot.EventCursor,
+	}, true
+}
+
+// stopAll stops every running channel and shuts down shared gateway processes.
+func (r *Runtime) stopAll(ctx context.Context) {
+	r.mu.Lock()
+	running := r.running
+	r.running = map[string]runningChannel{}
+	r.mu.Unlock()
+	for _, entry := range running {
+		stopChannel(ctx, entry.channel)
+	}
+	_ = syncWhatsAppGateway(ctx, false)
+}
+
+// stopChannel stops one platform channel and logs any shutdown error.
+func stopChannel(ctx context.Context, ch core.Channel) {
+	if err := ch.Stop(ctx); err != nil {
+		log.Printf("failed to stop chat channel %s:%s: %v", ch.ChannelID(), ch.AccountID(), err)
+	}
+}
--- a/internal/channels/bootstrap_test.go
+++ b/internal/channels/bootstrap_test.go
@@ -0,0 +1,179 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"ragflow/internal/channels/core"
+)
+
+func TestStartRetryDelay(t *testing.T) {
+	tests := []struct {
+		name     string
+		attempts int
+		want     time.Duration
+	}{
+		{name: "zero attempt uses first delay", attempts: 0, want: initialStartRetryDelay},
+		{name: "first attempt", attempts: 1, want: initialStartRetryDelay},
+		{name: "second attempt doubles", attempts: 2, want: 2 * initialStartRetryDelay},
+		{name: "fifth attempt", attempts: 5, want: 16 * initialStartRetryDelay},
+		{name: "bounded", attempts: 20, want: maxStartRetryDelay},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := startRetryDelay(tt.attempts); got != tt.want {
+				t.Fatalf("startRetryDelay(%d) = %s, want %s", tt.attempts, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestRecordStartFailure(t *testing.T) {
+	rt := NewRuntime()
+	now := time.Unix(100, 0)
+
+	rt.recordStartFailure("account-1", "fp-1", now)
+	first := rt.failed["account-1"]
+	if first.attempts != 1 {
+		t.Fatalf("attempts after first failure = %d, want 1", first.attempts)
+	}
+	if !first.nextRetryAt.Equal(now.Add(initialStartRetryDelay)) {
+		t.Fatalf("nextRetryAt after first failure = %s, want %s", first.nextRetryAt, now.Add(initialStartRetryDelay))
+	}
+
+	rt.recordStartFailure("account-1", "fp-1", now)
+	second := rt.failed["account-1"]
+	if second.attempts != 2 {
+		t.Fatalf("attempts after second failure = %d, want 2", second.attempts)
+	}
+	if !second.nextRetryAt.Equal(now.Add(2 * initialStartRetryDelay)) {
+		t.Fatalf("nextRetryAt after second failure = %s, want %s", second.nextRetryAt, now.Add(2*initialStartRetryDelay))
+	}
+
+	rt.recordStartFailure("account-1", "fp-2", now)
+	changed := rt.failed["account-1"]
+	if changed.attempts != 1 {
+		t.Fatalf("attempts after fingerprint change = %d, want 1", changed.attempts)
+	}
+}
+
+func TestClearStartFailure(t *testing.T) {
+	rt := NewRuntime()
+	rt.recordStartFailure("account-1", "fp-1", time.Unix(100, 0))
+
+	rt.clearStartFailure("account-1")
+
+	if _, ok := rt.failed["account-1"]; ok {
+		t.Fatal("clearStartFailure left failed entry behind")
+	}
+}
+
+func TestGatewayWorkdirDefaultContainsNodeEntrypoint(t *testing.T) {
+	t.Setenv("WHATSAPP_GATEWAY_WORKDIR", "")
+
+	entry := filepath.Join(gatewayWorkdir(), "index.js")
+	if _, err := os.Stat(entry); err != nil {
+		t.Fatalf("default gateway entry %s is not available: %v", entry, err)
+	}
+}
+
+func TestGatewayEnabledDefaultsToUserManaged(t *testing.T) {
+	t.Setenv("WHATSAPP_GATEWAY_ENABLED", "")
+
+	if gatewayEnabled() {
+		t.Fatal("gatewayEnabled() = true, want false by default")
+	}
+}
+
+func TestGatewayEnabledCanBeExplicitlyEnabled(t *testing.T) {
+	t.Setenv("WHATSAPP_GATEWAY_ENABLED", "true")
+
+	if !gatewayEnabled() {
+		t.Fatal("gatewayEnabled() = false, want true")
+	}
+}
+
+func TestGatewayEnabledRejectsInvalidValue(t *testing.T) {
+	t.Setenv("WHATSAPP_GATEWAY_ENABLED", "maybe")
+
+	if gatewayEnabled() {
+		t.Fatal("gatewayEnabled() = true for invalid value, want false")
+	}
+}
+
+func TestWhatsAppEnqueueDoesNotMarkDroppedMessageSeen(t *testing.T) {
+	ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"})
+	worker := &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)}
+	worker.queue <- core.IncomingMessage{MessageID: "queued"}
+	ch.workers["chat-1"] = worker
+
+	ok := ch.enqueueIncoming(context.Background(), core.IncomingMessage{
+		ChatID:    "chat-1",
+		MessageID: "dropped",
+	})
+
+	if ok {
+		t.Fatal("enqueueIncoming succeeded for a full queue")
+	}
+	if _, seen := ch.seen["dropped"]; seen {
+		t.Fatal("dropped message was marked seen")
+	}
+}
+
+func TestWhatsAppEnqueueMarksSeenAfterSuccessfulHandoff(t *testing.T) {
+	ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"})
+	ch.workers["chat-1"] = &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)}
+	msg := core.IncomingMessage{ChatID: "chat-1", MessageID: "message-1"}
+
+	if ok := ch.enqueueIncoming(context.Background(), msg); !ok {
+		t.Fatal("enqueueIncoming failed for an empty queue")
+	}
+	if _, seen := ch.seen[msg.MessageID]; !seen {
+		t.Fatal("successfully handed off message was not marked seen")
+	}
+	if ok := ch.enqueueIncoming(context.Background(), msg); ok {
+		t.Fatal("duplicate message was enqueued")
+	}
+}
+
+func TestWhatsAppRetireChatWorkerRequiresCurrentEmptyWorker(t *testing.T) {
+	ch := newWhatsAppChannel(whatsappAccount{AccountID: "account-1"})
+	worker := &whatsappChatWorker{queue: make(chan core.IncomingMessage, 1)}
+	worker.queue <- core.IncomingMessage{ChatID: "chat-1", MessageID: "message-1"}
+	ch.workers["chat-1"] = worker
+
+	if ch.retireChatWorker("chat-1", worker) {
+		t.Fatal("retireChatWorker retired a non-empty worker")
+	}
+	if ch.workers["chat-1"] != worker {
+		t.Fatal("non-empty worker was removed from ownership map")
+	}
+
+	<-worker.queue
+	if !ch.retireChatWorker("chat-1", worker) {
+		t.Fatal("retireChatWorker did not retire an empty current worker")
+	}
+	if _, ok := ch.workers["chat-1"]; ok {
+		t.Fatal("empty retired worker remains in ownership map")
+	}
+}
--- a/internal/channels/core/channel.go
+++ b/internal/channels/core/channel.go
@@ -0,0 +1,47 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package core
+
+import "context"
+
+type IncomingMessage struct {
+	Channel   string
+	AccountID string
+	ChatID    string
+	ChatType  string
+	MessageID string
+	SenderID  string
+	Text      string
+	Raw       map[string]any
+}
+
+type OutgoingMessage struct {
+	ChatID           string
+	Text             string
+	ReplyToMessageID string
+}
+
+type MessageHandler func(ctx context.Context, msg IncomingMessage) error
+
+type Channel interface {
+	ChannelID() string                                   // ChannelID return channel's name
+	AccountID() string                                   // AccountID return channel's channelID
+	SetMessageHandler(MessageHandler)                    // SetMessageHandler set channel's Message Handler
+	Start(ctx context.Context) error                     // Start the channel
+	Stop(ctx context.Context) error                      // Stop cancels channel event
+	Send(ctx context.Context, msg OutgoingMessage) error // Send posts an outgoing RAGFlow answer to the channel
+}
--- a/internal/channels/dingtalk.go
+++ b/internal/channels/dingtalk.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/discord.go
+++ b/internal/channels/discord.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/feishu.go
+++ b/internal/channels/feishu.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/line.go
+++ b/internal/channels/line.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/qqbot.go
+++ b/internal/channels/qqbot.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/telegram.go
+++ b/internal/channels/telegram.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/wecom.go
+++ b/internal/channels/wecom.go
@@ -0,0 +1,17 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
--- a/internal/channels/whatsapp.go
+++ b/internal/channels/whatsapp.go
@@ -0,0 +1,693 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"ragflow/internal/channels/core"
+)
+
+const (
+	defaultGatewayBaseURL = "http://127.0.0.1:3005"
+	defaultTimeout        = 30 * time.Second
+	startTimeout          = 2 * time.Minute
+	wsHandshakeTimeout    = 2 * time.Minute
+	wsReadTimeout         = 2 * time.Minute
+	wsPingInterval        = 30 * time.Second
+	defaultReconnect      = 3 * time.Second
+	messageTTL            = time.Hour
+	messageQueueSize      = 64
+	chatWorkerIdle        = 5 * time.Minute
+)
+
+type whatsappAccount struct {
+	AccountID      string
+	GatewayBaseURL string
+	GatewayToken   string
+	SessionKey     string
+	Timeout        time.Duration
+}
+
+type whatsappRuntimeSnapshot struct {
+	AccountID      string   `json:"account_id"`
+	SessionKey     string   `json:"session_key"`
+	Status         string   `json:"status"`
+	ConnectedAt    *float64 `json:"connected_at"`
+	QRUpdatedAt    *float64 `json:"qr_updated_at"`
+	QRDataURL      *string  `json:"qr_data_url"`
+	LastError      *string  `json:"last_error"`
+	SessionID      *string  `json:"session_id"`
+	LastSnapshotAt *float64 `json:"last_snapshot_at"`
+	GatewayBaseURL *string  `json:"gateway_base_url"`
+	EventCursor    int64    `json:"event_cursor"`
+}
+
+type whatsappChannel struct {
+	account whatsappAccount
+
+	mu             sync.Mutex
+	cancel         context.CancelFunc
+	handler        core.MessageHandler
+	client         *http.Client
+	status         string
+	lastErr        string
+	qrDataURL      string
+	qrUpdatedAt    float64
+	connectedAt    float64
+	lastSnapshotAt float64
+	sessionID      string
+	eventCursor    int64
+	seen           map[string]time.Time
+	workers        map[string]*whatsappChatWorker
+}
+
+type whatsappChatWorker struct {
+	queue chan core.IncomingMessage
+}
+
+var liveWhatsAppChannels sync.Map
+
+// newWhatsAppChannel creates a WhatsApp channel instance with default runtime settings filled in.
+func newWhatsAppChannel(account whatsappAccount) *whatsappChannel {
+	if account.GatewayBaseURL == "" {
+		account.GatewayBaseURL = defaultGatewayBaseURL
+	}
+	if account.SessionKey == "" {
+		account.SessionKey = account.AccountID
+	}
+	if account.Timeout <= 0 {
+		account.Timeout = defaultTimeout
+	}
+	return &whatsappChannel{
+		account: account,
+		client:  &http.Client{Timeout: account.Timeout},
+		status:  "stopped",
+		seen:    map[string]time.Time{},
+		workers: map[string]*whatsappChatWorker{},
+	}
+}
+
+// newWhatsAppChannelFromConfig builds a WhatsApp channel from chat_channel.config.credential.
+func newWhatsAppChannelFromConfig(accountID string, cfg map[string]any) (*whatsappChannel, error) {
+	timeout := defaultTimeout
+	if raw, ok := cfg["timeout_secs"]; ok {
+		switch v := raw.(type) {
+		case float64:
+			timeout = time.Duration(v) * time.Second
+		case int:
+			timeout = time.Duration(v) * time.Second
+		case string:
+			if n, err := strconv.Atoi(v); err == nil {
+				timeout = time.Duration(n) * time.Second
+			}
+		}
+	}
+	baseURL := firstString(cfg, "gateway_base_url", "gateway_url", "control_url")
+	token := firstString(cfg, "gateway_token", "token")
+	sessionKey := firstString(cfg, "session_key", "session_id")
+	if sessionKey == "" {
+		sessionKey = accountID
+	}
+	return newWhatsAppChannel(whatsappAccount{
+		AccountID:      accountID,
+		GatewayBaseURL: baseURL,
+		GatewayToken:   token,
+		SessionKey:     sessionKey,
+		Timeout:        timeout,
+	}), nil
+}
+
+// getWhatsAppRuntimeSnapshot returns the in-memory runtime snapshot for a live WhatsApp channel.
+func getWhatsAppRuntimeSnapshot(accountID string) (*whatsappRuntimeSnapshot, bool) {
+	raw, ok := liveWhatsAppChannels.Load(accountID)
+	if !ok {
+		return nil, false
+	}
+	return raw.(*whatsappChannel).Snapshot(), true
+}
+
+// ChannelID returns the platform identifier used by the chat-channel runtime.
+func (c *whatsappChannel) ChannelID() string {
+	return "whatsapp"
+}
+
+// AccountID returns the chat_channel.id bound to this WhatsApp runtime instance.
+func (c *whatsappChannel) AccountID() string {
+	return c.account.AccountID
+}
+
+// SetMessageHandler installs the RAGFlow bridge invoked for inbound WhatsApp messages.
+func (c *whatsappChannel) SetMessageHandler(handler core.MessageHandler) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.handler = handler
+}
+
+// Start begins the WhatsApp gateway session and event websocket loop.
+func (c *whatsappChannel) Start(ctx context.Context) error {
+	c.mu.Lock()
+	if c.cancel != nil {
+		c.mu.Unlock()
+		return nil
+	}
+	runCtx, cancel := context.WithCancel(ctx)
+	c.cancel = cancel
+	c.status = "connecting"
+	c.lastErr = ""
+	c.mu.Unlock()
+
+	liveWhatsAppChannels.Store(c.account.AccountID, c)
+	go c.run(runCtx)
+	return nil
+}
+
+// Stop cancels the WhatsApp event loop and asks the gateway to stop the session.
+func (c *whatsappChannel) Stop(ctx context.Context) error {
+	c.mu.Lock()
+	cancel := c.cancel
+	c.cancel = nil
+	c.status = "stopped"
+	c.lastErr = ""
+	c.qrDataURL = ""
+	c.qrUpdatedAt = 0
+	c.connectedAt = 0
+	c.lastSnapshotAt = 0
+	c.sessionID = ""
+	c.eventCursor = 0
+	c.seen = map[string]time.Time{}
+	c.workers = map[string]*whatsappChatWorker{}
+	c.mu.Unlock()
+
+	if cancel != nil {
+		cancel()
+	}
+	liveWhatsAppChannels.Delete(c.account.AccountID)
+	_ = c.requestJSON(ctx, http.MethodPost, fmt.Sprintf("whatsapp/%s/stop", url.PathEscape(c.sessionKey())), map[string]any{
+		"account_id":  c.account.AccountID,
+		"session_key": c.sessionKey(),
+	}, nil)
+	return nil
+}
+
+// Send posts an outgoing RAGFlow answer to the WhatsApp gateway.
+func (c *whatsappChannel) Send(ctx context.Context, msg core.OutgoingMessage) error {
+	if strings.TrimSpace(msg.ChatID) == "" {
+		return errors.New("chat_id is required")
+	}
+	payload := map[string]any{
+		"account_id":          c.account.AccountID,
+		"session_key":         c.sessionKey(),
+		"chat_id":             msg.ChatID,
+		"text":                msg.Text,
+		"reply_to_message_id": nil,
+	}
+	if msg.ReplyToMessageID != "" {
+		payload["reply_to_message_id"] = msg.ReplyToMessageID
+	}
+	return c.requestJSON(ctx, http.MethodPost, fmt.Sprintf("whatsapp/%s/send", url.PathEscape(c.sessionKey())), payload, nil)
+}
+
+// Snapshot copies the current WhatsApp runtime state for the runtime API.
+func (c *whatsappChannel) Snapshot() *whatsappRuntimeSnapshot {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	baseURL := strings.TrimRight(c.account.GatewayBaseURL, "/")
+	return &whatsappRuntimeSnapshot{
+		AccountID:      c.account.AccountID,
+		SessionKey:     c.sessionKey(),
+		Status:         c.status,
+		ConnectedAt:    floatPtr(c.connectedAt),
+		QRUpdatedAt:    floatPtr(c.qrUpdatedAt),
+		QRDataURL:      stringPtr(c.qrDataURL),
+		LastError:      stringPtr(c.lastErr),
+		SessionID:      stringPtr(c.sessionID),
+		LastSnapshotAt: floatPtr(c.lastSnapshotAt),
+		GatewayBaseURL: stringPtr(baseURL),
+		EventCursor:    c.eventCursor,
+	}
+}
+
+// run starts the gateway session and reconnects the websocket event stream when needed.
+func (c *whatsappChannel) run(ctx context.Context) {
+	if err := c.startSession(ctx); err != nil {
+		if ctx.Err() == nil {
+			c.setError(err)
+		}
+		return
+	}
+
+	for ctx.Err() == nil {
+		if err := c.runEvents(ctx); err != nil && ctx.Err() == nil {
+			c.setError(err)
+			log.Printf("[whatsapp:%s] event loop error: %v", c.account.AccountID, err)
+			time.Sleep(defaultReconnect)
+		}
+	}
+}
+
+// startSession asks the gateway to start the session and accepts an already-created session after timeout.
+func (c *whatsappChannel) startSession(ctx context.Context) error {
+	err := c.requestJSONWithTimeout(ctx, startTimeout, http.MethodPost, fmt.Sprintf("whatsapp/%s/start", url.PathEscape(c.sessionKey())), map[string]any{
+		"account_id":       c.account.AccountID,
+		"session_key":      c.sessionKey(),
+		"gateway_base_url": strings.TrimRight(c.account.GatewayBaseURL, "/"),
+	}, nil)
+	if err == nil {
+		return c.waitForStatus(ctx)
+	}
+	if statusErr := c.waitForStatus(ctx); statusErr == nil {
+		log.Printf("[whatsapp:%s] start request returned error after gateway created the session: %v", c.account.AccountID, err)
+		return nil
+	}
+	return err
+}
+
+// runEvents reads gateway websocket messages until the connection closes or the context ends.
+func (c *whatsappChannel) runEvents(ctx context.Context) error {
+	wsURL, err := c.eventsURL()
+	if err != nil {
+		return err
+	}
+	header := http.Header{}
+	if auth := c.authHeader(); auth != "" {
+		header.Set("Authorization", auth)
+	}
+	dialer := *websocket.DefaultDialer
+	dialer.HandshakeTimeout = wsHandshakeTimeout
+	conn, _, err := dialer.DialContext(ctx, wsURL, header)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	go func() {
+		<-ctx.Done()
+		_ = conn.Close()
+	}()
+	_ = conn.SetReadDeadline(time.Now().Add(wsReadTimeout))
+	conn.SetPongHandler(func(string) error {
+		return conn.SetReadDeadline(time.Now().Add(wsReadTimeout))
+	})
+	pingDone := make(chan struct{})
+	defer close(pingDone)
+	go func() {
+		ticker := time.NewTicker(wsPingInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-pingDone:
+				return
+			case <-ticker.C:
+				deadline := time.Now().Add(10 * time.Second)
+				if err := conn.WriteControl(websocket.PingMessage, nil, deadline); err != nil {
+					_ = conn.Close()
+					return
+				}
+			}
+		}
+	}()
+
+	for ctx.Err() == nil {
+		_, payload, err := conn.ReadMessage()
+		if err != nil {
+			return err
+		}
+		c.handleWSPayload(ctx, payload)
+	}
+	return ctx.Err()
+}
+
+// waitForStatus waits briefly until the gateway exposes the started session status.
+func (c *whatsappChannel) waitForStatus(ctx context.Context) error {
+	deadline := time.Now().Add(startTimeout)
+	for ctx.Err() == nil {
+		var response map[string]any
+		if err := c.requestJSON(ctx, http.MethodGet, fmt.Sprintf("whatsapp/%s/status", url.PathEscape(c.sessionKey())), nil, &response); err == nil {
+			if data, ok := response["data"].(map[string]any); ok {
+				c.applySnapshot(data)
+			}
+			return nil
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("WhatsApp session status did not become available within %s", startTimeout)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(defaultReconnect):
+		}
+	}
+	return ctx.Err()
+}
+
+// handleWSPayload dispatches a gateway websocket payload to snapshot or event handling.
+func (c *whatsappChannel) handleWSPayload(ctx context.Context, payload []byte) {
+	var obj map[string]any
+	if err := json.Unmarshal(payload, &obj); err != nil {
+		return
+	}
+	data, _ := obj["data"].(map[string]any)
+	switch strings.TrimSpace(fmt.Sprint(obj["type"])) {
+	case "snapshot":
+		c.applySnapshot(data)
+	case "event":
+		c.handleEvent(ctx, data)
+	}
+}
+
+// handleEvent converts a WhatsApp message event and queues it for ordered processing.
+func (c *whatsappChannel) handleEvent(ctx context.Context, item map[string]any) {
+	if fmt.Sprint(item["kind"]) != "message" {
+		return
+	}
+	messageID := strings.TrimSpace(fmt.Sprint(item["message_id"]))
+	if messageID == "" {
+		return
+	}
+	incoming := core.IncomingMessage{
+		Channel:   c.ChannelID(),
+		AccountID: c.account.AccountID,
+		ChatID:    fmt.Sprint(item["chat_id"]),
+		ChatType:  valueOrDefault(item["chat_type"], "p2p"),
+		MessageID: messageID,
+		SenderID:  fmt.Sprint(item["sender_id"]),
+		Text:      fmt.Sprint(item["text"]),
+		Raw:       item,
+	}
+	_ = c.enqueueIncoming(ctx, incoming)
+}
+
+// enqueueIncoming schedules message handling and marks the message seen after a successful handoff.
+func (c *whatsappChannel) enqueueIncoming(ctx context.Context, incoming core.IncomingMessage) bool {
+	if ctx.Err() != nil {
+		return false
+	}
+
+	now := time.Now()
+	var worker *whatsappChatWorker
+	startWorker := false
+	queueFull := false
+
+	c.mu.Lock()
+	c.pruneSeenLocked(now)
+	if _, ok := c.seen[incoming.MessageID]; ok {
+		c.mu.Unlock()
+		return false
+	}
+	if c.workers == nil {
+		c.workers = map[string]*whatsappChatWorker{}
+	}
+	worker = c.workers[incoming.ChatID]
+	if worker == nil {
+		worker = &whatsappChatWorker{queue: make(chan core.IncomingMessage, messageQueueSize)}
+		c.workers[incoming.ChatID] = worker
+		startWorker = true
+	}
+	select {
+	case worker.queue <- incoming:
+		c.seen[incoming.MessageID] = now
+		c.mu.Unlock()
+		if startWorker {
+			go c.runChatWorker(ctx, incoming.ChatID, worker)
+		}
+		return true
+	case <-ctx.Done():
+		c.mu.Unlock()
+	default:
+		queueFull = true
+		c.mu.Unlock()
+	}
+	if startWorker {
+		go c.runChatWorker(ctx, incoming.ChatID, worker)
+	}
+	if queueFull {
+		log.Printf("[whatsapp:%s] dropping message %s for chat %s: queue is full", c.account.AccountID, incoming.MessageID, incoming.ChatID)
+	}
+	return false
+}
+
+// runChatWorker processes one chat's inbound messages sequentially.
+func (c *whatsappChannel) runChatWorker(ctx context.Context, chatID string, worker *whatsappChatWorker) {
+	idle := time.NewTimer(chatWorkerIdle)
+	defer idle.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg := <-worker.queue:
+			if ctx.Err() != nil {
+				return
+			}
+			c.handleIncoming(ctx, msg)
+			resetTimer(idle, chatWorkerIdle)
+		case <-idle.C:
+			if c.retireChatWorker(chatID, worker) {
+				return
+			}
+			resetTimer(idle, chatWorkerIdle)
+		}
+	}
+}
+
+// retireChatWorker removes an idle worker only while it is still current and empty.
+func (c *whatsappChannel) retireChatWorker(chatID string, worker *whatsappChatWorker) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	current := c.workers[chatID]
+	if current != worker {
+		return true
+	}
+	if len(worker.queue) > 0 {
+		return false
+	}
+	if current == worker {
+		delete(c.workers, chatID)
+	}
+	return true
+}
+
+// handleIncoming invokes the installed RAGFlow bridge for one queued message.
+func (c *whatsappChannel) handleIncoming(ctx context.Context, incoming core.IncomingMessage) {
+	c.mu.Lock()
+	handler := c.handler
+	c.mu.Unlock()
+	if handler == nil {
+		return
+	}
+	if err := handler(ctx, incoming); err != nil {
+		log.Printf("[whatsapp:%s] message handler error: %v", c.account.AccountID, err)
+	}
+}
+
+// pruneSeenLocked removes expired duplicate-tracking entries while c.mu is held.
+func (c *whatsappChannel) pruneSeenLocked(now time.Time) {
+	for key, ts := range c.seen {
+		if now.Sub(ts) > messageTTL {
+			delete(c.seen, key)
+		}
+	}
+}
+
+// applySnapshot updates the cached connection and QR-code state from the gateway.
+func (c *whatsappChannel) applySnapshot(snapshot map[string]any) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.lastSnapshotAt = float64(time.Now().Unix())
+	c.status = valueOrDefault(snapshot["status"], "connecting")
+	c.lastErr = valueOrDefault(snapshot["last_error"], "")
+	c.qrDataURL = valueOrDefault(snapshot["qr_data_url"], "")
+	c.qrUpdatedAt = numeric(snapshot["qr_updated_at"])
+	c.connectedAt = numeric(snapshot["connected_at"])
+	c.sessionID = valueOrDefault(snapshot["session_id"], c.sessionID)
+	if cursor := int64(numeric(snapshot["event_cursor"])); cursor > c.eventCursor {
+		c.eventCursor = cursor
+	}
+	if c.status == "connected" {
+		c.lastErr = ""
+	}
+}
+
+// setError records a runtime failure in the snapshot exposed to the API.
+func (c *whatsappChannel) setError(err error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.status = "error"
+	c.lastErr = err.Error()
+	c.lastSnapshotAt = float64(time.Now().Unix())
+}
+
+// requestJSON sends an authenticated JSON request to the WhatsApp gateway.
+func (c *whatsappChannel) requestJSON(ctx context.Context, method, path string, body map[string]any, out any) error {
+	return c.requestJSONWithClient(ctx, c.client, method, path, body, out)
+}
+
+// requestJSONWithTimeout sends a JSON request with a one-off timeout override.
+func (c *whatsappChannel) requestJSONWithTimeout(ctx context.Context, timeout time.Duration, method, path string, body map[string]any, out any) error {
+	client := *c.client
+	client.Timeout = timeout
+	return c.requestJSONWithClient(ctx, &client, method, path, body, out)
+}
+
+// requestJSONWithClient sends an authenticated JSON request using the provided HTTP client.
+func (c *whatsappChannel) requestJSONWithClient(ctx context.Context, client *http.Client, method, path string, body map[string]any, out any) error {
+	payload, _ := json.Marshal(body)
+	req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.account.GatewayBaseURL, "/")+"/"+strings.TrimLeft(path, "/"), bytes.NewReader(payload))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if auth := c.authHeader(); auth != "" {
+		req.Header.Set("Authorization", auth)
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024*1024))
+	if resp.StatusCode >= 400 {
+		return fmt.Errorf("status: %d, response: %s", resp.StatusCode, string(respBody))
+	}
+	if out != nil && len(bytes.TrimSpace(respBody)) > 0 {
+		return json.Unmarshal(respBody, out)
+	}
+	return nil
+}
+
+// sessionKey returns the configured WhatsApp gateway session key.
+func (c *whatsappChannel) sessionKey() string {
+	if strings.TrimSpace(c.account.SessionKey) == "" {
+		return "default"
+	}
+	return c.account.SessionKey
+}
+
+// authHeader formats the optional gateway token as an Authorization header.
+func (c *whatsappChannel) authHeader() string {
+	token := strings.TrimSpace(c.account.GatewayToken)
+	if token == "" {
+		return ""
+	}
+	if strings.HasPrefix(strings.ToLower(token), "bearer ") {
+		return token
+	}
+	return "Bearer " + token
+}
+
+// eventsURL builds the websocket URL used to consume gateway events after the cursor.
+func (c *whatsappChannel) eventsURL() (string, error) {
+	base, err := url.Parse(strings.TrimRight(c.account.GatewayBaseURL, "/"))
+	if err != nil {
+		return "", err
+	}
+	switch base.Scheme {
+	case "http":
+		base.Scheme = "ws"
+	case "https":
+		base.Scheme = "wss"
+	}
+	base.Path = strings.TrimRight(base.Path, "/") + "/whatsapp/" + url.PathEscape(c.sessionKey()) + "/events/ws"
+	q := base.Query()
+	c.mu.Lock()
+	q.Set("after", strconv.FormatInt(c.eventCursor, 10))
+	c.mu.Unlock()
+	base.RawQuery = q.Encode()
+	return base.String(), nil
+}
+
+// firstString returns the first non-empty string value among several config keys.
+func firstString(cfg map[string]any, keys ...string) string {
+	for _, key := range keys {
+		if value := strings.TrimSpace(fmt.Sprint(cfg[key])); value != "" && value != "<nil>" {
+			return value
+		}
+	}
+	return ""
+}
+
+// valueOrDefault converts a snapshot value to string with an empty-value fallback.
+func valueOrDefault(value any, fallback string) string {
+	text := strings.TrimSpace(fmt.Sprint(value))
+	if text == "" || text == "<nil>" {
+		return fallback
+	}
+	return text
+}
+
+// numeric converts gateway numeric fields from common JSON-decoded Go types.
+func numeric(value any) float64 {
+	switch v := value.(type) {
+	case float64:
+		return v
+	case int:
+		return float64(v)
+	case int64:
+		return float64(v)
+	case json.Number:
+		n, _ := v.Float64()
+		return n
+	case string:
+		n, _ := strconv.ParseFloat(v, 64)
+		return n
+	default:
+		return 0
+	}
+}
+
+// stringPtr converts an empty string to nil for Python-compatible JSON null fields.
+func stringPtr(value string) *string {
+	if value == "" {
+		return nil
+	}
+	return &value
+}
+
+// floatPtr converts zero timestamps to nil for Python-compatible JSON null fields.
+func floatPtr(value float64) *float64 {
+	if value == 0 {
+		return nil
+	}
+	return &value
+}
+
+// resetTimer restarts a timer after draining any pending tick.
+func resetTimer(timer *time.Timer, delay time.Duration) {
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+	timer.Reset(delay)
+}
--- a/internal/channels/whatsapp_gateway.go
+++ b/internal/channels/whatsapp_gateway.go
@@ -0,0 +1,255 @@
+//
+//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+package channels
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	gatewayStartTimeout = 30 * time.Second
+	gatewayProbeEvery   = 500 * time.Millisecond
+	gatewayStopTimeout  = 10 * time.Second
+)
+
+type gatewayProcess struct {
+	cmd  *exec.Cmd
+	done chan struct{}
+}
+
+type gatewayRuntime struct {
+	mu      sync.Mutex
+	process *gatewayProcess
+}
+
+var gateway gatewayRuntime
+
+// syncWhatsAppGateway starts or stops the shared WhatsApp gateway process when management is enabled.
+func syncWhatsAppGateway(ctx context.Context, enabled bool) error {
+	if !enabled {
+		return gateway.stop()
+	}
+	if !gatewayEnabled() {
+		return nil
+	}
+	return gateway.start(ctx)
+}
+
+// gatewayEnabled reads the environment switch controlling gateway process management.
+func gatewayEnabled() bool {
+	raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_ENABLED"))
+	if raw == "" {
+		return false
+	}
+	v, err := strconv.ParseBool(raw)
+	return err == nil && v
+}
+
+// gatewayCommand resolves the command and working directory used to start the gateway.
+func gatewayCommand() ([]string, string) {
+	if raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_COMMAND")); raw != "" {
+		return strings.Fields(raw), gatewayWorkdir()
+	}
+
+	entry := filepath.Join(gatewayWorkdir(), "index.js")
+	if _, err := os.Stat(entry); err != nil {
+		return nil, gatewayWorkdir()
+	}
+	return []string{"node", entry}, gatewayWorkdir()
+}
+
+// gatewayWorkdir resolves the WhatsApp gateway-node working directory.
+func gatewayWorkdir() string {
+	if raw := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_WORKDIR")); raw != "" {
+		return raw
+	}
+	_, file, _, ok := runtime.Caller(0)
+	if !ok {
+		return "."
+	}
+	return filepath.Clean(filepath.Join(filepath.Dir(file), "../../api/channels/whatsapp/gateway-node"))
+}
+
+// start launches the gateway process if it is not already running.
+func (r *gatewayRuntime) start(ctx context.Context) error {
+	r.mu.Lock()
+	if r.process != nil {
+		process := r.process
+		select {
+		case <-process.done:
+			r.process = nil
+		default:
+			r.mu.Unlock()
+			if err := waitForGateway(ctx, gatewayStartTimeout); err != nil {
+				r.clearProcess(process)
+				_ = stopGatewayProcess(process)
+				return err
+			}
+			return nil
+		}
+	}
+	if gatewayReachable(ctx) {
+		r.mu.Unlock()
+		return nil
+	}
+
+	argv, cwd := gatewayCommand()
+	if len(argv) == 0 {
+		r.mu.Unlock()
+		return errors.New("WhatsApp gateway command is not configured")
+	}
+
+	cmd := exec.Command(argv[0], argv[1:]...)
+	cmd.Dir = cwd
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	cmd.Env = os.Environ()
+	if err := cmd.Start(); err != nil {
+		r.mu.Unlock()
+		return err
+	}
+	process := &gatewayProcess{cmd: cmd, done: make(chan struct{})}
+	r.process = process
+	r.mu.Unlock()
+
+	go func() {
+		err := cmd.Wait()
+		close(process.done)
+
+		r.mu.Lock()
+		owned := r.process == process
+		if r.process == process {
+			r.process = nil
+		}
+		r.mu.Unlock()
+		if err != nil && owned {
+			log.Printf("whatsapp gateway exited: %v", err)
+		}
+	}()
+	if err := waitForGateway(ctx, gatewayStartTimeout); err != nil {
+		r.clearProcess(process)
+		_ = stopGatewayProcess(process)
+		return err
+	}
+	return nil
+}
+
+// stop terminates the gateway process if this runtime started one.
+func (r *gatewayRuntime) stop() error {
+	r.mu.Lock()
+	process := r.process
+	r.process = nil
+	r.mu.Unlock()
+	return stopGatewayProcess(process)
+}
+
+// clearProcess removes process from the runtime if it is still the current managed process.
+func (r *gatewayRuntime) clearProcess(process *gatewayProcess) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.process == process {
+		r.process = nil
+	}
+}
+
+// stopGatewayProcess terminates one owned gateway process and waits for it to exit.
+func stopGatewayProcess(process *gatewayProcess) error {
+	if process == nil || process.cmd == nil || process.cmd.Process == nil {
+		return nil
+	}
+	select {
+	case <-process.done:
+		return nil
+	default:
+	}
+
+	if err := process.cmd.Process.Signal(os.Interrupt); err != nil {
+		_ = process.cmd.Process.Kill()
+		return err
+	}
+	select {
+	case <-process.done:
+		return nil
+	case <-time.After(gatewayStopTimeout):
+		_ = process.cmd.Process.Kill()
+	}
+	select {
+	case <-process.done:
+	case <-time.After(2 * time.Second):
+	}
+	return nil
+}
+
+// gatewayReachable reports whether an externally managed gateway is already listening.
+func gatewayReachable(ctx context.Context) bool {
+	probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(probeCtx, http.MethodGet, gatewayHealthURL(), nil)
+	if err != nil {
+		return false
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return false
+	}
+	defer resp.Body.Close()
+	return resp.StatusCode >= 200 && resp.StatusCode < 300
+}
+
+// waitForGateway blocks until the gateway health endpoint is ready or times out.
+func waitForGateway(ctx context.Context, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for {
+		if gatewayReachable(ctx) {
+			return nil
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("WhatsApp gateway did not become ready at %s within %s", gatewayHealthURL(), timeout)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(gatewayProbeEvery):
+		}
+	}
+}
+
+// gatewayHealthURL returns the health endpoint for the default managed gateway address.
+func gatewayHealthURL() string {
+	host := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_HOST"))
+	if host == "" {
+		host = "127.0.0.1"
+	}
+	port := strings.TrimSpace(os.Getenv("WHATSAPP_GATEWAY_PORT"))
+	if port == "" {
+		port = "3005"
+	}
+	return fmt.Sprintf("http://%s:%s/health", host, port)
+}
--- a/internal/dao/chat_channel.go
+++ b/internal/dao/chat_channel.go
@@ -58,3 +58,10 @@ func (dao *ChatChannelDAO) ListByTenantID(ctx context.Context, db *gorm.DB, tena

 	return results, err
 }
+
+// ListActive returns all enabled chat-channel rows for the runtime reconciler.
+func (dao *ChatChannelDAO) ListActive(ctx context.Context, db *gorm.DB) ([]*entity.ChatChannel, error) {
+	rows := make([]*entity.ChatChannel, 0)
+	err := db.WithContext(ctx).Where("status = ?", 1).Find(&rows).Error
+	return rows, err
+}
--- a/internal/handler/chat_channel.go
+++ b/internal/handler/chat_channel.go
@@ -30,6 +30,7 @@ type ChatChannelService interface {
 	List(ctx context.Context, tenantID string) ([]*entity.ChatChannelListResponse, error)
 	GetChatChannel(ctx context.Context, userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error)
 	UpdateChatChannel(ctx context.Context, userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error)
+	GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error)
 	DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error)
 }

@@ -49,7 +50,7 @@ func NewChatChannel() *ChatChannelHandler {
 type CreateChatChannelRequest struct {
 	Name    string         `json:"name" binding:"required"`
 	Channel string         `json:"channel" binding:"required"`
-	Config  entity.JSONMap `json:"config" binding:"required"`
+	Config  entity.JSONMap `json:"config"`
 	ChatID  *string        `json:"chat_id"`
 }

@@ -66,6 +67,9 @@ func (h *ChatChannelHandler) CreateChatChannel(c *gin.Context) {
 		common.ResponseWithCodeData(c, common.CodeDataError, nil, "Invalid request: "+err.Error())
 		return
 	}
+	if req.Config == nil {
+		req.Config = entity.JSONMap{}
+	}

 	ctx := c.Request.Context()

@@ -167,6 +171,35 @@ func (h *ChatChannelHandler) UpdateChatChannel(c *gin.Context) {
 	common.SuccessWithData(c, result, "success")
 }

+// GetChatChannelRuntime returns live runtime metadata for a running chat channel.
+func (h *ChatChannelHandler) GetChatChannelRuntime(c *gin.Context) {
+	user, errorCode, errorMessage := GetUser(c)
+	if errorCode != common.CodeSuccess {
+		common.ErrorWithCode(c, errorCode, errorMessage)
+		return
+	}
+
+	userID := strings.TrimSpace(user.ID)
+	if userID == "" {
+		common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "user_id is required")
+		return
+	}
+
+	channelID := strings.TrimSpace(c.Param("channel_id"))
+	if channelID == "" {
+		common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "channel_id is required")
+		return
+	}
+
+	ctx := c.Request.Context()
+	result, code, err := h.chatChannelService.GetChatChannelRuntime(ctx, userID, channelID)
+	if code != common.CodeSuccess || err != nil {
+		writeChatChannelError(c, code, chatChannelErrMsg(code, err))
+		return
+	}
+	common.SuccessWithData(c, result, "success")
+}
+
 // DeleteChatChannel handles DELETE /chat-channels/:channel_id.
 func (h *ChatChannelHandler) DeleteChatChannel(c *gin.Context) {
 	user, errorCode, errorMessage := GetUser(c)
--- a/internal/handler/chat_channel_test.go
+++ b/internal/handler/chat_channel_test.go
@@ -16,11 +16,12 @@ import (
 )

 type fakeChatChannelService struct {
-	createFn func(tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error)
-	listFn   func(tenantID string) ([]*entity.ChatChannelListResponse, error)
-	getFn    func(userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error)
-	updateFn func(userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error)
-	deleteFn func(userID, channelID string) (bool, common.ErrorCode, error)
+	createFn  func(tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error)
+	listFn    func(tenantID string) ([]*entity.ChatChannelListResponse, error)
+	getFn     func(userID, channelID string) (*entity.ChatChannel, common.ErrorCode, error)
+	updateFn  func(userID, channelID string, req map[string]interface{}) (*entity.ChatChannel, common.ErrorCode, error)
+	runtimeFn func(userID, channelID string) (map[string]any, common.ErrorCode, error)
+	deleteFn  func(userID, channelID string) (bool, common.ErrorCode, error)
 }

 func (f fakeChatChannelService) CreateChatChannel(ctx context.Context, tenantID, name, channelType string, config entity.JSONMap, chatID *string) (*entity.ChatChannel, error) {
@@ -51,6 +52,13 @@ func (f fakeChatChannelService) UpdateChatChannel(ctx context.Context, userID, c
 	return f.updateFn(userID, channelID, req)
 }

+func (f fakeChatChannelService) GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error) {
+	if f.runtimeFn == nil {
+		return nil, common.CodeServerError, errors.New("unexpected GetChatChannelRuntime call")
+	}
+	return f.runtimeFn(userID, channelID)
+}
+
 func (f fakeChatChannelService) DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error) {
 	if f.deleteFn == nil {
 		return false, common.CodeServerError, errors.New("unexpected DeleteChatChannel call")
@@ -327,3 +335,44 @@ func TestChatChannelHandlerDeleteChatChannelSuccess(t *testing.T) {
 		t.Fatalf("payload=%v", payload)
 	}
 }
+
+func TestChatChannelHandlerGetRuntimeSuccess(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	var gotUserID, gotChannelID string
+	h := &ChatChannelHandler{
+		chatChannelService: fakeChatChannelService{
+			runtimeFn: func(userID, channelID string) (map[string]any, common.ErrorCode, error) {
+				gotUserID = userID
+				gotChannelID = channelID
+				return map[string]any{"status": "waiting"}, common.CodeSuccess, nil
+			},
+		},
+	}
+
+	router := gin.New()
+	router.GET("/api/v1/chat-channels/:channel_id/runtime", func(c *gin.Context) {
+		c.Set("user", &entity.User{ID: "tenant-1"})
+		h.GetChatChannelRuntime(c)
+	})
+
+	resp := httptest.NewRecorder()
+	req := httptest.NewRequest(http.MethodGet, "/api/v1/chat-channels/cc-1/runtime", nil)
+	router.ServeHTTP(resp, req)
+
+	if gotUserID != "tenant-1" || gotChannelID != "cc-1" {
+		t.Fatalf("userID=%q channelID=%q", gotUserID, gotChannelID)
+	}
+
+	var payload map[string]interface{}
+	if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
+		t.Fatalf("unmarshal response: %v", err)
+	}
+	if payload["code"] != float64(common.CodeSuccess) {
+		t.Fatalf("payload=%v", payload)
+	}
+	data, _ := payload["data"].(map[string]interface{})
+	if data["status"] != "waiting" {
+		t.Fatalf("payload=%v", payload)
+	}
+}
--- a/internal/router/router.go
+++ b/internal/router/router.go
@@ -691,13 +691,14 @@ func (r *Router) Setup(engine *gin.Engine) {
 			}

 			// Chat Channel
-			chanChannel := v1.Group("/chat-channels")
+			chatChannel := v1.Group("/chat-channels")
 			{
-				chanChannel.POST("", r.chatChannelHandler.CreateChatChannel)
-				chanChannel.GET("", r.chatChannelHandler.ListChatChannel)
-				chanChannel.GET("/:channel_id", r.chatChannelHandler.GetChatChannel)
-				chanChannel.PATCH("/:channel_id", r.chatChannelHandler.UpdateChatChannel)
-				chanChannel.DELETE("/:channel_id", r.chatChannelHandler.DeleteChatChannel)
+				chatChannel.POST("", r.chatChannelHandler.CreateChatChannel)
+				chatChannel.GET("", r.chatChannelHandler.ListChatChannel)
+				chatChannel.GET("/:channel_id", r.chatChannelHandler.GetChatChannel)
+				chatChannel.PATCH("/:channel_id", r.chatChannelHandler.UpdateChatChannel)
+				chatChannel.DELETE("/:channel_id", r.chatChannelHandler.DeleteChatChannel)
+				chatChannel.GET("/:channel_id/runtime", r.chatChannelHandler.GetChatChannelRuntime)
 			}

 			// Langfuse tracing keys
--- a/internal/service/chat_channel.go
+++ b/internal/service/chat_channel.go
@@ -18,6 +18,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
+	"strings"
+	"sync"
+
 	"ragflow/internal/utility"

 	"ragflow/internal/common"
@@ -29,6 +33,7 @@ type ChatChannelService struct {
 	chatChannelDAO *dao.ChatChannelDAO
 	chatDAO        *dao.ChatDAO
 	userTenantDAO  *dao.UserTenantDAO
+	chatSessionSvc *ChatSessionService
 }

 func NewChatChannelService() *ChatChannelService {
@@ -36,9 +41,42 @@ func NewChatChannelService() *ChatChannelService {
 		chatChannelDAO: dao.NewChatChannel(),
 		chatDAO:        dao.NewChatDAO(),
 		userTenantDAO:  dao.NewUserTenantDAO(),
+		chatSessionSvc: NewChatSessionService(),
 	}
 }

+type ChatChannelIncomingMessage struct {
+	Channel   string
+	AccountID string
+	ChatID    string
+	ChatType  string
+	MessageID string
+	SenderID  string
+	Text      string
+}
+
+// ChatChannelRuntimeProvider reads runtime metadata for one live chat-channel instance.
+type ChatChannelRuntimeProvider func(channelID string) (map[string]any, bool)
+
+var (
+	chatChannelRuntimeProviderMu sync.RWMutex
+	chatChannelRuntimeProvider   ChatChannelRuntimeProvider
+)
+
+// SetChatChannelRuntimeProvider installs the runtime snapshot reader used by the chat-channel API.
+func SetChatChannelRuntimeProvider(provider ChatChannelRuntimeProvider) {
+	chatChannelRuntimeProviderMu.Lock()
+	defer chatChannelRuntimeProviderMu.Unlock()
+	chatChannelRuntimeProvider = provider
+}
+
+// getChatChannelRuntimeProvider returns the current runtime snapshot reader.
+func getChatChannelRuntimeProvider() ChatChannelRuntimeProvider {
+	chatChannelRuntimeProviderMu.RLock()
+	defer chatChannelRuntimeProviderMu.RUnlock()
+	return chatChannelRuntimeProvider
+}
+
 func (s *ChatChannelService) Insert(ctx context.Context, channel *entity.ChatChannel) error {
 	if channel == nil {
 		return errors.New("channel is nil")
@@ -216,6 +254,88 @@ func (s *ChatChannelService) UpdateChatChannel(ctx context.Context, userID, chan
 	return updated, common.CodeSuccess, nil
 }

+// GetChatChannelRuntime returns the live runtime metadata for a WhatsApp chat channel.
+func (s *ChatChannelService) GetChatChannelRuntime(ctx context.Context, userID, channelID string) (map[string]any, common.ErrorCode, error) {
+	conn, ok, err := s.accessible(ctx, userID, channelID)
+	if err != nil {
+		return nil, common.CodeServerError, err
+	}
+	if !ok {
+		return nil, common.CodeAuthenticationError, errors.New("no authorization")
+	}
+	if conn == nil {
+		return nil, common.CodeDataError, errors.New("can't find this chat channel")
+	}
+
+	if conn.Channel != "whatsapp" {
+		return nil, common.CodeDataError, errors.New("runtime snapshot is only available for WhatsApp")
+	}
+
+	if provider := getChatChannelRuntimeProvider(); provider != nil {
+		if snapshot, ok := provider(channelID); ok {
+			return snapshot, common.CodeSuccess, nil
+		}
+	}
+	return waitingRuntimeSnapshotMap(channelID), common.CodeSuccess, nil
+}
+
+// HandleIncomingMessage routes one external channel message through the bound RAGFlow dialog.
+func (s *ChatChannelService) HandleIncomingMessage(ctx context.Context, msg ChatChannelIncomingMessage) (string, error) {
+	if strings.TrimSpace(msg.Text) == "" {
+		return "", nil
+	}
+
+	channel, err := s.chatChannelDAO.GetByIDOnly(ctx, dao.DB, msg.AccountID)
+	if err != nil {
+		if dao.IsNotFoundErr(err) {
+			return "", nil
+		}
+		return "", err
+	}
+	if channel.ChatID == nil || strings.TrimSpace(*channel.ChatID) == "" {
+		return "", nil
+	}
+
+	dialog, err := s.chatDAO.GetByID(ctx, dao.DB, *channel.ChatID)
+	if err != nil {
+		if dao.IsNotFoundErr(err) {
+			return "", nil
+		}
+		return "", err
+	}
+
+	session, err := s.chatSessionSvc.GetOrCreateForChannel(ctx, dialog.ID, channel.ID, msg.ChatID)
+	if err != nil {
+		return "", err
+	}
+
+	result, err := s.chatSessionSvc.ChatCompletions(
+		ctx,
+		dialog.TenantID,
+		dialog.ID,
+		session.ID,
+		[]map[string]interface{}{{"role": "user", "content": msg.Text}},
+		"",
+		nil,
+		"",
+		nil,
+		map[string]interface{}{"quote": false},
+		false,
+		false,
+		false,
+		nil,
+	)
+	if err != nil {
+		log.Printf("chat channel %s completion failed: %v", channel.ID, err)
+		return "Sorry, I couldn't process your message right now.", nil
+	}
+	if result == nil {
+		return "", nil
+	}
+	answer, _ := result["answer"].(string)
+	return answer, nil
+}
+
 func (s *ChatChannelService) DeleteChatChannel(ctx context.Context, userID, channelID string) (bool, common.ErrorCode, error) {
 	channel, ok, err := s.accessible(ctx, userID, channelID)
 	if err != nil {
@@ -233,3 +353,20 @@ func (s *ChatChannelService) DeleteChatChannel(ctx context.Context, userID, chan
 	}
 	return true, common.CodeSuccess, nil
 }
+
+// waitingRuntimeSnapshotMap returns the API fallback payload before a runtime instance starts.
+func waitingRuntimeSnapshotMap(channelID string) map[string]any {
+	return map[string]any{
+		"account_id":       channelID,
+		"session_key":      channelID,
+		"status":           "waiting",
+		"connected_at":     nil,
+		"qr_updated_at":    nil,
+		"qr_data_url":      nil,
+		"last_error":       nil,
+		"session_id":       nil,
+		"last_snapshot_at": nil,
+		"gateway_base_url": nil,
+		"event_cursor":     int64(0),
+	}
+}
--- a/internal/service/chat_session.go
+++ b/internal/service/chat_session.go
@@ -18,6 +18,7 @@ package service

 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -1827,6 +1828,43 @@ func (s *ChatSessionService) createSessionForCompletion(ctx context.Context, cha
 	return session, nil
 }

+// GetOrCreateForChannel finds or creates the deterministic conversation for one external chat.
+func (s *ChatSessionService) GetOrCreateForChannel(ctx context.Context, dialogID, channelID, chatID string) (*entity.ChatSession, error) {
+	sessionID := channelSessionID(dialogID, channelID, chatID)
+	session, err := s.chatSessionDAO.GetByID(ctx, dao.DB, sessionID)
+	if err == nil {
+		return session, nil
+	}
+	if !dao.IsNotFoundErr(err) {
+		return nil, err
+	}
+
+	messagesJSON, _ := json.Marshal([]map[string]interface{}{})
+	referenceJSON, _ := json.Marshal([]interface{}{})
+	name := fmt.Sprintf("channel:%s:%s", channelID, chatID)
+	session = &entity.ChatSession{
+		ID:        sessionID,
+		DialogID:  dialogID,
+		Name:      &name,
+		Message:   messagesJSON,
+		Reference: referenceJSON,
+	}
+	if err = s.chatSessionDAO.Create(ctx, dao.DB, session); err != nil {
+		session, rereadErr := s.chatSessionDAO.GetByID(ctx, dao.DB, sessionID)
+		if rereadErr == nil {
+			return session, nil
+		}
+		return nil, err
+	}
+	return session, nil
+}
+
+// channelSessionID derives the stable SHA-256 conversation ID for a channel chat.
+func channelSessionID(dialogID, channelID, chatID string) string {
+	sum := sha256.Sum256([]byte(dialogID + ":" + channelID + ":" + chatID))
+	return fmt.Sprintf("%x", sum[:])[:32]
+}
+
 // appendSessionMessage appends the last user message to the session's message history.
 func (s *ChatSessionService) appendSessionMessage(session *entity.ChatSession, requestMsg []map[string]interface{}) *entity.ChatSession {
 	msgs := parseMessages(session.Message)