fix(channels): apply saved agent updates to existing conversations (#22093)

internal/service/agent.go

@@ -1906,6 +1906,20 @@ func (s *AgentService) RunAgent(ctx context.Context, canvasID, sessionID, versio
 		versionRow *entity.UserCanvasVersion
 		dsl        map[string]any
 	)
+	channelRun, _ := ctx.Value(channelAgentRunContextKey{}).(bool)
+	refreshSessionDSL := channelRun && version == ""
+	paused := s.runner != nil && s.runner.Peek(canvasID, sessionID)
+	var resumeInterruptID string
+	if userInput != nil && s.runTracker != nil {
+		interruptID, ok, err := s.runTracker.GetInterruptID(ctx, runID)
+		if err != nil && refreshSessionDSL {
+			return nil, fmt.Errorf("RunAgent: load pending interrupt: %w: %w", err, ErrAgentStorageError)
+		}
+		if err == nil && ok {
+			resumeInterruptID = interruptID
+			paused = true
+		}
+	}
 	if version != "" {
 		row, err := s.versionDAO.GetByID(ctx, dao.DB, version)
 		if err != nil {
@@ -1942,7 +1956,12 @@ func (s *AgentService) RunAgent(ctx context.Context, canvasID, sessionID, versio
 		}
 		sessionFound = existingSession != nil
 		if existingSession != nil && len(existingSession.DSL) > 0 {
-			dsl = dslpkg.NormalizeForRun(existingSession.DSL)
+			if refreshSessionDSL && !paused {
+				dsl = refreshChannelAgentDSL(dsl, existingSession.DSL)
+			} else {
+				// A paused execution must resume the graph that wrote its checkpoint.
+				dsl = dslpkg.NormalizeForRun(existingSession.DSL)
+			}
 		}
 	}
 	if err := validateAgentChatModels(ctx, agentTenantID, dsl); err != nil {
@@ -1975,6 +1994,9 @@ func (s *AgentService) RunAgent(ctx context.Context, canvasID, sessionID, versio
 		"session_id":     sessionID,
 		"user_id":        userID,
 	}
+	if refreshSessionDSL && !paused {
+		root["__force_new_run__"] = true
+	}
 	// Drop failed first-touch sessions only when there was no user question.
 	if (!sessionFound || newSession) && stringifyAgentUserInput(agentRunQuery(userInput)) == "" {
 		root["__drop_session_on_failure__"] = true
@@ -1984,11 +2006,9 @@ func (s *AgentService) RunAgent(ctx context.Context, canvasID, sessionID, versio
 	// cancellation identity.
 	// Recover a pending UserFillUp interrupt from Redis when this request
 	// lands on a different process.
-	if userInput != nil && s.runTracker != nil {
-		if interruptID, ok, ierr := s.runTracker.GetInterruptID(ctx, runID); ierr == nil && ok {
-			root["__resume_interrupt_id__"] = interruptID
-			root["__resume_data__"] = userInput
-		}
+	if resumeInterruptID != "" {
+		root["__resume_interrupt_id__"] = resumeInterruptID
+		root["__resume_data__"] = userInput
 	}
 	if userInput != nil {
 		root["user_input"] = userInput
@@ -2469,6 +2489,10 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv
 		if cpID != "" {
 			invokeOpts = []compose.Option{compose.WithCheckPointID(cpID)}
 		}
+		if forceNewRun, _ := root["__force_new_run__"].(bool); forceNewRun {
+			// Completed checkpoints retain node state from the previous graph.
+			invokeOpts = append(invokeOpts, compose.WithForceNewRun())
+		}
 		// On a resume, the user input is the resume payload for the
 		// previously-paused UserFillUp node — it does NOT represent
 		// a fresh sys.query. The 'BEGIN' node writes inputs["query"]
@@ -2820,6 +2844,41 @@ func (s *AgentService) persistAgentRunSession(
 	return s.api4ConversationDAO.UpdateHistory(ctx, dao.DB, sessionID, agentID, userID, updates, history)
 }
 
+// refreshChannelAgentDSL keeps conversation data while replacing the workflow.
+func refreshChannelAgentDSL(dsl, previous map[string]any) map[string]any {
+	dsl = dslpkg.ResetForNewSession(dsl)
+	for _, key := range []string{"history", "memory"} {
+		if value, ok := previous[key]; ok {
+			dsl[key] = value
+		}
+	}
+	globals, _ := dsl["globals"].(map[string]any)
+	if globals == nil {
+		globals = map[string]any{}
+		dsl["globals"] = globals
+	}
+	oldGlobals, _ := previous["globals"].(map[string]any)
+	variables, _ := dsl["variables"].(map[string]any)
+	oldVariables, _ := previous["variables"].(map[string]any)
+	for key, value := range oldGlobals {
+		switch key {
+		case "sys.history", "sys.conversation_turns":
+			globals[key] = value
+		default:
+			if !strings.HasPrefix(key, "env.") {
+				continue
+			}
+			name := strings.TrimPrefix(key, "env.")
+			if current, ok := globals[key]; ok &&
+				reflect.TypeOf(current) == reflect.TypeOf(value) &&
+				reflect.DeepEqual(variables[name], oldVariables[name]) {
+				globals[key] = value
+			}
+		}
+	}
+	return dsl
+}
+
 func buildPersistedAgentDSL(runDSL map[string]any, state *canvas.CanvasState) entity.JSONMap {
 	dsl := make(entity.JSONMap, len(runDSL)+3)
 	for key, value := range runDSL {

internal/service/chat_channel_agent.go

@@ -30,6 +30,8 @@ type chatChannelAgentRunner interface {
 	RunAgent(context.Context, string, string, string, any, []map[string]any) (<-chan canvas.RunEvent, error)
 }
 
+type channelAgentRunContextKey struct{}
+
 func normalizedChannelTarget(id *string) *string {
 	if id == nil || strings.TrimSpace(*id) == "" {
 		return nil
@@ -59,6 +61,7 @@ func (s *ChatChannelService) handleChannelAgentMessage(ctx context.Context, chan
 	// RunAgent creates and authorizes the persisted session; this boundary only
 	// supplies its stable identity for the first message and subsequent turns.
 	ctx = WithAgentSessionID(ctx, sessionID)
+	ctx = context.WithValue(ctx, channelAgentRunContextKey{}, true)
 	ctx = withAgentChannelName(ctx, channel.Name)
 	events, err := s.agentRunner.RunAgent(ctx, *channel.AgentID, sessionID, "", msg.Text, nil)
 	if err != nil {

internal/service/chat_channel_agent_test.go

@@ -3,10 +3,12 @@ package service
 import (
 	"context"
 	"errors"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/alicebob/miniredis/v2/server"
 	"github.com/redis/go-redis/v9"
 
 	"ragflow/internal/agent/canvas"
@@ -99,6 +101,12 @@ func TestChatChannelAgentRevokedAccess(t *testing.T) {
 }
 
 func TestChatChannelAgentRealEngineContinuationAndResume(t *testing.T) {
+	for _, mode := range []string{"local", "different-process", "without-tracker"} {
+		t.Run(mode, func(t *testing.T) { testChatChannelAgentUpdateDuringPause(t, mode) })
+	}
+}
+
+func testChatChannelAgentUpdateDuringPause(t *testing.T, mode string) {
 	db := setupServiceTestDB(t)
 	pushServiceDB(t, db)
 	if err := db.AutoMigrate(&entity.ChatChannel{}, &entity.Chat{}, &entity.APIToken{},
@@ -124,6 +132,9 @@ func TestChatChannelAgentRealEngineContinuationAndResume(t *testing.T) {
 	tracker, mr := newRunTrackerForTest(t, time.Hour)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
 	t.Cleanup(func() { _ = client.Close() })
+	if mode == "without-tracker" {
+		tracker = nil
+	}
 	runner := NewAgentServiceWithOptions(canvas.NewKvrocksCheckPointStoreWithClient(client, time.Hour), nil, tracker)
 	svc := NewChatChannelServiceWithAgentService(runner)
 	channel, err := svc.CreateChatChannel(ctx, "channel", "telegram", entity.JSONMap{}, nil, new("workflow"))
@@ -135,6 +146,14 @@ func TestChatChannelAgentRealEngineContinuationAndResume(t *testing.T) {
 	if err != nil || !strings.Contains(answer, "Your answer?") {
 		t.Fatalf("waiting: %q %v", answer, err)
 	}
+	updated := channelMessageDSL("updated")
+	if err := db.Model(&entity.UserCanvas{}).Where("id = ?", "workflow").Update("dsl", entity.JSONMap(updated)).Error; err != nil {
+		t.Fatal(err)
+	}
+	if mode == "different-process" {
+		// Recover the persisted interrupt without the first process's Runner.
+		svc.agentRunner = NewAgentServiceWithOptions(canvas.NewKvrocksCheckPointStoreWithClient(client, time.Hour), nil, tracker)
+	}
 	var session entity.API4Conversation
 	if err := db.First(&session, "id = ?", channelSessionID("workflow", channel.ID, "external-one")).Error; err != nil {
 		t.Fatalf("load session: %v", err)
@@ -147,9 +166,13 @@ func TestChatChannelAgentRealEngineContinuationAndResume(t *testing.T) {
 	if err != nil || answer != "got: hello" {
 		t.Fatalf("resume: %q %v", answer, err)
 	}
+	answer, err = svc.HandleIncomingMessage(t.Context(), msg, nil)
+	if err != nil || answer != "updated" {
+		t.Fatalf("fresh turn after resume: %q %v", answer, err)
+	}
 	msg.ChatID = "external-two"
 	answer, err = svc.HandleIncomingMessage(t.Context(), msg, nil)
-	if err != nil || !strings.Contains(answer, "Your answer?") {
+	if err != nil || answer != "updated" {
 		t.Fatalf("isolated conversation: %q %v", answer, err)
 	}
 	var count int64
@@ -165,6 +188,148 @@ func TestChatChannelAgentRealEngineContinuationAndResume(t *testing.T) {
 	}
 }
 
+func channelMessageDSL(text string) map[string]any {
+	return map[string]any{
+		"components": map[string]any{
+			"begin": map[string]any{"obj": map[string]any{"component_name": "Begin", "params": map[string]any{}}, "downstream": []any{"message"}},
+			"message": map[string]any{
+				"obj":      map[string]any{"component_name": "Message", "params": map[string]any{"text": text}},
+				"upstream": []any{"begin"},
+			},
+		},
+	}
+}
+
+func TestRefreshChannelAgentDSLRetainsOnlyCompatibleConversationState(t *testing.T) {
+	draft := channelMessageDSL("updated")
+	draft["globals"] = map[string]any{"env.keep": "default", "env.changed": "new", "env.retyped": 1.0, "env.added": true}
+	draft["variables"] = map[string]any{"changed": map[string]any{"type": "string", "value": "new"}}
+	old := map[string]any{
+		"history": []any{[]any{"user", "previous"}},
+		"memory":  []any{[]any{"remembered"}},
+		"path":    []any{"removed"},
+		"globals": map[string]any{
+			"env.keep": "remembered", "env.changed": "old", "env.retyped": "string", "env.removed": "old",
+			"sys.history": []any{"user: previous"}, "sys.conversation_turns": 3.0, "removed@output": "stale",
+		},
+		"variables": map[string]any{"changed": map[string]any{"type": "string", "value": "old"}},
+	}
+	got := refreshChannelAgentDSL(draft, old)
+	for _, key := range []string{"history", "memory"} {
+		if !reflect.DeepEqual(got[key], old[key]) {
+			t.Fatalf("%s = %#v, want %#v", key, got[key], old[key])
+		}
+	}
+	globals := got["globals"].(map[string]any)
+	for key, want := range map[string]any{
+		"env.keep": "remembered", "env.changed": "new", "env.retyped": 1.0, "env.added": true,
+		"sys.history": []any{"user: previous"}, "sys.conversation_turns": 3.0,
+	} {
+		if !reflect.DeepEqual(globals[key], want) {
+			t.Fatalf("%s = %#v, want %#v", key, globals[key], want)
+		}
+	}
+	for _, key := range []string{"env.removed", "removed@output"} {
+		if _, ok := globals[key]; ok {
+			t.Fatalf("retained obsolete state %s", key)
+		}
+	}
+	if len(got["path"].([]any)) != 0 || draft["globals"].(map[string]any)["env.keep"] != "default" {
+		t.Fatal("retained old execution path or mutated draft")
+	}
+}
+
+func TestChatChannelAgentUpdatesPreserveConversation(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	if err := db.AutoMigrate(&entity.ChatChannel{}, &entity.Chat{}, &entity.APIToken{},
+		&entity.TenantModelProvider{}, &entity.TenantModelInstance{}, &entity.TenantModel{}); err != nil {
+		t.Fatal(err)
+	}
+	ctx := serviceTestContext(t, "owner", "owner")
+	makeCanvasWithDSL(t, "workflow", "owner", "owner", "released", channelMessageDSL("v1"))
+	makeCanvasWithDSL(t, "other", "owner", "owner", "other-version", channelMessageDSL("other"))
+	tracker, mr := newRunTrackerForTest(t, time.Hour)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = client.Close() })
+	runner := NewAgentServiceWithOptions(canvas.NewKvrocksCheckPointStoreWithClient(client, time.Hour), nil, tracker)
+	svc := NewChatChannelServiceWithAgentService(runner)
+	channel, err := svc.CreateChatChannel(ctx, "channel", "telegram", entity.JSONMap{}, nil, new("workflow"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	msg := ChatChannelIncomingMessage{AccountID: channel.ID, ChatID: "external", Text: "first"}
+	if answer, err := svc.HandleIncomingMessage(t.Context(), msg, nil); err != nil || answer != "v1" {
+		t.Fatalf("first turn: %q %v", answer, err)
+	}
+	sessionID := channelSessionID("workflow", channel.ID, msg.ChatID)
+	session := getAPIConversationForTest(t, db, sessionID)
+	beforeHistory := session.DSL["history"].([]any)
+	if len(beforeHistory) != 2 {
+		t.Fatalf("first turn history = %#v", beforeHistory)
+	}
+	// Ordinary and explicit-version sessions retain their established snapshots.
+	run := func(id, version, want string) {
+		t.Helper()
+		events, err := runner.RunAgent(WithAgentSessionID(ctx, id), "workflow", id, version, "probe", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		messages, _, errs, _, _ := drainAgentEvents(t, events)
+		if len(errs) != 0 || len(messages) != 1 || messages[0].Content != want {
+			t.Fatalf("ordinary session %s: messages=%+v errors=%+v", id, messages, errs)
+		}
+	}
+	run("ordinary", "", "v1")
+	updated := channelMessageDSL("v2")
+	if err := db.Model(&entity.UserCanvas{}).Where("id = ?", "workflow").Update("dsl", entity.JSONMap(updated)).Error; err != nil {
+		t.Fatal(err)
+	}
+	msg.Text = "second"
+	if answer, err := svc.HandleIncomingMessage(t.Context(), msg, nil); err != nil || answer != "v2" {
+		t.Fatalf("updated same conversation: %q %v", answer, err)
+	}
+	session = getAPIConversationForTest(t, db, sessionID)
+	history := session.DSL["history"].([]any)
+	if len(history) != 4 || !reflect.DeepEqual(history[0], beforeHistory[0]) {
+		t.Fatalf("conversation history = %#v", history)
+	}
+	message := session.DSL["components"].(map[string]any)["message"].(map[string]any)
+	if text := message["params"].(map[string]any)["text"]; text != "v2" {
+		t.Fatalf("persisted graph text = %v", text)
+	}
+	run("ordinary", "", "v1")
+	run("explicit-version", "released", "v1")
+	run("new-ordinary", "", "v2")
+	for _, chatID := range []string{"new-external", "external"} {
+		msg.ChatID = chatID
+		if answer, err := svc.HandleIncomingMessage(t.Context(), msg, nil); err != nil || answer != "v2" {
+			t.Fatalf("conversation %s: %q %v", chatID, answer, err)
+		}
+	}
+	mr.Server().SetPreHook(func(peer *server.Peer, command string, args ...string) bool {
+		if strings.EqualFold(command, "HGET") && len(args) == 2 && args[1] == "interrupt_id" {
+			peer.WriteError("ERR unavailable interrupt state")
+			return true
+		}
+		return false
+	})
+	if _, err := svc.HandleIncomingMessage(t.Context(), msg, nil); !errors.Is(err, ErrAgentStorageError) {
+		t.Fatalf("unavailable interrupt state must prevent execution: %v", err)
+	}
+	mr.Server().SetPreHook(nil)
+	if _, _, err := svc.UpdateChatChannel(ctx, channel.ID, map[string]any{"agent_id": "other"}); err != nil {
+		t.Fatal(err)
+	}
+	if answer, err := svc.HandleIncomingMessage(t.Context(), msg, nil); err != nil || answer != "other" {
+		t.Fatalf("switched target: %q %v", answer, err)
+	}
+	otherSession := getAPIConversationForTest(t, db, channelSessionID("other", channel.ID, msg.ChatID))
+	if otherSession.DialogID != "other" || len(otherSession.DSL["history"].([]any)) != 2 {
+		t.Fatalf("switched target session = %+v", otherSession)
+	}
+}
+
 type channelAgentRunnerFunc func(context.Context, string, string, string, any, []map[string]any) (<-chan canvas.RunEvent, error)
 
 func (f channelAgentRunnerFunc) RunAgent(ctx context.Context, agent, session, version string, input any, files []map[string]any) (<-chan canvas.RunEvent, error) {