From 70354bfce0c547374d91b142a6b73f0f49c6d929 Mon Sep 17 00:00:00 2001 From: Logic Date: Thu, 30 Jul 2026 15:52:43 +0800 Subject: [PATCH 1/2] maintenance: scope AI conversations by creator --- .../hertzbeat/ai/dao/ChatConversationDao.java | 6 + .../service/impl/ConversationServiceImpl.java | 34 ++++-- .../impl/ConversationServiceImplTest.java | 114 ++++++++++++++++-- .../common/entity/ai/ChatConversation.java | 2 + .../entity/ai/ChatConversationTest.java | 45 +++++++ 5 files changed, 183 insertions(+), 18 deletions(-) create mode 100644 hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/entity/ai/ChatConversationTest.java diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/ChatConversationDao.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/ChatConversationDao.java index 971b51f7cb2..852f7d4f031 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/ChatConversationDao.java +++ b/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 { + + Optional findByIdAndCreator(Long id, String creator); + + List findAllByCreatorOrderByIdDesc(String creator); } diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImpl.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImpl.java index c674617dea2..1b6d3851bd9 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImpl.java +++ b/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> 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> 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 messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId); @@ -162,6 +161,7 @@ public Flux> 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 messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId); - conversation.setMessages(messages); - } + ChatConversation conversation = requireOwnedConversation(conversationId, requireCurrentUserId()); + List messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId); + conversation.setMessages(messages); return conversation; } @Override public List getAllConversations() { - List conversations = conversationDao.findAll(Sort.by(Sort.Direction.DESC, "id")); + List conversations = + conversationDao.findAllByCreatorOrderByIdDesc(requireCurrentUserId()); if (conversations.isEmpty()) { return conversations; } @@ -201,6 +200,7 @@ public List 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 messages = messageDao.findByConversationIdOrderByGmtCreateAsc(conversationId); @@ -212,7 +212,8 @@ public void deleteConversation(Long conversationId) { @Override public Boolean saveSecurityData(SecurityData securityData) { - Optional chatConversation = conversationDao.findById(securityData.getConversationId()); + Optional 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)); + } + } diff --git a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImplTest.java b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImplTest.java index b192073e45a..1a639cf23cc 100644 --- a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImplTest.java +++ b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/ConversationServiceImplTest.java @@ -18,9 +18,13 @@ package org.apache.hertzbeat.ai.service.impl; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -34,6 +38,7 @@ import org.apache.hertzbeat.ai.dao.SopScheduleDao; import org.apache.hertzbeat.ai.pojo.dto.ChatRequestContext; import org.apache.hertzbeat.ai.pojo.dto.ChatResponseChunk; +import org.apache.hertzbeat.ai.pojo.dto.SecurityData; import org.apache.hertzbeat.ai.service.ChatClientProviderService; import org.apache.hertzbeat.common.entity.ai.ChatConversation; import org.apache.hertzbeat.common.entity.ai.ChatMessage; @@ -78,29 +83,30 @@ void clearSecurityContext() { @Test void streamChatShouldKeepCompleteConversationHistory() { - SubjectSum subject = org.mockito.Mockito.mock(SubjectSum.class); - SurenessContextHolder.bindSubject(subject); + SubjectSum subject = bindSubject("alice"); ChatConversation conversation = ChatConversation.builder() .id(CONVERSATION_ID) - .title("已命名会话") + .title("Named conversation") + .creator("alice") .build(); List history = List.of( ChatMessage.builder() .id(11L) .conversationId(CONVERSATION_ID) .role("user") - .content("上一轮问题") + .content("Previous question") .build(), ChatMessage.builder() .id(12L) .conversationId(CONVERSATION_ID) .role("assistant") - .content("上一轮回答") + .content("Previous answer") .build()); AtomicLong messageId = new AtomicLong(20L); when(chatClientProviderService.isConfigured()).thenReturn(true); - when(conversationDao.findById(CONVERSATION_ID)).thenReturn(Optional.of(conversation)); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.of(conversation)); when(messageDao.findByConversationIdOrderByGmtCreateAsc(CONVERSATION_ID)).thenReturn(history); when(messageDao.save(any(ChatMessage.class))).thenAnswer(invocation -> { ChatMessage savedMessage = invocation.getArgument(0); @@ -108,10 +114,10 @@ void streamChatShouldKeepCompleteConversationHistory() { return savedMessage; }); when(chatClientProviderService.streamChat(any(ChatRequestContext.class))) - .thenReturn(Flux.just("本轮回答")); + .thenReturn(Flux.just("Current answer")); List> events = conversationService - .streamChat("本轮问题", CONVERSATION_ID) + .streamChat("Current question", CONVERSATION_ID) .collectList() .block(); @@ -128,12 +134,20 @@ void streamChatShouldKeepCompleteConversationHistory() { */ @Test void deleteConversationShouldRemoveSchedulesMessagesAndConversationInOrder() { + bindSubject("alice"); + ChatConversation conversation = ChatConversation.builder() + .id(CONVERSATION_ID) + .title("Owned conversation") + .creator("alice") + .build(); ChatMessage message = ChatMessage.builder() .id(11L) .conversationId(CONVERSATION_ID) .role("user") .content("message to delete") .build(); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.of(conversation)); when(messageDao.findByConversationIdOrderByGmtCreateAsc(CONVERSATION_ID)) .thenReturn(List.of(message)); @@ -144,4 +158,88 @@ void deleteConversationShouldRemoveSchedulesMessagesAndConversationInOrder() { deletionOrder.verify(messageDao).deleteAll(List.of(message)); deletionOrder.verify(conversationDao).deleteById(CONVERSATION_ID); } + + @Test + void listConversationsShouldExcludeOtherCreators() { + bindSubject("alice"); + ChatConversation ownedConversation = ChatConversation.builder() + .id(CONVERSATION_ID) + .title("Owned conversation") + .creator("alice") + .build(); + when(conversationDao.findAllByCreatorOrderByIdDesc("alice")) + .thenReturn(List.of(ownedConversation)); + when(messageDao.findByConversationIdInOrderByGmtCreateAsc(List.of(CONVERSATION_ID))) + .thenReturn(List.of()); + + List result = conversationService.getAllConversations(); + + assertEquals(List.of(ownedConversation), result); + verify(conversationDao).findAllByCreatorOrderByIdDesc("alice"); + } + + @Test + void getConversationShouldRejectAnotherCreator() { + bindSubject("alice"); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, + () -> conversationService.getConversation(CONVERSATION_ID)); + verify(messageDao, never()).findByConversationIdOrderByGmtCreateAsc(CONVERSATION_ID); + } + + @Test + void deleteConversationShouldRejectAnotherCreator() { + bindSubject("alice"); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, + () -> conversationService.deleteConversation(CONVERSATION_ID)); + verify(sopScheduleDao, never()).deleteByConversationId(CONVERSATION_ID); + verify(conversationDao, never()).deleteById(CONVERSATION_ID); + } + + @Test + void streamChatShouldRejectAnotherCreator() { + bindSubject("alice"); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, + () -> conversationService.streamChat("Current question", CONVERSATION_ID)); + verify(messageDao, never()).save(any(ChatMessage.class)); + } + + @Test + void createConversationShouldRecordCurrentCreator() { + bindSubject("alice"); + when(conversationDao.save(any(ChatConversation.class))) + .thenAnswer(invocation -> invocation.getArgument(0)); + + ChatConversation conversation = conversationService.createConversation(); + + assertEquals("alice", conversation.getCreator()); + } + + @Test + void saveSecurityDataShouldRejectAnotherCreator() { + bindSubject("alice"); + SecurityData securityData = new SecurityData(); + securityData.setConversationId(CONVERSATION_ID); + securityData.setSecurityData("sensitive-value"); + when(conversationDao.findByIdAndCreator(CONVERSATION_ID, "alice")) + .thenReturn(Optional.empty()); + + assertFalse(conversationService.saveSecurityData(securityData)); + verify(conversationDao, never()).save(any(ChatConversation.class)); + } + + private SubjectSum bindSubject(String principal) { + SubjectSum subject = mock(SubjectSum.class); + when(subject.getPrincipal()).thenReturn(principal); + SurenessContextHolder.bindSubject(subject); + return subject; + } } diff --git a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java index 22e2f946394..3236092eb4f 100644 --- a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java +++ b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java @@ -19,6 +19,7 @@ import static io.swagger.v3.oas.annotations.media.Schema.AccessMode.READ_ONLY; +import com.fasterxml.jackson.annotation.JsonIgnore; import io.swagger.v3.oas.annotations.media.Schema; import jakarta.persistence.Entity; import jakarta.persistence.EntityListeners; @@ -81,5 +82,6 @@ public class ChatConversation { @OneToMany(mappedBy = "conversation") private List messages; + @JsonIgnore private String securityData; } diff --git a/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/entity/ai/ChatConversationTest.java b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/entity/ai/ChatConversationTest.java new file mode 100644 index 00000000000..8d6cd93a23d --- /dev/null +++ b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/entity/ai/ChatConversationTest.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.hertzbeat.common.entity.ai; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.apache.hertzbeat.common.util.JsonUtil; +import org.junit.jupiter.api.Test; + +/** + * Tests AI conversation serialization. + */ +class ChatConversationTest { + + @Test + void serializationShouldExcludeStoredSecurityData() { + ChatConversation conversation = ChatConversation.builder() + .id(1L) + .title("Owned conversation") + .securityData("encrypted-value") + .build(); + + String json = JsonUtil.toJson(conversation); + + assertNotNull(json); + assertFalse(json.contains("securityData")); + assertFalse(json.contains("encrypted-value")); + } +} From d11f7a17661856450aa6aa2b27a39ec1c3776242 Mon Sep 17 00:00:00 2001 From: Logic Date: Thu, 30 Jul 2026 21:21:46 +0800 Subject: [PATCH 2/2] complete conversation ownership boundaries --- .../hertzbeat/ai/dao/SopScheduleDao.java | 14 +- .../ai/schedule/SopScheduleExecutor.java | 31 +++- .../ai/service/SopScheduleService.java | 9 + .../service/impl/SopScheduleServiceImpl.java | 100 ++++++++--- .../ai/tools/impl/MonitorToolsImpl.java | 12 +- .../ai/schedule/SopScheduleExecutorTest.java | 40 +++++ .../impl/SopScheduleServiceImplTest.java | 164 ++++++++++++++++++ .../ai/tools/impl/MonitorToolsImplTest.java | 77 ++++++++ .../common/entity/ai/ChatConversation.java | 5 +- .../common/entity/ai/SopSchedule.java | 1 + .../h2/V182__scope_sop_schedule_owners.sql | 24 +++ .../mysql/V182__scope_sop_schedule_owners.sql | 24 +++ .../V182__scope_sop_schedule_owners.sql | 24 +++ home/docs/start/upgrade.md | 26 +++ 14 files changed, 518 insertions(+), 33 deletions(-) create mode 100644 hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImplTest.java create mode 100644 hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImplTest.java create mode 100644 hertzbeat-startup/src/main/resources/db/migration/h2/V182__scope_sop_schedule_owners.sql create mode 100644 hertzbeat-startup/src/main/resources/db/migration/mysql/V182__scope_sop_schedule_owners.sql create mode 100644 hertzbeat-startup/src/main/resources/db/migration/postgresql/V182__scope_sop_schedule_owners.sql diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/SopScheduleDao.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/SopScheduleDao.java index 08861591e65..1b82536c1cb 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/dao/SopScheduleDao.java +++ b/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, JpaSpe * @param conversationId The conversation ID * @return List of schedules */ - List findByConversationId(Long conversationId); + List 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 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 findDueSchedules(@Param("currentTime") LocalDateTime currentTime); /** diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutor.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutor.java index be861e3d45e..afede329c02 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutor.java +++ b/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,7 +156,13 @@ 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(); @@ -150,6 +170,7 @@ private void executeSchedule(SopSchedule schedule) { .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. */ diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/SopScheduleService.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/SopScheduleService.java index 3f04938bf69..fa69c23699d 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/SopScheduleService.java +++ b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/SopScheduleService.java @@ -73,6 +73,15 @@ public interface SopScheduleService { */ List 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 diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImpl.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImpl.java index 88c8d0a23bd..ca4b50dcbf5 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImpl.java +++ b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImpl.java @@ -17,11 +17,16 @@ package org.apache.hertzbeat.ai.service.impl; +import com.usthe.sureness.subject.SubjectSum; +import com.usthe.sureness.util.SurenessContextHolder; import java.time.LocalDateTime; import java.util.List; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.apache.hertzbeat.ai.dao.ChatConversationDao; import org.apache.hertzbeat.ai.dao.SopScheduleDao; import org.apache.hertzbeat.ai.service.SopScheduleService; +import org.apache.hertzbeat.common.entity.ai.ChatConversation; import org.apache.hertzbeat.common.entity.ai.SopSchedule; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.support.CronExpression; @@ -36,23 +41,32 @@ public class SopScheduleServiceImpl implements SopScheduleService { private final SopScheduleDao sopScheduleDao; + private final ChatConversationDao conversationDao; @Autowired - public SopScheduleServiceImpl(SopScheduleDao sopScheduleDao) { + public SopScheduleServiceImpl(SopScheduleDao sopScheduleDao, + ChatConversationDao conversationDao) { this.sopScheduleDao = sopScheduleDao; + this.conversationDao = conversationDao; } @Override @Transactional public SopSchedule createSchedule(SopSchedule schedule) { - // Validate cron expression + String creator = requireCurrentUserId(); + requireOwnedConversation(schedule.getConversationId(), creator); validateCronExpression(schedule.getCronExpression()); - - // Calculate next run time - schedule.setNextRunTime(calculateNextRunTime(schedule.getCronExpression())); - schedule.setEnabled(schedule.getEnabled() != null ? schedule.getEnabled() : true); - - SopSchedule saved = sopScheduleDao.save(schedule); + + SopSchedule persisted = SopSchedule.builder() + .conversationId(schedule.getConversationId()) + .sopName(schedule.getSopName()) + .sopParams(schedule.getSopParams()) + .cronExpression(schedule.getCronExpression()) + .enabled(schedule.getEnabled() != null ? schedule.getEnabled() : true) + .nextRunTime(calculateNextRunTime(schedule.getCronExpression())) + .creator(creator) + .build(); + SopSchedule saved = sopScheduleDao.save(persisted); log.info("Created schedule {} for conversation {} with SOP {}", saved.getId(), saved.getConversationId(), saved.getSopName()); return saved; @@ -61,57 +75,59 @@ public SopSchedule createSchedule(SopSchedule schedule) { @Override @Transactional public SopSchedule updateSchedule(SopSchedule schedule) { - SopSchedule existing = sopScheduleDao.findById(schedule.getId()) + SopSchedule existing = sopScheduleDao.findByIdAndCreator( + schedule.getId(), requireCurrentUserId()) .orElseThrow(() -> new IllegalArgumentException("Schedule not found: " + schedule.getId())); - - // Update fields + existing.setSopName(schedule.getSopName()); existing.setSopParams(schedule.getSopParams()); - - // If cron expression changed, recalculate next run time + if (!existing.getCronExpression().equals(schedule.getCronExpression())) { validateCronExpression(schedule.getCronExpression()); existing.setCronExpression(schedule.getCronExpression()); existing.setNextRunTime(calculateNextRunTime(schedule.getCronExpression())); } - + if (schedule.getEnabled() != null) { existing.setEnabled(schedule.getEnabled()); } - + return sopScheduleDao.save(existing); } @Override @Transactional public void deleteSchedule(Long id) { + SopSchedule schedule = sopScheduleDao.findByIdAndCreator(id, requireCurrentUserId()) + .orElseThrow(() -> new IllegalArgumentException("Schedule not found: " + id)); log.info("Deleting schedule {}", id); - sopScheduleDao.deleteById(id); + sopScheduleDao.delete(schedule); } @Override public SopSchedule getSchedule(Long id) { - return sopScheduleDao.findById(id).orElse(null); + return sopScheduleDao.findByIdAndCreator(id, requireCurrentUserId()).orElse(null); } @Override public List getSchedulesByConversation(Long conversationId) { - return sopScheduleDao.findByConversationId(conversationId); + String creator = requireCurrentUserId(); + requireOwnedConversation(conversationId, creator); + return sopScheduleDao.findByConversationIdAndCreator(conversationId, creator); } @Override @Transactional public SopSchedule toggleSchedule(Long id, boolean enabled) { - SopSchedule schedule = sopScheduleDao.findById(id) + SopSchedule schedule = sopScheduleDao.findByIdAndCreator(id, requireCurrentUserId()) .orElseThrow(() -> new IllegalArgumentException("Schedule not found: " + id)); - + schedule.setEnabled(enabled); - - // If enabling, recalculate next run time + if (enabled) { schedule.setNextRunTime(calculateNextRunTime(schedule.getCronExpression())); } - + log.info("Schedule {} {} ", id, enabled ? "enabled" : "disabled"); return sopScheduleDao.save(schedule); } @@ -123,16 +139,34 @@ public List getDueSchedules() { @Override @Transactional - public void updateAfterExecution(Long id) { + public SopSchedule getScheduleForExecution(Long id) { SopSchedule schedule = sopScheduleDao.findById(id).orElse(null); + if (schedule == null || !Boolean.TRUE.equals(schedule.getEnabled())) { + return null; + } + if (StringUtils.isBlank(schedule.getCreator()) + || conversationDao.findByIdAndCreator( + schedule.getConversationId(), schedule.getCreator()).isEmpty()) { + schedule.setEnabled(false); + sopScheduleDao.save(schedule); + log.warn("Disabled schedule {} because its execution owner is missing", id); + return null; + } + return schedule; + } + + @Override + @Transactional + public void updateAfterExecution(Long id) { + SopSchedule schedule = getScheduleForExecution(id); if (schedule == null) { return; } - + schedule.setLastRunTime(LocalDateTime.now()); schedule.setNextRunTime(calculateNextRunTime(schedule.getCronExpression())); sopScheduleDao.save(schedule); - + log.debug("Updated schedule {} - Last run: {}, Next run: {}", id, schedule.getLastRunTime(), schedule.getNextRunTime()); } @@ -154,4 +188,18 @@ private LocalDateTime calculateNextRunTime(String cronExpression) { return null; } } + + 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)); + } } diff --git a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImpl.java b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImpl.java index 7886cf97b6d..1eb0ffd5961 100644 --- a/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImpl.java +++ b/hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImpl.java @@ -312,8 +312,16 @@ public String addMonitorProtected( // Query and add sensitive parameters if (conversationId != null) { - Optional chatConversation = conversationDao.findById(conversationId); - if (chatConversation.isPresent() && StringUtils.isNotEmpty(chatConversation.get().getSecurityData())) { + SubjectSum subject = McpContextHolder.getSubject(); + if (subject == null || subject.getPrincipal() == null) { + return "Error: Authenticated conversation context is required"; + } + Optional chatConversation = conversationDao.findByIdAndCreator( + conversationId, String.valueOf(subject.getPrincipal())); + if (chatConversation.isEmpty()) { + return "Error: Conversation not found or inaccessible"; + } + if (StringUtils.isNotEmpty(chatConversation.get().getSecurityData())) { List securityParams = JsonUtil.fromJson( AesUtil.aesDecode(chatConversation.get().getSecurityData()), new TypeReference>() { diff --git a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java index e4f6bcfe3cc..742fa0ee6b1 100644 --- a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java +++ b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java @@ -19,6 +19,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -75,6 +76,8 @@ void checkShouldContinueAfterOneScheduleFailsToUpdate() { .content("ok") .build(); when(scheduleService.getDueSchedules()).thenReturn(List.of(first, second)); + when(scheduleService.getScheduleForExecution(1L)).thenReturn(first, first); + when(scheduleService.getScheduleForExecution(2L)).thenReturn(second, second); when(skillRegistry.getSkill("daily_inspection")).thenReturn(definition); when(sopEngine.executeSync(any(SopDefinition.class), anyMap())).thenReturn(result); doThrow(new IllegalStateException("database unavailable")) @@ -83,6 +86,8 @@ void checkShouldContinueAfterOneScheduleFailsToUpdate() { executor.checkAndExecuteDueSchedules(); verify(sopEngine, times(2)).executeSync(any(SopDefinition.class), anyMap()); + verify(chatMessageDao, times(2)).save(argThat( + message -> "alice".equals(message.getCreator()))); verify(scheduleService).updateAfterExecution(2L); } @@ -90,6 +95,7 @@ void checkShouldContinueAfterOneScheduleFailsToUpdate() { void checkShouldRejectInvalidScheduleParameters() { SopSchedule schedule = schedule(1L, "not-json"); when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule)); + when(scheduleService.getScheduleForExecution(1L)).thenReturn(schedule, schedule); when(skillRegistry.getSkill("daily_inspection")) .thenReturn(SopDefinition.builder().name("daily_inspection").build()); @@ -100,12 +106,46 @@ void checkShouldRejectInvalidScheduleParameters() { verify(scheduleService).updateAfterExecution(1L); } + @Test + void checkShouldSkipScheduleWithoutValidatedOwner() { + SopSchedule schedule = schedule(1L, null); + when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule)); + when(scheduleService.getScheduleForExecution(1L)).thenReturn(null); + + executor.checkAndExecuteDueSchedules(); + + verifyNoInteractions(sopEngine, chatMessageDao); + verify(scheduleService, times(0)).updateAfterExecution(1L); + } + + @Test + void checkShouldNotDeliverWhenOwnerChangesDuringExecution() { + SopSchedule schedule = schedule(1L, null); + SopSchedule changedOwner = schedule(1L, null); + changedOwner.setCreator("bob"); + changedOwner.setConversationId(20L); + when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule)); + when(scheduleService.getScheduleForExecution(1L)) + .thenReturn(schedule, changedOwner, changedOwner); + when(skillRegistry.getSkill("daily_inspection")) + .thenReturn(SopDefinition.builder().name("daily_inspection").build()); + when(sopEngine.executeSync(any(SopDefinition.class), anyMap())) + .thenReturn(SopResult.builder().status("SUCCESS").content("ok").build()); + + executor.checkAndExecuteDueSchedules(); + + verify(sopEngine).executeSync(any(SopDefinition.class), anyMap()); + verifyNoInteractions(chatMessageDao); + verify(scheduleService).updateAfterExecution(1L); + } + private SopSchedule schedule(Long id, String params) { return SopSchedule.builder() .id(id) .conversationId(10L) .sopName("daily_inspection") .sopParams(params) + .creator("alice") .build(); } } diff --git a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImplTest.java b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImplTest.java new file mode 100644 index 00000000000..250dd542ff4 --- /dev/null +++ b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/service/impl/SopScheduleServiceImplTest.java @@ -0,0 +1,164 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to You 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 org.apache.hertzbeat.ai.service.impl; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.usthe.sureness.subject.SubjectSum; +import com.usthe.sureness.util.SurenessContextHolder; +import java.util.Optional; +import org.apache.hertzbeat.ai.dao.ChatConversationDao; +import org.apache.hertzbeat.ai.dao.SopScheduleDao; +import org.apache.hertzbeat.common.entity.ai.ChatConversation; +import org.apache.hertzbeat.common.entity.ai.SopSchedule; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Ownership contracts for user-facing SOP schedule operations. + */ +@ExtendWith(MockitoExtension.class) +class SopScheduleServiceImplTest { + + @Mock + private SopScheduleDao scheduleDao; + + @Mock + private ChatConversationDao conversationDao; + + private SopScheduleServiceImpl service; + + @BeforeEach + void setUp() { + service = new SopScheduleServiceImpl(scheduleDao, conversationDao); + SubjectSum subject = mock(SubjectSum.class); + lenient().when(subject.getPrincipal()).thenReturn("alice"); + SurenessContextHolder.bindSubject(subject); + } + + @AfterEach + void clearSubject() { + SurenessContextHolder.clear(); + } + + @Test + void createShouldNotTrustRequestCreator() { + SopSchedule request = schedule(1L, "bob"); + when(conversationDao.findByIdAndCreator(10L, "alice")) + .thenReturn(Optional.of(ChatConversation.builder() + .id(10L) + .creator("alice") + .build())); + when(scheduleDao.save(any(SopSchedule.class))) + .thenAnswer(invocation -> invocation.getArgument(0)); + + SopSchedule created = service.createSchedule(request); + + assertEquals("alice", created.getCreator()); + } + + @Test + void getShouldHideAnotherCreatorsSchedule() { + when(scheduleDao.findByIdAndCreator(1L, "alice")).thenReturn(Optional.empty()); + + assertNull(service.getSchedule(1L)); + } + + @Test + void listShouldRejectAnotherCreatorsConversation() { + when(conversationDao.findByIdAndCreator(10L, "alice")).thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, + () -> service.getSchedulesByConversation(10L)); + verify(scheduleDao, never()).findByConversationIdAndCreator(10L, "alice"); + } + + @Test + void deleteShouldNotRemoveAnotherCreatorsSchedule() { + when(scheduleDao.findByIdAndCreator(1L, "alice")).thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, () -> service.deleteSchedule(1L)); + verify(scheduleDao, never()).delete(any(SopSchedule.class)); + } + + @Test + void updateAndToggleShouldNotModifyAnotherCreatorsSchedule() { + when(scheduleDao.findByIdAndCreator(1L, "alice")).thenReturn(Optional.empty()); + + assertThrows(IllegalArgumentException.class, + () -> service.updateSchedule(schedule(1L, "bob"))); + assertThrows(IllegalArgumentException.class, + () -> service.toggleSchedule(1L, true)); + verify(scheduleDao, never()).save(any(SopSchedule.class)); + } + + @Test + void backgroundExecutionShouldDisableMissingOwner() { + SopSchedule schedule = schedule(1L, "legacy-owner"); + schedule.setEnabled(true); + when(scheduleDao.findById(1L)).thenReturn(Optional.of(schedule)); + when(conversationDao.findByIdAndCreator(10L, "legacy-owner")) + .thenReturn(Optional.empty()); + when(scheduleDao.save(schedule)).thenReturn(schedule); + + assertNull(service.getScheduleForExecution(1L)); + assertFalse(schedule.getEnabled()); + verify(scheduleDao).save(schedule); + } + + @Test + void backgroundExecutionUsesPersistedOwnerWithoutRequestSubject() { + SurenessContextHolder.clear(); + SopSchedule schedule = schedule(1L, "alice"); + schedule.setEnabled(true); + when(scheduleDao.findById(1L)).thenReturn(Optional.of(schedule)); + when(conversationDao.findByIdAndCreator(10L, "alice")) + .thenReturn(Optional.of(ChatConversation.builder() + .id(10L) + .creator("alice") + .build())); + + assertSame(schedule, service.getScheduleForExecution(1L)); + } + + private SopSchedule schedule(Long id, String creator) { + return SopSchedule.builder() + .id(id) + .conversationId(10L) + .sopName("daily_inspection") + .cronExpression("0 0 9 * * ?") + .creator(creator) + .build(); + } +} diff --git a/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImplTest.java b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImplTest.java new file mode 100644 index 00000000000..669868a0615 --- /dev/null +++ b/hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImplTest.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.hertzbeat.ai.tools.impl; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import com.usthe.sureness.subject.SubjectSum; +import java.util.Optional; +import org.apache.hertzbeat.ai.config.McpContextHolder; +import org.apache.hertzbeat.ai.dao.ChatConversationDao; +import org.apache.hertzbeat.manager.service.AppService; +import org.apache.hertzbeat.manager.service.MonitorService; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Verifies that protected monitor creation cannot load another user's + * conversation credentials. + */ +@ExtendWith(MockitoExtension.class) +class MonitorToolsImplTest { + + @Mock + private MonitorService monitorService; + + @Mock + private AppService appService; + + @Mock + private ChatConversationDao conversationDao; + + @InjectMocks + private MonitorToolsImpl monitorTools; + + @AfterEach + void clearContext() { + McpContextHolder.clear(); + } + + @Test + void protectedAddShouldRejectConversationOutsideCurrentCreator() { + SubjectSum subject = mock(SubjectSum.class); + when(subject.getPrincipal()).thenReturn("alice"); + McpContextHolder.setSubject(subject); + when(conversationDao.findByIdAndCreator(10L, "alice")).thenReturn(Optional.empty()); + + String result = monitorTools.addMonitorProtected( + 10L, "database", "mysql", 60, "{\"host\":\"db.local\"}", null); + + assertEquals("Error: Conversation not found or inaccessible", result); + verify(conversationDao).findByIdAndCreator(10L, "alice"); + verifyNoInteractions(monitorService); + } +} diff --git a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java index 3236092eb4f..0a8d3e51f59 100644 --- a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java +++ b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/ChatConversation.java @@ -26,6 +26,7 @@ import jakarta.persistence.GeneratedValue; import jakarta.persistence.GenerationType; import jakarta.persistence.Id; +import jakarta.persistence.Index; import jakarta.persistence.OneToMany; import jakarta.persistence.Table; import org.springframework.data.annotation.CreatedBy; @@ -48,7 +49,9 @@ @Builder @Entity @EntityListeners(AuditingEntityListener.class) -@Table(name = "hzb_ai_conversation") +@Table(name = "hzb_ai_conversation", indexes = { + @Index(name = "idx_ai_conversation_creator", columnList = "creator") +}) @AllArgsConstructor @NoArgsConstructor public class ChatConversation { diff --git a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/SopSchedule.java b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/SopSchedule.java index d0efbfb38f9..a362df38b16 100644 --- a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/SopSchedule.java +++ b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/entity/ai/SopSchedule.java @@ -51,6 +51,7 @@ @EntityListeners(AuditingEntityListener.class) @Table(name = "hzb_sop_schedule", indexes = { @Index(name = "idx_schedule_conversation_id", columnList = "conversation_id"), + @Index(name = "idx_schedule_creator_conversation", columnList = "creator, conversation_id"), @Index(name = "idx_schedule_enabled_next", columnList = "enabled, next_run_time") }) @AllArgsConstructor diff --git a/hertzbeat-startup/src/main/resources/db/migration/h2/V182__scope_sop_schedule_owners.sql b/hertzbeat-startup/src/main/resources/db/migration/h2/V182__scope_sop_schedule_owners.sql new file mode 100644 index 00000000000..a0aa6a82d89 --- /dev/null +++ b/hertzbeat-startup/src/main/resources/db/migration/h2/V182__scope_sop_schedule_owners.sql @@ -0,0 +1,24 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to You 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. + +UPDATE hzb_sop_schedule +SET enabled = 0 +WHERE creator IS NULL + OR TRIM(creator) = ''; + +CREATE INDEX IF NOT EXISTS idx_schedule_creator_conversation + ON hzb_sop_schedule(creator, conversation_id); diff --git a/hertzbeat-startup/src/main/resources/db/migration/mysql/V182__scope_sop_schedule_owners.sql b/hertzbeat-startup/src/main/resources/db/migration/mysql/V182__scope_sop_schedule_owners.sql new file mode 100644 index 00000000000..3d0a986ea18 --- /dev/null +++ b/hertzbeat-startup/src/main/resources/db/migration/mysql/V182__scope_sop_schedule_owners.sql @@ -0,0 +1,24 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to You 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. + +UPDATE hzb_sop_schedule +SET enabled = 0 +WHERE creator IS NULL + OR TRIM(creator) = ''; + +CREATE INDEX idx_schedule_creator_conversation + ON hzb_sop_schedule(creator, conversation_id); diff --git a/hertzbeat-startup/src/main/resources/db/migration/postgresql/V182__scope_sop_schedule_owners.sql b/hertzbeat-startup/src/main/resources/db/migration/postgresql/V182__scope_sop_schedule_owners.sql new file mode 100644 index 00000000000..44aaf712207 --- /dev/null +++ b/hertzbeat-startup/src/main/resources/db/migration/postgresql/V182__scope_sop_schedule_owners.sql @@ -0,0 +1,24 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to You 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. + +UPDATE hzb_sop_schedule +SET enabled = 0 +WHERE creator IS NULL + OR BTRIM(creator) = ''; + +CREATE INDEX idx_schedule_creator_conversation + ON hzb_sop_schedule(creator, conversation_id); diff --git a/home/docs/start/upgrade.md b/home/docs/start/upgrade.md index e7363fc4245..6c5a083ca97 100644 --- a/home/docs/start/upgrade.md +++ b/home/docs/start/upgrade.md @@ -43,4 +43,30 @@ Apache HertzBeat's metadata information is stored in H2 or Mysql, PostgreSQL rel - `bin/shutdown.sh` stops the HertzBeat process and downloads the new installation package - Refer to [Installation package to install HertzBeat](./package-deploy) to start with the new installation package and configure the database connection in `application.yml` +## AI Schedule Ownership After Upgrade + +AI conversations without a recorded creator are isolated and do not appear in +any user's conversation list. Scheduled AI SOP tasks are owned by the creator +of their target conversation. During upgrade, schedules without a recorded +creator are disabled. Schedules without a target conversation or whose creator +does not match the conversation creator are disabled before they can execute. +The records remain in the database so an administrator can recover them after +verifying the intended owner. + +Ownerless schedules are disabled by the database migration. Missing +conversations and creator mismatches are rechecked and disabled before every +background execution. + +Before re-enabling a legacy schedule: + +1. Back up the metadata database. +2. Verify the owner of the target row in `hzb_ai_conversation`. +3. Set the same verified principal in the conversation and schedule `creator` + columns. +4. Re-enable only the reviewed schedule. + +Do not assign all legacy rows to a shared account. A schedule is executed only +while its stored creator still owns the target conversation; ownership +mismatches are disabled automatically. + **HAVE FUN**