maintenance: scope AI conversations by creator

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/ChatConversationDao.java

@@ -17,6 +17,8 @@
 
 package org.apache.hertzbeat.ai.dao;
 
+import java.util.List;
+import java.util.Optional;
 import org.apache.hertzbeat.common.entity.ai.ChatConversation;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.stereotype.Repository;
@@ -26,4 +28,8 @@
  */
 @Repository
 public interface ChatConversationDao extends JpaRepository<ChatConversation, Long> {
+
+    Optional<ChatConversation> findByIdAndCreator(Long id, String creator);
+
+    List<ChatConversation> findAllByCreatorOrderByIdDesc(String creator);
 }

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/SopScheduleDao.java

@@ -19,6 +19,7 @@
 
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.Optional;
 import org.apache.hertzbeat.common.entity.ai.SopSchedule;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
@@ -37,14 +38,23 @@ public interface SopScheduleDao extends JpaRepository<SopSchedule, Long>, JpaSpe
      * @param conversationId The conversation ID
      * @return List of schedules
      */
-    List<SopSchedule> findByConversationId(Long conversationId);
+    List<SopSchedule> findByConversationIdAndCreator(Long conversationId, String creator);
+
+    /**
+     * Find a schedule only when it belongs to the supplied creator.
+     * @param id schedule identity
+     * @param creator authenticated creator
+     * @return matching schedule
+     */
+    Optional<SopSchedule> findByIdAndCreator(Long id, String creator);
 
     /**
      * Find all enabled schedules that are due for execution.
      * @param currentTime The current time to compare against
      * @return List of due schedules
      */
-    @Query("SELECT s FROM SopSchedule s WHERE s.enabled = true AND s.nextRunTime <= :currentTime")
+    @Query("SELECT s FROM SopSchedule s "
+            + "WHERE s.enabled = true AND s.creator IS NOT NULL AND s.nextRunTime <= :currentTime")
     List<SopSchedule> findDueSchedules(@Param("currentTime") LocalDateTime currentTime);
 
     /**

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutor.java

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.ai.dao.ChatMessageDao;
 import org.apache.hertzbeat.ai.service.SopScheduleService;
@@ -98,6 +99,12 @@ public void checkAndExecuteDueSchedules() {
      * Execute a single scheduled SOP and push result to conversation.
      */
     private void executeSchedule(SopSchedule schedule) {
+        schedule = sopScheduleService.getScheduleForExecution(schedule.getId());
+        if (schedule == null) {
+            return;
+        }
+        String executionCreator = schedule.getCreator();
+        Long executionConversationId = schedule.getConversationId();
         log.info("Executing scheduled SOP {} for conversation {}", 
                 schedule.getSopName(), schedule.getConversationId());
 
@@ -123,7 +130,13 @@ private void executeSchedule(SopSchedule schedule) {
 
             // Execute SOP
             SopResult result = sopEngine.executeSync(definition, params);
-
+
+            SopSchedule deliverySchedule = sopScheduleService.getScheduleForExecution(schedule.getId());
+            if (!hasSameExecutionTarget(deliverySchedule, executionCreator, executionConversationId)) {
+                log.warn("Schedule {} lost its execution owner before result delivery", schedule.getId());
+                return;
+            }
+
             // Create push message
             String messageContent = formatPushMessage(schedule, result);
 
@@ -132,6 +145,7 @@ private void executeSchedule(SopSchedule schedule) {
                     .conversationId(schedule.getConversationId())
                     .role(ROLE_SYSTEM_PUSH)
                     .content(messageContent)
+                    .creator(schedule.getCreator())
                     .build();
 
             chatMessageDao.save(pushMessage);
@@ -142,14 +156,21 @@ private void executeSchedule(SopSchedule schedule) {
         } catch (Exception e) {
             log.error("Failed to execute scheduled SOP {} for conversation {}", 
                     schedule.getSopName(), schedule.getConversationId(), e);
-
+
+            SopSchedule deliverySchedule = sopScheduleService.getScheduleForExecution(schedule.getId());
+            if (!hasSameExecutionTarget(deliverySchedule, executionCreator, executionConversationId)) {
+                log.warn("Schedule {} lost its execution owner before error delivery", schedule.getId());
+                return;
+            }
+
             // Still save an error message
             String errorContent = SopMessageUtil.getMessage("schedule.push.error.prefix") + " " + schedule.getSopName() 
                     + "\n\n" + SopMessageUtil.getMessage("schedule.push.error.label") + " " + e.getMessage();
             ChatMessage errorMessage = ChatMessage.builder()
                     .conversationId(schedule.getConversationId())
                     .role(ROLE_SYSTEM_PUSH)
                     .content(errorContent)
+                    .creator(schedule.getCreator())
                     .build();
             chatMessageDao.save(errorMessage);
 
@@ -159,6 +180,12 @@ private void executeSchedule(SopSchedule schedule) {
         }
     }
 
+    private boolean hasSameExecutionTarget(SopSchedule schedule, String creator, Long conversationId) {
+        return schedule != null
+                && Objects.equals(creator, schedule.getCreator())
+                && Objects.equals(conversationId, schedule.getConversationId());
+    }
+
     /**
      * Format the push message content with SOP result.
      */

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/SopScheduleService.java

@@ -73,6 +73,15 @@ public interface SopScheduleService {
      */
     List<SopSchedule> getDueSchedules();
 
+    /**
+     * Re-read a schedule for background execution and verify that its persisted
+     * creator still owns the target conversation. This method does not depend
+     * on a request-thread subject.
+     * @param id schedule ID
+     * @return validated schedule, or {@code null} when it must not execute
+     */
+    SopSchedule getScheduleForExecution(Long id);
+
     /**
      * Update the execution times after a schedule runs.
      * @param id The schedule ID

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImpl.java

@@ -33,7 +33,6 @@
 import org.apache.hertzbeat.common.entity.ai.ChatMessage;
 import org.apache.hertzbeat.common.util.AesUtil;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.domain.Sort;
 import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -66,6 +65,8 @@ public class ConversationServiceImpl implements ConversationService {
 
     @Override
     public Flux<ServerSentEvent<ChatResponseChunk>> streamChat(String message, Long conversationId) {
+        String creator = requireCurrentUserId();
+        ChatConversation conversation = requireOwnedConversation(conversationId, creator);
 
         // Check if provider is properly configured
         if (!chatClientProviderService.isConfigured()) {
@@ -79,8 +80,6 @@ public Flux<ServerSentEvent<ChatResponseChunk>> streamChat(String message, Long
         }
 
         log.info("Starting streaming conversation: {}", conversationId);
-        ChatConversation conversation = conversationDao.findById(conversationId)
-            .orElseThrow(() -> new IllegalArgumentException("Conversation not found: " + conversationId));
 
         // Manually load messages for conversation history
         List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
@@ -162,6 +161,7 @@ public Flux<ServerSentEvent<ChatResponseChunk>> streamChat(String message, Long
     public ChatConversation createConversation() {
         ChatConversation conversation = new ChatConversation();
         conversation.setTitle("conversation-" + UUID.randomUUID().toString().substring(0, 4));
+        conversation.setCreator(requireCurrentUserId());
         return conversationDao.save(conversation);
     }
 
@@ -170,17 +170,16 @@ public ChatConversation getConversation(Long conversationId) {
         if (conversationId == null) {
             return null;
         }
-        ChatConversation conversation = conversationDao.findById(conversationId).orElse(null);
-        if (conversation != null) {
-            List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
-            conversation.setMessages(messages);
-        }
+        ChatConversation conversation = requireOwnedConversation(conversationId, requireCurrentUserId());
+        List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
+        conversation.setMessages(messages);
         return conversation;
     }
 
     @Override
     public List<ChatConversation> getAllConversations() {
-        List<ChatConversation> conversations = conversationDao.findAll(Sort.by(Sort.Direction.DESC, "id"));
+        List<ChatConversation> conversations =
+            conversationDao.findAllByCreatorOrderByIdDesc(requireCurrentUserId());
         if (conversations.isEmpty()) {
             return conversations;
         }
@@ -201,6 +200,7 @@ public List<ChatConversation> getAllConversations() {
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void deleteConversation(Long conversationId) {
+        requireOwnedConversation(conversationId, requireCurrentUserId());
         // Delete associated schedules first to prevent tasks from writing orphaned messages.
         sopScheduleDao.deleteByConversationId(conversationId);
         List<ChatMessage> messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId);
@@ -212,7 +212,8 @@ public void deleteConversation(Long conversationId) {
 
     @Override
     public Boolean saveSecurityData(SecurityData securityData) {
-        Optional<ChatConversation> chatConversation = conversationDao.findById(securityData.getConversationId());
+        Optional<ChatConversation> chatConversation = conversationDao.findByIdAndCreator(
+            securityData.getConversationId(), requireCurrentUserId());
         if (chatConversation.isPresent()) {
             ChatConversation conversation = chatConversation.get();
             conversation.setSecurityData(AesUtil.aesEncode(securityData.getSecurityData()));
@@ -222,4 +223,17 @@ public Boolean saveSecurityData(SecurityData securityData) {
         return false;
     }
 
+    private String requireCurrentUserId() {
+        SubjectSum subject = SurenessContextHolder.getBindSubject();
+        if (subject == null || subject.getPrincipal() == null) {
+            throw new IllegalStateException("No authenticated user");
+        }
+        return String.valueOf(subject.getPrincipal());
+    }
+
+    private ChatConversation requireOwnedConversation(Long conversationId, String creator) {
+        return conversationDao.findByIdAndCreator(conversationId, creator)
+            .orElseThrow(() -> new IllegalArgumentException("Conversation not found: " + conversationId));
+    }
+
 }