fix(go-agent): preserve canvas system state across turns (#17010)

## Summary - Preserve request-scoped system variables such as files and user IDs during Canvas execution. - Persist conversation history, turn counts, and tool memory in the session DSL across turns. - Parse agent uploads into `sys.files` and align system variable rendering with Python. ## Testing - `bash build.sh --test ./internal/agent/...` - `bash build.sh --test ./internal/service/...` <img width="1896" height="1232" alt="image" src="https://github.com/user-attachments/assets/b420cd97-53c3-470f-a3e1-d39cea26a213" />
2026-07-18 21:57:27 +08:00 · 2026-07-17 15:53:58 +08:00
parent 7336c27814
commit 1b77e3ebcd
28 changed files with 2363 additions and 174 deletions
--- 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
--- 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")
+	}
+}
--- 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)
--- a/internal/agent/canvas/decode_history_test.go
+++ 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)
+	}
+}
--- 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
 	}
--- 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)
 			}
 		}
 	}
--- 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) {
--- 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 {
--- 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
--- 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
--- 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()))
 	}
 }
--- 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.
--- 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, &copyItem); 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.
--- a/internal/agent/runtime/state_clone_test.go
+++ 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)
+	}
+}
--- 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()

--- 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)
+}
--- 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{
--- 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)
+	}
+}
--- 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.
--- 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.
--- 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;
--- a/internal/service/agent_persistence_test.go
+++ 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)
+		}
+	}
+}
--- 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")
+	}
+}
--- 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)
+	}
+}
--- 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
--- 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)
 	}
--- 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 {
--- 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")