fix(go-agent): use session IDs for cancellation and context flow (#17462)

## Summary

- Propagate request contexts through Agent Canvas execution and external
calls.
- Replace internal task IDs with session IDs while retaining `task_id`
as a wire alias.
- Complete session-scoped cancellation with Redis lease and token
validation.

## Testing

- Go backend tests passed.

<img width="1176" height="574" alt="image"
src="https://github.com/user-attachments/assets/b86560be-9b8d-45bb-97e9-921dffab8ebe"
/>

internal/handler/agent.go

@@ -37,6 +37,7 @@ import (
 	"ragflow/internal/entity"
 	"ragflow/internal/service"
 	"ragflow/internal/service/file"
+	"ragflow/internal/utility"
 
 	dslpkg "ragflow/internal/agent/dsl"
 )
@@ -200,8 +201,9 @@ func (h *AgentHandler) ListAgents(c *gin.Context) {
 // mapAgentError normalises service-layer errors onto the existing
 // {code, data, message} response envelope used by every other handler.
 //
-// Three classes:
+// Four classes:
 //   - service.ErrAgentNotOwner  -> "Only the owner..."        (DELETE only, 103)
+//   - service.ErrAgentSessionBusy -> "session already running" (103)
 //   - dao.ErrUserCanvasNotFound -> "Make sure you have permission..."  (103)
 //   - service.ErrAgentStorageError -> "Internal storage error"  (500)
 //
@@ -219,6 +221,9 @@ func mapAgentError(err error) (common.ErrorCode, string) {
 	if errors.Is(err, service.ErrAgentNotOwner) {
 		return common.CodeOperatingError, "Only the owner of the agent is authorized for this operation."
 	}
+	if errors.Is(err, service.ErrAgentSessionBusy) {
+		return common.CodeOperatingError, "This agent session is already running."
+	}
 	if errors.Is(err, dao.ErrUserCanvasNotFound) ||
 		errors.Is(err, dao.ErrUserCanvasVersionNotFound) {
 		return common.CodeOperatingError, "Make sure you have permission to access the agent."
@@ -390,6 +395,11 @@ func (h *AgentHandler) RunAgent(c *gin.Context) {
 	canvasID := c.Param("canvas_id")
 	version := c.Query("version")
 	sessionID := c.Query("session_id")
+	if sessionID == "" {
+		// Allocate the ordinary-Agent session identity at the HTTP boundary.
+		// Persistence of the session record remains owned by AgentService.RunAgent.
+		sessionID = utility.GenerateToken()
+	}
 	userInput := readUserInput(c)
 
 	events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, canvasID, sessionID, version, userInput, nil)
@@ -469,21 +479,24 @@ func sanitiseRunEventError(data string) string {
 	return data
 }
 
-// CancelAgent signals the in-flight run to stop.
-// @Summary Cancel Agent Run
+// CancelSessionRun cancels one ordinary Agent run by session id.
+// @Summary Cancel Agent Session Run
 // @Tags agents
 // @Produce json
-// @Param canvas_id path string true "canvas id"
+// @Param session_id path string true "session id"
 // @Success 200 {object} map[string]interface{}
-// @Router /api/v1/agents/{canvas_id}/run [delete]
-func (h *AgentHandler) CancelAgent(c *gin.Context) {
+// @Router /api/v1/tasks/{session_id}/cancel [post]
+func (h *AgentHandler) CancelSessionRun(c *gin.Context) {
 	user, code, msg := GetUser(c)
 	if code != common.CodeSuccess {
 		common.ResponseWithCodeData(c, code, nil, msg)
 		return
 	}
-	canvasID := c.Param("canvas_id")
-	if err := h.agentService.CancelAgent(c.Request.Context(), user.ID, canvasID); err != nil {
+	if h.agentService == nil {
+		common.ResponseWithCodeData(c, common.CodeServerError, nil, "agent service unavailable")
+		return
+	}
+	if err := h.agentService.CancelSessionRun(c.Request.Context(), user.ID, c.Param("session_id")); err != nil {
 		ec, em := mapAgentError(err)
 		common.ResponseWithCodeData(c, ec, nil, em)
 		return
@@ -1007,6 +1020,12 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
 			zap.String("session_id", req.SessionID),
 		}, userInputMeta(userInput)...)...,
 	)
+	if req.SessionID == "" {
+		// Keep the effective session available to the non-stream response and
+		// to the task_id=session_id wire alias even when the canvas emits no
+		// events (for example an empty query).
+		req.SessionID = utility.GenerateToken()
+	}
 
 	events, err := h.chatRunner.RunAgent(c.Request.Context(), user.ID, req.AgentID, req.SessionID, "", userInput, req.Files)
 	if err != nil {
@@ -1044,7 +1063,6 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
 				zap.String("session_id", req.SessionID),
 				zap.String("event_type", ev.Type),
 				zap.String("message_id", ev.MessageID),
-				zap.String("task_id", ev.TaskID),
 			)
 			if err := service.WriteChatbotRunEvent(c.Writer, ev); err != nil {
 				common.Debug("agent chat completions: client disconnected",
@@ -1183,7 +1201,7 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
 		"data":       ansData,
 		"message_id": finalAns.MessageID,
 		"created_at": finalAns.CreatedAt,
-		"task_id":    finalAns.TaskID,
+		"task_id":    finalAns.SessionID,
 		"session_id": finalAns.SessionID,
 	}
 	common.SuccessWithData(c, result, "success")

internal/handler/agent_test.go

@@ -452,9 +452,6 @@ func (f *fullFakeAgentService) RunAgent(context.Context, string, string, string,
 	close(ch)
 	return ch, nil
 }
-func (f *fullFakeAgentService) CancelAgent(context.Context, string, string) error {
-	return nil
-}
 func (f *fullFakeAgentService) PublishAgent(context.Context, string, string, *service.PublishAgentRequest) (*entity.UserCanvasVersion, error) {
 	return f.version, nil
 }
@@ -500,7 +497,6 @@ func TestAgentHandler_RoutesRegistered(t *testing.T) {
 	g.PUT("/:canvas_id", func(c *gin.Context) { c.Status(http.StatusOK) })
 	g.DELETE("/:canvas_id", func(c *gin.Context) { c.Status(http.StatusOK) })
 	g.POST("/:canvas_id/run", func(c *gin.Context) { c.Status(http.StatusOK) })
-	g.DELETE("/:canvas_id/run", func(c *gin.Context) { c.Status(http.StatusOK) })
 	g.POST("/:canvas_id/publish", func(c *gin.Context) { c.Status(http.StatusOK) })
 	g.GET("/:canvas_id/versions", func(c *gin.Context) { c.Status(http.StatusOK) })
 	g.GET("/:canvas_id/versions/:version_id", func(c *gin.Context) { c.Status(http.StatusOK) })
@@ -516,14 +512,13 @@ func TestAgentHandler_RoutesRegistered(t *testing.T) {
 		{http.MethodPut, "/api/v1/agents/abc"},
 		{http.MethodDelete, "/api/v1/agents/abc"},
 		{http.MethodPost, "/api/v1/agents/abc/run"},
-		{http.MethodDelete, "/api/v1/agents/abc/run"},
 		{http.MethodPost, "/api/v1/agents/abc/publish"},
 		{http.MethodGet, "/api/v1/agents/abc/versions"},
 		{http.MethodGet, "/api/v1/agents/abc/versions/v1"},
 		{http.MethodDelete, "/api/v1/agents/abc/versions/v1"},
 	}
-	if len(routes) != 11 {
-		t.Fatalf("expected 11 routes, listed %d", len(routes))
+	if len(routes) != 10 {
+		t.Fatalf("expected 10 routes, listed %d", len(routes))
 	}
 	for _, rt := range routes {
 		w := httptest.NewRecorder()
@@ -731,7 +726,8 @@ func (s *stubChatRunner) RunAgent(_ context.Context, _, _, _, _ string, _ any, _
 // path: the handler streams canvas.RunEvent frames as
 // `data: {...}\n\n` with a trailing `data:[DONE]\n\n` terminator.
 // The frame shape is the Python agent-canvas envelope
-// {event,message_id,task_id,session_id,data:{content}}. See
+// {event,message_id,task_id,session_id,data:{content}}. task_id is a wire alias
+// for session_id. See
 // service.WriteChatbotRunEvent.
 //
 // The stubChatRunner emits one `message` frame and one `done` frame
@@ -748,7 +744,7 @@ func TestAgentChatCompletions_StreamSetsContentType(t *testing.T) {
 	c.Set("user_id", "u1")
 
 	runner := &stubChatRunner{events: []canvas.RunEvent{
-		{Type: "message", MessageID: "msg-1", TaskID: "task-1", SessionID: "sess-1", Data: `{"content":"hi back","reference":[]}`},
+		{Type: "message", MessageID: "msg-1", SessionID: "sess-1", Data: `{"content":"hi back","reference":[]}`},
 		{Type: "done", Data: ""},
 	}}
 	h := &AgentHandler{chatRunner: runner}
@@ -760,7 +756,7 @@ func TestAgentChatCompletions_StreamSetsContentType(t *testing.T) {
 	body := w.Body.String()
 	if !strings.Contains(body, `"event":"message"`) ||
 		!strings.Contains(body, `"message_id":"msg-1"`) ||
-		!strings.Contains(body, `"task_id":"task-1"`) ||
+		!strings.Contains(body, `"task_id":"sess-1"`) ||
 		!strings.Contains(body, `"session_id":"sess-1"`) ||
 		!strings.Contains(body, `"content":"hi back"`) {
 		t.Errorf("body should contain flat agent event with content, got %q", body)
@@ -781,7 +777,7 @@ func TestAgentChatCompletions_StreamAddsDoneWhenRunnerCloses(t *testing.T) {
 	c.Set("user_id", "u1")
 
 	runner := &stubChatRunner{events: []canvas.RunEvent{
-		{Type: "message", MessageID: "msg-1", TaskID: "task-1", SessionID: "sess-1", Data: `{"content":"hi back"}`},
+		{Type: "message", MessageID: "msg-1", SessionID: "sess-1", Data: `{"content":"hi back"}`},
 	}}
 	h := &AgentHandler{chatRunner: runner}
 	h.AgentChatCompletions(c)
@@ -807,7 +803,7 @@ func TestRunAgent_StreamAddsDoneWhenRunnerCloses(t *testing.T) {
 	c.Set("user_id", "u1")
 
 	runner := &stubChatRunner{events: []canvas.RunEvent{
-		{Type: "message", MessageID: "msg-1", TaskID: "task-1", SessionID: "sess-1", Data: `{"content":"hi back"}`},
+		{Type: "message", MessageID: "msg-1", SessionID: "sess-1", Data: `{"content":"hi back"}`},
 	}}
 	h := &AgentHandler{chatRunner: runner}
 	h.RunAgent(c)
@@ -837,7 +833,7 @@ func TestAgentChatCompletions_DefaultBranchNonStreaming(t *testing.T) {
 	c.Set("user_id", "u1")
 
 	runner := &stubChatRunner{events: []canvas.RunEvent{
-		{Type: "message", MessageID: "msg-2", TaskID: "task-2", SessionID: "sess-2", Data: `{"content":"hello back","reference":[]}`},
+		{Type: "message", MessageID: "msg-2", SessionID: "sess-2", Data: `{"content":"hello back","reference":[]}`},
 		{Type: "done", Data: ""},
 	}}
 	h := &AgentHandler{chatRunner: runner}
@@ -860,6 +856,44 @@ func TestAgentChatCompletions_DefaultBranchNonStreaming(t *testing.T) {
 	}
 }
 
+type emptySessionCaptureRunner struct {
+	sessionID string
+}
+
+func (r *emptySessionCaptureRunner) RunAgent(_ context.Context, _, _, sessionID, _ string, _ any, _ []map[string]interface{}) (<-chan canvas.RunEvent, error) {
+	r.sessionID = sessionID
+	ch := make(chan canvas.RunEvent)
+	close(ch)
+	return ch, nil
+}
+
+func TestAgentChatCompletions_EmptyOutputReturnsGeneratedSession(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":""}`))
+	c.Request.Header.Set("Content-Type", "application/json")
+	c.Set("user", &entity.User{ID: "u1"})
+	c.Set("user_id", "u1")
+
+	runner := &emptySessionCaptureRunner{}
+	h := &AgentHandler{chatRunner: runner}
+	h.AgentChatCompletions(c)
+
+	if runner.sessionID == "" {
+		t.Fatal("handler passed an empty session id to RunAgent")
+	}
+	var response map[string]any
+	if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	data, _ := response["data"].(map[string]any)
+	if got, _ := data["session_id"].(string); got != runner.sessionID {
+		t.Fatalf("empty-output session_id = %q, want %q", got, runner.sessionID)
+	}
+}
+
 // TestAgentChatCompletions_DerivesUserInputFromMessages covers the
 // fallback path: the request omits `query` but supplies `messages`
 // with a trailing user message. The handler must use that message's

internal/handler/agent_wait_for_user_test.go

@@ -114,7 +114,6 @@ func (f *waitFakeAgentService) RunAgent(ctx context.Context, userID, canvasID, s
 	}), nil
 }
 
-func (f *waitFakeAgentService) CancelAgent(context.Context, string, string) error { return nil }
 func (f *waitFakeAgentService) PublishAgent(context.Context, string, string, *service.PublishAgentRequest) (*entity.UserCanvasVersion, error) {
 	return &entity.UserCanvasVersion{}, nil
 }

internal/handler/agent_webhook.go

@@ -68,6 +68,8 @@ import (
 	"ragflow/internal/agent/canvas"
 	"ragflow/internal/common"
 	rediscli "ragflow/internal/engine/redis"
+	"ragflow/internal/service"
+	"ragflow/internal/utility"
 	"strconv"
 	"strings"
 	"time"
@@ -224,7 +226,7 @@ func (h *AgentHandler) Webhook(c *gin.Context) {
 		}
 		// Detached background run — does NOT inherit c.Request.Context()
 		// so a client disconnect does not cancel the canvas run.
-		go h.runWebhookDetached(cv, clean, isTest, startTs)
+		go h.runWebhookDetached(c.Request.Context(), cv, clean, isTest, startTs)
 		c.Data(status, contentType, payload)
 		return
 	}
@@ -494,17 +496,19 @@ func renderImmediatelyResponse(cfg map[string]any) (int, string, []byte, error)
 	return status, "text/plain", []byte(bodyTpl), nil
 }
 
-// runWebhookDetached runs the canvas in the background. It uses
-// context.Background() with a 5-minute timeout (NOT
-// c.Request.Context()) so a client disconnect does NOT cancel the run.
+// runWebhookDetached runs the canvas with a five-minute timeout. It preserves
+// request-scoped context values while intentionally detaching cancellation so
+// a client disconnect does not cancel an Immediate webhook run.
 // Trace events are appended to the redis key when isTest is true.
 //
 // Mirrors python: agent_api.py:2123-2175 (the asyncio.create_task body
 // inside the Immediately branch).
 func (h *AgentHandler) runWebhookDetached(
-	cv *entity.UserCanvas, payload map[string]any, isTest bool, startTs time.Time,
+	parent context.Context, cv *entity.UserCanvas, payload map[string]any, isTest bool, startTs time.Time,
 ) {
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	sessionID := utility.GenerateToken()
+	parent = service.WithAgentSessionID(parent, sessionID)
+	ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), 5*time.Minute)
 	defer cancel()
 
 	events, err := h.loader.RunAgentWithWebhook(ctx, cv.UserID, cv.ID, payload)
@@ -513,11 +517,14 @@ func (h *AgentHandler) runWebhookDetached(
 			zap.String("canvas", cv.ID),
 			zap.Error(err))
 		if isTest {
-			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "error", Data: mustJSON(map[string]any{"message": err.Error()})})
+			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "error", SessionID: sessionID, Data: mustJSON(map[string]any{"message": err.Error()})})
 		}
 		return
 	}
 	for ev := range events {
+		if ev.SessionID == "" {
+			ev.SessionID = sessionID
+		}
 		if isTest {
 			appendWebhookTrace(cv.ID, startTs, ev)
 		}
@@ -538,21 +545,28 @@ func (h *AgentHandler) runWebhookSync(
 	isTest bool, startTs time.Time,
 ) webhookSyncResult {
 	status := 200
+	sessionID := utility.GenerateToken()
+	ctx = service.WithAgentSessionID(ctx, sessionID)
 	events, err := h.loader.RunAgentWithWebhook(ctx, cv.UserID, cv.ID, payload)
 	if err != nil {
 		if isTest {
-			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "error", Data: mustJSON(map[string]any{"message": err.Error()})})
-			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "finished", Data: mustJSON(map[string]any{"success": false})})
+			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "error", SessionID: sessionID, Data: mustJSON(map[string]any{"message": err.Error()})})
+			appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "finished", SessionID: sessionID, Data: mustJSON(map[string]any{"success": false})})
 		}
 		return webhookSyncResult{status: http.StatusBadRequest, body: gin.H{
-			"code":    400,
-			"message": err.Error(),
-			"success": false,
+			"code":       400,
+			"message":    err.Error(),
+			"success":    false,
+			"task_id":    sessionID,
+			"session_id": sessionID,
 		}}
 	}
 
 	contents := []string{}
 	for ev := range events {
+		if ev.SessionID == "" {
+			ev.SessionID = sessionID
+		}
 		if isTest {
 			appendWebhookTrace(cv.ID, startTs, ev)
 		}
@@ -585,12 +599,14 @@ func (h *AgentHandler) runWebhookSync(
 	}
 	final := strings.Join(contents, "")
 	if isTest {
-		appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "finished", Data: mustJSON(map[string]any{"success": true})})
+		appendWebhookTrace(cv.ID, startTs, canvas.RunEvent{Type: "finished", SessionID: sessionID, Data: mustJSON(map[string]any{"success": true})})
 	}
 	return webhookSyncResult{status: status, body: gin.H{
-		"message": final,
-		"success": true,
-		"code":    status,
+		"message":    final,
+		"success":    true,
+		"code":       status,
+		"task_id":    sessionID,
+		"session_id": sessionID,
 	}}
 }
 
@@ -647,10 +663,8 @@ func appendWebhookTrace(agentID string, startTs time.Time, ev canvas.RunEvent) {
 	if ev.MessageID != "" {
 		eventRecord["message_id"] = ev.MessageID
 	}
-	if ev.TaskID != "" {
-		eventRecord["task_id"] = ev.TaskID
-	}
 	if ev.SessionID != "" {
+		eventRecord["task_id"] = ev.SessionID
 		eventRecord["session_id"] = ev.SessionID
 	}
 	entry["events"] = append(events, eventRecord)