diff --git a/internal/service/agent.go b/internal/service/agent.go index 82536be4c0..eed0f972a4 100644 --- a/internal/service/agent.go +++ b/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 { diff --git a/internal/service/chat_channel_agent.go b/internal/service/chat_channel_agent.go index 94e9308da0..b5a20648e2 100644 --- a/internal/service/chat_channel_agent.go +++ b/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 { diff --git a/internal/service/chat_channel_agent_test.go b/internal/service/chat_channel_agent_test.go index 73c4c4ebdb..d14a202e46 100644 --- a/internal/service/chat_channel_agent_test.go +++ b/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) {