diff --git a/internal/agent/canvas/canvas.go b/internal/agent/canvas/canvas.go index 839d0e6670..1442fe99ec 100644 --- a/internal/agent/canvas/canvas.go +++ b/internal/agent/canvas/canvas.go @@ -48,6 +48,7 @@ type Canvas struct { Components map[string]CanvasComponent `json:"components"` Path []string `json:"path"` History []map[string]any `json:"history,omitempty"` + Memory []map[string]any `json:"memory,omitempty"` Retrieval map[string]any `json:"retrieval,omitempty"` Globals map[string]any `json:"globals,omitempty"` // NodeParents preserves the front-end graph's grouping metadata diff --git a/internal/agent/canvas/canvas_test.go b/internal/agent/canvas/canvas_test.go index c9db0bd831..c50eac7c5f 100644 --- a/internal/agent/canvas/canvas_test.go +++ b/internal/agent/canvas/canvas_test.go @@ -17,6 +17,7 @@ package canvas import ( "context" + "strings" "testing" ) @@ -86,3 +87,57 @@ func TestBeginToMessage_Smoke(t *testing.T) { t.Fatalf("template resolve: got %q want %q", got, "hello world") } } + +// TestBuildWorkflow_PreservesRequestSysValues verifies that per-request sys +// values take precedence over the empty defaults stored in a real Canvas DSL. +// The service parses uploads and sets user_id before Invoke; GenLocalState must +// not replace that state and let the first statePre restore stale DSL values. +func TestBuildWorkflow_PreservesRequestSysValues(t *testing.T) { + dsl := &Canvas{ + Globals: map[string]any{ + "sys.files": []any{}, + "sys.user_id": "", + }, + Components: map[string]CanvasComponent{ + "begin_0": { + Obj: CanvasComponentObj{ComponentName: "Begin", Params: map[string]any{}}, + Downstream: []string{"message_0"}, + }, + "message_0": { + Obj: CanvasComponentObj{ComponentName: "Message", Params: map[string]any{ + "text": "{{sys.files}} {{sys.user_id}}", + }}, + Upstream: []string{"begin_0"}, + }, + }, + Path: []string{"begin_0", "message_0"}, + } + + cc, err := Compile(context.Background(), dsl) + if err != nil { + t.Fatalf("Compile: %v", err) + } + + state := NewCanvasState("run-request-sys", "task-request-sys") + state.Sys["files"] = []string{"File: notes.txt\nContent as following:\nhello"} + state.Sys["user_id"] = "user-1" + ctx := withState(context.Background(), state) + if _, err := cc.Workflow.Invoke(ctx, map[string]any{"query": "hello"}); err != nil { + t.Fatalf("Invoke: %v", err) + } + + files, err := state.GetVar("sys.files") + if err != nil { + t.Fatalf("GetVar(sys.files): %v", err) + } + if got, ok := files.([]string); !ok || len(got) != 1 || !strings.Contains(got[0], "notes.txt") { + t.Fatalf("sys.files = %#v, want current request upload", files) + } + userID, err := state.GetVar("sys.user_id") + if err != nil { + t.Fatalf("GetVar(sys.user_id): %v", err) + } + if userID != "user-1" { + t.Fatalf("sys.user_id = %#v, want %q", userID, "user-1") + } +} diff --git a/internal/agent/canvas/decode.go b/internal/agent/canvas/decode.go index ee4f8c9bbe..d04d823999 100644 --- a/internal/agent/canvas/decode.go +++ b/internal/agent/canvas/decode.go @@ -44,6 +44,8 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) { if p, ok := dsl["globals"].(map[string]any); ok { c.Globals = p } + c.History = decodeHistory(dsl["history"]) + c.Memory = decodeMemory(dsl["memory"]) if graph, ok := dsl["graph"].(map[string]any); ok { if nodes, ok := graph["nodes"].([]any); ok { for _, raw := range nodes { @@ -83,6 +85,115 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) { return c, nil } +func decodeHistory(raw any) []map[string]any { + items, ok := raw.([]any) + if !ok || len(items) == 0 { + return []map[string]any{} + } + history := make([]map[string]any, 0, len(items)) + for _, item := range items { + var role string + var payload any + switch value := item.(type) { + case []any: // Persisted Canvas shape: [role, payload]. + if len(value) < 2 { + continue + } + role, _ = value[0].(string) + payload = value[1] + case map[string]any: // Runtime shape: {role, content/payload}. + role, _ = value["role"].(string) + if preserved, exists := value["payload"]; exists { + payload = preserved + } else { + payload = value["content"] + } + default: + continue + } + if role == "" { + continue + } + history = append(history, map[string]any{ + "role": role, + "content": decodedHistoryContent(payload), + "payload": payload, + }) + } + return history +} + +// decodeMemory restores [user, assistant, tool summary] entries. +func decodeMemory(raw any) []map[string]any { + items, ok := raw.([]any) + if !ok || len(items) == 0 { + return []map[string]any{} + } + memory := make([]map[string]any, 0, len(items)) + for _, item := range items { + switch value := item.(type) { + case []any: // [[user_query, assistant_response, tool_summary], ...] + if len(value) < 3 { + continue + } + memory = append(memory, map[string]any{ + "user": value[0], + "assistant": value[1], + "summary": value[2], + }) + case map[string]any: + memory = append(memory, map[string]any{ + "user": value["user"], + "assistant": value["assistant"], + "summary": value["summary"], + }) + } + } + return memory +} + +func decodedHistoryContent(payload any) string { + switch value := payload.(type) { + case nil: + return "" + case string: + return value + case map[string]any: + content, _ := value["content"].(string) + return content + default: + return fmt.Sprint(value) + } +} + +// EncodeHistory converts runtime conversation entries back to the Python DSL +// list-of-pairs shape: [[role, payload], ...]. +func EncodeHistory(history []map[string]any) []any { + out := make([]any, 0, len(history)) + for _, entry := range history { + role, _ := entry["role"].(string) + if role == "" { + continue + } + payload, exists := entry["payload"] + if !exists { + payload = entry["content"] + } + out = append(out, []any{role, payload}) + } + return out +} + +// EncodeMemory converts runtime tool-call memory back to Python's +// [[user, assistant, summary], ...] DSL shape. +func EncodeMemory(memory []map[string]any) []any { + out := make([]any, 0, len(memory)) + for _, entry := range memory { + out = append(out, []any{entry["user"], entry["assistant"], entry["summary"]}) + } + return out +} + func decodeComponentFields(comp map[string]any) (name string, params map[string]any, downstream []string, upstream []string) { if obj, ok := comp["obj"].(map[string]any); ok { name, _ = obj["component_name"].(string) diff --git a/internal/agent/canvas/decode_history_test.go b/internal/agent/canvas/decode_history_test.go new file mode 100644 index 0000000000..b5b2dd6b04 --- /dev/null +++ b/internal/agent/canvas/decode_history_test.go @@ -0,0 +1,87 @@ +// +// 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 canvas + +import ( + "reflect" + "testing" +) + +func TestDecodeFromDSLHistoryRoundTrip(t *testing.T) { + dslHistory := []any{ + []any{"user", "你好"}, + []any{"assistant", map[string]any{"content": "您好", "downloads": []any{"file-1"}}}, + } + dslMemory := []any{ + []any{"question", "answer", "searched the knowledge base"}, + } + dsl := map[string]any{ + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Begin", + "params": map[string]any{}, + }, + }, + }, + "history": dslHistory, + "memory": dslMemory, + } + + decoded, err := DecodeFromDSL(dsl) + if err != nil { + t.Fatalf("DecodeFromDSL: %v", err) + } + if len(decoded.History) != 2 { + t.Fatalf("history length = %d, want 2", len(decoded.History)) + } + if got := decoded.History[1]["content"]; got != "您好" { + t.Fatalf("assistant content = %v, want 您好", got) + } + if !reflect.DeepEqual(EncodeHistory(decoded.History), dslHistory) { + t.Fatalf("history round trip = %#v, want %#v", EncodeHistory(decoded.History), dslHistory) + } + if !reflect.DeepEqual(EncodeMemory(decoded.Memory), dslMemory) { + t.Fatalf("memory round trip = %#v, want %#v", EncodeMemory(decoded.Memory), dslMemory) + } +} + +func TestDecodeFromDSLSkipsMalformedHistoryEntries(t *testing.T) { + dsl := map[string]any{ + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Begin", + "params": map[string]any{}, + }, + }, + }, + "history": []any{ + []any{"user"}, + 42, + []any{"assistant", "valid"}, + }, + } + + decoded, err := DecodeFromDSL(dsl) + if err != nil { + t.Fatalf("DecodeFromDSL: %v", err) + } + if len(decoded.History) != 1 || decoded.History[0]["role"] != "assistant" { + t.Fatalf("decoded history = %#v, want one assistant entry", decoded.History) + } +} diff --git a/internal/agent/canvas/scheduler.go b/internal/agent/canvas/scheduler.go index 67dff27e4c..eb8fcecd5f 100644 --- a/internal/agent/canvas/scheduler.go +++ b/internal/agent/canvas/scheduler.go @@ -151,6 +151,12 @@ func statePre(ctx context.Context, in map[string]any, state *CanvasState) (map[s // the upstream outputs the state post handler already wrote. if state != nil { if ctxState, _, _ := runtime.GetStateFromContext[*runtime.CanvasState](ctx); ctxState != nil && ctxState != state { + localHistory := state.SnapshotHistory() + contextHistory := ctxState.SnapshotHistory() + localMemory := state.SnapshotMemory() + contextMemory := ctxState.SnapshotMemory() + localSysHistory := state.SnapshotSysHistory() + contextSysHistory := ctxState.SnapshotSysHistory() for cpnID, bucket := range state.Outputs { for k, v := range bucket { ctxState.SetVar(cpnID, k, v) @@ -166,6 +172,23 @@ func statePre(ctx context.Context, in map[string]any, state *CanvasState) (map[s for k, v := range globalsNS { ctxState.Globals[k] = v } + if len(contextHistory) >= len(localHistory) { + state.SetHistory(contextHistory) + } else { + ctxState.SetHistory(localHistory) + } + if len(contextMemory) >= len(localMemory) { + state.SetMemory(contextMemory) + } else { + ctxState.SetMemory(localMemory) + } + if len(contextSysHistory) >= len(localSysHistory) { + state.SetSysHistory(contextSysHistory) + ctxState.SetSysHistory(contextSysHistory) + } else { + state.SetSysHistory(localSysHistory) + ctxState.SetSysHistory(localSysHistory) + } } } snapshot := state.Snapshot() @@ -215,6 +238,8 @@ func statePost(ctx context.Context, out map[string]any, state *CanvasState) (map state.Sys = sysNS state.Env = envNS state.Globals = globalsNS + state.SetHistory(ctxState.SnapshotHistory()) + state.SetMemory(ctxState.SnapshotMemory()) } return out, nil } @@ -366,9 +391,16 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string return nil, fmt.Errorf("canvas: no components") } - // GenLocalState seeds each run with a fresh *CanvasState. eino calls - // this once per run and threads the result through StatePre/Post - // handlers via context. + // GenLocalState copies the request-initialized *CanvasState when the + // caller attached one to the context. The service layer populates that + // state with per-run sys values (query, files, user_id) and persisted + // history/memory before Invoke; replacing it here with DSL globals would + // make the first statePre copy stale defaults such as sys.files=[] back + // over the request values. + // + // Callers that do not attach a state still get a fresh DSL-seeded state. + // eino calls this once per run and threads the result through + // StatePre/Post handlers. // // The initial env/sys values come from c.Globals (the DSL-level // "globals" map) so that env.* references like "env.counter" resolve @@ -378,7 +410,23 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string // seeding here mirrors the Python canvas.__init__ → // self.globals["env.counter"] = 0 path. globals := c.Globals - genState := func(_ context.Context) *CanvasState { + genState := func(runCtx context.Context) *CanvasState { + if ctxState, _, _ := runtime.GetStateFromContext[*runtime.CanvasState](runCtx); ctxState != nil { + st := NewCanvasState(ctxState.RunID, ctxState.TaskID) + for cpnID, bucket := range ctxState.Snapshot() { + for key, value := range bucket { + st.SetVar(cpnID, key, value) + } + } + sysNS, envNS, globalsNS := ctxState.SnapshotNamespaces() + st.Sys = sysNS + st.Env = envNS + st.Globals = globalsNS + st.Path = append([]string(nil), ctxState.Path...) + st.SetHistory(ctxState.SnapshotHistory()) + st.SetMemory(ctxState.SnapshotMemory()) + return st + } st := NewCanvasState("", "") if globals != nil { for k, v := range globals { @@ -391,6 +439,8 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string } } } + st.SetHistory(c.History) + st.SetMemory(c.Memory) st.EnsureSysDate() return st } diff --git a/internal/agent/component/agent.go b/internal/agent/component/agent.go index 7278623df6..357e69ddee 100644 --- a/internal/agent/component/agent.go +++ b/internal/agent/component/agent.go @@ -575,7 +575,7 @@ func (c *AgentComponent) Invoke(ctx context.Context, inputs map[string]any) (map // what the Agent runner actually consumes. if p.OptimizeMultiTurn { if state, _, sErr := runtime.GetStateFromContext[*runtime.CanvasState](ctx); sErr == nil && state != nil { - if rephrased, err := optimizeMultiTurnQuestion(ctx, p, state.History); err == nil && rephrased != "" { + if rephrased, err := optimizeMultiTurnQuestion(ctx, p, state.SnapshotPriorHistory()); err == nil && rephrased != "" { p.UserPrompt = rephrased } } @@ -584,15 +584,12 @@ func (c *AgentComponent) Invoke(ctx context.Context, inputs map[string]any) (map msg, err := agentRunner(ctx, p) // Tool-call memory summarization. After the ReAct loop // completes, summarize the tool calls via an LLM and append to - // the canvas state's History so downstream turns (history - // window) see the prior tool usage as prior assistant turns. + // the canvas state's Memory. Conversation History is reserved for + // actual user/assistant turns maintained by the canvas service. if err == nil && msg != nil { if state, _, sErr := runtime.GetStateFromContext[*runtime.CanvasState](ctx); sErr == nil && state != nil { if summary, sErr2 := addToolCallMemory(ctx, p, msg); sErr2 == nil && summary != "" { - state.History = append(state.History, map[string]any{ - "role": "assistant", - "content": summary, - }) + state.AppendMemory(p.UserPrompt, msg.Content, summary) } } } diff --git a/internal/agent/component/history_window_test.go b/internal/agent/component/history_window_test.go index 1cac8b818e..e7d731059f 100644 --- a/internal/agent/component/history_window_test.go +++ b/internal/agent/component/history_window_test.go @@ -153,6 +153,38 @@ func TestLLM_Invoke_HistoryWindow_PrependsFromState(t *testing.T) { } } +func TestLLM_Invoke_HistoryWindow_DoesNotDuplicateCurrentUser(t *testing.T) { + stub := &stubInvoker{resp: &ChatInvokeResponse{Content: "ok", Model: "echo"}} + withStubInvoker(t, stub) + + state := runtime.NewCanvasState("rid", "tid") + state.AppendHistory("user", "earlier") + state.AppendHistory("assistant", map[string]any{"content": "earlier reply"}) + state.AppendCurrentUser("now") + c := NewLLMComponent(LLMParam{ + ModelID: "echo", + UserPrompt: "now", + MessageHistoryWindowSize: 5, + }) + ctx := runtime.WithState(context.Background(), state) + if _, err := c.Invoke(ctx, map[string]any{}); err != nil { + t.Fatalf("Invoke: %v", err) + } + + if len(stub.captured.Messages) != 3 { + t.Fatalf("messages = %#v, want two prior turns plus one current user", stub.captured.Messages) + } + currentCount := 0 + for _, message := range stub.captured.Messages { + if message.Content == "now" { + currentCount++ + } + } + if currentCount != 1 { + t.Fatalf("current user occurrence count = %d, want 1", currentCount) + } +} + // TestLLM_Invoke_HistoryWindow_ZeroIsNoop: when window is 0, history is // not prepended even if present in state. func TestLLM_Invoke_HistoryWindow_ZeroIsNoop(t *testing.T) { diff --git a/internal/agent/component/llm.go b/internal/agent/component/llm.go index 09c1b90423..e596c631ee 100644 --- a/internal/agent/component/llm.go +++ b/internal/agent/component/llm.go @@ -464,7 +464,7 @@ func (c *LLMComponent) Invoke(ctx context.Context, inputs map[string]any) (map[s // this is a no-op. if p.MessageHistoryWindowSize > 0 { if state, _, sErr := runtime.GetStateFromContext[*runtime.CanvasState](ctx); sErr == nil && state != nil { - msgs = prependHistory(msgs, state.History, p.MessageHistoryWindowSize) + msgs = prependHistory(msgs, state.SnapshotPriorHistory(), p.MessageHistoryWindowSize) } } // Apply message fitting (trim to context window) after all @@ -810,23 +810,27 @@ func extractDataImages(values []string) []string { return out } -// collectSysFiles splits sys.files from canvas globals into text parts +// collectSysFiles splits sys.files from canvas state into text parts // and image data URIs. The caller is responsible for handling any // {sys.files} placeholder replacement in the prompts. func collectSysFiles(state *runtime.CanvasState) (textParts, imageURIs []string) { - files, ok := state.Globals["sys.files"] + files, ok := state.Sys["files"] if !ok { return nil, nil } - fileList, ok := files.([]any) - if !ok || len(fileList) == 0 { - return nil, nil - } - for _, f := range fileList { - s, ok := f.(string) - if !ok { - continue + var fileList []string + switch values := files.(type) { + case []string: + fileList = values + case []any: + fileList = make([]string, 0, len(values)) + for _, value := range values { + if s, ok := value.(string); ok { + fileList = append(fileList, s) + } } + } + for _, s := range fileList { if strings.HasPrefix(s, "data:image/") { imageURIs = append(imageURIs, s) } else { diff --git a/internal/agent/component/message.go b/internal/agent/component/message.go index 3593e421dc..817efe0608 100644 --- a/internal/agent/component/message.go +++ b/internal/agent/component/message.go @@ -180,14 +180,17 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m text = fallbackMessageText(inputs) } - resolved, err := runtime.ResolveTemplate(text, state) - if err != nil { - return nil, fmt.Errorf("Message: %w", err) - } + // Message renders values for display, so keep Python's tolerant behavior: + // missing references become empty strings and structured values use JSON. + // Parameter-binding components continue to use the strict resolver. + resolved := runtime.ResolveTemplateForDisplay(text, state) // Extract downloads. Walks inputs for download-info maps so // callers can attach binaries to the message body. downloads := ExtractDownloads(resolved) + if downloads == nil { + downloads = make([]DownloadInfo, 0) + } if len(downloads) > 0 && downloadInfoString(resolved) { resolved = "" } @@ -214,9 +217,12 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m }) } - out := map[string]any{"content": rendered} - if len(downloads) > 0 { - out["downloads"] = downloads + // Python's Message output schema always contains downloads, including an + // empty list. Keeping the key is also important for the full terminal + // output recorded in Canvas history between conversation turns. + out := map[string]any{ + "content": rendered, + "downloads": downloads, } // auto_play TTS dispatch. The audio bytes are returned under diff --git a/internal/agent/component/message_test.go b/internal/agent/component/message_test.go index 33d7d3e29e..cace320e32 100644 --- a/internal/agent/component/message_test.go +++ b/internal/agent/component/message_test.go @@ -46,6 +46,27 @@ func TestMessage_ResolveTemplate(t *testing.T) { if _, ok := out["streamed_chunks"]; ok { t.Errorf("streamed_chunks must not be present, got %v", out["streamed_chunks"]) } + if downloads, ok := out["downloads"].([]DownloadInfo); !ok || len(downloads) != 0 { + t.Errorf("downloads = %#v, want an empty []DownloadInfo", out["downloads"]) + } +} + +func TestMessage_ResolveListReferenceAsJSON(t *testing.T) { + c, _ := NewMessageComponent(nil) + state := canvas.NewCanvasState("run-list", "task-list") + state.SetVar("list_0", "result", []any{"user: 1"}) + ctx := withStateForTest(context.Background(), state) + + out, err := c.Invoke(ctx, map[string]any{ + "text": "{{list_0@result}}", + "stream": false, + }) + if err != nil { + t.Fatalf("Invoke: %v", err) + } + if got, _ := out["content"].(string); got != `["user: 1"]` { + t.Fatalf("content = %q, want JSON list", got) + } } // TestMessage_Stream confirms the Stream() contract: the returned diff --git a/internal/agent/component/tool_call_memory_test.go b/internal/agent/component/tool_call_memory_test.go index f34e520178..0d9a808efe 100644 --- a/internal/agent/component/tool_call_memory_test.go +++ b/internal/agent/component/tool_call_memory_test.go @@ -45,7 +45,8 @@ func TestAddToolCallMemory_NoToolCalls(t *testing.T) { } // TestAddToolCallMemory_SummarizesAndAppendsToState: end-to-end — -// stub returns a summary; Agent.Invoke appends it to state.History. +// stub returns a summary; Agent.Invoke appends it to state.Memory without +// polluting the user/assistant conversation history. func TestAddToolCallMemory_SummarizesAndAppendsToState(t *testing.T) { stub := &stubInvoker{resp: &ChatInvokeResponse{ Content: "the assistant searched docs and found 3 results", @@ -77,15 +78,19 @@ func TestAddToolCallMemory_SummarizesAndAppendsToState(t *testing.T) { if stub.calls != 1 { t.Errorf("expected 1 LLM call (the memory summary), got %d", stub.calls) } - if len(state.History) != 1 { - t.Fatalf("expected 1 history entry, got %d: %+v", len(state.History), state.History) + memory := state.SnapshotMemory() + if len(memory) != 1 { + t.Fatalf("expected 1 memory entry, got %d: %+v", len(memory), memory) } - h := state.History[0] - if h["role"] != "assistant" { - t.Errorf("history role=%v, want assistant", h["role"]) + if len(state.SnapshotHistory()) != 0 { + t.Fatalf("tool summary polluted conversation history: %+v", state.SnapshotHistory()) } - if !strings.Contains(h["content"].(string), "searched docs") { - t.Errorf("history content missing summary: %q", h["content"]) + entry := memory[0] + if entry["user"] != "do it" { + t.Errorf("memory user=%v, want do it", entry["user"]) + } + if !strings.Contains(entry["summary"].(string), "searched docs") { + t.Errorf("memory summary missing tool result: %q", entry["summary"]) } } @@ -112,7 +117,7 @@ func TestAddToolCallMemory_LLMFailure(t *testing.T) { if err != nil { t.Fatalf("Invoke should not error when memory summary fails: %v", err) } - if len(state.History) != 0 { - t.Errorf("expected no history entry on summary failure, got %d", len(state.History)) + if len(state.SnapshotMemory()) != 0 { + t.Errorf("expected no memory entry on summary failure, got %d", len(state.SnapshotMemory())) } } diff --git a/internal/agent/component/vision_test.go b/internal/agent/component/vision_test.go index c952fd0959..3997ebc432 100644 --- a/internal/agent/component/vision_test.go +++ b/internal/agent/component/vision_test.go @@ -22,6 +22,8 @@ import ( "testing" "github.com/cloudwego/eino/schema" + + "ragflow/internal/agent/runtime" ) // TestToEinoMessages_PreservesUserInputMultiContent guards against the @@ -223,6 +225,20 @@ func TestBuildMessagesWithImages_EmptyImages_ReturnsTextMessage(t *testing.T) { } } +func TestCollectSysFiles_ReadsSysNamespaceAndStringSlices(t *testing.T) { + state := runtime.NewCanvasState("run-1", "task-1") + state.Sys["files"] = []string{"document text", "data:image/png;base64,cG5n"} + state.Globals["sys.files"] = []any{"stale global value"} + + texts, images := collectSysFiles(state) + if !reflect.DeepEqual(texts, []string{"document text"}) { + t.Fatalf("texts = %#v, want sys.files text", texts) + } + if !reflect.DeepEqual(images, []string{"data:image/png;base64,cG5n"}) { + t.Fatalf("images = %#v, want sys.files image", images) + } +} + // TestBuildMessagesWithImages_WithImages_UsesUserInputMultiContent: // when images are present, the user message is built with a Text part // followed by an Image part per URI. diff --git a/internal/agent/runtime/state.go b/internal/agent/runtime/state.go index e7cc19b9f6..70706b26fc 100644 --- a/internal/agent/runtime/state.go +++ b/internal/agent/runtime/state.go @@ -33,6 +33,7 @@ package runtime import ( "encoding/json" "fmt" + "reflect" "sort" "strings" "sync" @@ -51,22 +52,25 @@ import ( // - Env : env.* namespace (deployment-time constants) // - Path : entry-point sequence (Begin nodes) // - History : conversation history (chat-flow agents) +// - Memory : tool-call summaries kept separate from conversation turns // - Retrieval : aggregate retrieval result (chunks, doc_aggs) // - Globals : cross-canvas-instance globals // - CancelFlag : set when cancel signal received; nodes may poll // - RunID : unique per-run identifier (used by RunTracker + CheckPointStore) type CanvasState struct { - mu sync.RWMutex - Outputs map[string]map[string]any - Sys map[string]any - Env map[string]any - Path []string - History []map[string]any - Retrieval map[string]any - Globals map[string]any - CancelFlag *atomic.Bool - RunID string - TaskID string + mu sync.RWMutex + activeHistoryIndex int + Outputs map[string]map[string]any + Sys map[string]any + Env map[string]any + Path []string + History []map[string]any + Memory []map[string]any + Retrieval map[string]any + Globals map[string]any + CancelFlag *atomic.Bool + RunID string + TaskID string } // NewCanvasState returns a zero-valued CanvasState with all maps allocated. @@ -74,16 +78,18 @@ type CanvasState struct { // even before any cancel signal has been wired. func NewCanvasState(runID, taskID string) *CanvasState { s := &CanvasState{ - Outputs: make(map[string]map[string]any), - Sys: make(map[string]any), - Env: make(map[string]any), - Path: []string{}, - History: []map[string]any{}, - Retrieval: make(map[string]any), - Globals: make(map[string]any), - CancelFlag: &atomic.Bool{}, - RunID: runID, - TaskID: taskID, + activeHistoryIndex: -1, + Outputs: make(map[string]map[string]any), + Sys: make(map[string]any), + Env: make(map[string]any), + Path: []string{}, + History: []map[string]any{}, + Memory: []map[string]any{}, + Retrieval: make(map[string]any), + Globals: make(map[string]any), + CancelFlag: &atomic.Bool{}, + RunID: runID, + TaskID: taskID, } s.EnsureSysDate() return s @@ -125,16 +131,18 @@ func init() { // place. The CancelFlag is round-tripped as a bool (atomic.Bool can't // be marshalled directly without a wrapper). type canvasStateJSON struct { - Outputs map[string]map[string]any `json:"outputs"` - Sys map[string]any `json:"sys,omitempty"` - Env map[string]any `json:"env,omitempty"` - Path []string `json:"path,omitempty"` - History []map[string]any `json:"history,omitempty"` - Retrieval map[string]any `json:"retrieval,omitempty"` - Globals map[string]any `json:"globals,omitempty"` - CancelFlag bool `json:"cancel_flag"` - RunID string `json:"run_id"` - TaskID string `json:"task_id"` + ActiveHistoryIndex *int `json:"active_history_index,omitempty"` + Outputs map[string]map[string]any `json:"outputs"` + Sys map[string]any `json:"sys,omitempty"` + Env map[string]any `json:"env,omitempty"` + Path []string `json:"path,omitempty"` + History []map[string]any `json:"history,omitempty"` + Memory []map[string]any `json:"memory,omitempty"` + Retrieval map[string]any `json:"retrieval,omitempty"` + Globals map[string]any `json:"globals,omitempty"` + CancelFlag bool `json:"cancel_flag"` + RunID string `json:"run_id"` + TaskID string `json:"task_id"` } // MarshalJSON serialises the CanvasState for eino's StatePre/Post @@ -156,17 +164,24 @@ type canvasStateJSON struct { func (s *CanvasState) MarshalJSON() ([]byte, error) { s.mu.RLock() defer s.mu.RUnlock() + var activeHistoryIndex *int + if s.activeHistoryIndex >= 0 { + index := s.activeHistoryIndex + activeHistoryIndex = &index + } snap := canvasStateJSON{ - Outputs: s.Outputs, - Sys: s.Sys, - Env: s.Env, - Path: s.Path, - History: s.History, - Retrieval: s.Retrieval, - Globals: s.Globals, - CancelFlag: s.CancelFlag != nil && s.CancelFlag.Load(), - RunID: s.RunID, - TaskID: s.TaskID, + ActiveHistoryIndex: activeHistoryIndex, + Outputs: s.Outputs, + Sys: s.Sys, + Env: s.Env, + Path: s.Path, + History: s.History, + Memory: s.Memory, + Retrieval: s.Retrieval, + Globals: s.Globals, + CancelFlag: s.CancelFlag != nil && s.CancelFlag.Load(), + RunID: s.RunID, + TaskID: s.TaskID, } // Use SafeJSONMarshal to handle non-serializable values (funcs, // channels) that may have leaked into state maps. Mirrors the @@ -197,6 +212,11 @@ func (s *CanvasState) UnmarshalJSON(b []byte) error { } s.Path = snap.Path s.History = snap.History + s.activeHistoryIndex = -1 + if snap.ActiveHistoryIndex != nil { + s.activeHistoryIndex = *snap.ActiveHistoryIndex + } + s.Memory = snap.Memory if snap.Retrieval != nil { s.Retrieval = snap.Retrieval } @@ -309,6 +329,278 @@ func (s *CanvasState) SnapshotNamespaces() (sys map[string]any, env map[string]a return sys, env, globals } +// SetHistory replaces the conversation history with a defensive copy. +func (s *CanvasState) SetHistory(history []map[string]any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.History = cloneMapSlice(history) + s.activeHistoryIndex = -1 +} + +// AppendHistory adds one user or assistant turn. payload preserves the +// Python DSL value while content is the text consumed by Go LLM components. +func (s *CanvasState) AppendHistory(role string, payload any) { + if s == nil || role == "" { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.appendHistory(role, payload) + s.activeHistoryIndex = -1 +} + +// AppendCurrentUser adds the user prompt for the in-flight turn and records +// its exact history index. SnapshotPriorHistory uses this identity instead of +// guessing that any trailing user entry must be the current prompt. +func (s *CanvasState) AppendCurrentUser(payload any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.activeHistoryIndex = s.appendHistory("user", payload) +} + +func (s *CanvasState) appendHistory(role string, payload any) int { + payload = cloneJSONValue(payload) + s.History = append(s.History, map[string]any{ + "role": role, + "content": historyContent(payload), + "payload": payload, + }) + return len(s.History) - 1 +} + +// SnapshotHistory returns a defensive copy of all conversation turns. +func (s *CanvasState) SnapshotHistory() []map[string]any { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + return cloneMapSlice(s.History) +} + +// SnapshotPriorHistory returns completed turns before the current in-flight +// user input. Python appends the current user before workflow execution but +// excludes it when prepending history to the same LLM request. +func (s *CanvasState) SnapshotPriorHistory() []map[string]any { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + history := cloneMapSlice(s.History) + if s.activeHistoryIndex >= 0 && s.activeHistoryIndex == len(history)-1 { + return history[:s.activeHistoryIndex] + } + return history +} + +// SetMemory replaces tool-call memory with a defensive copy. +func (s *CanvasState) SetMemory(memory []map[string]any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.Memory = cloneMapSlice(memory) +} + +// AppendMemory records one tool-call summary without polluting conversation +// history used by message-history windows. +func (s *CanvasState) AppendMemory(user, assistant, summary string) { + if s == nil || summary == "" { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.Memory = append(s.Memory, map[string]any{ + "user": user, + "assistant": assistant, + "summary": summary, + }) +} + +// SnapshotMemory returns a defensive copy of tool-call memory. +func (s *CanvasState) SnapshotMemory() []map[string]any { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + return cloneMapSlice(s.Memory) +} + +// AppendSysHistory appends a rendered entry to sys.history while accepting +// both []any and []string values decoded from existing DSLs. +func (s *CanvasState) AppendSysHistory(entry string) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Sys == nil { + s.Sys = make(map[string]any) + } + var history []any + switch value := s.Sys["history"].(type) { + case []any: + history = append(history, value...) + case []string: + history = make([]any, 0, len(value)+1) + for _, item := range value { + history = append(history, item) + } + } + s.Sys["history"] = append(history, entry) +} + +// SetSysHistory replaces sys.history with a defensive copy. +func (s *CanvasState) SetSysHistory(history []any) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Sys == nil { + s.Sys = make(map[string]any) + } + s.Sys["history"] = append([]any(nil), history...) +} + +// SnapshotSysHistory returns sys.history in its canonical []any wire shape. +func (s *CanvasState) SnapshotSysHistory() []any { + if s == nil { + return nil + } + s.mu.RLock() + defer s.mu.RUnlock() + switch value := s.Sys["history"].(type) { + case []any: + return append([]any(nil), value...) + case []string: + out := make([]any, 0, len(value)) + for _, item := range value { + out = append(out, item) + } + return out + default: + return []any{} + } +} + +// IncrementConversationTurns advances sys.conversation_turns once for the +// current run. JSON-backed DSLs commonly decode numbers as float64, while +// tests and programmatic callers often use int, so preserve either shape. +func (s *CanvasState) IncrementConversationTurns() { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.Sys == nil { + s.Sys = make(map[string]any) + } + switch turns := s.Sys["conversation_turns"].(type) { + case int: + s.Sys["conversation_turns"] = turns + 1 + case int32: + s.Sys["conversation_turns"] = turns + 1 + case int64: + s.Sys["conversation_turns"] = turns + 1 + case float32: + s.Sys["conversation_turns"] = turns + 1 + case float64: + s.Sys["conversation_turns"] = turns + 1 + default: + s.Sys["conversation_turns"] = 1 + } +} + +func cloneMapSlice(items []map[string]any) []map[string]any { + if items == nil { + return nil + } + if len(items) == 0 { + return []map[string]any{} + } + out := make([]map[string]any, 0, len(items)) + for _, item := range items { + out = append(out, cloneJSONValue(item).(map[string]any)) + } + return out +} + +func cloneJSONValue(value any) any { + return cloneJSONReflect(reflect.ValueOf(value)) +} + +func cloneJSONReflect(value reflect.Value) any { + if !value.IsValid() { + return nil + } + switch value.Kind() { + case reflect.Interface, reflect.Pointer: + if value.IsNil() { + return nil + } + return cloneJSONReflect(value.Elem()) + case reflect.Map: + if value.IsNil() { + return map[string]any(nil) + } + copyItem := make(map[string]any, value.Len()) + iter := value.MapRange() + for iter.Next() { + copyItem[fmt.Sprint(iter.Key().Interface())] = cloneJSONReflect(iter.Value()) + } + return copyItem + case reflect.Slice: + if value.IsNil() { + return []any(nil) + } + fallthrough + case reflect.Array: + copyItem := make([]any, value.Len()) + for index := range value.Len() { + copyItem[index] = cloneJSONReflect(value.Index(index)) + } + return copyItem + case reflect.Struct: + raw, err := json.Marshal(value.Interface()) + if err != nil { + return value.Interface() + } + var copyItem any + if err := json.Unmarshal(raw, ©Item); err != nil { + return value.Interface() + } + return copyItem + default: + return value.Interface() + } +} + +func historyContent(payload any) string { + switch value := payload.(type) { + case nil: + return "" + case string: + return value + case map[string]any: + if content, ok := value["content"].(string); ok { + return content + } + return "" + default: + return fmt.Sprint(value) + } +} + // RecordOutput stores payload under Outputs[cpnID][bucket]. Used by the // StatePostHandler to persist a node's result so downstream nodes can // resolve {{cpnID@bucket.x}} references against it. diff --git a/internal/agent/runtime/state_clone_test.go b/internal/agent/runtime/state_clone_test.go new file mode 100644 index 0000000000..88e2884264 --- /dev/null +++ b/internal/agent/runtime/state_clone_test.go @@ -0,0 +1,64 @@ +// +// 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 runtime_test + +import ( + "testing" + + "ragflow/internal/agent/component" + "ragflow/internal/agent/runtime" +) + +func TestCanvasStateSnapshotClonesTypedJSONComposites(t *testing.T) { + t.Parallel() + var nilMap map[string]string + var nilDownloads []component.DownloadInfo + payload := map[string]any{ + "typed_map": map[string]string{"status": "ready"}, + "nested": []map[string][]string{{"tags": {"one", "two"}}}, + "downloads": []component.DownloadInfo{{ + DocID: "doc-1", + Filename: "report.pdf", + }}, + "nil_map": nilMap, + "nil_downloads": nilDownloads, + } + state := runtime.NewCanvasState("run-copy", "task-copy") + state.AppendHistory("assistant", payload) + + cloned := state.SnapshotHistory()[0]["payload"].(map[string]any) + cloned["typed_map"].(map[string]any)["status"] = "changed" + cloned["nested"].([]any)[0].(map[string]any)["tags"].([]any)[0] = "changed" + cloned["downloads"].([]any)[0].(map[string]any)["filename"] = "changed.pdf" + if cloned["nil_map"].(map[string]any) != nil { + t.Fatalf("nil typed map was not preserved: %#v", cloned["nil_map"]) + } + if cloned["nil_downloads"].([]any) != nil { + t.Fatalf("nil typed slice was not preserved: %#v", cloned["nil_downloads"]) + } + + unchanged := state.SnapshotHistory()[0]["payload"].(map[string]any) + if got := unchanged["typed_map"].(map[string]any)["status"]; got != "ready" { + t.Fatalf("typed map mutation leaked into state: %v", got) + } + if got := unchanged["nested"].([]any)[0].(map[string]any)["tags"].([]any)[0]; got != "one" { + t.Fatalf("nested typed slice mutation leaked into state: %v", got) + } + if got := unchanged["downloads"].([]any)[0].(map[string]any)["filename"]; got != "report.pdf" { + t.Fatalf("typed struct slice mutation leaked into state: %v", got) + } +} diff --git a/internal/agent/runtime/state_test.go b/internal/agent/runtime/state_test.go index 43d22212c9..d4896b32a1 100644 --- a/internal/agent/runtime/state_test.go +++ b/internal/agent/runtime/state_test.go @@ -66,6 +66,31 @@ func TestCanvasState_MarshalUnmarshalJSON(t *testing.T) { } } +func TestCanvasStateCheckpointPreservesCurrentUserMarker(t *testing.T) { + t.Parallel() + src := NewCanvasState("run-checkpoint", "task-checkpoint") + src.AppendHistory("user", "previous question") + src.AppendHistory("assistant", "previous answer") + src.AppendCurrentUser("current question") + + raw, err := json.Marshal(src) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + var dst CanvasState + if err := json.Unmarshal(raw, &dst); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + prior := dst.SnapshotPriorHistory() + if len(prior) != 2 { + t.Fatalf("prior history length after restore = %d, want 2", len(prior)) + } + if got := prior[1]["content"]; got != "previous answer" { + t.Fatalf("prior assistant content after restore = %v, want previous answer", got) + } +} + // TestCanvasState_MarshalJSON_DoesNotLeakMutex pins the wire-shape // invariant: the unexported `mu sync.RWMutex` field must not appear // in the JSON output. If a future maintainer adds a `json:"mu"` tag @@ -85,6 +110,118 @@ func TestCanvasState_MarshalJSON_DoesNotLeakMutex(t *testing.T) { } } +func TestCanvasStateConversationHistory(t *testing.T) { + t.Parallel() + state := NewCanvasState("run-history", "task-history") + state.AppendHistory("user", "previous question") + state.AppendHistory("assistant", map[string]any{"content": "previous answer"}) + state.AppendCurrentUser("current question") + state.AppendSysHistory("user: previous question") + state.AppendSysHistory("assistant: {'content': 'previous answer'}") + state.AppendSysHistory("user: current question") + + prior := state.SnapshotPriorHistory() + if len(prior) != 2 { + t.Fatalf("prior history length = %d, want 2", len(prior)) + } + if got := prior[1]["content"]; got != "previous answer" { + t.Fatalf("prior assistant content = %v, want previous answer", got) + } + if got := state.SnapshotSysHistory(); len(got) != 3 || got[2] != "user: current question" { + t.Fatalf("sys.history = %#v", got) + } +} + +func TestCanvasStatePriorHistoryPreservesPersistedUser(t *testing.T) { + t.Parallel() + state := NewCanvasState("run-history", "task-history") + state.AppendHistory("user", "persisted unanswered question") + + prior := state.SnapshotPriorHistory() + if len(prior) != 1 || prior[0]["content"] != "persisted unanswered question" { + t.Fatalf("prior history = %#v, want persisted user turn", prior) + } + + state.AppendCurrentUser("current question") + prior = state.SnapshotPriorHistory() + if len(prior) != 1 || prior[0]["content"] != "persisted unanswered question" { + t.Fatalf("prior history = %#v, want only current user excluded", prior) + } +} + +func TestCanvasStateIncrementConversationTurns(t *testing.T) { + t.Parallel() + state := NewCanvasState("run-turns", "task-turns") + + state.IncrementConversationTurns() + if got := state.Sys["conversation_turns"]; got != 1 { + t.Fatalf("conversation_turns = %#v, want 1", got) + } + state.IncrementConversationTurns() + if got := state.Sys["conversation_turns"]; got != 2 { + t.Fatalf("conversation_turns = %#v, want 2", got) + } + + state.Sys["conversation_turns"] = float64(4) + state.IncrementConversationTurns() + if got := state.Sys["conversation_turns"]; got != float64(5) { + t.Fatalf("JSON conversation_turns = %#v, want float64(5)", got) + } +} + +func TestCanvasStateHistorySnapshotsDeepCopyPayload(t *testing.T) { + t.Parallel() + payload := map[string]any{ + "content": "answer", + "nil": nil, + "nil_map": map[string]any(nil), + "nil_list": []any(nil), + "metadata": map[string]any{ + "tags": []any{"one", map[string]any{"name": "nested"}}, + }, + } + state := NewCanvasState("run-copy", "task-copy") + state.AppendHistory("assistant", payload) + + snapshot := state.SnapshotHistory() + snapshotPayload := snapshot[0]["payload"].(map[string]any) + metadata := snapshotPayload["metadata"].(map[string]any) + tags := metadata["tags"].([]any) + metadata["new"] = true + tags[0] = "changed" + tags[1].(map[string]any)["name"] = "changed" + + unchanged := state.SnapshotHistory()[0]["payload"].(map[string]any) + unchangedMetadata := unchanged["metadata"].(map[string]any) + unchangedTags := unchangedMetadata["tags"].([]any) + if unchanged["nil"] != nil || unchanged["nil_map"].(map[string]any) != nil || unchanged["nil_list"].([]any) != nil { + t.Fatalf("nil values were not preserved: %#v", unchanged) + } + if _, exists := unchangedMetadata["new"]; exists || unchangedTags[0] != "one" || unchangedTags[1].(map[string]any)["name"] != "nested" { + t.Fatalf("snapshot mutation leaked into state: %#v", unchanged) + } + + payload["metadata"].(map[string]any)["source"] = "caller mutation" + unchanged = state.SnapshotHistory()[0]["payload"].(map[string]any) + if _, exists := unchanged["metadata"].(map[string]any)["source"]; exists { + t.Fatalf("caller mutation leaked into state: %#v", unchanged) + } +} + +func TestCanvasStateMemoryIsSeparateFromHistory(t *testing.T) { + t.Parallel() + state := NewCanvasState("run-memory", "task-memory") + state.AppendMemory("question", "answer", "used search") + + if len(state.SnapshotHistory()) != 0 { + t.Fatalf("memory polluted history: %#v", state.SnapshotHistory()) + } + memory := state.SnapshotMemory() + if len(memory) != 1 || memory[0]["summary"] != "used search" { + t.Fatalf("memory = %#v", memory) + } +} + func TestCanvasState_EnsureSysDate(t *testing.T) { t.Parallel() diff --git a/internal/agent/runtime/template.go b/internal/agent/runtime/template.go index fa57e90bff..d15dcff89c 100644 --- a/internal/agent/runtime/template.go +++ b/internal/agent/runtime/template.go @@ -23,8 +23,10 @@ package runtime import ( + "encoding/json" "fmt" "regexp" + "strings" ) // VarRefPattern matches the RAGFlow variable reference syntax. @@ -131,6 +133,24 @@ func ResolveTemplateForDisplay(s string, state *CanvasState) string { if v == nil { return "" } - return fmt.Sprintf("%v", v) + return stringifyDisplayValue(v) }) } + +// stringifyDisplayValue mirrors Message._stringify_message_value in Python: +// strings pass through unchanged, while lists, maps, numbers, and booleans +// use JSON syntax. In particular, a []string{"user: 1"} reference must render +// as ["user: 1"], not Go's fmt representation [user: 1]. +func stringifyDisplayValue(value any) string { + if text, ok := value.(string); ok { + return text + } + + var buf strings.Builder + encoder := json.NewEncoder(&buf) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(value); err == nil { + return strings.TrimSuffix(buf.String(), "\n") + } + return fmt.Sprintf("%v", value) +} diff --git a/internal/handler/agent.go b/internal/handler/agent.go index 59945ca51f..767f407f2a 100644 --- a/internal/handler/agent.go +++ b/internal/handler/agent.go @@ -65,7 +65,7 @@ type agentFileService interface { // NewAgentHandler assigns the concrete *service.AgentService — which // satisfies this interface because its RunAgent signature matches. type chatAgentService interface { - RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any) (<-chan canvas.RunEvent, error) + RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any, files []map[string]interface{}) (<-chan canvas.RunEvent, error) } // documentAccessChecker is the minimal surface RerunAgent needs @@ -393,7 +393,7 @@ func (h *AgentHandler) RunAgent(c *gin.Context) { sessionID := c.Query("session_id") userInput := readUserInput(c) - events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, canvasID, sessionID, version, userInput) + events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, canvasID, sessionID, version, userInput, nil) if err != nil { ec, em := mapAgentError(err) common.ResponseWithCodeData(c, ec, nil, em) @@ -848,6 +848,7 @@ type agentChatCompletionsRequest struct { Model string `json:"model"` Messages []map[string]interface{} `json:"messages"` ReturnTrace bool `json:"return_trace"` + Files []map[string]interface{} `json:"files"` } // extractLastUserContent returns the content of the last message in @@ -1004,7 +1005,7 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) { }, userInputMeta(userInput)...)..., ) - events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, req.AgentID, req.SessionID, "", userInput) + events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, req.AgentID, req.SessionID, "", userInput, req.Files) if err != nil { common.Warn("agent chat completions: RunAgent failed", append([]zap.Field{ diff --git a/internal/handler/agent_test.go b/internal/handler/agent_test.go index e98e316c29..02be7b0868 100644 --- a/internal/handler/agent_test.go +++ b/internal/handler/agent_test.go @@ -715,7 +715,7 @@ type stubChatRunner struct { err error } -func (s *stubChatRunner) RunAgent(_ context.Context, _, _, _, _ string, _ any) (<-chan canvas.RunEvent, error) { +func (s *stubChatRunner) RunAgent(_ context.Context, _, _, _, _ string, _ any, _ []map[string]interface{}) (<-chan canvas.RunEvent, error) { if s.err != nil { return nil, s.err } @@ -933,15 +933,19 @@ func TestAgentChatCompletions_DerivesStructuredUserInputFromInputs(t *testing.T) } } -// captureChatRunner records the userInput it was called with and +// captureChatRunner records the userInput and files it was called with and // returns an empty (closed) channel. Used to assert on argument // derivation without exercising the runner. type captureChatRunner struct { - captured *any + captured *any + capturedFiles *[]map[string]interface{} } -func (c *captureChatRunner) RunAgent(_ context.Context, _, _, _, _ string, userInput any) (<-chan canvas.RunEvent, error) { +func (c *captureChatRunner) RunAgent(_ context.Context, _, _, _, _ string, userInput any, files []map[string]interface{}) (<-chan canvas.RunEvent, error) { *c.captured = userInput + if c.capturedFiles != nil { + *c.capturedFiles = files + } ch := make(chan canvas.RunEvent) close(ch) return ch, nil @@ -1188,3 +1192,70 @@ type stubDocService struct { func (s *stubDocService) Accessible(_, _ string) bool { return s.accessible } + +// TestAgentChatCompletions_FilesDeserialized verifies that when the +// JSON request body contains a `files` field, the +// agentChatCompletionsRequest struct deserializes it correctly. +// Mirrors Python's req.get("files", []) at agent_api.py:1313. +func TestAgentChatCompletions_FilesDeserialized(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + body := `{ + "agent_id": "a1", + "query": "hi", + "files": [ + {"id": "file-1", "name": "resume.txt", "mime_type": "text/plain", "created_by": "u1"} + ] + }` + c.Request = httptest.NewRequest("POST", "/api/v1/agents/chat/completions", + strings.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + c.Set("user", &entity.User{ID: "u1"}) + c.Set("user_id", "u1") + + var captured any + var capturedFiles []map[string]interface{} + runner := &captureChatRunner{captured: &captured, capturedFiles: &capturedFiles} + h := &AgentHandler{chatRunner: runner} + h.AgentChatCompletions(c) + + if len(capturedFiles) != 1 { + t.Fatalf("capturedFiles length = %d, want 1", len(capturedFiles)) + } + if id, _ := capturedFiles[0]["id"].(string); id != "file-1" { + t.Errorf("capturedFiles[0][\"id\"] = %q, want %q", id, "file-1") + } + if name, _ := capturedFiles[0]["name"].(string); name != "resume.txt" { + t.Errorf("capturedFiles[0][\"name\"] = %q, want %q", name, "resume.txt") + } + mime, _ := capturedFiles[0]["mime_type"].(string) + if mime != "text/plain" { + t.Errorf("capturedFiles[0][\"mime_type\"] = %q, want %q", mime, "text/plain") + } +} + +// TestAgentChatCompletions_EmptyFilesNil verifies that when the JSON +// request body does NOT include `files`, the handler passes nil to +// RunAgent (no crash, no spurious slice). Mirrors Python's behavior +// where req.get("files", []) defaults to []. +func TestAgentChatCompletions_EmptyFilesNil(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("POST", "/api/v1/agents/chat/completions", + strings.NewReader(`{"agent_id":"a1","query":"hi"}`)) + c.Request.Header.Set("Content-Type", "application/json") + c.Set("user", &entity.User{ID: "u1"}) + c.Set("user_id", "u1") + + var captured any + var capturedFiles []map[string]interface{} + runner := &captureChatRunner{captured: &captured, capturedFiles: &capturedFiles} + h := &AgentHandler{chatRunner: runner} + h.AgentChatCompletions(c) + + if capturedFiles != nil { + t.Errorf("capturedFiles = %v, want nil when files not in request", capturedFiles) + } +} diff --git a/internal/handler/agent_wait_for_user_test.go b/internal/handler/agent_wait_for_user_test.go index 6eee7a8b85..430278aff3 100644 --- a/internal/handler/agent_wait_for_user_test.go +++ b/internal/handler/agent_wait_for_user_test.go @@ -94,7 +94,7 @@ func (f *waitFakeAgentService) DeleteAgent(context.Context, string, string) erro // RunAgent mimics service.AgentService.RunAgent for the test // driver. It loads the canvas (a no-op in tests), builds a RunFunc // from the supplied stub, and hands off to the orchestrator. -func (f *waitFakeAgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any) (<-chan canvas.RunEvent, error) { +func (f *waitFakeAgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any, _ []map[string]interface{}) (<-chan canvas.RunEvent, error) { _ = ctx _ = userID _ = version @@ -151,7 +151,7 @@ func waitForUserRoutes(svc *waitFakeAgentService) *gin.Engine { canvasID := c.Param("canvas_id") sessionID := c.Query("session_id") userInput := c.Query("user_input") - events, err := svc.RunAgent(c.Request.Context(), "user-wait", canvasID, sessionID, "", userInput) + events, err := svc.RunAgent(c.Request.Context(), "user-wait", canvasID, sessionID, "", userInput, nil) if err != nil { // We never expect a non-nil err from the fake, // but be defensive. diff --git a/internal/handler/bot_test.go b/internal/handler/bot_test.go index 5febcaceef..7e12236a58 100644 --- a/internal/handler/bot_test.go +++ b/internal/handler/bot_test.go @@ -492,6 +492,29 @@ func TestAgentbotCompletion_ResumesSession(t *testing.T) { } } +func TestAgentbotCompletion_BindsFileDescriptors(t *testing.T) { + var capturedReq service.AgentbotCompletionRequest + stub := &stubBotService{ + agentbotCompleteFn: func(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) (<-chan canvas.RunEvent, common.ErrorCode, error) { + capturedReq = req + ch := make(chan canvas.RunEvent) + close(ch) + return ch, common.CodeSuccess, nil + }, + } + r := botTestEngine(stub) + _ = doJSON(r, http.MethodPost, "/api/v1/agentbots/a1/completions", `{ + "question":"hi", + "files":[{"id":"upload-1","name":"notes.txt","mime_type":"text/plain","created_by":"user-1"}] + }`) + if len(capturedReq.Files) != 1 { + t.Fatalf("files = %#v, want one descriptor", capturedReq.Files) + } + if capturedReq.Files[0]["id"] != "upload-1" || capturedReq.Files[0]["created_by"] != "user-1" { + t.Fatalf("file descriptor = %#v", capturedReq.Files[0]) + } +} + // ----- AgentbotInputs tests (criteria 21, 22, 23) ----- // TestAgentbotInputs_OK covers criterion 21. diff --git a/internal/service/agent.go b/internal/service/agent.go index 09f5767ac2..111018e8bd 100644 --- a/internal/service/agent.go +++ b/internal/service/agent.go @@ -23,6 +23,7 @@ import ( "fmt" "ragflow/internal/utility" "reflect" + "sort" "strings" "sync" "time" @@ -85,7 +86,7 @@ func (s *AgentService) RunAgentWithWebhook( if payload != nil { ctx = context.WithValue(ctx, webhookPayloadKey{}, payload) } - return s.RunAgent(ctx, userID, canvasID, "", "", "") + return s.RunAgent(ctx, userID, canvasID, "", "", "", nil) } func emitAgentMessageEvents(emit func(string, string), answer, thinking string, reference any) { @@ -863,11 +864,12 @@ func (s *AgentService) DeleteVersion(ctx context.Context, userID, canvasID, vers // The per-run RunFunc is built by buildRunFunc — see its doc comment // for the full production chain (real Compile/Invoke, resume path, // error-layering contract). -func (s *AgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any) (<-chan canvas.RunEvent, error) { +func (s *AgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID, version string, userInput any, files []map[string]interface{}) (<-chan canvas.RunEvent, error) { canvasRow, err := s.loadCanvasForUser(ctx, userID, canvasID) if err != nil { return nil, err } + newSession := sessionID == "" if sessionID == "" { sessionID = utility.GenerateToken() } @@ -958,6 +960,23 @@ func (s *AgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID if dsl == nil { dsl = normalisedDSLForRun(versionRow) } + if sessionID != "" && s.api4ConversationDAO != nil { + session, sessionErr := s.api4ConversationDAO.GetBySessionID(sessionID, canvasID) + if sessionErr != nil { + return nil, fmt.Errorf("RunAgent: load session %q: %w: %w", sessionID, sessionErr, ErrAgentStorageError) + } + if session != nil && session.UserID != userID { + return nil, fmt.Errorf("RunAgent: session %q not found: %w", sessionID, dao.ErrUserCanvasNotFound) + } + if session != nil && len(session.DSL) > 0 { + dsl = dslpkg.NormalizeForRun(session.DSL) + } + } + if newSession && len(dsl) > 0 { + if err := s.createAgentRunSession(sessionID, userID, canvasID, dsl, versionRow); err != nil { + return nil, fmt.Errorf("RunAgent: create session %q: %w: %w", sessionID, err, ErrAgentStorageError) + } + } run := s.buildRunFunc(canvasID, versionRow, dsl) @@ -970,6 +989,9 @@ func (s *AgentService) RunAgent(ctx context.Context, userID, canvasID, sessionID if userInput != nil { root["user_input"] = userInput } + if len(files) > 0 { + root["files"] = files + } if dsl != nil { root["__dsl_present__"] = true } @@ -1181,14 +1203,28 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv } } } + state.SetHistory(c.History) + state.SetMemory(c.Memory) state.EnsureSysDate() state.Sys["query"] = userInput + state.AppendCurrentUser(userInput) + state.AppendSysHistory("user: " + renderUserHistoryValue(userInput)) if uid, ok := root["user_id"].(string); ok && uid != "" { state.Sys["user_id"] = uid } if tid, ok := root["tenant_id"].(string); ok && tid != "" { state.Sys["tenant_id"] = tid } + if rawFiles, ok := root["files"].([]map[string]interface{}); ok && len(rawFiles) > 0 { + fileSvc := NewFileService() + files, ferr := fileSvc.parseAgentUploads(userID, rawFiles, beginLayoutRecognize(c)) + if ferr != nil { + s.markRunFailed(ctx2, runID, "parse files: "+ferr.Error()) + return nil, fmt.Errorf("parse agent files: %w", ferr) + } + state.Sys["files"] = files + } + state.IncrementConversationTurns() ctx2 = runtime.WithState(ctx2, state) // Resume path. The user input is the resume payload for the @@ -1271,7 +1307,8 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv if isResume && resumeID != "" { wfInput = "" } - _, err = cc.Workflow.Invoke(ctx2, map[string]any{"query": wfInput}, invokeOpts...) + workflowOutput, invokeErr := cc.Workflow.Invoke(ctx2, map[string]any{"query": wfInput}, invokeOpts...) + err = invokeErr if cpID != "" && s.runTracker != nil { _ = s.runTracker.AttachCheckpoint(ctx2, runID, cpID) @@ -1308,6 +1345,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv } } referencePayload := agentRunReferencePayload(state, legacyReference) + assistantOutput := terminalCanvasOutput(c, state, workflowOutput, answer, downloads) runtime.FinalizeAgentMessage(ctx2) messageEventsEmitted := runtime.AgentMessageEventsEmitted(ctx2) @@ -1322,7 +1360,12 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv if canvas.IsInterruptError(err) { s.markRunFailed(ctx2, runID, "interrupt: "+err.Error()) if answer != "" { - s.persistAgentRunSession(canvasID, sessionID, messageID, userInput, answer, referencePayload) + appendAssistantHistory(state, partialAssistantOutput(answer, downloads)) + } + if persistErr := s.persistAgentRunSession(canvasID, userID, sessionID, messageID, userInput, answer, referencePayload, dsl, state, answer != ""); persistErr != nil { + return nil, fmt.Errorf("persist interrupted agent session: %w: %w", persistErr, ErrAgentStorageError) + } + if answer != "" { if !messageEventsEmitted { emitAgentMessageEvents(emit, answer, thinking, referencePayload) } @@ -1335,7 +1378,11 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv return state, err } if shouldTreatAsCompletedLoopRun(err, answer) { - s.persistAgentRunSession(canvasID, sessionID, messageID, userInput, answer, referencePayload) + appendAssistantHistory(state, assistantOutput) + if persistErr := s.persistAgentRunSession(canvasID, userID, sessionID, messageID, userInput, answer, referencePayload, dsl, state, true); persistErr != nil { + s.markRunFailed(ctx2, runID, "persist session: "+persistErr.Error()) + return nil, fmt.Errorf("persist agent session: %w: %w", persistErr, ErrAgentStorageError) + } if !messageEventsEmitted { emitAgentMessageEvents(emit, answer, thinking, referencePayload) } @@ -1365,7 +1412,11 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv } // Emit message + message_end (mirrors Python's ans dict). - s.persistAgentRunSession(canvasID, sessionID, messageID, userInput, answer, referencePayload) + appendAssistantHistory(state, assistantOutput) + if persistErr := s.persistAgentRunSession(canvasID, userID, sessionID, messageID, userInput, answer, referencePayload, dsl, state, true); persistErr != nil { + s.markRunFailed(ctx2, runID, "persist session: "+persistErr.Error()) + return nil, fmt.Errorf("persist agent session: %w: %w", persistErr, ErrAgentStorageError) + } if !messageEventsEmitted { emitAgentMessageEvents(emit, answer, thinking, referencePayload) } @@ -1394,6 +1445,44 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv } } +func beginLayoutRecognize(c *canvas.Canvas) string { + if c == nil { + return "" + } + for _, comp := range c.Components { + if !strings.EqualFold(comp.Obj.ComponentName, "Begin") { + continue + } + layout, _ := comp.Obj.Params["layout_recognize"].(string) + return layout + } + return "" +} + +func (s *AgentService) createAgentRunSession( + sessionID, userID, agentID string, + runDSL map[string]any, + versionRow *entity.UserCanvasVersion, +) error { + if s == nil || s.api4ConversationDAO == nil { + return errors.New("agent session storage is not configured") + } + source := "agent" + session := &entity.API4Conversation{ + ID: sessionID, + DialogID: agentID, + UserID: userID, + Message: json.RawMessage(`[]`), + Reference: json.RawMessage(`[]`), + Source: &source, + DSL: entity.JSONMap(runDSL), + } + if versionRow != nil { + session.VersionTitle = versionRow.Title + } + return s.api4ConversationDAO.Create(session) +} + // runIDFor builds the per-run CanvasState identifier: canvasID // alone for first-touch runs, canvasID + sessionID for resumed runs // (so two concurrent sessions on the same canvas don't collide in @@ -1428,24 +1517,34 @@ func emptyDownloadValue(value any) bool { } } -func (s *AgentService) persistAgentRunSession(agentID, sessionID, messageID string, userInput any, answer string, reference map[string]interface{}) { +func (s *AgentService) persistAgentRunSession( + agentID, userID, sessionID, messageID string, + userInput any, + answer string, + reference map[string]interface{}, + runDSL map[string]any, + state *canvas.CanvasState, + appendAssistantMessage bool, +) error { if sessionID == "" || s == nil || s.api4ConversationDAO == nil || dao.DB == nil { - return + return nil } session, err := s.api4ConversationDAO.GetBySessionID(sessionID, agentID) if err != nil { common.Warn("agent run: load session for update failed", zap.String("agent_id", agentID), zap.String("session_id", sessionID), zap.Error(err)) - return + return nil } - if session == nil { - return + if session == nil || session.UserID != userID { + return nil } messages := parseAgentSessionMessages(session.Message) now := time.Now().Unix() if text := stringifyAgentUserInput(userInput); text != "" { messages = append(messages, map[string]interface{}{"role": "user", "content": text, "id": utility.GenerateToken(), "created_at": now}) } - messages = append(messages, map[string]interface{}{"role": "assistant", "content": answer, "id": messageID, "created_at": now}) + if appendAssistantMessage { + messages = append(messages, map[string]interface{}{"role": "assistant", "content": answer, "id": messageID, "created_at": now}) + } if raw, err := json.Marshal(messages); err == nil { session.Message = raw } @@ -1454,9 +1553,54 @@ func (s *AgentService) persistAgentRunSession(agentID, sessionID, messageID stri if raw, err := json.Marshal(references); err == nil { session.Reference = raw } - if err := s.api4ConversationDAO.Update(session); err != nil { - common.Warn("agent run: update session failed", zap.String("agent_id", agentID), zap.String("session_id", sessionID), zap.Error(err)) + if state != nil { + session.DSL = buildPersistedAgentDSL(runDSL, state) } + return s.api4ConversationDAO.Update(session) +} + +func buildPersistedAgentDSL(runDSL map[string]any, state *canvas.CanvasState) entity.JSONMap { + dsl := make(entity.JSONMap, len(runDSL)+3) + for key, value := range runDSL { + dsl[key] = value + } + if state == nil { + return dsl + } + + globals := make(map[string]any) + if existing, ok := dsl["globals"].(map[string]any); ok { + for key, value := range existing { + globals[key] = value + } + } + sysValues, envValues, globalValues := state.SnapshotNamespaces() + for key := range globals { + switch { + case strings.HasPrefix(key, "sys."): + if value, exists := sysValues[strings.TrimPrefix(key, "sys.")]; exists { + globals[key] = value + } + case strings.HasPrefix(key, "env."): + if value, exists := envValues[strings.TrimPrefix(key, "env.")]; exists { + globals[key] = value + } + default: + if value, exists := globalValues[key]; exists { + globals[key] = value + } + } + } + for _, key := range []string{"query", "user_id", "conversation_turns", "files", "history", "date"} { + if value, exists := sysValues[key]; exists { + globals["sys."+key] = value + } + } + + dsl["globals"] = globals + dsl["history"] = canvas.EncodeHistory(state.SnapshotHistory()) + dsl["memory"] = canvas.EncodeMemory(state.SnapshotMemory()) + return dsl } func agentRunReferencePayload(state *canvas.CanvasState, legacyChunks []interface{}) map[string]interface{} { @@ -1489,6 +1633,163 @@ func stringifyAgentUserInput(userInput any) string { } } +func appendAssistantHistory(state *canvas.CanvasState, payload map[string]any) { + if state == nil { + return + } + state.AppendHistory("assistant", payload) + state.AppendSysHistory("assistant: " + pythonHistoryRepr(payload)) +} + +func partialAssistantOutput(answer string, downloads any) map[string]any { + output := map[string]any{"content": answer} + if !emptyDownloadValue(downloads) { + output["downloads"] = downloads + } + return output +} + +func terminalCanvasOutput( + c *canvas.Canvas, + state *canvas.CanvasState, + workflowOutput map[string]any, + answer string, + downloads any, +) map[string]any { + terminalIDs := make([]string, 0) + if c != nil { + for cpnID, component := range c.Components { + if len(component.Downstream) == 0 { + terminalIDs = append(terminalIDs, cpnID) + } + } + } + sort.Strings(terminalIDs) + for _, cpnID := range terminalIDs { + if output, ok := workflowOutput[cpnID].(map[string]any); ok && len(output) > 0 { + return cloneCanvasOutput(output) + } + } + if state != nil { + snapshot := state.Snapshot() + for _, cpnID := range terminalIDs { + if output := snapshot[cpnID]; len(output) > 0 { + return cloneCanvasOutput(output) + } + } + } + if len(workflowOutput) > 0 { + return cloneCanvasOutput(workflowOutput) + } + fallback := map[string]any{"content": answer} + if !emptyDownloadValue(downloads) { + fallback["downloads"] = downloads + } + return fallback +} + +func cloneCanvasOutput(input map[string]any) map[string]any { + output := make(map[string]any, len(input)) + for key, value := range input { + switch key { + case "__cpn_id__", "state", "__legacy_noop__": + continue + } + output[key] = value + } + return output +} + +func renderUserHistoryValue(value any) string { + switch value := value.(type) { + case string: + return value + case map[string]any: + var buf strings.Builder + encoder := json.NewEncoder(&buf) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(value); err != nil { + return fmt.Sprint(value) + } + return strings.TrimSuffix(buf.String(), "\n") + default: + return pythonHistoryRepr(value) + } +} + +func pythonHistoryRepr(value any) string { + switch item := value.(type) { + case nil: + return "None" + case string: + replacer := strings.NewReplacer( + "\\", "\\\\", + "'", "\\'", + "\n", "\\n", + "\r", "\\r", + "\t", "\\t", + ) + return "'" + replacer.Replace(item) + "'" + case bool: + if item { + return "True" + } + return "False" + case map[string]any: + keys := make([]string, 0, len(item)) + for key := range item { + keys = append(keys, key) + } + sort.Slice(keys, func(i, j int) bool { + leftPriority := pythonOutputKeyPriority(keys[i]) + rightPriority := pythonOutputKeyPriority(keys[j]) + if leftPriority != rightPriority { + return leftPriority < rightPriority + } + return keys[i] < keys[j] + }) + parts := make([]string, 0, len(keys)) + for _, key := range keys { + parts = append(parts, pythonHistoryRepr(key)+": "+pythonHistoryRepr(item[key])) + } + return "{" + strings.Join(parts, ", ") + "}" + case []any: + parts := make([]string, 0, len(item)) + for _, child := range item { + parts = append(parts, pythonHistoryRepr(child)) + } + return "[" + strings.Join(parts, ", ") + "]" + case []string: + parts := make([]string, 0, len(item)) + for _, child := range item { + parts = append(parts, pythonHistoryRepr(child)) + } + return "[" + strings.Join(parts, ", ") + "]" + default: + return fmt.Sprint(item) + } +} + +// pythonOutputKeyPriority reconstructs the order produced by Python's +// ComponentParamBase output dictionaries: declared business outputs first, +// followed by the timing fields added by ComponentBase.invoke(). Message +// declares content then downloads, which is the terminal payload most often +// persisted in conversation history. +func pythonOutputKeyPriority(key string) int { + switch key { + case "content": + return 0 + case "downloads": + return 1 + case "_created_time": + return 3 + case "_elapsed_time": + return 4 + default: + return 2 + } +} + // tenantIDFromRoot returns the optional run-tracker tenant id that // RunAgent populated on the root map. Runtime components use // root["tenant_id"] / state.Sys["tenant_id"] for the caller tenant; diff --git a/internal/service/agent_persistence_test.go b/internal/service/agent_persistence_test.go new file mode 100644 index 0000000000..9c9910bd09 --- /dev/null +++ b/internal/service/agent_persistence_test.go @@ -0,0 +1,92 @@ +// +// 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" + "encoding/json" + "errors" + "strings" + "testing" + + "ragflow/internal/agent/canvas" + "ragflow/internal/dao" + "ragflow/internal/entity" +) + +func TestAgentRunSessionUpdateFailurePreventsSuccessEvents(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate(&entity.API4Conversation{}); err != nil { + t.Fatalf("migrate: %v", err) + } + originalDB := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = originalDB }) + + if err := testDB.Create(&entity.API4Conversation{ + ID: "session-update-failure", + DialogID: "canvas-update-failure", + UserID: "user-1", + Message: json.RawMessage(`[]`), + Reference: json.RawMessage(`[]`), + }).Error; err != nil { + t.Fatalf("create session: %v", err) + } + if err := testDB.Exec(` + CREATE TRIGGER fail_agent_session_update + BEFORE UPDATE ON api_4_conversation + BEGIN + SELECT RAISE(FAIL, 'forced update failure'); + END + `).Error; err != nil { + t.Fatalf("create update trigger: %v", err) + } + + dsl := map[string]any{ + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{"component_name": "Begin", "params": map[string]any{}}, + "downstream": []any{"message_0"}, + }, + "message_0": map[string]any{ + "obj": map[string]any{"component_name": "Message", "params": map[string]any{"text": "hello {{sys.query}}"}}, + "upstream": []any{"begin_0"}, + }, + }, + "path": []any{"begin_0", "message_0"}, + } + events := make(chan canvas.RunEvent, 32) + _, err := NewAgentService().buildRunFunc("canvas-update-failure", nil, dsl)(context.Background(), map[string]any{ + "__events__": events, + "__message_id__": "message-1", + "__session_id__": "session-update-failure", + "user_id": "user-1", + "user_input": "world", + }) + if !errors.Is(err, ErrAgentStorageError) { + t.Fatalf("run error = %v, want ErrAgentStorageError", err) + } + if !strings.Contains(err.Error(), "forced update failure") { + t.Fatalf("run error = %v, want underlying update failure", err) + } + close(events) + for event := range events { + if event.Type == "message_end" || event.Type == "workflow_finished" { + t.Fatalf("persistence failure emitted success event %q", event.Type) + } + } +} diff --git a/internal/service/agent_run_e2e_test.go b/internal/service/agent_run_e2e_test.go index 09160f5b40..99b98a417b 100644 --- a/internal/service/agent_run_e2e_test.go +++ b/internal/service/agent_run_e2e_test.go @@ -40,6 +40,7 @@ package service import ( "context" "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -51,6 +52,7 @@ import ( _ "ragflow/internal/agent/component" // blank import: registers factories via component.init() "ragflow/internal/dao" "ragflow/internal/entity" + "ragflow/internal/storage" "github.com/alicebob/miniredis/v2" "github.com/redis/go-redis/v9" @@ -169,6 +171,7 @@ func TestRunAgent_RealCanvas_BeginMessage(t *testing.T) { t.Cleanup(func() { dao.DB = orig }) dsl := map[string]any{ + "globals": map[string]any{"sys.files": []any{"stale file"}}, "components": map[string]any{ "begin_0": map[string]any{ "obj": map[string]any{ @@ -200,8 +203,7 @@ func TestRunAgent_RealCanvas_BeginMessage(t *testing.T) { "canvas-hello", "session-hello", "", // latest version - "world", - ) + "world", nil) if err != nil { t.Fatalf("RunAgent: %v", err) } @@ -223,6 +225,275 @@ func TestRunAgent_RealCanvas_BeginMessage(t *testing.T) { } } +func TestRunAgent_SessionHistoryFeedsSysHistoryAndPersists(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + dsl := map[string]any{ + "globals": map[string]any{ + "sys.conversation_turns": 0, + "sys.files": []any{"existing file"}, + "sys.history": []any{}, + "sys.user_id": "", + }, + "history": []any{}, + "memory": []any{}, + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Begin", + "params": map[string]any{}, + }, + "downstream": []any{"history_0"}, + }, + "history_0": map[string]any{ + "obj": map[string]any{ + "component_name": "ListOperations", + "params": map[string]any{ + "query": "sys.history", + "operations": "sort", + }, + }, + "upstream": []any{"begin_0"}, + "downstream": []any{"message_0"}, + }, + "message_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Message", + "params": map[string]any{ + "text": "{{history_0@result}}", + }, + }, + "upstream": []any{"history_0"}, + }, + }, + "path": []any{"begin_0", "history_0", "message_0"}, + } + makeCanvasWithDSL(t, "canvas-history", "user-1", "tenant-1", "v-history", dsl) + if err := testDB.Create(&entity.API4Conversation{ + ID: "session-history", + DialogID: "canvas-history", + UserID: "user-1", + Message: json.RawMessage(`[]`), + Reference: json.RawMessage(`[]`), + }).Error; err != nil { + t.Fatalf("create session: %v", err) + } + + svc := NewAgentService() + run := func(input string) canvas.MessageEvent { + t.Helper() + events, err := svc.RunAgent(context.Background(), "user-1", "canvas-history", "session-history", "", input, nil) + if err != nil { + t.Fatalf("RunAgent(%q): %v", input, err) + } + messages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) != 0 || len(waiting) != 0 || !done { + t.Fatalf("RunAgent(%q): messages=%+v waiting=%+v errors=%+v done=%v", input, messages, waiting, errs, done) + } + if len(messages) != 1 { + t.Fatalf("RunAgent(%q): message count = %d, want 1", input, len(messages)) + } + return messages[0] + } + + first := run("hi") + if first.Content != `["user: hi"]` { + t.Fatalf("first content = %q, want JSON-rendered sys.history", first.Content) + } + var afterFirst entity.API4Conversation + if err := testDB.Where("id = ?", "session-history").First(&afterFirst).Error; err != nil { + t.Fatalf("reload session after first run: %v", err) + } + if components, ok := afterFirst.DSL["components"].(map[string]any); !ok || len(components) != 3 { + t.Fatalf("persisted DSL lost runtime components: %#v", afterFirst.DSL) + } + second := run("again") + var secondHistory []string + if err := json.Unmarshal([]byte(second.Content), &secondHistory); err != nil { + t.Fatalf("second content = %q, want a JSON string list: %v", second.Content, err) + } + if len(secondHistory) != 3 { + t.Fatalf("second history = %#v, want assistant plus two user entries", secondHistory) + } + if !strings.HasPrefix(secondHistory[0], "assistant: ") || + !strings.Contains(secondHistory[0], `'content': '["user: hi"]'`) || + !strings.Contains(secondHistory[0], `'downloads': []`) { + t.Fatalf("assistant history entry = %q, want persisted Message content and downloads", secondHistory[0]) + } + if secondHistory[1] != "user: again" || secondHistory[2] != "user: hi" { + t.Fatalf("sorted user history = %#v, want [user: again, user: hi]", secondHistory[1:]) + } + + var session entity.API4Conversation + if err := testDB.Where("id = ?", "session-history").First(&session).Error; err != nil { + t.Fatalf("reload session: %v", err) + } + history, ok := session.DSL["history"].([]any) + if !ok || len(history) != 4 { + t.Fatalf("persisted history = %#v, want four user/assistant entries", session.DSL["history"]) + } + globals, _ := session.DSL["globals"].(map[string]any) + sysHistory, ok := globals["sys.history"].([]any) + if !ok || len(sysHistory) != 4 { + t.Fatalf("persisted sys.history = %#v, want four rendered entries", globals["sys.history"]) + } + if turns := globals["sys.conversation_turns"]; turns != 2 && turns != float64(2) { + t.Fatalf("persisted sys.conversation_turns = %#v, want 2", turns) + } + if globals["sys.user_id"] != "user-1" { + t.Fatalf("persisted sys.user_id = %#v, want user-1", globals["sys.user_id"]) + } + files, ok := globals["sys.files"].([]any) + if !ok || len(files) != 1 || files[0] != "existing file" { + t.Fatalf("persisted sys.files = %#v, want existing file", globals["sys.files"]) + } +} + +func TestRunAgent_NewSessionPersistsHistoryForNextTurn(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.API4Conversation{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + dsl := map[string]any{ + "globals": map[string]any{"sys.history": []any{}}, + "history": []any{}, + "memory": []any{}, + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{"component_name": "Begin", "params": map[string]any{}}, + "downstream": []any{"message_0"}, + }, + "message_0": map[string]any{ + "obj": map[string]any{"component_name": "Message", "params": map[string]any{"text": "{{sys.history}}"}}, + "upstream": []any{"begin_0"}, + }, + }, + "path": []any{"begin_0", "message_0"}, + } + makeCanvasWithDSL(t, "canvas-new-session", "user-1", "tenant-1", "v-new-session", dsl) + + svc := NewAgentService() + events, err := svc.RunAgent(context.Background(), "user-1", "canvas-new-session", "", "", "1", nil) + if err != nil { + t.Fatalf("first RunAgent: %v", err) + } + firstMessages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) != 0 || len(waiting) != 0 || !done || len(firstMessages) != 1 { + t.Fatalf("first run: messages=%+v waiting=%+v errors=%+v done=%v", firstMessages, waiting, errs, done) + } + if firstMessages[0].Content != `["user: 1"]` { + t.Fatalf("first content = %q", firstMessages[0].Content) + } + + var session entity.API4Conversation + if err := testDB.Where("dialog_id = ? AND user_id = ?", "canvas-new-session", "user-1").First(&session).Error; err != nil { + t.Fatalf("new session was not persisted: %v", err) + } + if session.ID == "" { + t.Fatal("persisted session has an empty ID") + } + + events, err = svc.RunAgent(context.Background(), "user-1", "canvas-new-session", session.ID, "", "1", nil) + if err != nil { + t.Fatalf("second RunAgent: %v", err) + } + secondMessages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) != 0 || len(waiting) != 0 || !done || len(secondMessages) != 1 { + t.Fatalf("second run: messages=%+v waiting=%+v errors=%+v done=%v", secondMessages, waiting, errs, done) + } + var history []string + if err := json.Unmarshal([]byte(secondMessages[0].Content), &history); err != nil { + t.Fatalf("second content = %q: %v", secondMessages[0].Content, err) + } + if len(history) != 3 || history[0] != "user: 1" || !strings.HasPrefix(history[1], "assistant: ") || history[2] != "user: 1" { + t.Fatalf("second history = %#v, want first user, assistant, current user", history) + } +} + +func TestRunAgent_RejectsSessionOwnedByAnotherUser(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.API4Conversation{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + dsl := map[string]any{ + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{"component_name": "Begin", "params": map[string]any{}}, + "downstream": []any{"message_0"}, + }, + "message_0": map[string]any{ + "obj": map[string]any{"component_name": "Message", "params": map[string]any{"text": "safe"}}, + "upstream": []any{"begin_0"}, + }, + }, + "path": []any{"begin_0", "message_0"}, + } + makeCanvasWithDSL(t, "canvas-session-owner", "user-1", "tenant-1", "v-session-owner", dsl) + foreignMessage := json.RawMessage(`[{"role":"assistant","content":"foreign"}]`) + if err := testDB.Create(&entity.API4Conversation{ + ID: "session-foreign", + DialogID: "canvas-session-owner", + UserID: "user-2", + Message: foreignMessage, + Reference: json.RawMessage(`[]`), + DSL: entity.JSONMap(dsl), + }).Error; err != nil { + t.Fatalf("create foreign session: %v", err) + } + + events, err := NewAgentService().RunAgent( + context.Background(), + "user-1", + "canvas-session-owner", + "session-foreign", + "", + "attempted overwrite", + nil, + ) + if err == nil { + t.Fatal("RunAgent accepted a session owned by another user") + } + if events != nil { + t.Fatalf("events = %#v, want nil for rejected session", events) + } + if !errors.Is(err, dao.ErrUserCanvasNotFound) { + t.Fatalf("error = %v, want not-found authorization sentinel", err) + } + + var unchanged entity.API4Conversation + if err := testDB.Where("id = ?", "session-foreign").First(&unchanged).Error; err != nil { + t.Fatalf("reload foreign session: %v", err) + } + if string(unchanged.Message) != string(foreignMessage) { + t.Fatalf("foreign session message was overwritten: %s", unchanged.Message) + } +} + // TestRunAgent_RealCanvas_WaitForUserResume pins the resume path. // It publishes a 3-component DSL (Begin → Message → UserFillUp), // invokes RunAgent twice on the same (canvas, session), and asserts: @@ -257,6 +528,8 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { t.Cleanup(func() { dao.DB = orig }) dsl := map[string]any{ + "globals": map[string]any{"sys.history": []any{}}, + "history": []any{}, "components": map[string]any{ "begin_0": map[string]any{ "obj": map[string]any{ @@ -283,6 +556,16 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { "path": []any{"begin_0", "user_fill_up_0", "message_0"}, } makeCanvasWithDSL(t, "canvas-fillup", "user-1", "tenant-1", "v-fillup", dsl) + if err := testDB.Create(&entity.API4Conversation{ + ID: "session-fillup", + DialogID: "canvas-fillup", + UserID: "user-1", + Message: json.RawMessage(`[]`), + Reference: json.RawMessage(`[]`), + DSL: entity.JSONMap(dsl), + }).Error; err != nil { + t.Fatalf("create session: %v", err) + } // v3.6.1 Gap #1 fix: a real checkpoint store + serializer is // REQUIRED for the second Invoke to actually resume from @@ -315,8 +598,7 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { "canvas-fillup", "session-fillup", "", - "please ask", - ) + "please ask", nil) if err != nil { t.Fatalf("RunAgent run 1: %v", err) } @@ -330,6 +612,14 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { if waiting1[0].CpnID == "" { t.Error("run 1: waiting_for_user event has empty cpn_id") } + var interrupted entity.API4Conversation + if err := testDB.Where("id = ?", "session-fillup").First(&interrupted).Error; err != nil { + t.Fatalf("run 1: reload session: %v", err) + } + interruptedHistory, _ := interrupted.DSL["history"].([]any) + if len(interruptedHistory) != 1 { + t.Fatalf("run 1: persisted history = %#v, want only the user turn", interrupted.DSL["history"]) + } // Run 2: with the checkpoint store wired, eino loads the // saved state from run 1, the engine targets the UserFillUp @@ -345,8 +635,7 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { "canvas-fillup", "session-fillup", // SAME sessionID as run 1 "", - "my follow-up", - ) + "my follow-up", nil) if err != nil { t.Fatalf("RunAgent run 2: %v", err) } @@ -366,6 +655,97 @@ func TestRunAgent_RealCanvas_WaitForUserResume(t *testing.T) { if !strings.Contains(messages2[0].Content, "got: my follow-up") { t.Errorf("run 2: Content = %q, want substring %q", messages2[0].Content, "got: my follow-up") } + var completed entity.API4Conversation + if err := testDB.Where("id = ?", "session-fillup").First(&completed).Error; err != nil { + t.Fatalf("run 2: reload session: %v", err) + } + completedHistory, _ := completed.DSL["history"].([]any) + if len(completedHistory) != 3 { + t.Fatalf("run 2: persisted history = %#v, want two user turns and one assistant turn", completed.DSL["history"]) + } +} + +func TestRunAgent_InterruptPersistsPartialAssistantHistory(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.API4Conversation{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + dsl := map[string]any{ + "globals": map[string]any{"sys.history": []any{}}, + "history": []any{}, + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{"component_name": "Begin", "params": map[string]any{}}, + "downstream": []any{"prompt_0"}, + }, + "prompt_0": map[string]any{ + "obj": map[string]any{"component_name": "Message", "params": map[string]any{"text": "partial answer"}}, + "upstream": []any{"begin_0"}, + "downstream": []any{"user_fill_up_0"}, + }, + "user_fill_up_0": map[string]any{ + "obj": map[string]any{"component_name": "UserFillUp", "params": map[string]any{"enable_tips": true}}, + "upstream": []any{"prompt_0"}, + }, + }, + "path": []any{"begin_0", "prompt_0", "user_fill_up_0"}, + } + makeCanvasWithDSL(t, "canvas-partial", "user-1", "tenant-1", "v-partial", dsl) + if err := testDB.Create(&entity.API4Conversation{ + ID: "session-partial", + DialogID: "canvas-partial", + UserID: "user-1", + Message: json.RawMessage(`[]`), + Reference: json.RawMessage(`[]`), + DSL: entity.JSONMap(dsl), + }).Error; err != nil { + t.Fatalf("create session: %v", err) + } + + events, err := NewAgentService().RunAgent( + context.Background(), + "user-1", + "canvas-partial", + "session-partial", + "", + "question", + nil, + ) + if err != nil { + t.Fatalf("RunAgent: %v", err) + } + messages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) != 0 || len(waiting) != 1 || !done { + t.Fatalf("messages=%+v waiting=%+v errors=%+v done=%v", messages, waiting, errs, done) + } + if len(messages) != 1 || messages[0].Content != "partial answer" { + t.Fatalf("partial messages = %+v", messages) + } + + var session entity.API4Conversation + if err := testDB.Where("id = ?", "session-partial").First(&session).Error; err != nil { + t.Fatalf("reload session: %v", err) + } + history, ok := session.DSL["history"].([]any) + if !ok || len(history) != 2 { + t.Fatalf("persisted history = %#v, want user and partial assistant", session.DSL["history"]) + } + assistant, ok := history[1].([]any) + if !ok || len(assistant) != 2 || assistant[0] != "assistant" { + t.Fatalf("assistant history = %#v", history[1]) + } + payload, ok := assistant[1].(map[string]any) + if !ok || payload["content"] != "partial answer" { + t.Fatalf("assistant payload = %#v, want partial answer", assistant[1]) + } } func TestRunAgent_RealCanvas_WaitForUserResume_EventSemantics(t *testing.T) { @@ -425,8 +805,7 @@ func TestRunAgent_RealCanvas_WaitForUserResume_EventSemantics(t *testing.T) { "canvas-fillup-events", "session-fillup-events", "", - "please ask", - ) + "please ask", nil) if err != nil { t.Fatalf("RunAgent run 1: %v", err) } @@ -449,8 +828,7 @@ func TestRunAgent_RealCanvas_WaitForUserResume_EventSemantics(t *testing.T) { "canvas-fillup-events", "session-fillup-events", "", - "my follow-up", - ) + "my follow-up", nil) if err != nil { t.Fatalf("RunAgent run 2: %v", err) } @@ -568,8 +946,7 @@ func TestRunAgent_RealCanvas_GroupedParallelOuterFollower(t *testing.T) { "canvas-parallel", "session-parallel", "", - "a,b,c", - ) + "a,b,c", nil) if err != nil { t.Fatalf("RunAgent: %v", err) } @@ -639,8 +1016,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) { "canvas-all", "session-all-loop", "", - "loop", - ) + "loop", nil) if err != nil { t.Fatalf("RunAgent run 1: %v", err) } @@ -680,8 +1056,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) { "canvas-all", "session-all-loop", "", - "loop", - ) + "loop", nil) if err != nil { t.Fatalf("RunAgent run 2: %v", err) } @@ -703,7 +1078,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume(t *testing.T) { } events3, err := svc.RunAgent( - context.Background(), "user-1", "canvas-all", "session-all-loop", "", "1", + context.Background(), "user-1", "canvas-all", "session-all-loop", "", "1", nil, ) if err != nil { t.Fatalf("RunAgent run 3: %v", err) @@ -767,8 +1142,7 @@ func TestRunAgent_AllFixture_LoopInterruptResume_MultiTurn(t *testing.T) { "canvas-all-multi", sessionID, "", - input, - ) + input, nil) if err != nil { t.Fatalf("RunAgent run %d (%q): %v", i+1, input, err) } @@ -861,8 +1235,7 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) { "canvas-all-iteration", sessionID, "", - "iteration", - ) + "iteration", nil) if err != nil { t.Fatalf("RunAgent run 1: %v", err) } @@ -891,8 +1264,7 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) { "canvas-all-iteration", sessionID, "", - "iteration", - ) + "iteration", nil) if err != nil { t.Fatalf("RunAgent run 2: %v", err) } @@ -914,7 +1286,7 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) { } events3, err := svc.RunAgent( - context.Background(), "user-1", "canvas-all-iteration", sessionID, "", "a,b,c,d,e", + context.Background(), "user-1", "canvas-all-iteration", sessionID, "", "a,b,c,d,e", nil, ) if err != nil { t.Fatalf("RunAgent run 3: %v", err) @@ -927,12 +1299,9 @@ func TestRunAgent_AllFixture_IterationFormatsItems(t *testing.T) { t.Fatalf("run 3: waiting=%+v done=%v", waiting3, done3) } content := messages2[0].Content - // Go renders []any{"a","b","c","d","e"} via fmt.Sprintf("%v", ...) - // as "[a b c d e]" (space-separated, no commas, no quotes). The - // DSL template "输入数组: {StringTransform:SplitCSV@result}" - // therefore produces "输入数组: [a b c d e]" with the leading - // space the author put in the DSL between ':' and '{'. - want := "迭代结束。\n输入数组: [a b c d e]\n格式化输出(lines):[0: a 1: b 2: c 3: d 4: e]" + // Python Message._stringify_message_value renders non-string template + // values as JSON, so list references keep commas and quotes. + want := "迭代结束。\n输入数组: [\"a\",\"b\",\"c\",\"d\",\"e\"]\n格式化输出(lines):[\"0: a\",\"1: b\",\"2: c\",\"3: d\",\"4: e\"]" if content != want { t.Fatalf("run 2: Content = %q, want %q", content, want) } @@ -976,8 +1345,7 @@ func TestRunAgent_AllFixture_VarAssigner(t *testing.T) { "canvas-all-var-assigner", "session-all-var-assigner", "", - "var_assigner", - ) + "var_assigner", nil) if err != nil { t.Fatalf("RunAgent: %v", err) } @@ -1001,6 +1369,7 @@ func TestRunAgent_AllFixture_VarAssigner(t *testing.T) { "session-all-var-assigner", "", "var_assigner", + nil, ) if err != nil { t.Fatalf("RunAgent resume: %v", err) @@ -1060,8 +1429,7 @@ func TestRunAgent_AllFixture_DataOps(t *testing.T) { "canvas-all-data-ops", "session-all-data-ops", "", - "data_ops", - ) + "data_ops", nil) if err != nil { t.Fatalf("RunAgent: %v", err) } @@ -1085,6 +1453,7 @@ func TestRunAgent_AllFixture_DataOps(t *testing.T) { "session-all-data-ops", "", "data_ops", + nil, ) if err != nil { t.Fatalf("RunAgent resume: %v", err) @@ -1107,10 +1476,10 @@ func TestRunAgent_AllFixture_DataOps(t *testing.T) { // not the legacy hashableKey first-field). desc + score picks // Alpha(0.91), Beta(0.88), Gamma(0.76) regardless of input order. // Head(2) then takes Alpha, Beta. - if !strings.Contains(messages[0].Content, "first=map[id:1 score:0.91 tag:demo title:Alpha]") { + if !strings.Contains(messages[0].Content, `first={"id":1,"score":0.91,"tag":"demo","title":"Alpha"}`) { t.Errorf("Content = %q, want first=Alpha (sort_by=score desc top-1)", messages[0].Content) } - if !strings.Contains(messages[0].Content, "last=map[id:2 score:0.88 tag:demo title:Beta]") { + if !strings.Contains(messages[0].Content, `last={"id":2,"score":0.88,"tag":"demo","title":"Beta"}`) { t.Errorf("Content = %q, want last=Beta (sort_by=score desc top-2)", messages[0].Content) } } @@ -1166,8 +1535,7 @@ func TestRunAgent_RealCanvas_CompileFails(t *testing.T) { "canvas-bogus", "session-bogus", "", - "hello", - ) + "hello", nil) if err != nil { t.Fatalf("RunAgent returned sync error: %v", err) } @@ -1239,8 +1607,7 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) { "canvas-all-categorize", "session-all-categorize", "", - "categorize", - ) + "categorize", nil) if err != nil { t.Fatalf("RunAgent run 1: %v", err) } @@ -1268,8 +1635,7 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) { "canvas-all-categorize", "session-all-categorize", "", - "categorize", - ) + "categorize", nil) if err != nil { t.Fatalf("RunAgent run 2: %v", err) } @@ -1304,6 +1670,7 @@ func TestRunAgent_AllFixture_CategorizeResume(t *testing.T) { "session-all-categorize", "", "hello", + nil, ) if err != nil { t.Fatalf("RunAgent run 3: %v", err) @@ -1400,8 +1767,7 @@ func TestRunAgent_RealCanvas_InvokeFails(t *testing.T) { "canvas-invoke-fail", "session-invoke-fail", "", - "hello", - ) + "hello", nil) if err != nil { t.Fatalf("RunAgent returned sync error: %v", err) } @@ -1505,8 +1871,7 @@ func TestRunAgent_RunTracker_AttachCheckpoint_CallSequence(t *testing.T) { "canvas-cp", "session-cp", "", // latest version - "world", - ) + "world", nil) if err != nil { t.Fatalf("RunAgent: %v", err) } @@ -1566,3 +1931,283 @@ func TestRunAgent_RunTracker_AttachCheckpoint_CallSequence(t *testing.T) { t.Error("finished_at is empty — MarkSucceeded missing the timestamp?") } } + +// TestRunAgent_FilesPopulateIteration verifies the full upload object -> +// sys.files -> Parallel/Iteration item path. +func TestRunAgent_FilesPopulateIteration(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.UserTenant{}, + &entity.APIToken{}, + &entity.API4Conversation{}, + &entity.TenantModelProvider{}, + &entity.TenantModelInstance{}, + &entity.TenantModel{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + canvasID := "canvas-files-e2e" + sessionID := "session-files-e2e" + versionID := "v-files-e2e" + memory := storage.NewMemoryStorage() + if err := memory.Put("user-1-downloads", "upload-1", []byte("iteration payload")); err != nil { + t.Fatalf("put upload: %v", err) + } + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(memory) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + dsl := map[string]any{ + "globals": map[string]any{ + "sys.files": []any{}, + "sys.user_id": "", + }, + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Begin", + "params": map[string]any{"layout_recognize": "Plain Text"}, + }, + "downstream": []any{"parallel_0"}, + }, + "parallel_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Parallel", + "params": map[string]any{ + "items_ref": "sys.files", + "outputs": map[string]any{ + "lines": map[string]any{"ref": "format_0@result"}, + }, + }, + }, + "upstream": []any{"begin_0"}, + "downstream": []any{"message_0"}, + }, + "iteration_item_0": map[string]any{ + "obj": map[string]any{ + "component_name": "IterationItem", + "params": map[string]any{}, + }, + "upstream": []any{"parallel_0"}, + "downstream": []any{"format_0"}, + "parent_id": "parallel_0", + }, + "format_0": map[string]any{ + "obj": map[string]any{ + "component_name": "StringTransform", + "params": map[string]any{ + "method": "merge", + "script": "{item}", + "delimiters": []any{"|"}, + }, + }, + "upstream": []any{"iteration_item_0"}, + "parent_id": "parallel_0", + }, + "message_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Message", + "params": map[string]any{"content": []any{"{parallel_0@lines}"}}, + }, + "upstream": []any{"parallel_0"}, + }, + }, + "path": []any{"begin_0", "parallel_0", "message_0"}, + "graph": map[string]any{ + "nodes": []any{ + map[string]any{"id": "iteration_item_0", "parentId": "parallel_0"}, + map[string]any{"id": "format_0", "parentId": "parallel_0"}, + }, + }, + } + makeCanvasWithDSL(t, canvasID, "user-1", "tenant-1", versionID, dsl) + + testFiles := []map[string]interface{}{ + { + "id": "upload-1", + "name": "notes.txt", + "mime_type": "text/plain", + "created_by": "user-1", + }, + } + + svc := NewAgentService() + events, err := svc.RunAgent( + context.Background(), + "user-1", + canvasID, + sessionID, + "", + "hello-files", testFiles) + if err != nil { + t.Fatalf("RunAgent with files: %v", err) + } + messages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) > 0 { + t.Fatalf("unexpected error events with files: %+v", errs) + } + if len(waiting) > 0 { + t.Fatalf("unexpected waiting_for_user events with files: %+v", waiting) + } + if len(messages) != 1 { + t.Fatalf("expected 1 message event with files, got %d", len(messages)) + } + if !strings.Contains(messages[0].Content, "File: notes.txt") || !strings.Contains(messages[0].Content, "iteration payload") { + t.Errorf("Content = %q, want parsed upload from iteration", messages[0].Content) + } + if !done { + t.Error("missing terminator done event with files") + } +} + +func TestRunAgent_MissingUploadEmitsError(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.UserTenant{}, + &entity.APIToken{}, + &entity.API4Conversation{}, + &entity.TenantModelProvider{}, + &entity.TenantModelInstance{}, + &entity.TenantModel{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + origDB := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = origDB }) + + memory := storage.NewMemoryStorage() + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(memory) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + dsl := 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": "should not run"}}, + "upstream": []any{"begin"}, + }, + }, + "path": []any{"begin", "message"}, + } + makeCanvasWithDSL(t, "canvas-missing-upload", "user-1", "tenant-1", "v-missing-upload", dsl) + + events, err := NewAgentService().RunAgent( + context.Background(), + "user-1", + "canvas-missing-upload", + "session-missing-upload", + "", + "hello", + []map[string]interface{}{{ + "id": "missing", + "name": "missing.txt", + "mime_type": "text/plain", + "created_by": "user-1", + }}, + ) + if err != nil { + t.Fatalf("RunAgent: %v", err) + } + messages, _, errs, done := drainAgentEvents(t, events) + if len(messages) != 0 { + t.Fatalf("messages = %+v, want none", messages) + } + if len(errs) != 1 || !strings.Contains(errs[0].Message, "parse agent files") { + t.Fatalf("errors = %+v, want parse agent files error", errs) + } + if !done { + t.Fatal("missing done event") + } +} + +// TestRunAgent_NoFilesRunsNormally verifies that the files-aware +// RunAgent path does not regress when no files are passed (nil +// parameter). This is a counterpart to the existing +// TestRunAgent_RealCanvas_BeginMessage to ensure backward +// compatibility. +func TestRunAgent_NoFilesRunsNormally(t *testing.T) { + testDB := setupServiceTestDB(t) + if err := testDB.AutoMigrate( + &entity.UserCanvas{}, + &entity.UserCanvasVersion{}, + &entity.UserTenant{}, + &entity.APIToken{}, + &entity.API4Conversation{}, + &entity.TenantModelProvider{}, + &entity.TenantModelInstance{}, + &entity.TenantModel{}, + ); err != nil { + t.Fatalf("migrate: %v", err) + } + orig := dao.DB + dao.DB = testDB + t.Cleanup(func() { dao.DB = orig }) + + canvasID := "canvas-nofiles-e2e" + sessionID := "session-nofiles-e2e" + versionID := "v-nofiles-e2e" + + dsl := map[string]any{ + "components": map[string]any{ + "begin_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Begin", + "params": map[string]any{}, + }, + "downstream": []any{"message_0"}, + }, + "message_0": map[string]any{ + "obj": map[string]any{ + "component_name": "Message", + "params": map[string]any{"text": "echo: {{sys.query}}"}, + }, + "upstream": []any{"begin_0"}, + }, + }, + "path": []any{"begin_0", "message_0"}, + } + makeCanvasWithDSL(t, canvasID, "user-1", "tenant-1", versionID, dsl) + + svc := NewAgentService() + events, err := svc.RunAgent( + context.Background(), + "user-1", + canvasID, + sessionID, + "", + "hello-nofiles", nil) + if err != nil { + t.Fatalf("RunAgent without files: %v", err) + } + messages, waiting, errs, done := drainAgentEvents(t, events) + if len(errs) > 0 { + t.Fatalf("unexpected error events without files: %+v", errs) + } + if len(waiting) > 0 { + t.Fatalf("unexpected waiting_for_user events without files: %+v", waiting) + } + if len(messages) != 1 { + t.Fatalf("expected 1 message event without files, got %d", len(messages)) + } + if !strings.Contains(messages[0].Content, "hello-nofiles") { + t.Errorf("Content = %q, want substring %q", messages[0].Content, "hello-nofiles") + } + if !done { + t.Error("missing terminator done event without files") + } +} diff --git a/internal/service/agent_test.go b/internal/service/agent_test.go index aff57333d9..34b2b2e512 100644 --- a/internal/service/agent_test.go +++ b/internal/service/agent_test.go @@ -536,8 +536,7 @@ func TestRunAgent_VersionBelongsToOtherCanvas(t *testing.T) { "canvas-1", // we're running canvas-1… "", // session ID auto-generated "v-on-canvas-2", // …with a version that belongs to canvas-2 - "hi", - ) + "hi", nil) if err == nil { t.Fatal("expected error when version belongs to a different canvas (IDOR guard)") } @@ -583,8 +582,7 @@ func TestRunAgent_VersionNotFound(t *testing.T) { "canvas-1", "", "does-not-exist", - "hi", - ) + "hi", nil) if err == nil { t.Fatal("expected error when explicit version id does not exist") } @@ -641,8 +639,7 @@ func TestRunAgent_NoVersionPublishedPlaceholder(t *testing.T) { "canvas-empty", "test-session", "", // no explicit version → use GetLatest, which returns ErrUserCanvasVersionNotFound - "hi", - ) + "hi", nil) if err != nil { t.Fatalf("RunAgent should proceed with placeholder when no version published: %v", err) } @@ -751,8 +748,7 @@ func TestRunAgent_StorageErrorFromCanvasAccess(t *testing.T) { "canvas-1", "", "", - "hi", - ) + "hi", nil) if err == nil { t.Fatal("expected storage error from closed DB") } @@ -1832,3 +1828,22 @@ func TestDeleteAgentSessionItem_RejectsIDOR(t *testing.T) { t.Fatalf("session was deleted despite IDOR rejection: %+v", verify) } } + +func TestAgentHistoryRenderingMatchesPythonShapes(t *testing.T) { + user := renderUserHistoryValue(map[string]any{ + "content": "你好", + "count": 2, + }) + if user != `{"content":"你好","count":2}` { + t.Fatalf("rendered user history = %q", user) + } + + assistant := pythonHistoryRepr(map[string]any{ + "content": "it's ready\nnext", + "ok": true, + }) + want := `{'content': 'it\'s ready\nnext', 'ok': True}` + if assistant != want { + t.Fatalf("rendered assistant history = %q, want %q", assistant, want) + } +} diff --git a/internal/service/bot.go b/internal/service/bot.go index 9be1960899..b88cbcd5f2 100644 --- a/internal/service/bot.go +++ b/internal/service/bot.go @@ -145,24 +145,21 @@ func (s *BotService) AgentbotCompletion( if _, err := s.loadCanvas(ctx, tenantID, agentID); err != nil { return nil, common.CodeDataError, err } - // Compose the canvas user input from req.UserInput (the - // `inputs` dict body field) plus the top-level `question` and - // `files` fields. The python canvas_service.completion at - // api/db/services/canvas_service.py:313 reads all three; the - // previous code dropped question/files, so a body like + // Compose the canvas user input from req.UserInput (the `inputs` + // dict body field) plus the top-level `question`. Files remain a + // separate RunAgent argument so they can populate sys.files. The + // Python canvas_service.completion reads the same three fields + // separately; the previous code dropped question/files, so a body like // `{"question":"hi"}` reached the canvas with empty inputs. - userInput := make(map[string]any, len(req.UserInput)+2) + userInput := make(map[string]any, len(req.UserInput)+1) for k, v := range req.UserInput { userInput[k] = v } if req.Question != "" { userInput["question"] = req.Question } - if len(req.Files) > 0 { - userInput["files"] = req.Files - } ch, err := s.agentService.RunAgent(ctx, tenantID, agentID, - req.SessionID, "", userInput) + req.SessionID, "", userInput, req.Files) if err != nil { return nil, common.CodeDataError, err } @@ -178,12 +175,10 @@ type AgentbotCompletionRequest struct { SessionID string `json:"session_id"` UserID string `json:"user_id"` Stream bool `json:"stream"` - // UserInput is the dict-shaped root input the canvas run expects - // (mirrors the python "question"/"files"/"inputs" trio collapsed - // into one map). - UserInput map[string]any `json:"inputs"` - Question string `json:"question"` - Files []string `json:"files"` + // UserInput is the dict-shaped root input the canvas run expects. + UserInput map[string]any `json:"inputs"` + Question string `json:"question"` + Files []map[string]interface{} `json:"files"` } // ChatbotCompletionRequest is the request body for diff --git a/internal/service/document_test.go b/internal/service/document_test.go index e6f6d7b297..938d3fa47b 100644 --- a/internal/service/document_test.go +++ b/internal/service/document_test.go @@ -396,6 +396,7 @@ func setupServiceTestDB(t *testing.T) *gorm.DB { &entity.User{}, &entity.Tenant{}, &entity.UserTenant{}, + &entity.API4Conversation{}, ); err != nil { t.Fatalf("failed to migrate: %v", err) } diff --git a/internal/service/file.go b/internal/service/file.go index a6f8822123..a7954d77f1 100644 --- a/internal/service/file.go +++ b/internal/service/file.go @@ -19,6 +19,7 @@ package service import ( "context" "encoding/base64" + "encoding/json" "fmt" "html" "io" @@ -1123,6 +1124,91 @@ func (s *FileService) GetFileContents(uid string, fileDicts []map[string]interfa return texts, images, nil } +// parseAgentUploads resolves descriptors returned by upload_info from the +// caller's downloads bucket and converts them to sys.files values. +func (s *FileService) parseAgentUploads(userID string, fileDicts []map[string]interface{}, layoutRecognize string) ([]string, error) { + storageImpl := storage.GetStorageFactory().GetStorage() + if storageImpl == nil { + return nil, fmt.Errorf("storage not initialized") + } + + contents := make([]string, 0, len(fileDicts)) + for i, fd := range fileDicts { + id, _ := fd["id"].(string) + name, _ := fd["name"].(string) + mimeType, _ := fd["mime_type"].(string) + createdBy, _ := fd["created_by"].(string) + if id == "" || name == "" || mimeType == "" || createdBy == "" { + return nil, fmt.Errorf("file %d: id, name, mime_type, and created_by are required", i) + } + if createdBy != userID { + return nil, fmt.Errorf("file %q: created_by does not match the current user", name) + } + + data, err := storageImpl.Get(createdBy+"-downloads", id) + if err != nil { + return nil, fmt.Errorf("file %q: read upload: %w", name, err) + } + if len(data) == 0 { + return nil, fmt.Errorf("file %q: upload is empty", name) + } + + mediaType := strings.ToLower(strings.TrimSpace(strings.Split(mimeType, ";")[0])) + if strings.HasPrefix(mediaType, "image/") { + contents = append(contents, "data:"+mediaType+";base64,"+base64.StdEncoding.EncodeToString(data)) + continue + } + + content, err := parseAgentUploadContent(name, data, layoutRecognize) + if err != nil { + return nil, fmt.Errorf("file %q: parse upload: %w", name, err) + } + contents = append(contents, content) + } + return contents, nil +} + +func parseAgentUploadContent(filename string, data []byte, layoutRecognize string) (string, error) { + content := string(data) + fileType := utility.GetFileType(filename) + if fileType != utility.FileTypeOTHER { + fp, err := parser.GetParser(fileType) + if err != nil { + return "", err + } + if configurable, ok := fp.(interface{ ConfigureFromSetup(map[string]any) }); ok { + configurable.ConfigureFromSetup(map[string]any{"layout_recognize": layoutRecognize}) + } + res := fp.ParseWithResult(filename, data) + if res.Err != nil { + return "", res.Err + } + switch res.OutputFormat { + case "text": + content = res.Text + case "markdown": + content = res.Markdown + case "html": + content = res.HTML + case "json": + parts := make([]string, 0, len(res.JSON)) + for _, item := range res.JSON { + if text, ok := item["text"].(string); ok { + parts = append(parts, text) + continue + } + raw, err := json.Marshal(item) + if err != nil { + return "", err + } + parts = append(parts, string(raw)) + } + content = strings.Join(parts, "\n") + } + } + return fmt.Sprintf("\n -----------------\nFile: %s\nContent as following: \n%s", filename, content), nil +} + // parseFileContent tries to parse a file's contents using the appropriate parser. // Falls back to returning raw text if no parser is available. func parseFileContent(filename string, data []byte) string { diff --git a/internal/service/file_test.go b/internal/service/file_test.go index cf814e646f..58799548a7 100644 --- a/internal/service/file_test.go +++ b/internal/service/file_test.go @@ -229,6 +229,67 @@ func TestFileService_GetFileContents_Accessible(t *testing.T) { } } +func TestFileService_ParseAgentUploads_TextAndImageInRequestOrder(t *testing.T) { + memory := storage.NewMemoryStorage() + if err := memory.Put("user-1-downloads", "text-id", []byte("uploaded text")); err != nil { + t.Fatalf("put text: %v", err) + } + if err := memory.Put("user-1-downloads", "image-id", []byte("png")); err != nil { + t.Fatalf("put image: %v", err) + } + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(memory) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + contents, err := testFileService().parseAgentUploads("user-1", []map[string]interface{}{ + {"id": "text-id", "name": "notes.txt", "mime_type": "text/plain", "created_by": "user-1"}, + {"id": "image-id", "name": "photo.bin", "mime_type": "image/png", "created_by": "user-1"}, + }, "Plain Text") + if err != nil { + t.Fatalf("ParseAgentUploads: %v", err) + } + if len(contents) != 2 { + t.Fatalf("contents length = %d, want 2", len(contents)) + } + if !strings.Contains(contents[0], "File: notes.txt") || !strings.Contains(contents[0], "uploaded text") { + t.Fatalf("unexpected text content: %q", contents[0]) + } + if contents[1] != "data:image/png;base64,cG5n" { + t.Fatalf("unexpected image content: %q", contents[1]) + } +} + +func TestFileService_ParseAgentUploads_RejectsForeignOwner(t *testing.T) { + memory := storage.NewMemoryStorage() + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(memory) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + _, err := testFileService().parseAgentUploads("user-1", []map[string]interface{}{ + {"id": "file-id", "name": "secret.txt", "mime_type": "text/plain", "created_by": "user-2"}, + }, "") + if err == nil || !strings.Contains(err.Error(), "created_by does not match") { + t.Fatalf("error = %v, want created_by mismatch", err) + } +} + +func TestFileService_ParseAgentUploads_MissingObjectFails(t *testing.T) { + memory := storage.NewMemoryStorage() + factory := storage.GetStorageFactory() + originalStorage := factory.GetStorage() + factory.SetStorage(memory) + t.Cleanup(func() { factory.SetStorage(originalStorage) }) + + _, err := testFileService().parseAgentUploads("user-1", []map[string]interface{}{ + {"id": "missing", "name": "missing.txt", "mime_type": "text/plain", "created_by": "user-1"}, + }, "") + if err == nil || !strings.Contains(err.Error(), "read upload") { + t.Fatalf("error = %v, want storage read failure", err) + } +} + func TestFileService_DownloadAgentFile_Success(t *testing.T) { // Setup mock storage expectedBlob := []byte("fake file content")