fix(agent): wire memory persistence in Message component (Go runtime) (#17256)

cmd/ragflow_server.go

@@ -49,7 +49,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"go.uber.org/zap"
 
-	_ "ragflow/internal/agent/component"
+	"ragflow/internal/agent/component"
 	"ragflow/internal/common"
 	"ragflow/internal/dao"
 	"ragflow/internal/engine"
@@ -705,6 +705,10 @@ func startServer(config *server.Config) {
 	mcpService := service.NewMCPService()
 	modelProviderService := service.NewModelProviderService()
 
+	// Wire the real MemorySaver so the Message component can persist
+	// conversation turns to memory stores declared in the canvas DSL.
+	component.SetMemorySaver(service.NewMemorySaverAdapter(memoryService))
+
 	// Initialize doc engine for skill search
 	docEngine := engine.Get()
 	documentDAO := dao.NewDocumentDAO()

internal/agent/component/message.go

@@ -56,6 +56,8 @@ type MessageComponent struct {
 	autoPlay     audio.Engine
 	voice        string
 	lang         string
+	memoryIDs    []string
+	userID       string
 }
 
 // NewMessageComponent constructs a Message component. The params map
@@ -82,6 +84,8 @@ func NewMessageComponent(params map[string]any) (Component, error) {
 		format = OutputFormat(v)
 	}
 	engine, voice, lang := extractAudioConfig(params)
+	memIDs := extractMemoryIDsFromAny(params["memory_ids"])
+	userID, _ := params["user_id"].(string)
 	return &MessageComponent{
 		name:         componentNameMessage,
 		text:         tpl,
@@ -89,6 +93,8 @@ func NewMessageComponent(params map[string]any) (Component, error) {
 		autoPlay:     engine,
 		voice:        voice,
 		lang:         lang,
+		memoryIDs:    memIDs,
+		userID:       userID,
 	}, nil
 }
 
@@ -285,22 +291,38 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m
 	// memory service returns ErrMemoryServiceMissing which we
 	// surface under outputs["memory_error"] so the message still
 	// flows.
-	memSave, _ := inputs["memory_save"].(bool)
-	if memSave {
-		memIDs := extractMemoryIDs(inputs)
-		if len(memIDs) > 0 {
-			saver := GetMemorySaver()
-			saveErr := saver.Save(ctx, MemorySaveRequest{
-				MemoryIDs:     memIDs,
-				AgentID:       state.TaskID,
-				SessionID:     state.RunID,
-				UserInput:     stringFromStateSys(state, "query"),
-				AgentResponse: rendered,
-			})
-			if saveErr != nil {
-				out["memory_error"] = saveErr.Error()
-				common.Error("Message: memory_save failed", saveErr)
-			}
+	//
+	// The effective memory IDs come from inputs (runtime override)
+	// or fall back to the DSL-declared m.memoryIDs. This matches
+	// the Python Message component, which saves whenever
+	// memory_ids is non-empty.
+	memIDs := extractMemoryIDs(inputs)
+	if len(memIDs) == 0 {
+		memIDs = m.memoryIDs
+	}
+	if len(memIDs) > 0 {
+		userID := stringFromStateSys(state, "user_id")
+		if userID == "" {
+			userID = m.userID
+		}
+		// If userID is a canvas variable reference (e.g. "{cpn@user_id}"),
+		// resolve it against the current state. Mirrors Python's
+		// agent/component/message.py:569-571.
+		if userID != "" && runtime.VarRefPattern.MatchString(userID) {
+			userID = runtime.ResolveTemplateForDisplay(userID, state)
+		}
+		saver := GetMemorySaver()
+		saveErr := saver.Save(ctx, MemorySaveRequest{
+			MemoryIDs:     memIDs,
+			UserID:        userID,
+			AgentID:       state.TaskID,
+			SessionID:     state.RunID,
+			UserInput:     stringFromStateSys(state, "query"),
+			AgentResponse: rendered,
+		})
+		if saveErr != nil {
+			out["memory_error"] = saveErr.Error()
+			common.Error("Message: memory_save failed", saveErr)
 		}
 	}
 
@@ -310,10 +332,13 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m
 // extractMemoryIDs normalises a memory_ids value from inputs /
 // params. Accepts []string and []any[string].
 func extractMemoryIDs(inputs map[string]any) []string {
-	v, ok := inputs["memory_ids"]
-	if !ok {
-		return nil
-	}
+	return extractMemoryIDsFromAny(inputs["memory_ids"])
+}
+
+// extractMemoryIDsFromAny normalises a memory_ids value from any
+// source (DSL params or runtime inputs). Accepts []string and
+// []any[string].
+func extractMemoryIDsFromAny(v any) []string {
 	switch x := v.(type) {
 	case []string:
 		return x
@@ -362,7 +387,7 @@ func fallbackMessageText(inputs map[string]any) string {
 func isMessageInfraInput(key string) bool {
 	switch key {
 	case "state", "__cpn_id__", "__legacy_noop__", "_created_time", "_elapsed_time",
-		"output_format", "voice", "lang", "auto_play", "memory_save", "stream":
+		"output_format", "voice", "lang", "auto_play", "memory_save", "memory_ids", "user_id", "stream":
 		return true
 	default:
 		return false

internal/agent/component/message_phase8b_test.go

@@ -285,6 +285,113 @@ func TestMessage_MemorySave_Success(t *testing.T) {
 	}
 }
 
+// TestMessage_MemorySave_FromDSLParams: memory_ids declared in the
+// DSL params (constructor) are honoured even when the runtime inputs
+// map does not carry memory_ids. This is the production path where
+// the pipeline only provides upstream component outputs.
+func TestMessage_MemorySave_FromDSLParams(t *testing.T) {
+	var saved MemorySaveRequest
+	saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error {
+		saved = req
+		return nil
+	})
+	SetMemorySaver(&saveFn)
+	defer SetMemorySaver(nil)
+
+	c, _ := NewMessageComponent(map[string]any{
+		"text":       "hi",
+		"memory_ids": []string{"dsl-m1", "dsl-m2"},
+	})
+	state := canvas.NewCanvasState("run-dsl", "task-dsl")
+	state.Sys["query"] = "hello?"
+	ctx := withStateForTest(context.Background(), state)
+
+	// Inputs simulate what the pipeline actually provides: only upstream
+	// outputs, NO memory_ids or memory_save keys.
+	_, err := c.Invoke(ctx, map[string]any{
+		"text":   "hi",
+		"stream": false,
+	})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if len(saved.MemoryIDs) != 2 || saved.MemoryIDs[0] != "dsl-m1" || saved.MemoryIDs[1] != "dsl-m2" {
+		t.Errorf("MemoryIDs from DSL params: %+v", saved.MemoryIDs)
+	}
+	if saved.AgentResponse != "hi" {
+		t.Errorf("AgentResponse: got %q, want hi", saved.AgentResponse)
+	}
+}
+
+// TestMessage_MemorySave_UserIDVariable: when user_id is a canvas
+// variable reference (e.g. "{begin@user_id}"), it is resolved
+// against the current state before being passed to the saver.
+// Mirrors Python agent/component/message.py:569-571.
+func TestMessage_MemorySave_UserIDVariable(t *testing.T) {
+	var saved MemorySaveRequest
+	saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error {
+		saved = req
+		return nil
+	})
+	SetMemorySaver(&saveFn)
+	defer SetMemorySaver(nil)
+
+	c, _ := NewMessageComponent(map[string]any{
+		"text":    "hi",
+		"user_id": "{begin@user_id}",
+	})
+	state := canvas.NewCanvasState("run-uid", "task-uid")
+	state.Sys["query"] = "hello?"
+	state.SetVar("begin", "user_id", "resolved-user-123")
+	ctx := withStateForTest(context.Background(), state)
+
+	_, err := c.Invoke(ctx, map[string]any{
+		"text":        "hi",
+		"stream":      false,
+		"memory_save": true,
+		"memory_ids":  []string{"m1"},
+	})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if saved.UserID != "resolved-user-123" {
+		t.Errorf("UserID: got %q, want resolved-user-123", saved.UserID)
+	}
+}
+
+// TestMessage_MemorySave_UserIDLiteral: a plain (non-variable)
+// user_id is passed through as-is.
+func TestMessage_MemorySave_UserIDLiteral(t *testing.T) {
+	var saved MemorySaveRequest
+	saveFn := memSaverFunc(func(_ context.Context, req MemorySaveRequest) error {
+		saved = req
+		return nil
+	})
+	SetMemorySaver(&saveFn)
+	defer SetMemorySaver(nil)
+
+	c, _ := NewMessageComponent(map[string]any{
+		"text":    "hi",
+		"user_id": "plain-user-456",
+	})
+	state := canvas.NewCanvasState("run-uid2", "task-uid2")
+	state.Sys["query"] = "hello?"
+	ctx := withStateForTest(context.Background(), state)
+
+	_, err := c.Invoke(ctx, map[string]any{
+		"text":        "hi",
+		"stream":      false,
+		"memory_save": true,
+		"memory_ids":  []string{"m1"},
+	})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if saved.UserID != "plain-user-456" {
+		t.Errorf("UserID: got %q, want plain-user-456", saved.UserID)
+	}
+}
+
 // memSaverFunc adapts a closure to the MemorySaver interface so
 // tests can record the call inline.
 type memSaverFunc func(ctx context.Context, req MemorySaveRequest) error

internal/service/memory_saver_adapter.go

@@ -0,0 +1,63 @@
+//
+//  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 service
+
+import (
+	"context"
+	"fmt"
+
+	"ragflow/internal/agent/component"
+)
+
+// memorySaverAdapter bridges the component.MemorySaver interface to
+// MemoryService.AddMessage. It is installed via component.SetMemorySaver
+// at boot time so that the Message component can persist conversation
+// turns to memory stores declared in the canvas DSL.
+type memorySaverAdapter struct {
+	svc *MemoryService
+}
+
+// NewMemorySaverAdapter returns a component.MemorySaver backed by the
+// given MemoryService.
+func NewMemorySaverAdapter(svc *MemoryService) component.MemorySaver {
+	return &memorySaverAdapter{svc: svc}
+}
+
+// Save implements component.MemorySaver. It delegates to
+// MemoryService.AddMessage which handles access filtering, message
+// construction, embedding, and async task queueing — the same pipeline
+// used by the REST API add_message endpoint.
+func (a *memorySaverAdapter) Save(ctx context.Context, req component.MemorySaveRequest) error {
+	if a == nil || a.svc == nil {
+		return fmt.Errorf("memory: saver adapter not initialised")
+	}
+	msg := MemoryMessage{
+		UserID:        req.UserID,
+		AgentID:       req.AgentID,
+		SessionID:     req.SessionID,
+		UserInput:     req.UserInput,
+		AgentResponse: req.AgentResponse,
+	}
+	ok, detail, err := a.svc.AddMessage(ctx, req.UserID, req.MemoryIDs, msg)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return fmt.Errorf("memory: %s", detail)
+	}
+	return nil
+}