diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java index db0a2ae8..34b9f0a3 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java @@ -1410,6 +1410,17 @@ public class ConversationWindowManager { ChatResponse response = chatModel.call(new Prompt(promptMessages, options)); if (response != null && response.getResult() != null && response.getResult().getOutput() != null) { + String finishReason = response.getResult().getMetadata() != null + ? response.getResult().getMetadata().getFinishReason() : null; + if ("length".equalsIgnoreCase(finishReason)) { + // A non-empty response can still be structurally incomplete + // when the provider exhausts max tokens. Persisting it would + // poison every later iterative summary. + log.warn("[ConversationWindow] LLM 摘要因 token 上限被截断，丢弃结果, conv={}", + conversationId); + setSummaryCooldown(conversationId); + return null; + } String summary = response.getResult().getOutput().getText(); if (summary != null && !summary.isBlank()) { // 成功：保存摘要供下次迭代更新，清除冷却 diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java index a74d0fc9..437f1c3d 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/AbstractCronResultDelivery.java @@ -96,36 +96,44 @@ public abstract class AbstractCronResultDelivery implements CronResultDelivery { // ---------- SQL state-machine helpers ---------- /** - * Atomic SQL CAS: transition delivery_status from {@code NONE} or - * {@code PENDING} → {@code PENDING}. Returns true iff this instance won - * the race. NONE-eligibility lets fresh runs claim without a separate - * "first-time" branch; PENDING-eligibility covers the rare same-instance - * retry inside the listener. + * Atomic SQL CAS: transition delivery_status from {@code NONE} (or legacy + * {@code NULL}) to {@code PENDING}. An already-pending row is owned by the + * worker that claimed it and must never be claimable again. * *

SQL semantics gotcha: {@code IN (...)} never matches NULL. Legacy * rows from before V57 (pre-RFC) may have null delivery_status, so the - * predicate explicitly tests {@code IS NULL OR IN (NONE, PENDING)} via + * predicate explicitly tests {@code IS NULL OR = NONE} via * a nested OR group rather than putting null inside the IN list. */ private boolean claimRun(CronJobRunEntity run) { return runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) .and(w -> w.isNull(CronJobRunEntity::getDeliveryStatus) - .or().in(CronJobRunEntity::getDeliveryStatus, "NONE", "PENDING")) + .or().eq(CronJobRunEntity::getDeliveryStatus, "NONE")) .set(CronJobRunEntity::getDeliveryStatus, "PENDING")) == 1; } private void markDelivered(CronJobRunEntity run, DeliveryOutcome o) { - runMapper.update(null, new LambdaUpdateWrapper() + int updated = runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) + .eq(CronJobRunEntity::getDeliveryStatus, "PENDING") .set(CronJobRunEntity::getDeliveryStatus, "DELIVERED") .set(CronJobRunEntity::getDeliveryTarget, o.target())); + if (updated == 0) { + log.warn("[CronDelivery] Run {} lost its PENDING fence before success was persisted", + run.getId()); + } } private void markNotDelivered(CronJobRunEntity run, Exception e) { - runMapper.update(null, new LambdaUpdateWrapper() + int updated = runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) + .eq(CronJobRunEntity::getDeliveryStatus, "PENDING") .set(CronJobRunEntity::getDeliveryStatus, "NOT_DELIVERED") .set(CronJobRunEntity::getDeliveryError, StrUtil.maxLength(e.getMessage(), 500))); + if (updated == 0) { + log.warn("[CronDelivery] Run {} lost its PENDING fence before failure was persisted", + run.getId()); + } } } diff --git a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java index 7d5efb41..8eed2c05 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/delivery/CronRunStaleCleanup.java @@ -21,7 +21,7 @@ import java.time.LocalDateTime; * {@code NOT_DELIVERED} with {@code stale-pending-timeout} reason. * Covers listener crashes / OOMs / forced kills after a successful * {@code claimRun()} but before {@code markDelivered}. - *

  • {@code status='running'} older than 30 min → mark {@code failed} + *
  • {@code status='running'} without a heartbeat for 2 min → mark {@code failed} * with {@code stale-running-timeout}. Covers * {@code CronJobLifecycleService.markRunFailed()} itself failing under * DB jitter (the LLM call already terminated by then).
    • @@ -38,7 +38,7 @@ public class CronRunStaleCleanup { private final CronJobRunMapper runMapper; private static final Duration DELIVERY_STALE = Duration.ofMinutes(15); - private static final Duration RUN_STALE = Duration.ofMinutes(30); + private static final Duration RUN_STALE = Duration.ofMinutes(2); /** * RFC-03 Lane G2: in a multi-instance deployment, the sweep is purely @@ -62,7 +62,7 @@ public class CronRunStaleCleanup { int staleRunning = runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getStatus, "running") - .lt(CronJobRunEntity::getStartedAt, now.minus(RUN_STALE)) + .apply("COALESCE(heartbeat_at, started_at) < {0}", now.minus(RUN_STALE)) .set(CronJobRunEntity::getStatus, "failed") .set(CronJobRunEntity::getFinishedAt, now) .set(CronJobRunEntity::getErrorMessage, "stale-running-timeout")); diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java index 5fa8d813..b90f6393 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java @@ -71,7 +71,9 @@ public class CronJobLifecycleService { run.setConversationId(conversationId); run.setStatus("running"); run.setTriggerType(triggerType != null ? triggerType : "scheduled"); - run.setStartedAt(LocalDateTime.now()); + LocalDateTime now = LocalDateTime.now(); + run.setStartedAt(now); + run.setHeartbeatAt(now); run.setDeliveryStatus("NONE"); runMapper.insert(run); @@ -130,6 +132,7 @@ public class CronJobLifecycleService { String message = error != null && error.getMessage() != null ? error.getMessage() : "unknown error"; runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) + .eq(CronJobRunEntity::getStatus, "running") .set(CronJobRunEntity::getStatus, "failed") .set(CronJobRunEntity::getFinishedAt, LocalDateTime.now()) .set(CronJobRunEntity::getErrorMessage, StrUtil.maxLength(message, 1000))); @@ -148,7 +151,9 @@ public class CronJobLifecycleService { run.setConversationId(null); run.setStatus("running"); run.setTriggerType(triggerType != null ? triggerType : "scheduled"); - run.setStartedAt(LocalDateTime.now()); + LocalDateTime now = LocalDateTime.now(); + run.setStartedAt(now); + run.setHeartbeatAt(now); run.setDeliveryStatus("NONE"); runMapper.insert(run); return run; @@ -162,12 +167,16 @@ public class CronJobLifecycleService { */ @Transactional(propagation = Propagation.REQUIRES_NEW) public void markRunSucceeded(CronJobRunEntity run, String description) { - runMapper.update(null, new LambdaUpdateWrapper() + int updated = runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) + .eq(CronJobRunEntity::getStatus, "running") .set(CronJobRunEntity::getStatus, "succeeded") .set(CronJobRunEntity::getFinishedAt, LocalDateTime.now()) .set(CronJobRunEntity::getErrorMessage, description != null ? StrUtil.maxLength(description, 1000) : null)); + if (updated == 0) { + log.warn("[CronLifecycle] Run {} lost its running fence before system completion", run.getId()); + } } /** @@ -205,11 +214,17 @@ public class CronJobLifecycleService { int totalTokens = chatResult != null ? chatResult.promptTokens() + chatResult.completionTokens() : 0; - runMapper.update(null, new LambdaUpdateWrapper() + int updated = runMapper.update(null, new LambdaUpdateWrapper() .eq(CronJobRunEntity::getId, run.getId()) + .eq(CronJobRunEntity::getStatus, "running") .set(CronJobRunEntity::getStatus, "succeeded") .set(CronJobRunEntity::getFinishedAt, LocalDateTime.now()) .set(totalTokens > 0, CronJobRunEntity::getTokenUsage, totalTokens)); + if (updated == 0) { + log.warn("[CronLifecycle] Run {} lost its running fence before completion; dropping late result", + run.getId()); + return; + } if (silent) { // No-op run: persist a short marker so the tasks_ diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java index 796eb306..11004e11 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java @@ -42,6 +42,7 @@ import vip.mate.wiki.service.WikiProcessingService; public class CronJobRunner { private final CronJobLifecycleService lifecycle; + private final CronRunHeartbeatService heartbeat; private final AgentService agentService; private final CronChatOriginFactory originFactory; private final vip.mate.cron.CronConversationResolver conversationResolver; @@ -144,14 +145,16 @@ public class CronJobRunner { try { ChatOrigin origin = originFactory.from( job, conversationId, started.originMessageId()); - chatResult = runAgent(job, userMessage, origin, conversationId); + try (CronRunHeartbeatService.Lease ignored = heartbeat.begin(run.getId())) { + chatResult = runAgent(job, userMessage, origin, conversationId); + } result = new AssistantMessage(chatResult.content()); } catch (Exception e) { log.error("[CronRunner] runAgent failed for job {}: {}", job.getId(), e.getMessage(), e); try { lifecycle.markRunFailed(run, e); } catch (Exception markErr) { - // CronRunStaleCleanup will sweep status='running' rows older than 30 min. + // CronRunStaleCleanup will recover a run after its heartbeat expires. log.warn("[CronRunner] markRunFailed itself failed for run {}: {} (stale-cleanup will recover)", run.getId(), markErr.getMessage()); } diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronRunHeartbeatService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronRunHeartbeatService.java new file mode 100644 index 00000000..a0151a4e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronRunHeartbeatService.java @@ -0,0 +1,107 @@ +package vip.mate.cron.service; + +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.time.Clock; +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** Maintains a durable liveness signal while a cron run is inside a long agent call. */ +@Slf4j +@Service +public class CronRunHeartbeatService { + + static final Duration DEFAULT_INTERVAL = Duration.ofSeconds(30); + + private final CronJobRunMapper runMapper; + private final ScheduledExecutorService scheduler; + private final Duration interval; + private final Clock clock; + private final boolean ownsScheduler; + + @Autowired + public CronRunHeartbeatService(CronJobRunMapper runMapper) { + this(runMapper, newScheduler(), DEFAULT_INTERVAL, Clock.systemDefaultZone(), true); + } + + CronRunHeartbeatService(CronJobRunMapper runMapper, + ScheduledExecutorService scheduler, + Duration interval, + Clock clock, + boolean ownsScheduler) { + this.runMapper = Objects.requireNonNull(runMapper, "runMapper"); + this.scheduler = Objects.requireNonNull(scheduler, "scheduler"); + this.interval = Objects.requireNonNull(interval, "interval"); + this.clock = Objects.requireNonNull(clock, "clock"); + this.ownsScheduler = ownsScheduler; + if (interval.isZero() || interval.isNegative()) { + throw new IllegalArgumentException("heartbeat interval must be positive"); + } + } + + /** + * Start refreshing one run. The returned lease is idempotent and must be + * closed when the long call exits, including exceptional exits. + */ + public Lease begin(Long runId) { + Objects.requireNonNull(runId, "runId"); + long periodMillis = interval.toMillis(); + ScheduledFuture future = scheduler.scheduleAtFixedRate( + () -> safeTouch(runId), periodMillis, periodMillis, TimeUnit.MILLISECONDS); + AtomicBoolean closed = new AtomicBoolean(); + return () -> { + if (closed.compareAndSet(false, true)) { + future.cancel(false); + } + }; + } + + private void safeTouch(Long runId) { + try { + int updated = runMapper.update(null, new LambdaUpdateWrapper() + .eq(CronJobRunEntity::getId, runId) + .eq(CronJobRunEntity::getStatus, "running") + .set(CronJobRunEntity::getHeartbeatAt, LocalDateTime.now(clock))); + if (updated == 0) { + log.debug("[CronHeartbeat] Run {} is no longer running; heartbeat ignored", runId); + } + } catch (RuntimeException e) { + // ScheduledExecutorService suppresses all later ticks if a task + // escapes with an exception. Keep the liveness loop recoverable. + log.warn("[CronHeartbeat] Failed to refresh run {}: {}", runId, e.getMessage()); + } + } + + @PreDestroy + void shutdown() { + if (ownsScheduler) { + scheduler.shutdownNow(); + } + } + + private static ScheduledExecutorService newScheduler() { + return Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "cron-run-heartbeat"); + thread.setDaemon(true); + return thread; + }); + } + + @FunctionalInterface + public interface Lease extends AutoCloseable { + @Override + void close(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java b/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java index 91dbab9e..5afd2596 100644 --- a/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/dashboard/model/CronJobRunEntity.java @@ -16,6 +16,8 @@ public class CronJobRunEntity { /** scheduled / manual */ private String triggerType; private LocalDateTime startedAt; + /** Last durable liveness signal while the run is executing. */ + private LocalDateTime heartbeatAt; private LocalDateTime finishedAt; private String errorMessage; private Integer tokenUsage; diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V190__cron_run_heartbeat.sql b/mateclaw-server/src/main/resources/db/migration/h2/V190__cron_run_heartbeat.sql new file mode 100644 index 00000000..392c1aac --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V190__cron_run_heartbeat.sql @@ -0,0 +1,6 @@ +-- Durable liveness for long cron runs. Stale cleanup falls back to started_at +-- for pre-migration rows whose heartbeat_at is null. +ALTER TABLE mate_cron_job_run ADD COLUMN IF NOT EXISTS heartbeat_at TIMESTAMP; + +CREATE INDEX IF NOT EXISTS idx_cron_run_status_heartbeat + ON mate_cron_job_run(status, heartbeat_at); diff --git a/mateclaw-server/src/main/resources/db/migration/kingbase/V190__cron_run_heartbeat.sql b/mateclaw-server/src/main/resources/db/migration/kingbase/V190__cron_run_heartbeat.sql new file mode 100644 index 00000000..7290e2ed --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/kingbase/V190__cron_run_heartbeat.sql @@ -0,0 +1,5 @@ +-- Durable liveness for long cron runs (Kingbase/PostgreSQL dialect). +ALTER TABLE mate_cron_job_run ADD COLUMN IF NOT EXISTS heartbeat_at TIMESTAMP; + +CREATE INDEX IF NOT EXISTS idx_cron_run_status_heartbeat + ON mate_cron_job_run(status, heartbeat_at); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V190__cron_run_heartbeat.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V190__cron_run_heartbeat.sql new file mode 100644 index 00000000..d336d786 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V190__cron_run_heartbeat.sql @@ -0,0 +1,16 @@ +-- Durable liveness for long cron runs (MySQL dialect). +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' + AND COLUMN_NAME = 'heartbeat_at'); +SET @s := IF(@c = 0, + 'ALTER TABLE mate_cron_job_run ADD COLUMN heartbeat_at TIMESTAMP NULL', + 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; + +SET @c := (SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mate_cron_job_run' + AND INDEX_NAME = 'idx_cron_run_status_heartbeat'); +SET @s := IF(@c = 0, + 'CREATE INDEX idx_cron_run_status_heartbeat ON mate_cron_job_run(status, heartbeat_at)', + 'SELECT 1'); +PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt; diff --git a/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSummaryBudgetTest.java b/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSummaryBudgetTest.java index 5c25415d..2c3bc2d8 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSummaryBudgetTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSummaryBudgetTest.java @@ -87,6 +87,28 @@ class ConversationWindowManagerSummaryBudgetTest { "iterative-update: literal placeholder must not leak into the SystemMessage (the bug regression guard)"); } + @Test + @DisplayName("Token-limit summaries are rejected and never become iterative state") + void lengthFinishReasonRejectsTruncatedSummary() throws Exception { + stubResponse("TRUNCATED BUT NONEMPTY", "length"); + + String result = invokeGenerateSummary("conv-truncated", null); + + assertNull(result); + assertFalse(previousSummaries().containsKey("conv-truncated")); + } + + @Test + @DisplayName("Normally stopped summaries remain accepted") + void stopFinishReasonAcceptsCompleteSummary() throws Exception { + stubResponse("COMPLETE SUMMARY", "stop"); + + String result = invokeGenerateSummary("conv-complete", null); + + assertEquals("COMPLETE SUMMARY", result); + assertEquals("COMPLETE SUMMARY", previousSummaries().get("conv-complete")); + } + /** * Reflectively invoke the private {@code generateSummary} method and capture * the {@link Prompt} sent to the mocked {@link ChatModel}. @@ -106,4 +128,26 @@ class ConversationWindowManagerSummaryBudgetTest { org.mockito.Mockito.verify(chatModel).call(captor.capture()); return captor.getValue(); } + + private String invokeGenerateSummary(String conversationId, String memoryExtra) throws Exception { + List oldMessages = List.of(new UserMessage("hello"), new UserMessage("world")); + Method method = ConversationWindowManager.class.getDeclaredMethod( + "generateSummary", List.class, ChatModel.class, String.class, int.class, String.class); + method.setAccessible(true); + return (String) method.invoke(manager, oldMessages, chatModel, conversationId, 1500, memoryExtra); + } + + private void stubResponse(String text, String finishReason) { + Generation generation = new Generation( + new org.springframework.ai.chat.messages.AssistantMessage(text), + ChatGenerationMetadata.builder().finishReason(finishReason).build()); + when(chatModel.call(any(Prompt.class))).thenReturn(new ChatResponse(List.of(generation))); + } + + @SuppressWarnings("unchecked") + private ConcurrentHashMap previousSummaries() throws Exception { + Field field = ConversationWindowManager.class.getDeclaredField("previousSummaries"); + field.setAccessible(true); + return (ConcurrentHashMap) field.get(manager); + } } diff --git a/mateclaw-server/src/test/java/vip/mate/cron/delivery/AbstractCronResultDeliveryTest.java b/mateclaw-server/src/test/java/vip/mate/cron/delivery/AbstractCronResultDeliveryTest.java index 49c8155a..9daaa2df 100644 --- a/mateclaw-server/src/test/java/vip/mate/cron/delivery/AbstractCronResultDeliveryTest.java +++ b/mateclaw-server/src/test/java/vip/mate/cron/delivery/AbstractCronResultDeliveryTest.java @@ -1,18 +1,21 @@ package vip.mate.cron.delivery; import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.AbstractWrapper; import com.baomidou.mybatisplus.core.conditions.Wrapper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.springframework.ai.chat.messages.AssistantMessage; import vip.mate.cron.model.CronJobEntity; import vip.mate.dashboard.model.CronJobRunEntity; import vip.mate.dashboard.repository.CronJobRunMapper; import java.util.HashSet; +import java.util.List; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -101,6 +104,75 @@ class AbstractCronResultDeliveryTest { verify(runMapper, times(1)).update(any(), any(Wrapper.class)); // only the failed claim } + @Test + void claimRun_acceptsOnlyFreshOrLegacyDeliveryState() { + when(runMapper.update(any(), any(Wrapper.class))).thenReturn(0); + + AbstractCronResultDelivery strategy = new AbstractCronResultDelivery(runMapper) { + @Override public boolean supports(CronJobEntity j) { return true; } + @Override protected DeliveryOutcome doDeliver( + CronJobEntity j, AssistantMessage r, CronJobRunEntity ignored) { + return DeliveryOutcome.delivered("never"); + } + }; + + strategy.deliver(job, new AssistantMessage("hi"), run); + + @SuppressWarnings("rawtypes") + ArgumentCaptor captor = ArgumentCaptor.forClass(Wrapper.class); + verify(runMapper).update(isNull(), captor.capture()); + Wrapper claim = captor.getValue(); + assertTrue(claim.getSqlSegment().contains("delivery_status IS NULL")); + assertTrue(whereValues(claim).contains("NONE")); + assertFalse(whereValues(claim).contains("PENDING"), + "an already-PENDING delivery must not be claimable again"); + } + + @Test + void deliveredTerminalWrite_requiresPendingFence() { + when(runMapper.update(any(), any(Wrapper.class))).thenReturn(1, 1); + + AbstractCronResultDelivery strategy = new AbstractCronResultDelivery(runMapper) { + @Override public boolean supports(CronJobEntity j) { return true; } + @Override protected DeliveryOutcome doDeliver( + CronJobEntity j, AssistantMessage r, CronJobRunEntity ignored) { + return DeliveryOutcome.delivered("user-x"); + } + }; + + strategy.deliver(job, new AssistantMessage("hi"), run); + + @SuppressWarnings("rawtypes") + ArgumentCaptor captor = ArgumentCaptor.forClass(Wrapper.class); + verify(runMapper, times(2)).update(isNull(), captor.capture()); + List writes = captor.getAllValues(); + assertTrue(whereValues(writes.get(1)).contains("PENDING"), + "late success must not overwrite a stale-cleanup terminal state"); + } + + @Test + void failedTerminalWrite_requiresPendingFence() { + when(runMapper.update(any(), any(Wrapper.class))).thenReturn(1, 1); + + AbstractCronResultDelivery strategy = new AbstractCronResultDelivery(runMapper) { + @Override public boolean supports(CronJobEntity j) { return true; } + @Override protected DeliveryOutcome doDeliver( + CronJobEntity j, AssistantMessage r, CronJobRunEntity ignored) { + throw new IllegalStateException("delivery failed"); + } + }; + + assertThrows(IllegalStateException.class, + () -> strategy.deliver(job, new AssistantMessage("hi"), run)); + + @SuppressWarnings("rawtypes") + ArgumentCaptor captor = ArgumentCaptor.forClass(Wrapper.class); + verify(runMapper, times(2)).update(isNull(), captor.capture()); + List writes = captor.getAllValues(); + assertTrue(whereValues(writes.get(1)).contains("PENDING"), + "late failure must not overwrite a stale-cleanup terminal state"); + } + @Test void deliver_doDeliverThrows_marksNotDeliveredAndRethrows() { // Claim returns 1, then markNotDelivered returns 1 @@ -173,4 +245,16 @@ class AbstractCronResultDeliveryTest { pool.shutdownNow(); } } + + private static Set whereValues(Wrapper rawWrapper) { + AbstractWrapper wrapper = (AbstractWrapper) rawWrapper; + String where = wrapper.getSqlSegment(); + Set values = new HashSet<>(); + wrapper.getParamNameValuePairs().forEach((key, value) -> { + if (where.contains(key)) { + values.add(value); + } + }); + return values; + } } diff --git a/mateclaw-server/src/test/java/vip/mate/cron/delivery/CronRunStaleCleanupTest.java b/mateclaw-server/src/test/java/vip/mate/cron/delivery/CronRunStaleCleanupTest.java new file mode 100644 index 00000000..86ba310c --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/cron/delivery/CronRunStaleCleanupTest.java @@ -0,0 +1,44 @@ +package vip.mate.cron.delivery; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.*; + +class CronRunStaleCleanupTest { + + @BeforeAll + static void initMpLambdaCache() { + MybatisConfiguration cfg = new MybatisConfiguration(); + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(cfg, ""), CronJobRunEntity.class); + } + + @Test + void runningSweep_usesHeartbeatWithLegacyStartedAtFallback() { + CronJobRunMapper mapper = mock(CronJobRunMapper.class); + when(mapper.update(isNull(), any(Wrapper.class))).thenReturn(0); + CronRunStaleCleanup cleanup = new CronRunStaleCleanup(mapper); + + cleanup.sweep(); + + @SuppressWarnings("rawtypes") + ArgumentCaptor captor = ArgumentCaptor.forClass(Wrapper.class); + verify(mapper, times(2)).update(isNull(), captor.capture()); + List writes = captor.getAllValues(); + String runningWhere = writes.get(1).getSqlSegment(); + assertTrue(runningWhere.contains("status"), runningWhere); + assertTrue(runningWhere.contains("COALESCE(heartbeat_at, started_at)"), runningWhere); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobLifecycleFenceTest.java b/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobLifecycleFenceTest.java new file mode 100644 index 00000000..397d9503 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobLifecycleFenceTest.java @@ -0,0 +1,72 @@ +package vip.mate.cron.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.context.ApplicationEventPublisher; +import vip.mate.cron.model.CronJobEntity; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; +import vip.mate.i18n.I18nService; +import vip.mate.memory.event.ConversationCompletionPublisher; +import vip.mate.workspace.conversation.ConversationService; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.*; + +class CronJobLifecycleFenceTest { + + @BeforeAll + static void initMpLambdaCache() { + MybatisConfiguration cfg = new MybatisConfiguration(); + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(cfg, ""), CronJobRunEntity.class); + } + + @Test + void failedTerminalWrite_isFencedByRunningStatus() { + Fixture fixture = new Fixture(); + fixture.service.markRunFailed(run(), new IllegalStateException("failed")); + + @SuppressWarnings("rawtypes") + ArgumentCaptor captor = ArgumentCaptor.forClass(Wrapper.class); + verify(fixture.mapper).update(isNull(), captor.capture()); + assertTrue(captor.getValue().getSqlSegment().contains("status")); + } + + @Test + void lateCompletion_dropsMessagesAndEventsAfterFenceIsLost() { + Fixture fixture = new Fixture(); + when(fixture.mapper.update(isNull(), any(Wrapper.class))).thenReturn(0); + CronJobEntity job = new CronJobEntity(); + job.setId(1L); + + fixture.service.finishRunAndPublish(job, run(), "request", + new AssistantMessage("late result"), "cron-1", false); + + verifyNoInteractions(fixture.conversations, fixture.completionPublisher, fixture.events); + } + + private static CronJobRunEntity run() { + CronJobRunEntity run = new CronJobRunEntity(); + run.setId(42L); + run.setConversationId("cron-1"); + return run; + } + + private static final class Fixture { + private final CronJobRunMapper mapper = mock(CronJobRunMapper.class); + private final ConversationService conversations = mock(ConversationService.class); + private final ConversationCompletionPublisher completionPublisher = + mock(ConversationCompletionPublisher.class); + private final ApplicationEventPublisher events = mock(ApplicationEventPublisher.class); + private final CronJobLifecycleService service = new CronJobLifecycleService( + mapper, conversations, completionPublisher, events, mock(I18nService.class)); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobOriginPropagationTest.java b/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobOriginPropagationTest.java index 1f5e7532..f5cdab56 100644 --- a/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobOriginPropagationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/cron/service/CronJobOriginPropagationTest.java @@ -17,6 +17,7 @@ import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageEntity; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -36,8 +37,9 @@ class CronJobOriginPropagationTest { @Test void lifecycleReturnsThePersistedUserMessageIdWithoutSavingTwice() { ConversationService conversations = mock(ConversationService.class); + CronJobRunMapper runMapper = mock(CronJobRunMapper.class); CronJobLifecycleService lifecycle = new CronJobLifecycleService( - mock(CronJobRunMapper.class), conversations, + runMapper, conversations, mock(ConversationCompletionPublisher.class), mock(ApplicationEventPublisher.class), mock(I18nService.class)); CronJobEntity job = job(); @@ -50,6 +52,8 @@ class CronJobOriginPropagationTest { job, "do work", "scheduled", CONVERSATION_ID); assertEquals(MESSAGE_ID, result.originMessageId()); + assertEquals(result.run().getStartedAt(), result.run().getHeartbeatAt(), + "a new run must be live before the first scheduled heartbeat"); verify(conversations, times(1)).saveMessage(CONVERSATION_ID, "user", "do work"); } @@ -59,6 +63,8 @@ class CronJobOriginPropagationTest { AgentService agentService = mock(AgentService.class); CronChatOriginFactory originFactory = mock(CronChatOriginFactory.class); CronConversationResolver resolver = mock(CronConversationResolver.class); + CronRunHeartbeatService heartbeat = mock(CronRunHeartbeatService.class); + CronRunHeartbeatService.Lease lease = mock(CronRunHeartbeatService.Lease.class); CronJobEntity job = job(); CronJobRunEntity run = new CronJobRunEntity(); run.setId(55L); @@ -68,18 +74,49 @@ class CronJobOriginPropagationTest { when(lifecycle.startRun(job, "do work", "scheduled", CONVERSATION_ID)) .thenReturn(new CronJobLifecycleService.StartResult(run, MESSAGE_ID)); when(originFactory.from(job, CONVERSATION_ID, MESSAGE_ID)).thenReturn(origin); + when(heartbeat.begin(55L)).thenReturn(lease); when(agentService.chatWithUsage(eq(AGENT_ID), anyString(), eq(CONVERSATION_ID), eq(origin))) .thenReturn(AgentService.ChatResult.contentOnly("done")); - CronJobRunner runner = new CronJobRunner(lifecycle, agentService, originFactory, resolver, + CronJobRunner runner = new CronJobRunner(lifecycle, heartbeat, agentService, originFactory, resolver, mock(WikiProcessingService.class), new ObjectMapper()); runner.executeJob(job); verify(originFactory).from(job, CONVERSATION_ID, MESSAGE_ID); + verify(heartbeat).begin(55L); + verify(lease).close(); verify(agentService).chatWithUsage(eq(AGENT_ID), anyString(), eq(CONVERSATION_ID), eq(origin)); verify(agentService, never()).chatWithUsage(eq(AGENT_ID), anyString(), eq(CONVERSATION_ID)); } + @Test + void runnerClosesHeartbeatWhenAgentFails() { + CronJobLifecycleService lifecycle = mock(CronJobLifecycleService.class); + CronRunHeartbeatService heartbeat = mock(CronRunHeartbeatService.class); + CronRunHeartbeatService.Lease lease = mock(CronRunHeartbeatService.Lease.class); + AgentService agentService = mock(AgentService.class); + CronChatOriginFactory originFactory = mock(CronChatOriginFactory.class); + CronConversationResolver resolver = mock(CronConversationResolver.class); + CronJobEntity job = job(); + CronJobRunEntity run = new CronJobRunEntity(); + run.setId(55L); + ChatOrigin origin = ChatOrigin.cron(CONVERSATION_ID, WORKSPACE_ID, null, null, null); + when(resolver.resolve(job)).thenReturn(CONVERSATION_ID); + when(lifecycle.startRun(job, "do work", "scheduled", CONVERSATION_ID)) + .thenReturn(new CronJobLifecycleService.StartResult(run, MESSAGE_ID)); + when(originFactory.from(job, CONVERSATION_ID, MESSAGE_ID)).thenReturn(origin); + when(heartbeat.begin(55L)).thenReturn(lease); + when(agentService.chatWithUsage(eq(AGENT_ID), anyString(), eq(CONVERSATION_ID), eq(origin))) + .thenThrow(new IllegalStateException("provider timeout")); + CronJobRunner runner = new CronJobRunner(lifecycle, heartbeat, agentService, originFactory, resolver, + mock(WikiProcessingService.class), new ObjectMapper()); + + runner.executeJob(job); + + verify(lease).close(); + verify(lifecycle).markRunFailed(eq(run), any(IllegalStateException.class)); + } + private static CronJobEntity job() { CronJobEntity job = new CronJobEntity(); job.setId(JOB_ID); diff --git a/mateclaw-server/src/test/java/vip/mate/cron/service/CronRunHeartbeatServiceTest.java b/mateclaw-server/src/test/java/vip/mate/cron/service/CronRunHeartbeatServiceTest.java new file mode 100644 index 00000000..2c66c1ba --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/cron/service/CronRunHeartbeatServiceTest.java @@ -0,0 +1,86 @@ +package vip.mate.cron.service; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.conditions.AbstractWrapper; +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import vip.mate.dashboard.model.CronJobRunEntity; +import vip.mate.dashboard.repository.CronJobRunMapper; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class CronRunHeartbeatServiceTest { + + @BeforeAll + static void initMpLambdaCache() { + MybatisConfiguration cfg = new MybatisConfiguration(); + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(cfg, ""), CronJobRunEntity.class); + } + + @Test + void begin_schedulesFencedHeartbeat_andLeaseClosesIdempotently() { + CronJobRunMapper mapper = mock(CronJobRunMapper.class); + ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class); + @SuppressWarnings("unchecked") + ScheduledFuture future = mock(ScheduledFuture.class); + ArgumentCaptor task = ArgumentCaptor.forClass(Runnable.class); + doReturn(future).when(scheduler) + .scheduleAtFixedRate(task.capture(), eq(30_000L), eq(30_000L), eq(TimeUnit.MILLISECONDS)); + when(mapper.update(isNull(), any(Wrapper.class))).thenReturn(1); + Clock clock = Clock.fixed(Instant.parse("2026-09-02T08:00:00Z"), ZoneOffset.UTC); + CronRunHeartbeatService service = new CronRunHeartbeatService( + mapper, scheduler, Duration.ofSeconds(30), clock, false); + + CronRunHeartbeatService.Lease lease = service.begin(42L); + task.getValue().run(); + + @SuppressWarnings("rawtypes") + ArgumentCaptor wrapperCaptor = ArgumentCaptor.forClass(Wrapper.class); + verify(mapper).update(isNull(), wrapperCaptor.capture()); + AbstractWrapper wrapper = (AbstractWrapper) wrapperCaptor.getValue(); + Set values = Set.copyOf(wrapper.getParamNameValuePairs().values()); + String where = wrapper.getSqlSegment(); + assertTrue(where.contains("id"), () -> "heartbeat must target one run: " + where); + assertTrue(where.contains("status"), () -> "heartbeat must not revive a terminal run: " + where); + assertTrue(values.contains(LocalDateTime.of(2026, 9, 2, 8, 0)), + () -> "missing fixed heartbeat timestamp in " + values); + + lease.close(); + lease.close(); + verify(future, times(1)).cancel(false); + } + + @Test + void heartbeatFailure_doesNotKillSchedulerTask() { + CronJobRunMapper mapper = mock(CronJobRunMapper.class); + ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class); + @SuppressWarnings("unchecked") + ScheduledFuture future = mock(ScheduledFuture.class); + ArgumentCaptor task = ArgumentCaptor.forClass(Runnable.class); + doReturn(future).when(scheduler) + .scheduleAtFixedRate(task.capture(), anyLong(), anyLong(), any()); + when(mapper.update(isNull(), any(Wrapper.class))).thenThrow(new IllegalStateException("db jitter")); + CronRunHeartbeatService service = new CronRunHeartbeatService( + mapper, scheduler, Duration.ofSeconds(30), Clock.systemUTC(), false); + + service.begin(7L); + + assertDoesNotThrow(() -> task.getValue().run()); + } +}