feat: Add stop session functionality to chat API and UI

- Implemented a new API endpoint to stop ongoing chat sessions, allowing users to cancel message generation.
- Enhanced InputField component to include a stop button, enabling users to halt the assistant's response generation.
- Updated chat view to manage the assistant message ID and loading state effectively during stop events.
- Improved event handling in the backend to support stop events, ensuring proper session management and user feedback.

frontend/src/api/chat/index.ts

@@ -46,4 +46,8 @@ export async function delSession(session_id: string) {
 
 export async function getSession(session_id: string) {
   return get(`/api/v1/sessions/${session_id}`);
+}
+
+export async function stopSession(session_id: string, message_id: string) {
+  return post(`/api/v1/sessions/${session_id}/stop`, { message_id });
 }

frontend/src/components/Input-field.vue

@@ -6,6 +6,7 @@ import { MessagePlugin } from "tdesign-vue-next";
 import { useSettingsStore } from '@/stores/settings';
 import { useUIStore } from '@/stores/ui';
 import { listKnowledgeBases } from '@/api/knowledge-base';
+import { stopSession } from '@/api/chat';
 import KnowledgeBaseSelector from './KnowledgeBaseSelector.vue';
 
 const route = useRoute();
@@ -20,6 +21,14 @@ const props = defineProps({
   isReplying: {
     type: Boolean,
     required: false
+  },
+  sessionId: {
+    type: String,
+    required: false
+  },
+  assistantMessageId: {
+    type: String,
+    required: false
   }
 });
 
@@ -68,7 +77,7 @@ watch(() => route.params.kbId, (newKbId) => {
   }
 });
 
-const emit = defineEmits(['send-msg']);
+const emit = defineEmits(['send-msg', 'stop-generation']);
 
 const createSession = (val: string) => {
   if (!val.trim()) {
@@ -157,6 +166,32 @@ const removeKb = (kbId: string) => {
   settingsStore.removeKnowledgeBase(kbId);
 }
 
+const handleStop = async () => {
+  if (!props.sessionId) {
+    MessagePlugin.warning('会话 ID 不存在');
+    return;
+  }
+  
+  if (!props.assistantMessageId) {
+    console.error('[Stop] Assistant message ID is empty');
+    MessagePlugin.warning('无法获取消息 ID，请刷新页面后重试');
+    return;
+  }
+  
+  console.log('[Stop] Stopping generation for message:', props.assistantMessageId);
+  
+  // 发送 stop 事件，通知父组件立即清除 loading 状态
+  emit('stop-generation');
+  
+  try {
+    await stopSession(props.sessionId, props.assistantMessageId);
+    MessagePlugin.success('已停止生成');
+  } catch (error) {
+    console.error('Failed to stop session:', error);
+    MessagePlugin.error('停止失败，请重试');
+  }
+}
+
 onBeforeRouteUpdate((to, from, next) => {
   clearvalue()
   next()
@@ -248,13 +283,33 @@ onBeforeRouteUpdate((to, from, next) => {
         </div>
       </div>
 
-      <!-- 右侧发送按钮 -->
+      <!-- 右侧控制按钮组 -->
+      <div class="control-right">
+        <!-- 停止按钮（仅在回复中时显示） -->
+        <t-tooltip 
+          v-if="isReplying"
+          content="停止生成"
+          placement="top"
+        >
+          <div 
+            @click="handleStop" 
+            class="control-btn stop-btn"
+          >
+            <svg width="16" height="16" viewBox="0 0 16 16" fill="currentColor">
+              <rect x="5" y="5" width="6" height="6" rx="1" />
+            </svg>
+          </div>
+        </t-tooltip>
+
+        <!-- 发送按钮 -->
       <div 
+          v-if="!isReplying"
         @click="createSession(query)" 
         class="control-btn send-btn"
         :class="{ 'disabled': !query.length || selectedKbIds.length === 0 }"
       >
         <img src="../assets/img/sending-aircraft.svg" alt="发送" />
+        </div>
       </div>
     </div>
 
@@ -492,6 +547,104 @@ const getImgSrc = (url: string) => {
   }
 }
 
+.control-right {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+}
+
+.stop-btn {
+  width: 28px;
+  height: 28px;
+  padding: 0;
+  background: rgba(7, 192, 95, 0.08);
+  color: #07C05F;
+  border: 1.5px solid rgba(7, 192, 95, 0.2);
+  position: relative;
+  box-shadow: 0 1px 3px rgba(0, 0, 0, 0.05);
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  animation: stopBtnPulse 2s ease-in-out infinite;
+  
+  &:hover {
+    background: rgba(7, 192, 95, 0.12);
+    border-color: #07C05F;
+    transform: scale(1.05);
+    box-shadow: 0 2px 4px rgba(7, 192, 95, 0.2);
+    animation: none;
+  }
+  
+  &:active {
+    transform: scale(0.95);
+    background: rgba(7, 192, 95, 0.15);
+  }
+  
+  svg {
+    display: none;
+  }
+  
+  &::before {
+    content: '';
+    width: 12px;
+    height: 12px;
+    background: #07C05F;
+    border-radius: 50%;
+    display: block;
+    animation: stopDotPulse 2s ease-in-out infinite;
+  }
+  
+  &:hover::before {
+    animation: none;
+  }
+  
+  &::after {
+    content: '';
+    position: absolute;
+    width: 12px;
+    height: 12px;
+    border-radius: 50%;
+    border: 2px solid #07C05F;
+    animation: stopRipple 2s ease-out infinite;
+  }
+  
+  &:hover::after {
+    animation: none;
+    opacity: 0;
+  }
+}
+
+@keyframes stopBtnPulse {
+  0%, 100% {
+    box-shadow: 0 1px 3px rgba(0, 0, 0, 0.05);
+  }
+  50% {
+    box-shadow: 0 2px 4px rgba(7, 192, 95, 0.1);
+  }
+}
+
+@keyframes stopDotPulse {
+  0%, 100% {
+    transform: scale(1);
+    opacity: 1;
+  }
+  50% {
+    transform: scale(0.92);
+    opacity: 0.9;
+  }
+}
+
+@keyframes stopRipple {
+  0% {
+    transform: scale(0.9);
+    opacity: 0.4;
+  }
+  100% {
+    transform: scale(2);
+    opacity: 0;
+  }
+}
+
 .send-btn {
   width: 28px;
   height: 28px;

frontend/src/views/chat/components/AgentStreamDisplay.vue

@@ -175,7 +175,7 @@ const isLastThinking = (eventId: string): boolean => {
   return eventId === lastThinkingEventId.value;
 };
 
-// Check if conversation is done (based on answer event with done=true)
+// Check if conversation is done (based on answer event with done=true or stop event)
 const isConversationDone = computed(() => {
   const stream = eventStream.value;
   if (!stream || stream.length === 0) {
@@ -183,7 +183,14 @@ const isConversationDone = computed(() => {
     return false;
   }
   
-  // Only check for answer event with done=true
+  // Check for stop event (user cancelled)
+  const stopEvent = stream.find((e: any) => e.type === 'stop');
+  if (stopEvent) {
+    console.log('[Collapse] Found stop event, conversation done');
+    return true;
+  }
+  
+  // Check for answer event with done=true
   const answerEvents = stream.filter((e: any) => e.type === 'answer');
   const doneAnswer = answerEvents.find((e: any) => e.done === true);
   

frontend/src/views/chat/index.vue

@@ -18,7 +18,13 @@
             </div>
         </div>
         <div style="min-height: 115px; margin: 16px auto 4px;width: 100%;max-width: 800px;">
-            <InputField @send-msg="sendMsg" :isReplying="isReplying"></InputField>
+            <InputField 
+                @send-msg="sendMsg" 
+                @stop-generation="handleStopGeneration"
+                :isReplying="isReplying" 
+                :sessionId="session_id"
+                :assistantMessageId="currentAssistantMessageId"
+            ></InputField>
         </div>
     </div>
 </template>
@@ -46,6 +52,7 @@ const created_at = ref('');
 const limit = ref(20);
 const messagesList = reactive([]);
 const isReplying = ref(false);
+const currentAssistantMessageId = ref(''); // 当前正在生成的 assistant message ID
 const scrollLock = ref(false);
 const isNeedTitle = ref(false);
 const isFirstEnter = ref(true);
@@ -208,7 +215,13 @@ const handleMsgList = async (data, isScrollType = false, newScrollHeight) => {
     }
     if (messagesList[messagesList.length - 1] && !messagesList[messagesList.length - 1].is_completed) {
         isReplying.value = true;
-        await startStream({ session_id: session_id.value, query: messagesList[messagesList.length - 1].id, method: 'GET', url: '/api/v1/sessions/continue-stream' });
+        // 保存正在 stream 的消息 ID，以便停止时使用
+        const lastMessage = messagesList[messagesList.length - 1];
+        if (lastMessage.role === 'assistant') {
+            currentAssistantMessageId.value = lastMessage.id;
+            console.log('[Continue Stream] Set assistant message ID:', lastMessage.id);
+        }
+        await startStream({ session_id: session_id.value, query: lastMessage.id, method: 'GET', url: '/api/v1/sessions/continue-stream' });
     }
 
 }
@@ -220,6 +233,15 @@ const checkmenuTitle = (session_id) => {
     });
 }
 // 发送消息
+// 处理停止生成事件 - 立即清除 loading 状态
+const handleStopGeneration = () => {
+    console.log('[Stop Generation] Immediately clearing loading state');
+    loading.value = false;
+    isReplying.value = false;
+    // 注意：不在这里清空 currentAssistantMessageId，因为需要它来调用 API
+    // API 调用成功后，后端的 stop 事件会清空它
+};
+
 const sendMsg = async (value) => {
     userquery.value = value;
     isReplying.value = true;
@@ -238,6 +260,8 @@ const sendMsg = async (value) => {
         MessagePlugin.warning('请至少选择一个知识库');
         isReplying.value = false;
         loading.value = false;
+        // 清空当前 assistant message ID
+        currentAssistantMessageId.value = '';
         // Remove the user message that was just added
         messagesList.pop();
         return;
@@ -262,6 +286,8 @@ watch(error, (newError) => {
         MessagePlugin.error(newError);
         isReplying.value = false;
         loading.value = false;
+        // 清空当前 assistant message ID
+        currentAssistantMessageId.value = '';
     }
 });
 
@@ -274,13 +300,20 @@ onChunk((data) => {
         done: data.done,
         content_length: data.content?.length || 0,
         content_preview: data.content ? data.content.substring(0, 50) : '',
-        data: data.data
+        data: data.data,
+        session_id: data.session_id,
+        assistant_message_id: data.assistant_message_id
     });
     
-    // 处理 agent query 事件 - 保持 loading 状态
+    // 处理 agent query 事件 - 保存 assistant message ID 并保持 loading 状态
     if (data.response_type === 'agent_query') {
+        if (data.assistant_message_id) {
+            currentAssistantMessageId.value = data.assistant_message_id;
+            console.log('[Agent Query] Saved assistant message ID:', data.assistant_message_id);
+        }
         console.log('[Agent Query Event]', {
-            session_id: data.data?.session_id,
+            session_id: data.session_id || data.data?.session_id,
+            assistant_message_id: data.assistant_message_id,
             query: data.data?.query,
             request_id: data.data?.request_id
         });
@@ -308,12 +341,22 @@ onChunk((data) => {
     // 判断是否是 Agent 模式的响应
     const isAgentResponse = data.response_type === 'thinking' || 
                            data.response_type === 'tool_call' || 
-                           data.response_type === 'references';
+                           data.response_type === 'references' ||
+                           data.response_type === 'stop';
     
-    // Agent 模式处理
+    // Agent 模式处理（包括 stop 事件）
     if (isAgentResponse || messagesList[messagesList.length - 1]?.isAgentMode) {
         // 在 handleAgentChunk 中处理 loading 状态
         handleAgentChunk(data);
+        
+        // 对于 stop 事件，额外处理全局状态
+        if (data.response_type === 'stop') {
+            console.log('[Stop Event] Generation stopped');
+            loading.value = false;
+            isReplying.value = false;
+            // 清空当前 assistant message ID
+            currentAssistantMessageId.value = '';
+        }
         return;
     }
     
@@ -352,6 +395,8 @@ onChunk((data) => {
         // 如果标题还未生成，前端会通过 SSE 事件接收
         isReplying.value = false;
         fullContent.value = "";
+        // 清空当前 assistant message ID
+        currentAssistantMessageId.value = '';
     }
     updateAssistantSession(obj);
 })
@@ -566,11 +611,30 @@ const handleAgentChunk = (data) => {
                 // 完成
                 isReplying.value = false;
                 fullContent.value = '';
+                // 清空当前 assistant message ID
+                currentAssistantMessageId.value = '';
                 
                 // 标题生成已改为异步事件推送，不再需要在这里手动调用
                 // 如果标题还未生成，前端会通过 SSE 事件接收
             }
             break;
+            
+        case 'stop':
+            // 停止事件 - 添加到事件流并标记对话完成
+            console.log('[Agent] Stop event received');
+            if (!message.agentEventStream) message.agentEventStream = [];
+            
+            // Add stop event to stream
+            message.agentEventStream.push({
+                type: 'stop',
+                timestamp: Date.now(),
+                reason: data.data?.reason || 'user_requested'
+            });
+            
+            // Mark conversation as stopped
+            isReplying.value = false;
+            fullContent.value = '';
+            break;
     }
     
     scrollToBottom();

internal/event/event.go

@@ -59,6 +59,9 @@ const (
 
 	// Session events
 	EventSessionTitle EventType = "session_title" // 会话标题更新
+
+	// Control events
+	EventStop EventType = "stop" // 停止对话生成
 )
 
 // Event represents an event in the system

internal/event/event_data.go

@@ -201,3 +201,10 @@ type SessionTitleData struct {
 	SessionID string `json:"session_id"`
 	Title     string `json:"title"`
 }
+
+// StopData represents stop generation request data
+type StopData struct {
+	SessionID string `json:"session_id"`
+	MessageID string `json:"message_id"`
+	Reason    string `json:"reason,omitempty"` // Optional reason for stopping
+}

internal/handler/session.go

@@ -820,9 +820,39 @@ func (h *SessionHandler) handleKnowledgeQARequest(
 	c.Header("Connection", "keep-alive")
 	c.Header("X-Accel-Buffering", "no")
 
+	// Write initial agent_query event to StreamManager
+	agentQueryEvent := interfaces.StreamEvent{
+		ID:        fmt.Sprintf("query-%d", time.Now().UnixNano()),
+		Type:      types.ResponseTypeAgentQuery,
+		Content:   "",
+		Done:      true,
+		Timestamp: time.Now(),
+		Data: map[string]interface{}{
+			"session_id":           sessionID,
+			"assistant_message_id": assistantMessage.ID,
+		},
+	}
+	if err := h.streamManager.AppendEvent(ctx, sessionID, assistantMessage.ID, agentQueryEvent); err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+			"message_id": assistantMessage.ID,
+		})
+		// Non-fatal error, continue
+	}
+
 	// Create dedicated EventBus for this request
 	eventBus := event.NewEventBus()
-	asyncCtx := logger.CloneContext(ctx)
+	// Create cancellable context for async operations
+	asyncCtx, cancel := context.WithCancel(logger.CloneContext(ctx))
+
+	// Register stop event handler to cancel the context
+	eventBus.On(event.EventStop, func(ctx context.Context, evt event.Event) error {
+		logger.Infof(ctx, "Received stop event, cancelling async operations for session: %s", sessionID)
+		cancel()
+		assistantMessage.Content = "用户停止了本次对话"
+		h.completeAssistantMessage(ctx, assistantMessage)
+		return nil
+	})
 
 	// Create stream handler with dedicated EventBus
 	streamHandler := NewAgentStreamHandler(
@@ -848,6 +878,7 @@ func (h *SessionHandler) handleKnowledgeQARequest(
 		if data.Done {
 			logger.Infof(asyncCtx, "Knowledge QA service completed for session: %s", sessionID)
 			h.completeAssistantMessage(asyncCtx, assistantMessage)
+			cancel() // Clean up context
 			return nil
 		}
 		return nil
@@ -878,7 +909,7 @@ func (h *SessionHandler) handleKnowledgeQARequest(
 	}()
 
 	// Handle events for SSE (blocking until connection is done)
-	h.handleAgentEventsForSSEWithHandler(ctx, c, sessionID, assistantMessage.ID, requestID)
+	h.handleAgentEventsForSSEWithHandler(ctx, c, sessionID, assistantMessage.ID, requestID, eventBus)
 }
 
 // AgentQA handles agent-based question answering with conversation history and streaming
@@ -1067,11 +1098,32 @@ func (h *SessionHandler) AgentQA(c *gin.Context) {
 	logger.Infof(ctx, "Calling agent QA service, session ID: %s", sessionID)
 
 	// SSE headers already set earlier when sending agent_query event
+	// Write initial agent_query event to StreamManager
+	agentQueryEvent := interfaces.StreamEvent{
+		ID:        fmt.Sprintf("query-%d", time.Now().UnixNano()),
+		Type:      types.ResponseTypeAgentQuery,
+		Content:   "",
+		Done:      true,
+		Timestamp: time.Now(),
+		Data: map[string]interface{}{
+			"session_id":           sessionID,
+			"assistant_message_id": assistantMessage.ID,
+		},
+	}
+	if err := h.streamManager.AppendEvent(ctx, sessionID, assistantMessage.ID, agentQueryEvent); err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+			"message_id": assistantMessage.ID,
+		})
+		// Non-fatal error, continue
+	}
 
 	eventBus := event.NewEventBus()
+	// Create cancellable context for async operations
+	asyncCtx, cancel := context.WithCancel(logger.CloneContext(ctx))
 	// Create stream handler with dedicated EventBus BEFORE calling AgentQA
 	streamHandler := NewAgentStreamHandler(
-		ctx, sessionID, assistantMessage.ID, requestID,
+		asyncCtx, sessionID, assistantMessage.ID, requestID,
 		assistantMessage, h.streamManager, eventBus,
 	)
 
@@ -1081,10 +1133,18 @@ func (h *SessionHandler) AgentQA(c *gin.Context) {
 	// Start async title generation if session has no title
 	if session.Title == "" {
 		logger.Infof(ctx, "Session has no title, starting async title generation, session ID: %s", sessionID)
-		h.sessionService.GenerateTitleAsync(ctx, session, request.Query, eventBus)
+		h.sessionService.GenerateTitleAsync(asyncCtx, session, request.Query, eventBus)
 	}
 
-	asyncCtx := logger.CloneContext(ctx)
+	// Register stop event handler to cancel the context
+	eventBus.On(event.EventStop, func(ctx context.Context, evt event.Event) error {
+		logger.Warnf(asyncCtx, "Received stop event, cancelling async operations for session: %s", sessionID)
+		cancel()
+		assistantMessage.Content = "用户停止了本次对话"
+		h.completeAssistantMessage(ctx, assistantMessage)
+		return nil
+	})
+
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
@@ -1111,7 +1171,7 @@ func (h *SessionHandler) AgentQA(c *gin.Context) {
 	}()
 
 	// Handle events for SSE (blocking until connection is done)
-	h.handleAgentEventsForSSEWithHandler(ctx, c, sessionID, assistantMessage.ID, requestID)
+	h.handleAgentEventsForSSEWithHandler(ctx, c, sessionID, assistantMessage.ID, requestID, eventBus)
 }
 
 // handleAgentEventsForSSEWithHandler handles agent events for SSE streaming using an existing handler
@@ -1121,6 +1181,7 @@ func (h *SessionHandler) handleAgentEventsForSSEWithHandler(
 	ctx context.Context,
 	c *gin.Context,
 	sessionID, assistantMessageID, requestID string,
+	eventBus *event.EventBus,
 ) {
 	ticker := time.NewTicker(100 * time.Millisecond)
 	defer ticker.Stop()
@@ -1148,13 +1209,53 @@ func (h *SessionHandler) handleAgentEventsForSSEWithHandler(
 			// Send any new events
 			streamCompleted := false
 			for _, evt := range events {
+				// Check for stop event
+				if evt.Type == types.ResponseType(event.EventStop) {
+					log.Infof("Detected stop event, triggering stop via EventBus for session=%s", sessionID)
+
+					// Emit stop event to the EventBus to trigger context cancellation
+					if eventBus != nil {
+						eventBus.Emit(ctx, event.Event{
+							Type:      event.EventStop,
+							SessionID: sessionID,
+							Data: event.StopData{
+								SessionID: sessionID,
+								MessageID: assistantMessageID,
+								Reason:    "user_requested",
+							},
+						})
+					}
+
+					// Send stop notification to frontend
+					c.SSEvent("message", &types.StreamResponse{
+						ID:           requestID,
+						ResponseType: "stop",
+						Content:      "Generation stopped by user",
+						Done:         true,
+					})
+					c.Writer.Flush()
+					return
+				}
+
 				// Build StreamResponse from StreamEvent
+				// For agent_query event, extract session_id and assistant_message_id from Data
+				var sessionIDFromEvent, assistantMessageIDFromEvent string
+				if evt.Type == types.ResponseTypeAgentQuery {
+					if sid, ok := evt.Data["session_id"].(string); ok {
+						sessionIDFromEvent = sid
+					}
+					if amid, ok := evt.Data["assistant_message_id"].(string); ok {
+						assistantMessageIDFromEvent = amid
+					}
+				}
 				response := &types.StreamResponse{
-					ID:           requestID,
-					ResponseType: evt.Type,
-					Content:      evt.Content,
-					Done:         evt.Done,
-					Data:         evt.Data,
+					ID:                 requestID,
+					ResponseType:       evt.Type,
+					Content:            evt.Content,
+					Done:               evt.Done,
+					Data:               evt.Data,
+					SessionID:          sessionIDFromEvent,
+					AssistantMessageID: assistantMessageIDFromEvent,
 				}
 
 				// Special handling for references event
@@ -1199,3 +1300,114 @@ func (h *SessionHandler) handleAgentEventsForSSEWithHandler(
 		}
 	}
 }
+
+// StopSessionRequest represents the stop session request
+type StopSessionRequest struct {
+	MessageID string `json:"message_id" binding:"required"`
+}
+
+// StopSession handles the stop generation request
+func (h *SessionHandler) StopSession(c *gin.Context) {
+	ctx := logger.CloneContext(c.Request.Context())
+	sessionID := c.Param("session_id")
+
+	if sessionID == "" {
+		c.JSON(400, gin.H{"error": "Session ID is required"})
+		return
+	}
+
+	// Parse request body to get message_id
+	var req StopSessionRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+		})
+		c.JSON(400, gin.H{"error": "message_id is required"})
+		return
+	}
+
+	assistantMessageID := req.MessageID
+	logger.Infof(ctx, "Stop generation request for session: %s, message: %s", sessionID, assistantMessageID)
+
+	// Get tenant ID from context
+	tenantID, exists := c.Get(types.TenantIDContextKey.String())
+	if !exists {
+		logger.Error(ctx, "Failed to get tenant ID")
+		c.JSON(401, gin.H{"error": "Unauthorized"})
+		return
+	}
+	tenantIDUint := tenantID.(uint)
+
+	// Verify message ownership and status
+	message, err := h.messageService.GetMessage(ctx, sessionID, assistantMessageID)
+	if err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+			"message_id": assistantMessageID,
+		})
+		c.JSON(404, gin.H{"error": "Message not found"})
+		return
+	}
+
+	// Verify message belongs to this session (double check)
+	if message.SessionID != sessionID {
+		logger.Warnf(ctx, "Message %s does not belong to session %s", assistantMessageID, sessionID)
+		c.JSON(403, gin.H{"error": "Message does not belong to this session"})
+		return
+	}
+
+	// Verify message belongs to the current tenant
+	session, err := h.sessionService.GetSession(ctx, sessionID)
+	if err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+		})
+		c.JSON(404, gin.H{"error": "Session not found"})
+		return
+	}
+
+	if session.TenantID != tenantIDUint {
+		logger.Warnf(ctx, "Session %s does not belong to tenant %d", sessionID, tenantIDUint)
+		c.JSON(403, gin.H{"error": "Access denied"})
+		return
+	}
+
+	// Check if message is already completed (stopped)
+	if message.IsCompleted {
+		logger.Infof(ctx, "Message %s is already completed, no need to stop", assistantMessageID)
+		c.JSON(200, gin.H{
+			"success": true,
+			"message": "Message already completed",
+		})
+		return
+	}
+
+	// Write stop event to StreamManager for distributed support
+	stopEvent := interfaces.StreamEvent{
+		ID:        fmt.Sprintf("stop-%d", time.Now().UnixNano()),
+		Type:      types.ResponseType(event.EventStop),
+		Content:   "",
+		Done:      true,
+		Timestamp: time.Now(),
+		Data: map[string]interface{}{
+			"session_id": sessionID,
+			"message_id": assistantMessageID,
+			"reason":     "user_requested",
+		},
+	}
+
+	if err := h.streamManager.AppendEvent(ctx, sessionID, assistantMessageID, stopEvent); err != nil {
+		logger.ErrorWithFields(ctx, err, map[string]interface{}{
+			"session_id": sessionID,
+			"message_id": assistantMessageID,
+		})
+		c.JSON(500, gin.H{"error": "Failed to write stop event"})
+		return
+	}
+
+	logger.Infof(ctx, "Stop event written successfully for session: %s, message: %s", sessionID, assistantMessageID)
+	c.JSON(200, gin.H{
+		"success": true,
+		"message": "Generation stopped",
+	})
+}

internal/router/router.go

@@ -180,6 +180,7 @@ func RegisterSessionRoutes(r *gin.RouterGroup, handler *handler.SessionHandler)
 		sessions.PUT("/:id", handler.UpdateSession)
 		sessions.DELETE("/:id", handler.DeleteSession)
 		sessions.POST("/:session_id/generate_title", handler.GenerateTitle)
+		sessions.POST("/:session_id/stop", handler.StopSession)
 		// 继续接收活跃流
 		sessions.GET("/continue-stream/:session_id", handler.ContinueStream)
 	}

internal/types/chat.go

@@ -72,6 +72,10 @@ type StreamResponse struct {
 	Done bool `json:"done"`
 	// Knowledge references
 	KnowledgeReferences References `json:"knowledge_references,omitempty"`
+	// Session ID (for agent_query event)
+	SessionID string `json:"session_id,omitempty"`
+	// Assistant Message ID (for agent_query event)
+	AssistantMessageID string `json:"assistant_message_id,omitempty"`
 	// Tool calls for streaming (partial)
 	ToolCalls []LLMToolCall `json:"tool_calls,omitempty"`
 	// Additional metadata for enhanced display