fix(go-agent): align logs auth with Python and harden event data (#17617)

## Summary

- Align Go agent logs access with the Python behavior by checking tenant
access to the target agent instead of requiring the token to be bound
to that agent.
- Marshal canvas lifecycle event payloads safely so non-serializable
inputs do not break Thinking expansion in the UI.

  ## Testing

- `CGO_ENABLED=0 go test -count=1 -v -run
"Handler|Attachment|Logs|Chatbot|Agentbot|BuildWorkflow|NodeLifecycle|VarRef|Resolve|Extract"./internal/handler/
./internal/agent/canvas/...` PASS

internal/agent/canvas/scheduler.go

@@ -20,7 +20,6 @@ package canvas
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
@@ -285,7 +284,7 @@ func nodeStartedAt(ctx context.Context, state *CanvasState, cpnID, componentName
 		state.Sys["_node_start_"+cpnID] = now
 		state.Sys["_node_inputs_"+cpnID] = sanitizeNodeInputs(inputs)
 	}
-	nsData, _ := json.Marshal(NodeStartedData{
+	nsData, err := runtime.SafeJSONMarshal(NodeStartedData{
 		Inputs:        sanitizeNodeInputs(inputs),
 		CreatedAt:     now,
 		ComponentID:   cpnID,
@@ -293,6 +292,15 @@ func nodeStartedAt(ctx context.Context, state *CanvasState, cpnID, componentName
 		ComponentType: componentType,
 		Thoughts:      "",
 	})
+	if err != nil {
+		common.Warn("node_started marshal failed",
+			zap.String("cpnID", cpnID),
+			zap.String("componentName", componentName),
+			zap.Error(err),
+		)
+		nsData = []byte(fmt.Sprintf(`{"component_id":%q,"component_name":%q,"component_type":%q}`,
+			cpnID, componentName, componentType))
+	}
 	meta := GetRunMeta(ctx)
 	msgID, sessionID := "", ""
 	if meta != nil {
@@ -346,7 +354,7 @@ func nodeFinishedNow(ctx context.Context, state *CanvasState, cpnID, componentNa
 		nfErr = nodeErr.Error()
 	}
 
-	nfData, _ := json.Marshal(NodeFinishedData{
+	nfData, err := runtime.SafeJSONMarshal(NodeFinishedData{
 		Inputs:        inputs,
 		Outputs:       outputs,
 		ComponentID:   cpnID,
@@ -356,6 +364,15 @@ func nodeFinishedNow(ctx context.Context, state *CanvasState, cpnID, componentNa
 		ElapsedTime:   elapsed,
 		CreatedAt:     now,
 	})
+	if err != nil {
+		common.Warn("node_finished marshal failed",
+			zap.String("cpnID", cpnID),
+			zap.String("componentName", componentName),
+			zap.Error(err),
+		)
+		nfData = []byte(fmt.Sprintf(`{"component_id":%q,"component_name":%q,"component_type":%q}`,
+			cpnID, componentName, componentType))
+	}
 	meta := GetRunMeta(ctx)
 	msgID, sessionID := "", ""
 	if meta != nil {

internal/agent/canvas/scheduler_test.go

@@ -3,6 +3,7 @@ package canvas
 
 import (
 	"context"
+	"encoding/json"
 	"strings"
 	"testing"
 )
@@ -245,3 +246,89 @@ func TestBuildWorkflow_ErrorsOnSelfEdge(t *testing.T) {
 		t.Fatalf("expected 'self-edge' in error, got: %v", err)
 	}
 }
+
+func TestNodeLifecycleEventsSafeMarshalNonSerializableInputs(t *testing.T) {
+	events := make(chan RunEvent, 2)
+	ctx := WithRunMeta(context.Background(), &RunMeta{
+		Events:    events,
+		MessageID: "msg-1",
+		SessionID: "session-1",
+	})
+	state := NewCanvasState("run-1", "session-1")
+	inputs := map[string]any{
+		"query":    "hi",
+		"callback": func() {},
+	}
+
+	nodeStartedAt(ctx, state, "agent_0", "Agent", "Agent", inputs)
+	nodeFinishedNow(ctx, state, "agent_0", "Agent", "Agent", nil)
+
+	for i, wantType := range []string{"node_started", "node_finished"} {
+		select {
+		case ev := <-events:
+			if ev.Type != wantType {
+				t.Fatalf("event[%d].Type = %q, want %q", i, ev.Type, wantType)
+			}
+			var payload map[string]any
+			if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {
+				t.Fatalf("%s Data should be valid JSON, got %q: %v", wantType, ev.Data, err)
+			}
+			if payload["component_type"] != "Agent" {
+				t.Fatalf("%s component_type = %v, want Agent; payload=%v", wantType, payload["component_type"], payload)
+			}
+			in, ok := payload["inputs"].(map[string]any)
+			if !ok {
+				t.Fatalf("%s inputs = %T %v, want object", wantType, payload["inputs"], payload["inputs"])
+			}
+			if _, ok := in["callback"]; !ok {
+				t.Fatalf("%s inputs missing sanitized callback key: %v", wantType, in)
+			}
+			if in["callback"] != nil {
+				t.Fatalf("%s callback = %T %v, want nil after safe JSON cleanup", wantType, in["callback"], in["callback"])
+			}
+		default:
+			t.Fatalf("missing %s event", wantType)
+		}
+	}
+}
+
+func TestNodeLifecycleEventsFallbackMarshalErrorPreservesIdentity(t *testing.T) {
+	events := make(chan RunEvent, 2)
+	ctx := WithRunMeta(context.Background(), &RunMeta{
+		Events:    events,
+		MessageID: "msg-2",
+		SessionID: "session-2",
+	})
+	state := NewCanvasState("run-2", "session-2")
+	inputs := map[string]any{
+		"query":  "hi",
+		"broken": complex(1, 2),
+	}
+
+	nodeStartedAt(ctx, state, "agent_1", "Agent", "Agent", inputs)
+	nodeFinishedNow(ctx, state, "agent_1", "Agent", "Agent", nil)
+
+	for i, wantType := range []string{"node_started", "node_finished"} {
+		select {
+		case ev := <-events:
+			if ev.Type != wantType {
+				t.Fatalf("event[%d].Type = %q, want %q", i, ev.Type, wantType)
+			}
+			var payload map[string]any
+			if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {
+				t.Fatalf("%s Data should be valid JSON, got %q: %v", wantType, ev.Data, err)
+			}
+			if payload["component_id"] != "agent_1" {
+				t.Fatalf("%s component_id = %v, want agent_1; payload=%v", wantType, payload["component_id"], payload)
+			}
+			if payload["component_name"] != "Agent" {
+				t.Fatalf("%s component_name = %v, want Agent; payload=%v", wantType, payload["component_name"], payload)
+			}
+			if payload["component_type"] != "Agent" {
+				t.Fatalf("%s component_type = %v, want Agent; payload=%v", wantType, payload["component_type"], payload)
+			}
+		default:
+			t.Fatalf("missing %s event", wantType)
+		}
+	}
+}

internal/handler/bot.go

@@ -18,14 +18,11 @@ package handler
 
 import (
 	"context"
-	"encoding/json"
-	"fmt"
 
 	"github.com/gin-gonic/gin"
 
 	"ragflow/internal/agent/canvas"
 	"ragflow/internal/common"
-	"ragflow/internal/engine/redis"
 	"ragflow/internal/service"
 )
 
@@ -48,6 +45,7 @@ type botService interface {
 		ec common.ErrorCode, err error)
 	AgentbotCompletion(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) (
 		<-chan canvas.RunEvent, common.ErrorCode, error)
+	AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error)
 	ChatbotCompletion(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) (
 		<-chan service.ChatbotSSEFrame, common.ErrorCode, error)
 }
@@ -230,13 +228,12 @@ func (h *BotHandler) ChatbotCompletion(c *gin.Context) {
 
 // GetAgentbotLogs GET /api/v1/agentbots/<agent_id>/logs/<message_id>
 //
-// Beta-token sibling of GetAgentLogs. The shared/embedded chat
-// page's "Thinking" button hits this endpoint because the share
-// flow authenticates with a beta APIToken (no session JWT) and
-// the regular /api/v1/agents/<id>/logs/<msg> requires @login_required.
-// Mirrors python bot_api.py:agent_bot_logs.
+// Beta-token sibling of GetAgentLogs. The shared/embedded chat page's
+// "Thinking" button hits this endpoint because the share flow authenticates
+// with a beta APIToken (no session JWT). Mirrors python bot_api.py:agent_bot_logs.
 func (h *BotHandler) GetAgentbotLogs(c *gin.Context) {
-	if _, code, msg := GetUser(c); code != common.CodeSuccess {
+	user, code, msg := GetUser(c)
+	if code != common.CodeSuccess {
 		common.ResponseWithCodeData(c, code, nil, msg)
 		return
 	}
@@ -245,38 +242,25 @@ func (h *BotHandler) GetAgentbotLogs(c *gin.Context) {
 		common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "agent_id is required")
 		return
 	}
-	boundAgentID, _ := c.Get("agent_id")
-	boundAgentIDStr, _ := boundAgentID.(string)
-	if boundAgentIDStr == "" {
-		common.ResponseWithCodeData(c, common.CodeDataError, nil, "API token is not bound to an agent.")
-		return
-	}
-	if boundAgentIDStr != agentIDStr {
-		common.ResponseWithCodeData(c, common.CodeUnauthorized, nil, "API token is not authorized for this agent.")
-		return
-	}
 	messageID := c.Param("message_id")
 	if messageID == "" {
 		common.ResponseWithCodeData(c, common.CodeArgumentError, nil, "message_id is required")
 		return
 	}
-	key := fmt.Sprintf("%s-%s-logs", agentIDStr, messageID)
-	payload, rerr := redis.Get().Get(key)
-	// Surface Redis / decode failures instead of silently returning
-	// `{code: 0, data: {}}` — the previous form made the endpoint
-	// indistinguishable from "logs not yet written", which masked
-	// real outages and corrupted payloads from operators (PR review
-	// round 5, Major #6).
-	if rerr != nil {
-		common.ResponseWithCodeData(c, common.CodeServerError, nil, "failed to read agent logs")
-		return
-	}
-	data := map[string]interface{}{}
-	if payload != "" {
-		if uerr := json.Unmarshal([]byte(payload), &data); uerr != nil {
-			common.ResponseWithCodeData(c, common.CodeServerError, nil, "failed to decode agent logs")
+	// A beta token with DialogID is restricted to that agent. Tokens without
+	// DialogID remain tenant-scoped and are checked by AgentbotLogs through
+	// the existing Canvas access guard.
+	if boundAgentID, ok := c.Get("agent_id"); ok {
+		boundAgentIDStr, ok := boundAgentID.(string)
+		if ok && boundAgentIDStr != "" && boundAgentIDStr != agentIDStr {
+			common.ResponseWithCodeData(c, common.CodeUnauthorized, nil, "API token is not authorized for this agent.")
 			return
 		}
 	}
+	data, ec, err := h.botService.AgentbotLogs(c.Request.Context(), user.ID, agentIDStr, messageID)
+	if err != nil {
+		common.ResponseWithCodeData(c, ec, nil, err.Error())
+		return
+	}
 	common.SuccessWithData(c, data, "success")
 }

internal/handler/bot_test.go

@@ -42,6 +42,7 @@ type stubBotService struct {
 	chatbotInfoFn      func(ctx context.Context, tenantID, dialogID string) (string, string, string, string, bool, common.ErrorCode, error)
 	agentbotInputsFn   func(ctx context.Context, tenantID, agentID string) (string, string, string, string, map[string]any, common.ErrorCode, error)
 	agentbotCompleteFn func(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) (<-chan canvas.RunEvent, common.ErrorCode, error)
+	agentbotLogsFn     func(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error)
 	chatbotCompleteFn  func(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) (<-chan service.ChatbotSSEFrame, common.ErrorCode, error)
 }
 
@@ -66,6 +67,13 @@ func (s *stubBotService) AgentbotCompletion(ctx context.Context, tenantID, agent
 	return nil, common.CodeDataError, errors.New("not stubbed")
 }
 
+func (s *stubBotService) AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) {
+	if s.agentbotLogsFn != nil {
+		return s.agentbotLogsFn(ctx, tenantID, agentID, messageID)
+	}
+	return nil, common.CodeDataError, errors.New("not stubbed")
+}
+
 func (s *stubBotService) ChatbotCompletion(ctx context.Context, tenantID, dialogID string, req service.ChatbotCompletionRequest) (<-chan service.ChatbotSSEFrame, common.ErrorCode, error) {
 	if s.chatbotCompleteFn != nil {
 		return s.chatbotCompleteFn(ctx, tenantID, dialogID, req)
@@ -1209,7 +1217,7 @@ func TestGetAgentbotLogs_MissingRouteAgentID(t *testing.T) {
 	}
 }
 
-func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) {
+func TestGetAgentbotLogs_AllowsUnboundTokenForAccessibleAgent(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	w := httptest.NewRecorder()
 	c, _ := gin.CreateTestContext(w)
@@ -1222,6 +1230,99 @@ func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) {
 	}
 
 	h := NewBotHandler(nil)
+	h.botService = &stubBotService{agentbotLogsFn: func(_ context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) {
+		if tenantID != "u1" || agentID != "agent-a" || messageID != "msg-1" {
+			t.Fatalf("logs arguments = (%q, %q, %q)", tenantID, agentID, messageID)
+		}
+		return map[string]any{"events": []any{}}, common.CodeSuccess, nil
+	}}
+	h.GetAgentbotLogs(c)
+
+	var resp struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if resp.Code != int(common.CodeSuccess) {
+		t.Errorf("code = %d, want %d; message=%q", resp.Code, common.CodeSuccess, resp.Message)
+	}
+}
+
+func TestGetAgentbotLogs_AllowsMatchingBoundAgent(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("GET", "/api/v1/agentbots/agent-a/logs/msg-1", nil)
+	c.Set("user", &entity.User{ID: "u1"})
+	c.Set("agent_id", "agent-a")
+	c.Params = gin.Params{
+		{Key: "agent_id", Value: "agent-a"},
+		{Key: "message_id", Value: "msg-1"},
+	}
+
+	h := NewBotHandler(nil)
+	h.botService = &stubBotService{agentbotLogsFn: func(_ context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) {
+		return map[string]any{"events": []any{}}, common.CodeSuccess, nil
+	}}
+	h.GetAgentbotLogs(c)
+
+	var resp struct {
+		Code int `json:"code"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if resp.Code != int(common.CodeSuccess) {
+		t.Fatalf("code = %d, want %d; body = %s", resp.Code, common.CodeSuccess, w.Body.String())
+	}
+}
+
+func TestGetAgentbotLogs_DeniesMismatchedBoundAgent(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("GET", "/api/v1/agentbots/agent-b/logs/msg-1", nil)
+	c.Set("user", &entity.User{ID: "u1"})
+	c.Set("agent_id", "agent-a")
+	c.Params = gin.Params{
+		{Key: "agent_id", Value: "agent-b"},
+		{Key: "message_id", Value: "msg-1"},
+	}
+
+	h := NewBotHandler(nil)
+	h.botService = &stubBotService{agentbotLogsFn: func(context.Context, string, string, string) (map[string]any, common.ErrorCode, error) {
+		t.Fatal("AgentbotLogs should not be called for a mismatched bound agent")
+		return nil, common.CodeServerError, errors.New("unexpected call")
+	}}
+	h.GetAgentbotLogs(c)
+
+	var resp struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	}
+	_ = json.Unmarshal(w.Body.Bytes(), &resp)
+	if resp.Code != int(common.CodeUnauthorized) {
+		t.Fatalf("code = %d, want %d; body = %s", resp.Code, common.CodeUnauthorized, w.Body.String())
+	}
+	if !strings.Contains(resp.Message, "not authorized") {
+		t.Fatalf("message = %q, want bound-agent authorization error", resp.Message)
+	}
+}
+
+func TestGetAgentbotLogs_DeniesInaccessibleAgent(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest("GET",
+		"/api/v1/agentbots/agent-b/logs/msg-1", nil)
+	c.Set("user", &entity.User{ID: "u1"})
+	c.Params = gin.Params{
+		{Key: "agent_id", Value: "agent-b"},
+		{Key: "message_id", Value: "msg-1"},
+	}
+
+	h := NewBotHandler(nil)
+	h.botService = &stubBotService{agentbotLogsFn: func(context.Context, string, string, string) (map[string]any, common.ErrorCode, error) {
+		return nil, common.CodeDataError, errors.New("Can't find agent by ID: agent-b")
+	}}
 	h.GetAgentbotLogs(c)
 
 	var resp struct {
@@ -1232,37 +1333,8 @@ func TestGetAgentbotLogs_RequiresBoundAgentID(t *testing.T) {
 	if resp.Code != int(common.CodeDataError) {
 		t.Errorf("code = %d, want %d", resp.Code, common.CodeDataError)
 	}
-	if !strings.Contains(resp.Message, "not bound") {
-		t.Errorf("message = %q, want it to mention 'not bound'", resp.Message)
-	}
-}
-
-func TestGetAgentbotLogs_CrossAgentDenied(t *testing.T) {
-	gin.SetMode(gin.TestMode)
-	w := httptest.NewRecorder()
-	c, _ := gin.CreateTestContext(w)
-	c.Request = httptest.NewRequest("GET",
-		"/api/v1/agentbots/agent-b/logs/msg-1", nil)
-	c.Set("user", &entity.User{ID: "u1"})
-	c.Set("agent_id", "agent-a")
-	c.Params = gin.Params{
-		{Key: "agent_id", Value: "agent-b"},
-		{Key: "message_id", Value: "msg-1"},
-	}
-
-	h := NewBotHandler(nil)
-	h.GetAgentbotLogs(c)
-
-	var resp struct {
-		Code    int    `json:"code"`
-		Message string `json:"message"`
-	}
-	_ = json.Unmarshal(w.Body.Bytes(), &resp)
-	if resp.Code != int(common.CodeUnauthorized) {
-		t.Errorf("code = %d, want %d", resp.Code, common.CodeUnauthorized)
-	}
-	if !strings.Contains(resp.Message, "not authorized") {
-		t.Errorf("message = %q, want it to mention 'not authorized'", resp.Message)
+	if !strings.Contains(resp.Message, "Can't find agent") {
+		t.Errorf("message = %q, want an access denial", resp.Message)
 	}
 }
 
@@ -1275,7 +1347,6 @@ func TestGetAgentbotLogs_MissingMessageID(t *testing.T) {
 	c.Request = httptest.NewRequest("GET",
 		"/api/v1/agentbots/shared-x/logs/", nil)
 	c.Set("user", &entity.User{ID: "u1"})
-	c.Set("agent_id", "agent-real")
 	c.Params = gin.Params{{Key: "agent_id", Value: "agent-real"}}
 	// Gin's path param extraction returns "" for a missing
 	// segment so the handler must reject with CodeArgumentError.

internal/service/bot.go

@@ -24,6 +24,7 @@ package service
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"hash/fnv"
@@ -34,6 +35,7 @@ import (
 	"ragflow/internal/agent/dsl"
 	"ragflow/internal/common"
 	"ragflow/internal/dao"
+	"ragflow/internal/engine/redis"
 	"ragflow/internal/entity"
 )
 
@@ -178,6 +180,27 @@ func (s *BotService) AgentbotCompletion(
 	return ch, common.CodeSuccess, nil
 }
 
+// AgentbotLogs returns the stored execution timeline for an agentbot run.
+// Access is scoped to the caller's accessible tenants, matching the Python
+// agent_bot_logs endpoint.
+func (s *BotService) AgentbotLogs(ctx context.Context, tenantID, agentID, messageID string) (map[string]any, common.ErrorCode, error) {
+	if _, err := s.loadCanvas(ctx, tenantID, agentID); err != nil {
+		return nil, common.CodeDataError, err
+	}
+	payload, err := redis.Get().Get(fmt.Sprintf("%s-%s-logs", agentID, messageID))
+	if err != nil {
+		return nil, common.CodeServerError, errors.New("failed to read agent logs")
+	}
+	data := map[string]any{}
+	if payload == "" {
+		return data, common.CodeSuccess, nil
+	}
+	if err := json.Unmarshal([]byte(payload), &data); err != nil {
+		return nil, common.CodeServerError, errors.New("failed to decode agent logs")
+	}
+	return data, common.CodeSuccess, nil
+}
+
 // AgentbotCompletionRequest is the request body for
 // /api/v1/agentbots/<agent_id>/completions. We intentionally accept
 // the same fields the production /agents/chat/completions handler