fix(ai): improve SOP schedule error handling (#4254)

Signed-off-by: yuluo-yx <yuluo08290126@gmail.com>

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

@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import tools.jackson.core.type.TypeReference;
 
 /**
  * Scheduled executor that checks for due SOP schedules and executes them.
@@ -82,7 +83,11 @@ public class SopScheduleExecutor {
             log.info("Found {} due schedules to execute", dueSchedules.size());
             
             for (SopSchedule schedule : dueSchedules) {
-                executeSchedule(schedule);
+                try {
+                    executeSchedule(schedule);
+                } catch (Exception e) {
+                    log.error("Unexpected error processing scheduled SOP {}", schedule.getId(), e);
+                }
             }
         } catch (Exception e) {
             log.error("Error checking due schedules", e);
@@ -108,11 +113,12 @@ public class SopScheduleExecutor {
             // Parse parameters
             Map<String, Object> params = new HashMap<>();
             if (schedule.getSopParams() != null && !schedule.getSopParams().isEmpty()) {
-                try {
-                    params = JsonUtil.fromJson(schedule.getSopParams(), Map.class);
-                } catch (Exception e) {
-                    log.warn("Failed to parse SOP params: {}", schedule.getSopParams());
+                Map<String, Object> parsedParams = JsonUtil.fromJson(
+                        schedule.getSopParams(), new TypeReference<>() {});
+                if (parsedParams == null) {
+                    throw new IllegalArgumentException("SOP schedule parameters must be a valid JSON object");
                 }
+                params = parsedParams;
             }
             
             // Execute SOP

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/sop/registry/SopToolCallback.java

@@ -109,7 +109,7 @@ public class SopToolCallback implements ToolCallback {
         String defaultValue = parameter.getDefaultValue();
         try {
             return switch (mapType(parameter.getType())) {
-                case "boolean" -> Boolean.valueOf(defaultValue);
+                case "boolean" -> parseBooleanDefault(parameter, defaultValue);
                 case "integer" -> Long.valueOf(defaultValue);
                 case "number" -> Double.valueOf(defaultValue);
                 default -> defaultValue;
@@ -119,6 +119,18 @@ public class SopToolCallback implements ToolCallback {
         }
     }
 
+    private boolean parseBooleanDefault(SopParameter parameter, String defaultValue) {
+        String normalizedValue = defaultValue.trim();
+        if ("true".equalsIgnoreCase(normalizedValue)) {
+            return true;
+        }
+        if ("false".equalsIgnoreCase(normalizedValue)) {
+            return false;
+        }
+        throw new IllegalArgumentException(
+                "Invalid boolean default value for SOP parameter: " + parameter.getName());
+    }
+
     private String mapType(String type) {
         if (type == null) {
             return "string";

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/schedule/SopScheduleExecutorTest.java

@@ -0,0 +1,111 @@
+/*
+ * 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.schedule;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import org.apache.hertzbeat.ai.dao.ChatMessageDao;
+import org.apache.hertzbeat.ai.service.SopScheduleService;
+import org.apache.hertzbeat.ai.sop.engine.SopEngine;
+import org.apache.hertzbeat.ai.sop.model.SopDefinition;
+import org.apache.hertzbeat.ai.sop.model.SopResult;
+import org.apache.hertzbeat.ai.sop.registry.SkillRegistry;
+import org.apache.hertzbeat.common.entity.ai.ChatMessage;
+import org.apache.hertzbeat.common.entity.ai.SopSchedule;
+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;
+
+/**
+ * Verifies that due SOP schedules are isolated from each other and reject invalid parameters.
+ */
+@ExtendWith(MockitoExtension.class)
+class SopScheduleExecutorTest {
+
+    @Mock
+    private SopScheduleService scheduleService;
+
+    @Mock
+    private SopEngine sopEngine;
+
+    @Mock
+    private SkillRegistry skillRegistry;
+
+    @Mock
+    private ChatMessageDao chatMessageDao;
+
+    private SopScheduleExecutor executor;
+
+    @BeforeEach
+    void setUp() {
+        executor = new SopScheduleExecutor(scheduleService, sopEngine, skillRegistry, chatMessageDao);
+    }
+
+    @Test
+    void checkShouldContinueAfterOneScheduleFailsToUpdate() {
+        SopSchedule first = schedule(1L, null);
+        SopSchedule second = schedule(2L, null);
+        SopDefinition definition = SopDefinition.builder().name("daily_inspection").build();
+        SopResult result = SopResult.builder()
+                .status("SUCCESS")
+                .content("ok")
+                .build();
+        when(scheduleService.getDueSchedules()).thenReturn(List.of(first, second));
+        when(skillRegistry.getSkill("daily_inspection")).thenReturn(definition);
+        when(sopEngine.executeSync(any(SopDefinition.class), anyMap())).thenReturn(result);
+        doThrow(new IllegalStateException("database unavailable"))
+                .when(scheduleService).updateAfterExecution(1L);
+
+        executor.checkAndExecuteDueSchedules();
+
+        verify(sopEngine, times(2)).executeSync(any(SopDefinition.class), anyMap());
+        verify(scheduleService).updateAfterExecution(2L);
+    }
+
+    @Test
+    void checkShouldRejectInvalidScheduleParameters() {
+        SopSchedule schedule = schedule(1L, "not-json");
+        when(scheduleService.getDueSchedules()).thenReturn(List.of(schedule));
+        when(skillRegistry.getSkill("daily_inspection"))
+                .thenReturn(SopDefinition.builder().name("daily_inspection").build());
+
+        executor.checkAndExecuteDueSchedules();
+
+        verifyNoInteractions(sopEngine);
+        verify(chatMessageDao).save(any(ChatMessage.class));
+        verify(scheduleService).updateAfterExecution(1L);
+    }
+
+    private SopSchedule schedule(Long id, String params) {
+        return SopSchedule.builder()
+                .id(id)
+                .conversationId(10L)
+                .sopName("daily_inspection")
+                .sopParams(params)
+                .build();
+    }
+}

hertzbeat-ai/src/test/java/org/apache/hertzbeat/ai/sop/registry/SopToolCallbackTest.java

@@ -76,6 +76,23 @@ class SopToolCallbackTest {
         assertThrows(IllegalArgumentException.class, () -> callback.call("not-json"));
     }
 
+    @Test
+    void schemaShouldRejectInvalidBooleanDefault() {
+        SopParameter enabled = SopParameter.builder()
+                .name("enabled")
+                .type("boolean")
+                .defaultValue("yes")
+                .build();
+        SopDefinition definition = SopDefinition.builder()
+                .name("invalid-default")
+                .description("包含非法布尔默认值")
+                .parameters(List.of(enabled))
+                .build();
+
+        assertThrows(IllegalArgumentException.class,
+                () -> new SopToolCallback(definition, new RecordingEngine()));
+    }
+
     private SopDefinition definition() {
         SopParameter monitorId = SopParameter.builder()
                 .name("monitorId")