opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(goal): make long tasks recoverable and pace continuation dispatch

Browse Source
This commit is contained in:
matevip 2026-08-27 03:49:15 -04:00
parent 2090d09704
commit 01ed4a4fcd
32 changed files with 1777 additions and 284 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
View File

@ -44,6 +44,8 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.time.LocalDateTime;
import java.util.UUID;
/**
* Web 渠道聊天接口
@ -67,6 +69,7 @@ public class ChatController {
private final vip.mate.memory.identity.MemoryOwnerResolver memoryOwnerResolver;
private final vip.mate.workspace.core.service.ChatUploadLocationResolver uploadLocationResolver;
private final vip.mate.tool.document.preview.OfficePreviewService officePreviewService;
private final ConversationInputQueueStore inputQueue;
@org.springframework.beans.factory.annotation.Autowired
private ConversationTurnGate turnGate = new ConversationTurnGate();
@ -299,8 +302,8 @@ public class ChatController {
conversationService.getMessageCount(conversationId)));
// deny 是正常 turn 终结，用户可能在 awaiting_approval 阶段排了消息
ChatStreamTracker.CompletionResult denyCr = streamTracker.completeAndConsumeIfLast(conversationId);
if (denyCr.allDone() && denyCr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, denyCr.queuedInput(), username, requestBaseUrl);
if (denyCr.allDone() && hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, username, requestBaseUrl);
} else {
completeEmitterQuietly(emitter, approvalEmitterDone);
}
@ -313,8 +316,8 @@ public class ChatController {
broadcastEvent(conversationId, "done", Map.of("status", "completed"));
// 审批记录被另一个请求消费，但用户可能在等待期间排了消息
ChatStreamTracker.CompletionResult consumedNullCr = streamTracker.completeAndConsumeIfLast(conversationId);
if (consumedNullCr.allDone() && consumedNullCr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, consumedNullCr.queuedInput(), username, requestBaseUrl);
if (consumedNullCr.allDone() && hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, username, requestBaseUrl);
} else {
completeEmitterQuietly(emitter, approvalEmitterDone);
}
@ -418,8 +421,8 @@ public class ChatController {
} finally {
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
if (cr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, cr.queuedInput(), username, requestBaseUrl);
if (hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, username, requestBaseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, approvalEmitterDone);
@ -519,8 +522,8 @@ public class ChatController {
streamTracker.clearInterruptState(conversationId);
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
if (cr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, cr.queuedInput(), username, requestBaseUrl);
if (hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, approvalEmitterDone, username, requestBaseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, approvalEmitterDone);
@ -764,7 +767,7 @@ public class ChatController {
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
// RFC follow-up (2026-04-27): the previous guard
// cr.queuedInput() != null && (isInterruptFollowup || !wasStopped)
// hasQueuedInput(conversationId) && (isInterruptFollowup || !wasStopped)
// dropped legitimate queued messages when the user stopped
// the running turn and then sent a new message via the
// enqueue path (not the interrupt-with-followup path) —
@ -775,8 +778,8 @@ public class ChatController {
// run it" condition; align with them. If the user
// genuinely doesn't want continuation, no message would
// have been in messageQueue to begin with.
if (cr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, emitterDone, cr.queuedInput(), username, requestBaseUrl);
if (hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, emitterDone, username, requestBaseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
// 延迟关闭 emitter，确保最后的事件都已发送
@ -868,9 +871,9 @@ public class ChatController {
streamTracker.clearInterruptState(conversationId);
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
if (cr.queuedInput() != null) {
if (hasQueuedInput(conversationId)) {
// 无论中断类型，都消费排队消息（修复 Disposable 不可用时队列被丢弃的 bug）
startQueuedMessage(conversationId, emitter, emitterDone, cr.queuedInput(), username, requestBaseUrl);
startQueuedMessage(conversationId, emitter, emitterDone, username, requestBaseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, emitterDone);
@ -978,7 +981,7 @@ public class ChatController {
streamTracker.clearInterruptState(conversationId);
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
log.info("SSE doOnError cleanup: conversationId={}, allDone={}, isInterruptFollowup={}, hasQueued={}",
conversationId, cr.allDone(), isInterruptFollowup, cr.queuedInput() != null);
conversationId, cr.allDone(), isInterruptFollowup, hasQueuedInput(conversationId));
if (cr.allDone()) {
// RFC follow-up (2026-04-27): the previous guard
// cr.queuedInput()!=null && !(isUserStop && !isInterruptFollowup)
@ -992,8 +995,8 @@ public class ChatController {
// follow-up. Whoever puts a message in messageQueue means it
// — just run it. Aligns with doOnComplete and the 4 other
// queue-launch sites in this controller.
if (cr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, emitterDone, cr.queuedInput(), username, requestBaseUrl);
if (hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, emitterDone, username, requestBaseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, emitterDone);
@ -1108,17 +1111,24 @@ public class ChatController {
// 判断当前阶段（仅用于 reason 字段，行为对所有阶段一致：仅入队）
boolean isAwaitingApproval = approvalService.findPendingByConversation(conversationId) != null;
// 仅入队、不 dispose。延迟持久化到 startQueuedMessage（让 Asst-N 先在 doOnComplete 落库，
// 否则 listMessages ORDER BY create_time ASC 会把 Q(N+1) 排到 Asst-N 前面）
boolean queued = streamTracker.enqueueMessage(conversationId, message, agentId, false, contentParts);
// Commit the payload before publishing acceptance. The stream tracker is
// only a wake signal; the database row remains authoritative on restart.
var stored = inputQueue.enqueue(conversationId, agentId, username, message, contentParts,
LocalDateTime.now());
boolean queued = streamTracker.notifyQueuedInput(conversationId);
if (!queued) {
inputQueue.cancel(stored.id(), "stream_finished_before_queue_registration",
LocalDateTime.now());
}
log.info("Enqueued follow-up message during running turn: conversationId={}, user={}, queueSize={}, awaitingApproval={}",
conversationId, username, streamTracker.getQueueSize(conversationId), isAwaitingApproval);
conversationId, username, inputQueue.countQueued(conversationId), isAwaitingApproval);
return R.ok(Map.of(
"interrupted", false,
"queued", queued,
"queueSize", streamTracker.getQueueSize(conversationId),
"reason", isAwaitingApproval ? "awaiting_approval" : "queued"
"queueItemId", stored.id().toString(),
"queueSize", inputQueue.countQueued(conversationId),
"reason", queued ? (isAwaitingApproval ? "awaiting_approval" : "queued") : "no_active_stream"
));
}
@ -1424,21 +1434,30 @@ public class ChatController {
private Boolean regenerate;
}
/**
* 自动启动排队消息（interrupt-with-followup 或自然完成后的续跑逻辑）。
* 接受由 {@link ChatStreamTracker#completeAndConsumeIfLast} 预先消费的 QueuedInput 快照。
* 快照已脱离 RunState 生命周期，不受后续 complete/register 影响。
* 支持链式续跑：queued stream 自身完成时也通过 completeAndConsumeIfLast 检查并递归调用。
*/
/** Claims and starts the next durable input after the current stream finishes. */
private void startQueuedMessage(String conversationId, SseEmitter emitter, AtomicBoolean emitterDone,
ChatStreamTracker.QueuedInput preConsumedInput, String requesterId,
String baseUrl) {
String requesterId, String baseUrl) {
String queueClaimId = UUID.randomUUID().toString();
ConversationInputQueueStore.QueuedInput preConsumedInput = inputQueue
.claimNext(conversationId, queueClaimId, LocalDateTime.now())
.orElse(null);
if (preConsumedInput == null) {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, emitterDone);
return;
}
Long agentId = preConsumedInput.agentId() != null ? preConsumedInput.agentId() : 1L;
var queuedConversation = conversationService.findByConversationId(conversationId);
if (queuedConversation == null || !agentId.equals(queuedConversation.getAgentId())) {
inputQueue.release(preConsumedInput.id(), queueClaimId, LocalDateTime.now());
broadcastEvent(conversationId, "warning", Map.of(
"message", "排队消息对应的助手已变化，请确认后重试"));
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, emitterDone);
return;
}
// Rate Limit 防护：如果上一轮以 rate limit 错误结束，不立即续跑排队消息（必然再次 429）。
// 改为持久化用户消息 + 通知前端"稍后重试"，避免连锁 429 浪费配额。
String lastMessage = conversationService.getLastMessage(conversationId);
@ -1446,11 +1465,14 @@ public class ChatController {
|| lastMessage.contains("429") || lastMessage.contains("速率限制"))) {
log.warn("Skipping queued message after rate limit error: conversationId={}, lastMessage={}",
conversationId, lastMessage.substring(0, Math.min(50, lastMessage.length())));
// 持久化用户消息不丢失
if (preConsumedInput.message() != null && !preConsumedInput.message().isBlank()
&& !preConsumedInput.persisted()) {
conversationService.saveMessage(conversationId, "user", preConsumedInput.message());
if (preConsumedInput.persistedMessageId() == null) {
MessageEntity saved = conversationService.saveMessage(conversationId, "user",
preConsumedInput.message(), preConsumedInput.contentParts(), "queued");
if (saved != null) {
inputQueue.bindMessage(preConsumedInput.id(), queueClaimId, saved.getId(), LocalDateTime.now());
}
}
inputQueue.consume(preConsumedInput.id(), queueClaimId, LocalDateTime.now());
broadcastEvent(conversationId, "warning", Map.of(
"message", "上一轮请求触发了频率限制，排队消息已保存，请稍后重新发送"));
broadcastEvent(conversationId, "done", Map.of("status", "rate_limited"));
@ -1460,18 +1482,26 @@ public class ChatController {
}
String queuedMessage = preConsumedInput.message();
Long agentId = preConsumedInput.agentId() != null ? preConsumedInput.agentId() : 1L;
log.info("Starting queued message: conversationId={}, agentId={}, message={}",
conversationId, agentId, queuedMessage.substring(0, Math.min(30, queuedMessage.length())));
conversationId, agentId, queuedMessage == null ? "" : queuedMessage.substring(0, Math.min(30, queuedMessage.length())));
// 持久化排队的用户消息（含 contentParts；幂等：如果 /interrupt 已提前持久化则跳过）。
// 这里持久化是为了确保 user 消息在 assistant 消息（doOnError/doOnCancel 已写入）之后落库，
// 让 listMessages ORDER BY create_time ASC 后顺序正确：Q1 → Asst1 → Q2 → Asst2。
Long queuedOriginMessageId = null;
if (queuedMessage != null && !queuedMessage.isBlank() && !preConsumedInput.persisted()) {
Long queuedOriginMessageId = preConsumedInput.persistedMessageId();
if (queuedOriginMessageId == null) {
MessageEntity savedUser = conversationService.saveMessage(conversationId, "user", queuedMessage,
preConsumedInput.contentParts(), "queued");
queuedOriginMessageId = savedUser == null ? null : savedUser.getId();
if (queuedOriginMessageId == null
|| !inputQueue.bindMessage(preConsumedInput.id(), queueClaimId,
queuedOriginMessageId, LocalDateTime.now())) {
inputQueue.release(preConsumedInput.id(), queueClaimId, LocalDateTime.now());
throw new IllegalStateException("Queued input could not be bound to its persisted message");
}
}
if (!inputQueue.consume(preConsumedInput.id(), queueClaimId, LocalDateTime.now())) {
throw new IllegalStateException("Queued input claim was lost before execution");
}
// 广播 queued_input_started 事件
@ -1558,9 +1588,9 @@ public class ChatController {
} finally {
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
if (cr.queuedInput() != null) {
if (hasQueuedInput(conversationId)) {
// 链式续跑：queued stream 期间又排了新消息
startQueuedMessage(conversationId, emitter, emitterDone, cr.queuedInput(), requesterId, baseUrl);
startQueuedMessage(conversationId, emitter, emitterDone, requesterId, baseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
sseExecutor.execute(() -> {
@ -1604,8 +1634,8 @@ public class ChatController {
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
if (cr.queuedInput() != null) {
startQueuedMessage(conversationId, emitter, emitterDone, cr.queuedInput(), requesterId, baseUrl);
if (hasQueuedInput(conversationId)) {
startQueuedMessage(conversationId, emitter, emitterDone, requesterId, baseUrl);
} else {
conversationService.updateStreamStatus(conversationId, "idle");
completeEmitterQuietly(emitter, emitterDone);
@ -1618,6 +1648,10 @@ public class ChatController {
() -> emergencySaveAccumulator(conversationId, accumulator));
}
private boolean hasQueuedInput(String conversationId) {
return inputQueue.countQueued(conversationId) > 0;
}
/**
* Terminal error path for requests rejected before a stream is registered:
* emit an {@code error} + terminal {@code done} pair and complete the

72
mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
View File

@ -190,8 +190,8 @@ public class ChatStreamTracker {
/** 等待原因（审批等待时有值） */
volatile String waitingReason;
/** 排队的用户消息队列（支持多条排队消息，按序消费） */
final java.util.Queue<QueuedInput> messageQueue = new java.util.concurrent.ConcurrentLinkedQueue<>();
/** Wake signal only; queued input payloads live in the database. */
final AtomicBoolean queuedInputPending = new AtomicBoolean(false);
/**
* Emergency save callback registered by the SSE chain owner (ChatController).
@ -500,18 +500,7 @@ public class ChatStreamTracker {
}
if (current.done) {
stopHeartbeat(current);
RunState nextState = new RunState(id);
int carried = 0;
QueuedInput queued;
while ((queued = current.messageQueue.poll()) != null) {
nextState.messageQueue.offer(queued);
carried++;
}
if (carried > 0) {
log.info("[ChatStreamTracker] Carried {} queued message(s) into next run: {}",
carried, id);
}
return nextState;
return new RunState(id);
}
// Registration is a fresh lifecycle entrance. Refresh every
// stale-run input while holding the same lock cleanup uses to
@ -1296,10 +1285,8 @@ public class ChatStreamTracker {
}
}
/**
* 完成结果：包含是否全部完成、排队消息快照
*/
public record CompletionResult(boolean allDone, QueuedInput queuedInput) {}
/** Completion result for the current in-memory stream generation. */
public record CompletionResult(boolean allDone) {}
/**
* 标记一个 Flux 完成。仅在所有 Flux 都完成时才真正移除 RunState。
@ -1365,22 +1352,19 @@ public class ChatStreamTracker {
public CompletionResult completeAndConsumeIfLast(String conversationId) {
RunState state = runs.get(conversationId);
if (state == null) {
return new CompletionResult(true, null);
return new CompletionResult(true);
}
QueuedInput consumed = null;
ScheduledFuture<?> oldHeartbeat;
synchronized (state.lock) {
if (!isCurrent(state)) {
return new CompletionResult(false, null);
return new CompletionResult(false);
}
state.activeFluxCount = Math.max(0, state.activeFluxCount - 1);
if (state.activeFluxCount > 0) {
log.debug("Stream partially completed: {} (remaining flux={}, queuePreserved={})",
conversationId, state.activeFluxCount, !state.messageQueue.isEmpty());
return new CompletionResult(false, null);
log.debug("Stream partially completed: {} (remaining flux={}, queuedInputPending={})",
conversationId, state.activeFluxCount, state.queuedInputPending.get());
return new CompletionResult(false);
}
// 最后一个 Flux：在同一个锁内消费排队消息（取队首）
consumed = state.messageQueue.poll();
state.done = true;
state.cancellationHooks.clear();
state.termination.complete(null);
@ -1392,9 +1376,9 @@ public class ChatStreamTracker {
if (oldHeartbeat != null) {
oldHeartbeat.cancel(false);
}
log.debug("Stream fully completed: {} (hasQueuedSnapshot={}, kept in map for {}ms reconnect window)",
conversationId, consumed != null, DONE_RETENTION_MS);
return new CompletionResult(true, consumed);
log.debug("Stream fully completed: {} (queuedInputPending={}, kept in map for {}ms reconnect window)",
conversationId, state.queuedInputPending.get(), DONE_RETENTION_MS);
return new CompletionResult(true);
}
/**
@ -1497,7 +1481,7 @@ public class ChatStreamTracker {
"currentPhase", safe(state.currentPhase),
"waitingReason", safe(state.waitingReason),
"runningToolName", safe(state.runningToolName),
"queueLength", state.messageQueue.size(),
"queueLength", state.queuedInputPending.get() ? 1 : 0,
"timestamp", System.currentTimeMillis()
));
} catch (Exception e) {
@ -1642,8 +1626,7 @@ public class ChatStreamTracker {
synchronized (state.lock) {
Disposable d = state.disposable;
canInterrupt = d != null && !d.isDisposed();
// 无论是否可中断，都入队（支持多条排队消息）
state.messageQueue.offer(new QueuedInput(queuedMessage, agentId, persisted, contentParts));
state.queuedInputPending.set(true);
if (canInterrupt) {
state.interruptType = InterruptType.USER_INTERRUPT_WITH_FOLLOWUP;
state.stopRequested.set(true);
@ -1710,7 +1693,7 @@ public class ChatStreamTracker {
if (state == null || state.done) {
return false;
}
state.messageQueue.offer(new QueuedInput(message, agentId, persisted, contentParts));
state.queuedInputPending.set(true);
// broadcast 在锁外
try {
String json = objectMapper.writeValueAsString(Map.of(
@ -1740,9 +1723,7 @@ public class ChatStreamTracker {
* 从队列头部取出一条消息。
*/
public QueuedInput consumeQueuedInput(String conversationId) {
RunState state = runs.get(conversationId);
if (state == null) return null;
return state.messageQueue.poll();
return null;
}
/**
@ -1786,7 +1767,7 @@ public class ChatStreamTracker {
*/
public boolean hasQueuedMessage(String conversationId) {
RunState state = runs.get(conversationId);
return state != null && !state.messageQueue.isEmpty();
return state != null && state.queuedInputPending.get();
}
/**
@ -1794,7 +1775,20 @@ public class ChatStreamTracker {
*/
public int getQueueSize(String conversationId) {
RunState state = runs.get(conversationId);
return state != null ? state.messageQueue.size() : 0;
return state != null && state.queuedInputPending.get() ? 1 : 0;
}
/** Notify the live stream that durable queued input is ready to consume. */
public boolean notifyQueuedInput(String conversationId) {
RunState state = runs.get(conversationId);
if (state == null || state.done) return false;
state.queuedInputPending.set(true);
return true;
}
boolean hasQueuedInputNotification(String conversationId) {
RunState state = runs.get(conversationId);
return state != null && state.queuedInputPending.get();
}
// ===== Approval idempotency =====
@ -2224,7 +2218,7 @@ public class ChatStreamTracker {
int queue;
synchronized (s.lock) {
subs = s.subscribers.size();
queue = s.messageQueue.size();
queue = s.queuedInputPending.get() ? 1 : 0;
}
out.add(new RunSnapshot(
s.conversationId,

187
mateclaw-server/src/main/java/vip/mate/channel/web/ConversationInputQueueStore.java Normal file
View File

@ -0,0 +1,187 @@
package vip.mate.channel.web;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import vip.mate.workspace.conversation.model.MessageContentPart;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;
/** Database-backed FIFO for user input accepted while a conversation is busy. */
@Repository
public class ConversationInputQueueStore {
private static final TypeReference<List<MessageContentPart>> PARTS_TYPE = new TypeReference<>() {};
private final JdbcTemplate jdbc;
private final ObjectMapper mapper;
public ConversationInputQueueStore(JdbcTemplate jdbc, ObjectMapper mapper) {
this.jdbc = jdbc;
this.mapper = mapper;
}
public QueuedInput enqueue(String conversationId, Long agentId, String createdBy,
String message, List<MessageContentPart> contentParts,
LocalDateTime now) {
long id = IdWorker.getId();
jdbc.update("""
INSERT INTO mate_conversation_input_queue(
id,conversation_id,agent_id,created_by,message,content_parts,state,
created_at,updated_at)
VALUES(?,?,?,?,?,?,'queued',?,?)
""", id, conversationId, agentId, createdBy, message == null ? "" : message,
writeParts(contentParts), now, now);
return get(id);
}
public Optional<QueuedInput> claimNext(String conversationId, String attemptId,
LocalDateTime now) {
for (int tries = 0; tries < 8; tries++) {
List<Long> ids = jdbc.queryForList("""
SELECT id FROM mate_conversation_input_queue
WHERE conversation_id=? AND state='queued' ORDER BY id LIMIT 1
""", Long.class, conversationId);
if (ids.isEmpty()) return Optional.empty();
long id = ids.getFirst();
if (jdbc.update("""
UPDATE mate_conversation_input_queue
SET state='claimed',claimed_by_attempt_id=?,updated_at=?
WHERE id=? AND state='queued'
""", attemptId, now, id) == 1) {
return Optional.of(get(id));
}
}
return Optional.empty();
}
public boolean bindMessage(Long id, String attemptId, Long messageId, LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue
SET persisted_message_id=COALESCE(persisted_message_id,?),updated_at=?
WHERE id=? AND claimed_by_attempt_id=? AND state='claimed'
""", messageId, now, id, attemptId) == 1;
}
public boolean consume(Long id, String attemptId, LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue SET state='consumed',updated_at=?
WHERE id=? AND claimed_by_attempt_id=? AND state='claimed'
""", now, id, attemptId) == 1;
}
public boolean release(Long id, String attemptId, LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue
SET state='queued',claimed_by_attempt_id=NULL,updated_at=?
WHERE id=? AND claimed_by_attempt_id=? AND state='claimed'
""", now, id, attemptId) == 1;
}
public int releaseClaims(String attemptId,LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue
SET state='queued',claimed_by_attempt_id=NULL,updated_at=?
WHERE claimed_by_attempt_id=? AND state='claimed'
""",now,attemptId);
}
public int releaseClaimsBefore(LocalDateTime cutoff,LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue
SET state='queued',claimed_by_attempt_id=NULL,updated_at=?
WHERE state='claimed' AND updated_at<=?
""",now,cutoff);
}
public boolean cancel(Long id, String reason, LocalDateTime now) {
return jdbc.update("""
UPDATE mate_conversation_input_queue
SET state='cancelled',cancel_reason=?,updated_at=?
WHERE id=? AND state='queued'
""", bounded(reason), now, id) == 1;
}
public QueuedInput get(Long id) {
List<QueuedInput> rows = jdbc.query("""
SELECT * FROM mate_conversation_input_queue WHERE id=?
""", (rs, row) -> read(rs), id);
return rows.isEmpty() ? null : rows.getFirst();
}
public List<QueuedInput> listQueued(String conversationId) {
return jdbc.query("""
SELECT * FROM mate_conversation_input_queue
WHERE conversation_id=? AND state='queued' ORDER BY id
""", (rs, row) -> read(rs), conversationId);
}
public int countQueued(String conversationId) {
Integer count = jdbc.queryForObject("""
SELECT COUNT(*) FROM mate_conversation_input_queue
WHERE conversation_id=? AND state='queued'
""", Integer.class, conversationId);
return count == null ? 0 : count;
}
private QueuedInput read(ResultSet rs) throws SQLException {
return new QueuedInput(rs.getLong("id"), rs.getString("conversation_id"),
nullableLong(rs, "agent_id"), rs.getString("created_by"),
rs.getString("message"), readParts(rs.getString("content_parts")),
rs.getString("state"), rs.getString("claimed_by_attempt_id"),
nullableLong(rs, "persisted_message_id"), rs.getString("cancel_reason"),
time(rs, "created_at"), time(rs, "updated_at"));
}
private String writeParts(List<MessageContentPart> parts) {
try {
return mapper.writeValueAsString(parts == null ? List.of() : parts);
} catch (JsonProcessingException error) {
throw new IllegalArgumentException("Queued input contains invalid content parts", error);
}
}
private List<MessageContentPart> readParts(String json) {
if (json == null || json.isBlank()) return List.of();
try {
return mapper.readValue(json, PARTS_TYPE);
} catch (JsonProcessingException error) {
throw new IllegalStateException("Persisted queued input contains invalid content parts", error);
}
}
private static LocalDateTime time(ResultSet rs, String column) throws SQLException {
Timestamp value = rs.getTimestamp(column);
return value == null ? null : value.toLocalDateTime();
}
private static Long nullableLong(ResultSet rs, String column) throws SQLException {
long value = rs.getLong(column);
return rs.wasNull() ? null : value;
}
private static String bounded(String text) {
return text == null ? null : text.substring(0, Math.min(128, text.length()));
}
public record QueuedInput(
Long id,
String conversationId,
Long agentId,
String createdBy,
String message,
List<MessageContentPart> contentParts,
String state,
String claimedByAttemptId,
Long persistedMessageId,
String cancelReason,
LocalDateTime createdAt,
LocalDateTime updatedAt) {}
}

21
mateclaw-server/src/main/java/vip/mate/goal/config/GoalProperties.java
View File

@ -52,6 +52,27 @@ public class GoalProperties {
*/
private boolean allowAutoFollowup = true;
/** Maximum number of persistent goal segments executing in this backend instance. */
private int maxConcurrentSegments = 4;
public void setMaxConcurrentSegments(int maxConcurrentSegments) {
this.maxConcurrentSegments = Math.max(1, maxConcurrentSegments);
}
/** Runtime floor between two ordinary persistent-goal segments. */
private int minimumContinuationIntervalSeconds = 1;
public void setMinimumContinuationIntervalSeconds(int minimumContinuationIntervalSeconds) {
this.minimumContinuationIntervalSeconds = Math.max(1, minimumContinuationIntervalSeconds);
}
/** Instance-wide pause before claiming more work after a retryable provider failure. */
private int providerFailureGlobalBackoffSeconds = 30;
public void setProviderFailureGlobalBackoffSeconds(int providerFailureGlobalBackoffSeconds) {
this.providerFailureGlobalBackoffSeconds = Math.max(0, providerFailureGlobalBackoffSeconds);
}
/**
* Auto-derive a goal from a multi-step Plan-Execute plan. The Plan-Execute
* planner decomposes the request into steps and the step executor is a

34
mateclaw-server/src/main/java/vip/mate/goal/controller/GoalExecutionController.java
View File

@ -8,6 +8,8 @@ import org.springframework.web.bind.annotation.RestController;
import vip.mate.common.result.R;
import vip.mate.exception.MateClawException;
import vip.mate.goal.service.GoalContinuationStore;
import vip.mate.goal.service.GoalAttemptStore;
import vip.mate.goal.model.GoalAttempt;
import vip.mate.goal.service.GoalService;
import vip.mate.workspace.conversation.ConversationService;
@ -18,13 +20,37 @@ public class GoalExecutionController {
private final GoalService goals;
private final GoalContinuationStore store;
private final ConversationService conversations;
private final GoalAttemptStore attempts;
@GetMapping("/api/v1/goals/{id}/execution")
public R<GoalContinuationStore.Continuation> execution(@PathVariable Long id, Authentication auth) {
var goal=goals.getById(id);
if (auth==null || !conversations.isConversationOwner(goal.getConversationId(),auth.getName())) {
throw new MateClawException("err.goal.forbidden",403,"Not the conversation owner");
}
authorize(id,auth);
return R.ok(store.get(id));
}
@GetMapping("/api/v1/goals/{id}/execution/attempts")
public R<java.util.List<AttemptView>> attempts(@PathVariable Long id,Authentication auth) {
authorize(id,auth);
return R.ok(attempts.listRecent(id,50).stream().map(AttemptView::from).toList());
}
private void authorize(Long id,Authentication auth) {
var goal=goals.getById(id);
if(auth==null || !conversations.isConversationOwner(goal.getConversationId(),auth.getName())) {
throw new MateClawException("err.goal.forbidden",403,"Not the conversation owner");
}
}
public record AttemptView(String id,String parentAttemptId,String triggerType,String state,
Long inputItemId,Long assistantMessageId,String replaySafety,
String checkpointType,String finishReason,String errorCategory,
java.time.LocalDateTime startedAt,java.time.LocalDateTime finishedAt,
java.time.LocalDateTime createdAt,java.time.LocalDateTime updatedAt) {
static AttemptView from(GoalAttempt attempt) {
return new AttemptView(attempt.id(),attempt.parentAttemptId(),attempt.triggerType(),attempt.state(),
attempt.inputItemId(),attempt.assistantMessageId(),attempt.replaySafety(),attempt.checkpointType(),
attempt.finishReason(),attempt.errorCategory(),attempt.startedAt(),attempt.finishedAt(),
attempt.createdAt(),attempt.updatedAt());
}
}
}

33
mateclaw-server/src/main/java/vip/mate/goal/model/GoalAttempt.java Normal file
View File

@ -0,0 +1,33 @@
package vip.mate.goal.model;
import java.time.LocalDateTime;
import java.util.Set;
/** One immutable-identity execution attempt for a bounded goal segment. */
public record GoalAttempt(
String id,
Long goalId,
String conversationId,
String parentAttemptId,
String triggerType,
String state,
String leaseToken,
LocalDateTime leaseUntil,
Long inputItemId,
Long assistantMessageId,
String replaySafety,
String checkpointType,
String finishReason,
String errorCategory,
LocalDateTime startedAt,
LocalDateTime finishedAt,
LocalDateTime createdAt,
LocalDateTime updatedAt) {
private static final Set<String> TERMINAL_STATES =
Set.of("succeeded", "retryable", "blocked", "cancelled");
public boolean terminal() {
return TERMINAL_STATES.contains(state);
}
}

20
mateclaw-server/src/main/java/vip/mate/goal/model/SegmentOutcome.java Normal file
View File

@ -0,0 +1,20 @@
package vip.mate.goal.model;
/** Durable scheduling facts returned by one bounded goal segment. */
public sealed interface SegmentOutcome {
String reason();
default String finishReason() { return reason(); }
default boolean awaitingApproval() { return this instanceof AwaitApproval; }
default boolean evaluationUnavailable() { return this instanceof Retry retry
&& "evaluation".equals(retry.category()); }
record Continue(String reason) implements SegmentOutcome {}
record Defer(String reason, java.time.LocalDateTime nextRunAt) implements SegmentOutcome {}
record Complete(String reason) implements SegmentOutcome {}
record AwaitApproval(String reason) implements SegmentOutcome {}
record WaitInput(String reason) implements SegmentOutcome {}
record Retry(String category, String reason) implements SegmentOutcome {}
record Blocked(String category, String reason) implements SegmentOutcome {}
record Cancelled(String reason) implements SegmentOutcome {}
}

132
mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java Normal file
View File

@ -0,0 +1,132 @@
package vip.mate.goal.service;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
import vip.mate.goal.model.GoalAttempt;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
/** Fenced persistence for bounded goal execution attempts. */
@Repository
public class GoalAttemptStore {
private final JdbcTemplate jdbc;
public GoalAttemptStore(JdbcTemplate jdbc) {
this.jdbc = jdbc;
}
public GoalAttempt create(Long goalId, String conversationId, String parentAttemptId,
String triggerType, String leaseToken, LocalDateTime leaseUntil,
Long inputItemId, LocalDateTime now) {
String id = UUID.randomUUID().toString();
jdbc.update("""
INSERT INTO mate_goal_attempt(
attempt_id,goal_id,conversation_id,parent_attempt_id,trigger_type,state,
lease_token,lease_until,input_item_id,replay_safety,checkpoint_type,
created_at,updated_at)
VALUES(?,?,?,?,?,'claimed',?,?,?,'safe','claimed',?,?)
""", id, goalId, conversationId, parentAttemptId, triggerType, leaseToken,
leaseUntil, inputItemId, now, now);
return get(id);
}
public GoalAttempt get(String id) {
List<GoalAttempt> rows = jdbc.query("""
SELECT * FROM mate_goal_attempt WHERE attempt_id=?
""", (rs, row) -> read(rs), id);
return rows.isEmpty() ? null : rows.getFirst();
}
public List<GoalAttempt> listRecent(Long goalId, int limit) {
return jdbc.query("""
SELECT * FROM mate_goal_attempt WHERE goal_id=?
ORDER BY created_at DESC,attempt_id DESC LIMIT ?
""", (rs, row) -> read(rs), goalId, Math.max(1, Math.min(limit, 100)));
}
public boolean markRunning(String id, String leaseToken, LocalDateTime now) {
return jdbc.update("""
UPDATE mate_goal_attempt SET state='running',started_at=?,updated_at=?
WHERE attempt_id=? AND lease_token=? AND state='claimed'
""", now, now, id, leaseToken) == 1;
}
public boolean renew(String id, String leaseToken, LocalDateTime leaseUntil,
LocalDateTime now) {
return jdbc.update("""
UPDATE mate_goal_attempt SET lease_until=?,updated_at=?
WHERE attempt_id=? AND lease_token=? AND state IN ('claimed','running')
""", leaseUntil, now, id, leaseToken) == 1;
}
public boolean checkpoint(String id, String leaseToken, String replaySafety,
String checkpointType, Long assistantMessageId,
LocalDateTime now) {
return jdbc.update("""
UPDATE mate_goal_attempt
SET replay_safety=?,checkpoint_type=?,
assistant_message_id=COALESCE(?,assistant_message_id),updated_at=?
WHERE attempt_id=? AND lease_token=? AND state IN ('claimed','running')
""", replaySafety, checkpointType, assistantMessageId, now, id, leaseToken) == 1;
}
public boolean finish(String id, String leaseToken, String state, String finishReason,
String errorCategory, LocalDateTime now) {
if (!GoalAttemptTerminalState.valid(state)) {
throw new IllegalArgumentException("Unsupported terminal attempt state: " + state);
}
return jdbc.update("""
UPDATE mate_goal_attempt
SET state=?,finish_reason=?,error_category=?,finished_at=?,updated_at=?
WHERE attempt_id=? AND lease_token=? AND state IN ('claimed','running')
""", state, bounded(finishReason), bounded(errorCategory), now, now,
id, leaseToken) == 1;
}
public List<GoalAttempt> expired(LocalDateTime now, int limit) {
return jdbc.query("""
SELECT * FROM mate_goal_attempt
WHERE state IN ('claimed','running') AND lease_until<=?
ORDER BY lease_until,created_at LIMIT ?
""", (rs, row) -> read(rs), now, Math.max(1, Math.min(limit, 100)));
}
private static GoalAttempt read(ResultSet rs) throws SQLException {
return new GoalAttempt(
rs.getString("attempt_id"), rs.getLong("goal_id"),
rs.getString("conversation_id"), rs.getString("parent_attempt_id"),
rs.getString("trigger_type"), rs.getString("state"),
rs.getString("lease_token"), time(rs, "lease_until"),
nullableLong(rs, "input_item_id"), nullableLong(rs, "assistant_message_id"),
rs.getString("replay_safety"), rs.getString("checkpoint_type"),
rs.getString("finish_reason"), rs.getString("error_category"),
time(rs, "started_at"), time(rs, "finished_at"),
time(rs, "created_at"), time(rs, "updated_at"));
}
private static LocalDateTime time(ResultSet rs, String column) throws SQLException {
Timestamp value = rs.getTimestamp(column);
return value == null ? null : value.toLocalDateTime();
}
private static Long nullableLong(ResultSet rs, String column) throws SQLException {
long value = rs.getLong(column);
return rs.wasNull() ? null : value;
}
private static String bounded(String text) {
return text == null ? null : text.substring(0, Math.min(128, text.length()));
}
private static final class GoalAttemptTerminalState {
private static boolean valid(String state) {
return "succeeded".equals(state) || "retryable".equals(state)
|| "blocked".equals(state) || "cancelled".equals(state);
}
}
}

71
mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java
View File

@ -25,7 +25,15 @@ public class GoalContinuationStore {
public record Continuation(Long goalId, String conversationId, String state,
LocalDateTime nextRunAt, String leaseOwner,
LocalDateTime leaseUntil, int failures, String reason) {}
LocalDateTime leaseUntil, int failures, String reason,
String currentAttemptId, long revision) {
public Continuation(Long goalId, String conversationId, String state,
LocalDateTime nextRunAt, String leaseOwner,
LocalDateTime leaseUntil, int failures, String reason) {
this(goalId, conversationId, state, nextRunAt, leaseOwner, leaseUntil,
failures, reason, null, 0);
}
}
public void discover(LocalDateTime now) {
// Bounded discovery; another instance may insert the same goal concurrently.
@ -63,7 +71,8 @@ public class GoalContinuationStore {
public boolean claim(Long goalId, String token, LocalDateTime now, LocalDateTime until) {
return jdbc.update("""
UPDATE mate_goal_continuation SET state='running',lease_owner=?,lease_until=?,updated_at=?,wake_requested=FALSE
UPDATE mate_goal_continuation SET state='running',lease_owner=?,lease_until=?,updated_at=?,
wake_requested=FALSE,revision=revision+1
WHERE goal_id=? AND
((state IN ('queued','retry') AND next_run_at<=?)
OR (state='running' AND lease_until<=?))
@ -78,6 +87,53 @@ public class GoalContinuationStore {
""", until, goalId, token) == 1;
}
public boolean bindAttempt(Long goalId, String token, String attemptId, long expectedRevision) {
return jdbc.update("""
UPDATE mate_goal_continuation SET current_attempt_id=?,revision=revision+1,updated_at=?
WHERE goal_id=? AND lease_owner=? AND state='running'
AND current_attempt_id IS NULL AND revision=?
""", attemptId, LocalDateTime.now(), goalId, token, expectedRevision) == 1;
}
public boolean matchesFence(Long goalId, String token, String attemptId, long revision) {
Integer count=jdbc.queryForObject("""
SELECT COUNT(*) FROM mate_goal_continuation
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=?
AND revision=? AND state='running'
""",Integer.class,goalId,token,attemptId,revision);
return count!=null && count==1;
}
public boolean renewFenced(Long goalId,String token,String attemptId,long revision,LocalDateTime until) {
return jdbc.update("""
UPDATE mate_goal_continuation SET lease_until=?,updated_at=?
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=?
AND revision=? AND state='running'
""",until,LocalDateTime.now(),goalId,token,attemptId,revision)==1;
}
public boolean settleFenced(Long goalId,String token,String attemptId,long revision,String state,
LocalDateTime nextRunAt,int failures,String reason,LocalDateTime now) {
return jdbc.update("""
UPDATE mate_goal_continuation
SET state=CASE WHEN ?='waiting_approval' AND wake_requested=TRUE THEN 'queued' ELSE ? END,
next_run_at=?,failures=?,reason=?,wake_requested=FALSE,lease_owner=NULL,lease_until=NULL,
current_attempt_id=NULL,revision=revision+1,updated_at=?
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND revision=? AND state='running'
""",state,state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,revision)==1;
}
public boolean recoverExpired(Long goalId,String token,String attemptId,LocalDateTime expiredAt,
String state,LocalDateTime nextRunAt,int failures,String reason,LocalDateTime now) {
return jdbc.update("""
UPDATE mate_goal_continuation
SET state=?,next_run_at=?,failures=?,reason=?,wake_requested=FALSE,
lease_owner=NULL,lease_until=NULL,current_attempt_id=NULL,revision=revision+1,updated_at=?
WHERE goal_id=? AND lease_owner=? AND current_attempt_id=? AND state='running'
AND lease_until<=?
""",state,nextRunAt,failures,bounded(reason),now,goalId,token,attemptId,expiredAt)==1;
}
public boolean settle(Long goalId, String token, String state, LocalDateTime nextRunAt,
int failures, String reason) {
return jdbc.update("""
@ -90,21 +146,23 @@ public class GoalContinuationStore {
public void suspendConversation(String conversationId, String reason) {
jdbc.update("""
UPDATE mate_goal_continuation SET state='paused',reason=?,lease_owner=NULL,lease_until=NULL,
updated_at=? WHERE goal_id IN (SELECT id FROM mate_agent_goal WHERE conversation_id=?)
current_attempt_id=NULL,revision=revision+1,updated_at=?
WHERE goal_id IN (SELECT id FROM mate_agent_goal WHERE conversation_id=?)
""", bounded(reason), LocalDateTime.now(), conversationId);
}
public void resume(Long goalId, LocalDateTime now) {
jdbc.update("""
UPDATE mate_goal_continuation SET state='queued',next_run_at=?,failures=0,reason='resumed',
lease_owner=NULL,lease_until=NULL,updated_at=? WHERE goal_id=? AND state<>'running'
lease_owner=NULL,lease_until=NULL,current_attempt_id=NULL,revision=revision+1,updated_at=?
WHERE goal_id=? AND state<>'running'
""", now, now, goalId);
}
public void turnFinished(String conversationId, LocalDateTime now) {
jdbc.update("""
UPDATE mate_goal_continuation SET state=CASE WHEN state='waiting_approval' THEN 'queued' ELSE state END,
wake_requested=TRUE,next_run_at=?,reason='interactive_turn_finished',updated_at=?
wake_requested=TRUE,next_run_at=?,reason='interactive_turn_finished',revision=revision+1,updated_at=?
WHERE state IN ('waiting_approval','running') AND goal_id IN
(SELECT id FROM mate_agent_goal WHERE conversation_id=? AND status='active' AND deleted=0)
""",now,now,conversationId);
@ -114,7 +172,8 @@ public class GoalContinuationStore {
Timestamp until = rs.getTimestamp("lease_until");
return new Continuation(rs.getLong("goal_id"), rs.getString("conversation_id"), rs.getString("state"),
rs.getTimestamp("next_run_at").toLocalDateTime(), rs.getString("lease_owner"),
until == null ? null : until.toLocalDateTime(), rs.getInt("failures"), rs.getString("reason"));
until == null ? null : until.toLocalDateTime(), rs.getInt("failures"), rs.getString("reason"),
rs.getString("current_attempt_id"),rs.getLong("revision"));
}
private static String bounded(String text) {

111
mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java
View File

@ -16,14 +16,15 @@ import vip.mate.goal.config.GoalProperties;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.GoalEvaluationResult;
import vip.mate.goal.model.GoalStatus;
import vip.mate.goal.model.SegmentOutcome;
import java.time.Clock;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
/** Owns cross-turn liveness. Graph recursion limits bound segments, not goal lifetime. */
@Slf4j
@ -38,126 +39,130 @@ public class GoalContinuationSupervisor {
private final ChatStreamTracker streams;
private final Clock clock;
private final Executor executor;
private final ConcurrentHashMap<Long, String> active = new ConcurrentHashMap<>();
private final GoalRunCoordinator coordinator;
private final GoalRecoveryService recovery;
private final ConcurrentHashMap<Long, GoalRunCoordinator.ClaimedRun> active = new ConcurrentHashMap<>();
private final AtomicReference<LocalDateTime> providerBackoffUntil = new AtomicReference<>();
private volatile boolean closing;
@Autowired
public GoalContinuationSupervisor(GoalContinuationStore store, GoalService goals, GoalProperties properties,
GoalFollowupService followups, GoalSegmentRunner runner, RunningConversationRegistry running,
ChatStreamTracker streams) {
this(store,goals,properties,followups,runner,running,streams,Clock.systemDefaultZone(),
ChatStreamTracker streams,GoalRunCoordinator coordinator,GoalRecoveryService recovery) {
this(store,goals,properties,followups,runner,running,streams,coordinator,recovery,Clock.systemDefaultZone(),
Executors.newVirtualThreadPerTaskExecutor());
}
GoalContinuationSupervisor(GoalContinuationStore store, GoalService goals, GoalProperties properties,
GoalFollowupService followups, GoalSegmentRunner runner, RunningConversationRegistry running,
ChatStreamTracker streams, Clock clock, Executor executor) {
ChatStreamTracker streams,GoalRunCoordinator coordinator,GoalRecoveryService recovery,
Clock clock, Executor executor) {
this.store=store; this.goals=goals; this.properties=properties; this.followups=followups;
this.runner=runner; this.running=running; this.streams=streams; this.clock=clock; this.executor=executor;
this.runner=runner; this.running=running; this.streams=streams; this.coordinator=coordinator;this.recovery=recovery;
this.clock=clock; this.executor=executor;
}
@Scheduled(fixedDelayString="${mateclaw.goal.supervisor-poll-ms:5000}", initialDelayString="${mateclaw.goal.supervisor-poll-ms:5000}")
public void tick() {
if (closing || !properties.isEnabled() || !properties.isAllowAutoFollowup()) return;
LocalDateTime now = LocalDateTime.now(clock);
active.forEach((id, token) -> {
recovery.recoverExpired(now);
active.forEach((id, claimed) -> {
GoalEntity goal = goals.getById(id);
boolean cancelled = goal.getStatus()==GoalStatus.PAUSED || goal.getStatus()==GoalStatus.ABANDONED
|| !Boolean.TRUE.equals(goal.getAutoFollowupEnabled());
if (cancelled || !store.renew(id, token, now.plusSeconds(60))) runner.cancel(id);
if (cancelled || !coordinator.renew(claimed,now)) runner.cancel(id);
});
LocalDateTime backoffUntil=providerBackoffUntil.get();
if (backoffUntil!=null && now.isBefore(backoffUntil)) return;
store.discover(now);
for (var candidate : store.due(now, 20)) {
if (active.size() >= 4) break;
int maxConcurrent = properties.getMaxConcurrentSegments();
// Scan beyond the execution capacity: a due conversation may currently
// belong to an interactive user turn and must not starve later goals.
for (var candidate : store.due(now, Math.max(20,maxConcurrent))) {
if (active.size() >= maxConcurrent) break;
String conv = candidate.conversationId();
if (active.containsKey(candidate.goalId()) || running.isActive(conv)
|| streams.isRunning(conv)) continue;
GoalEntity goal = goals.getById(candidate.goalId());
if (!eligible(goal)) continue;
String token = UUID.randomUUID().toString();
if (active.putIfAbsent(goal.getId(),token) != null) continue;
if (active.containsKey(goal.getId())) continue;
GoalRunCoordinator.ClaimedRun claimed=coordinator.claim(candidate,goal,now);
if(claimed==null || active.putIfAbsent(goal.getId(),claimed)!=null) continue;
try {
if (!store.claim(goal.getId(),token,now,now.plusSeconds(60))) {
active.remove(goal.getId(),token); continue;
}
executor.execute(() -> execute(goal, candidate, token));
executor.execute(() -> execute(claimed));
} catch (RuntimeException error) {
active.remove(goal.getId(),token);
store.settle(goal.getId(),token,"retry",now.plusSeconds(5),candidate.failures()+1,"dispatch_failed");
active.remove(goal.getId(),claimed);
settle(claimed,new SegmentOutcome.Retry("dispatch","dispatch_failed"),now);
log.warn("Goal {} dispatch failed",goal.getId(),error);
}
}
}
private void execute(GoalEntity initial, GoalContinuationStore.Continuation candidate, String token) {
private void execute(GoalRunCoordinator.ClaimedRun claimed) {
GoalEntity initial=claimed.goal();
LocalDateTime now = LocalDateTime.now(clock);
try {
if (closing) return;
GoalEntity goal = goals.getById(initial.getId());
if (!eligible(goal)) {
settle(initial,token,"paused",now,0,"goal_not_runnable"); return;
settle(claimed,new SegmentOutcome.Cancelled("goal_not_runnable"),now); return;
}
var decision = followups.decide(goal,new GoalEvaluationResult(0,goal.getProgressSummary(),
GoalEvaluationResult.DECISION_CONTINUE,false,"",0,0,List.of(),null),now);
switch (decision.action()) {
case DEFER, RETRY -> {
settle(goal,token,"queued",decision.nextRunAt(),candidate.failures(),decision.reason()); return;
settle(claimed,new SegmentOutcome.Defer(decision.reason(),decision.nextRunAt()),now); return;
}
case BUDGET_LIMITED -> {
goals.markExhausted(goal.getId(),decision.reason());
settle(goal,token,"budget_limited",now,0,decision.reason()); return;
settle(claimed,new SegmentOutcome.Cancelled(decision.reason()),now); return;
}
case COMPLETE, DISABLED -> {
settle(goal,token,"paused",now,0,decision.reason()); return;
settle(claimed,new SegmentOutcome.Cancelled(decision.reason()),now); return;
}
case CONTINUE -> { }
}
GoalSegmentRunner.Result result = runner.run(goal,decision.prompt(),"running".equals(candidate.state()));
if(!coordinator.markRunning(claimed,now)) return;
SegmentOutcome outcome = runner.run(claimed,decision.prompt(),"running".equals(claimed.candidate().state()));
if (outcome instanceof SegmentOutcome.Retry retry
&& ("provider".equals(retry.category()) || "evaluation".equals(retry.category()))) {
activateProviderBackoff(LocalDateTime.now(clock));
}
// Shutdown cancellation is not user Stop: retain the lease for recovery.
if (closing) return;
GoalEntity fresh = goals.getById(goal.getId());
if (fresh.getStatus() == GoalStatus.COMPLETED) {
settle(goal,token,"completed",now,0,"goal_completed");
} else if (fresh.getStatus() == GoalStatus.PAUSED && goals.isBudgetExhausted(fresh)) {
settle(goal,token,"budget_limited",now,0,goals.exhaustionReason(fresh));
} else if (!eligible(fresh) || "stopped".equals(result.finishReason())) {
boolean waiting = fresh.getProgressSummary()!=null && fresh.getProgressSummary().startsWith("Waiting for input:");
settle(goal,token,waiting ? "waiting_input" : "paused",now,0,
waiting ? fresh.getProgressSummary() : "goal_paused_or_stopped");
} else if (result.awaitingApproval()) {
settle(goal,token,"waiting_approval",now.plusSeconds(5),0,"approval_required");
} else if ("error_fallback".equals(result.finishReason())) {
// The graph discarded the original error category; do not blindly replay tools.
goals.pause(goal.getId(),goal.getCreatedBy());
settle(goal,token,"blocked",now,candidate.failures()+1,"graph_error_requires_review");
} else if (result.evaluationUnavailable()) {
settle(goal,token,"retry",now.plusSeconds(30),candidate.failures()+1,"evaluation_unavailable");
} else {
int cooldown = fresh.getFollowupCooldownSeconds() == null ? 0 : fresh.getFollowupCooldownSeconds();
settle(goal,token,"queued",LocalDateTime.now(clock).plusSeconds(Math.max(1,cooldown)),0,"unfinished");
}
settle(claimed,outcome,LocalDateTime.now(clock));
} catch (RuntimeException error) {
// A shutdown/lost-lease cancellation is not a task failure. Leave the
// running lease for restart recovery; the runner saves partial evidence.
if (closing || Thread.currentThread().isInterrupted()) return;
int failures = Math.min(1000,candidate.failures()+1);
boolean transientError = retryable(error);
if (transientError) activateProviderBackoff(now);
if (!transientError) {
GoalEntity fresh=goals.getById(initial.getId());
if (eligible(fresh)) goals.pause(fresh.getId(),fresh.getCreatedBy());
}
long delay = Math.min(300,5L << Math.min(6,failures-1));
settle(initial,token,transientError ? "retry" : "blocked",now.plusSeconds(delay),failures,
transientError ? "transient_provider_error" : "execution_requires_review");
settle(claimed,transientError
? new SegmentOutcome.Retry("provider","transient_provider_error")
: new SegmentOutcome.Blocked("execution","execution_requires_review"),now);
log.warn("Goal {} segment failed ({})",initial.getId(),transientError ? "retry" : "blocked",error);
} finally {
active.remove(initial.getId(),token);
active.remove(initial.getId(),claimed);
}
}
private void settle(GoalEntity goal, String token, String state, LocalDateTime due, int failures, String reason) {
if (store.settle(goal.getId(),token,state,due,failures,reason)) {
streams.broadcastObject(goal.getConversationId(),"goal_continuation",store.get(goal.getId()));
private void activateProviderBackoff(LocalDateTime now) {
int seconds=properties.getProviderFailureGlobalBackoffSeconds();
if (seconds<=0) return;
LocalDateTime proposed=now.plusSeconds(seconds);
LocalDateTime effective=providerBackoffUntil.updateAndGet(current ->
current==null || current.isBefore(proposed) ? proposed : current);
log.warn("Goal dispatch paused until {} after retryable provider failure",effective);
}
private void settle(GoalRunCoordinator.ClaimedRun claimed,SegmentOutcome outcome,LocalDateTime now) {
if (coordinator.settle(claimed,outcome,now)) {
streams.broadcastObject(claimed.goal().getConversationId(),"goal_continuation",store.get(claimed.goal().getId()));
}
}

84
mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java Normal file
View File

@ -0,0 +1,84 @@
package vip.mate.goal.service;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import vip.mate.channel.web.ConversationInputQueueStore;
import vip.mate.goal.model.GoalAttempt;
import java.time.LocalDateTime;
/** Reconciles expired attempts from durable checkpoints before dispatching new work. */
@Service
public class GoalRecoveryService {
public enum RecoveryDecision {
RETRY_SAFE,
RESUME_FROM_EVIDENCE,
RECONCILE_MESSAGE,
BLOCK_UNCERTAIN_SIDE_EFFECT
}
private final GoalAttemptStore attempts;
private final GoalContinuationStore continuations;
private final ConversationInputQueueStore inputs;
private final GoalService goals;
private final LocalDateTime startupCutoff=LocalDateTime.now();
private volatile boolean orphanClaimsReleased;
public GoalRecoveryService(GoalAttemptStore attempts,GoalContinuationStore continuations,
ConversationInputQueueStore inputs,GoalService goals) {
this.attempts=attempts;this.continuations=continuations;this.inputs=inputs;this.goals=goals;
}
public RecoveryDecision classify(GoalAttempt attempt) {
if("tool_started".equals(attempt.checkpointType()) && "uncertain".equals(attempt.replaySafety())) {
return RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT;
}
if("message_saved".equals(attempt.checkpointType()) && attempt.assistantMessageId()!=null) {
return RecoveryDecision.RECONCILE_MESSAGE;
}
if("tool_completed".equals(attempt.checkpointType()) && "resolved".equals(attempt.replaySafety())) {
return RecoveryDecision.RESUME_FROM_EVIDENCE;
}
return RecoveryDecision.RETRY_SAFE;
}
public int recoverExpired(LocalDateTime now) {
if(!orphanClaimsReleased) {
synchronized(this) {
if(!orphanClaimsReleased) {
inputs.releaseClaimsBefore(startupCutoff,now);
orphanClaimsReleased=true;
}
}
}
int recovered=0;
for(GoalAttempt attempt:attempts.expired(now,100)) {
if(recover(attempt,now)) recovered++;
}
return recovered;
}
@Transactional
boolean recover(GoalAttempt attempt,LocalDateTime now) {
var continuation=continuations.get(attempt.goalId());
if(continuation==null || !attempt.id().equals(continuation.currentAttemptId())
|| !attempt.leaseToken().equals(continuation.leaseOwner())) return false;
RecoveryDecision decision=classify(attempt);
String attemptState=decision==RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT ? "blocked" : "retryable";
String projectionState=decision==RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT ? "blocked" : "retry";
String reason=decision==RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT
? "uncertain_tool_outcome_requires_review" : "restart_recovery";
if(!attempts.finish(attempt.id(),attempt.leaseToken(),attemptState,reason,
decision.name().toLowerCase(),now)) return false;
if(!continuations.recoverExpired(attempt.goalId(),attempt.leaseToken(),attempt.id(),now,
projectionState,now,continuation.failures()+1,reason,now)) {
throw new IllegalStateException("Expired goal projection changed during recovery");
}
inputs.releaseClaims(attempt.id(),now);
if(decision==RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT) {
var goal=goals.getById(attempt.goalId());
if(goal!=null) goals.pause(goal.getId(),goal.getCreatedBy());
}
return true;
}
}

143
mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java Normal file
View File

@ -0,0 +1,143 @@
package vip.mate.goal.service;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import vip.mate.goal.config.GoalProperties;
import vip.mate.goal.model.GoalAttempt;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.GoalStatus;
import vip.mate.goal.model.SegmentOutcome;
import java.time.LocalDateTime;
import java.util.UUID;
/** Owns fenced claim, renewal and settlement for one durable goal segment. */
@Service
public class GoalRunCoordinator {
private static final int LEASE_SECONDS=60;
private final GoalContinuationStore continuations;
private final GoalAttemptStore attempts;
private final GoalService goals;
private final GoalProperties properties;
public GoalRunCoordinator(GoalContinuationStore continuations,GoalAttemptStore attempts,GoalService goals,
GoalProperties properties) {
this.continuations=continuations;this.attempts=attempts;this.goals=goals;this.properties=properties;
}
public record ClaimedRun(GoalContinuationStore.Continuation candidate,GoalEntity goal,
GoalAttempt attempt,long revision) {}
@Transactional
public ClaimedRun claim(GoalContinuationStore.Continuation candidate,GoalEntity goal,LocalDateTime now) {
if(candidate==null || goal==null || candidate.currentAttemptId()!=null) return null;
String token=UUID.randomUUID().toString();
LocalDateTime until=now.plusSeconds(LEASE_SECONDS);
if(!continuations.claim(goal.getId(),token,now,until)) return null;
GoalContinuationStore.Continuation claimed=continuations.get(goal.getId());
String parentAttemptId=null;
if("restart_recovery".equals(candidate.reason())) {
var recent=attempts.listRecent(goal.getId(),1);
if(!recent.isEmpty()) parentAttemptId=recent.getFirst().id();
}
GoalAttempt attempt=attempts.create(goal.getId(),goal.getConversationId(),parentAttemptId,
"continuation",token,until,null,now);
if(!continuations.bindAttempt(goal.getId(),token,attempt.id(),claimed.revision())) {
throw new IllegalStateException("Goal attempt could not be bound to its continuation");
}
return new ClaimedRun(candidate,goal,attempt,claimed.revision()+1);
}
@Transactional
public boolean markRunning(ClaimedRun run,LocalDateTime now) {
if(!current(run)) return false;
return attempts.markRunning(run.attempt().id(),run.attempt().leaseToken(),now);
}
@Transactional
public boolean renew(ClaimedRun run,LocalDateTime now) {
LocalDateTime until=now.plusSeconds(LEASE_SECONDS);
if(!continuations.renewFenced(run.goal().getId(),run.attempt().leaseToken(),
run.attempt().id(),run.revision(),until)) return false;
return attempts.renew(run.attempt().id(),run.attempt().leaseToken(),until,now);
}
@Transactional
public boolean checkpoint(ClaimedRun run,String replaySafety,String checkpointType,
Long assistantMessageId,LocalDateTime now) {
if(!current(run)) return false;
return attempts.checkpoint(run.attempt().id(),run.attempt().leaseToken(),replaySafety,
checkpointType,assistantMessageId,now);
}
@Transactional
public boolean settle(ClaimedRun run,SegmentOutcome outcome,LocalDateTime now) {
if(!current(run)) return false;
GoalEntity fresh=goals.getById(run.goal().getId());
Settlement settlement=classify(run,outcome,fresh,now);
if((outcome instanceof SegmentOutcome.Continue || outcome instanceof SegmentOutcome.Complete)
&& !attempts.checkpoint(run.attempt().id(),run.attempt().leaseToken(),"resolved",
"evaluation_saved",null,now)) return false;
if(!attempts.finish(run.attempt().id(),run.attempt().leaseToken(),settlement.attemptState,
outcome.reason(),settlement.errorCategory,now)) return false;
if(!continuations.settleFenced(run.goal().getId(),run.attempt().leaseToken(),run.attempt().id(),
run.revision(),settlement.projectionState,settlement.nextRunAt,settlement.failures,
settlement.reason,now)) {
throw new IllegalStateException("Goal projection fence changed during settlement");
}
return true;
}
private boolean current(ClaimedRun run) {
return run!=null && continuations.matchesFence(run.goal().getId(),run.attempt().leaseToken(),
run.attempt().id(),run.revision());
}
private Settlement classify(ClaimedRun run,SegmentOutcome outcome,GoalEntity fresh,LocalDateTime now) {
int failures=run.candidate().failures();
if(fresh!=null && fresh.getStatus()==GoalStatus.COMPLETED || outcome instanceof SegmentOutcome.Complete) {
return new Settlement("succeeded","completed",now,0,"goal_completed",null);
}
if(fresh!=null && fresh.getStatus()==GoalStatus.PAUSED && goals.isBudgetExhausted(fresh)) {
return new Settlement("succeeded","budget_limited",now,0,goals.exhaustionReason(fresh),null);
}
if(outcome instanceof SegmentOutcome.AwaitApproval) {
return new Settlement("succeeded","waiting_approval",now,0,outcome.reason(),null);
}
if(outcome instanceof SegmentOutcome.WaitInput) {
return new Settlement("succeeded","waiting_input",now,0,outcome.reason(),null);
}
if(outcome instanceof SegmentOutcome.Retry retry) {
int nextFailures=Math.min(1000,failures+1);
long delay=Math.min(300,5L << Math.min(6,nextFailures-1));
return new Settlement("retryable","retry",now.plusSeconds(delay),nextFailures,
retry.reason(),retry.category());
}
if(outcome instanceof SegmentOutcome.Defer defer) {
return new Settlement("succeeded","queued",defer.nextRunAt(),failures,defer.reason(),null);
}
if(outcome instanceof SegmentOutcome.Blocked blocked) {
return new Settlement("blocked","blocked",now,Math.min(1000,failures+1),
blocked.reason(),blocked.category());
}
if(outcome instanceof SegmentOutcome.Cancelled || !eligible(fresh)) {
boolean waiting=fresh!=null && fresh.getProgressSummary()!=null
&& fresh.getProgressSummary().startsWith("Waiting for input:");
return new Settlement("cancelled",waiting ? "waiting_input" : "paused",now,0,
waiting ? fresh.getProgressSummary() : outcome.reason(),null);
}
int cooldown=fresh==null || fresh.getFollowupCooldownSeconds()==null ? 0 : fresh.getFollowupCooldownSeconds();
int delay=Math.max(properties.getMinimumContinuationIntervalSeconds(),cooldown);
return new Settlement("succeeded","queued",now.plusSeconds(delay),0,
outcome.reason(),null);
}
private static boolean eligible(GoalEntity goal) {
return goal!=null && goal.getStatus()==GoalStatus.ACTIVE
&& Boolean.TRUE.equals(goal.getPersistentExecution())
&& Boolean.TRUE.equals(goal.getAutoFollowupEnabled());
}
private record Settlement(String attemptState,String projectionState,LocalDateTime nextRunAt,
int failures,String reason,String errorCategory) {}
}

129
mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java
View File

@ -10,12 +10,17 @@ import vip.mate.agent.runtime.ConversationTurnGate;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.channel.web.AgentStreamAccumulator;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.channel.web.ConversationInputQueueStore;
import vip.mate.exception.MateClawException;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.SegmentOutcome;
import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.model.MessageEntity;
import java.util.Map;
import java.util.Objects;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicBoolean;
@ -30,6 +35,7 @@ public class GoalSegmentRunner {
private final ChatStreamTracker streams;
private final ObjectMapper mapper;
private final ConversationTurnGate gate;
private final ConversationInputQueueStore inputQueue;
private final ConcurrentHashMap<Long,Worker> workers=new ConcurrentHashMap<>();
private volatile boolean closing;
private static final class Worker {
@ -42,16 +48,17 @@ public class GoalSegmentRunner {
}
@org.springframework.beans.factory.annotation.Autowired
private GoalService goals;
@org.springframework.beans.factory.annotation.Autowired
private GoalRunCoordinator coordinator;
public GoalSegmentRunner(AgentService agents, ConversationService conversations,
ApprovalWorkflowService approvals, ChatStreamTracker streams, ObjectMapper mapper, ConversationTurnGate gate) {
ApprovalWorkflowService approvals, ChatStreamTracker streams, ObjectMapper mapper,
ConversationTurnGate gate, ConversationInputQueueStore inputQueue) {
this.agents=agents;this.conversations=conversations;this.approvals=approvals;
this.streams=streams;this.mapper=mapper;this.gate=gate;
this.streams=streams;this.mapper=mapper;this.gate=gate;this.inputQueue=inputQueue;
}
public record Result(String finishReason, boolean awaitingApproval, boolean evaluationUnavailable) {
public Result(String finishReason, boolean awaitingApproval) { this(finishReason,awaitingApproval,false); }
}
private record SegmentResult(String finishReason, boolean awaitingApproval, boolean evaluationUnavailable) {}
/** Cancel this worker only, never a newer conversation generation or a user turn. */
public void cancel(Long goalId) {
@ -78,17 +85,27 @@ public class GoalSegmentRunner {
// worker: it also performs JDBC I/O on shared embedded database channels.
}
public Result run(GoalEntity goal, String prompt, boolean recovered) {
public SegmentOutcome run(GoalEntity goal, String prompt, boolean recovered) {
return run(goal,prompt,recovered,null);
}
public SegmentOutcome run(GoalRunCoordinator.ClaimedRun claimed,String prompt,boolean recovered) {
return run(claimed.goal(),prompt,recovered,claimed);
}
private SegmentOutcome run(GoalEntity goal,String prompt,boolean recovered,
GoalRunCoordinator.ClaimedRun claimedRun) {
String convId=goal.getConversationId();
var permit=gate.tryAcquire(convId);
if (permit==null) throw new MateClawException("err.agent.conversation_busy",409,"Conversation is busy");
Worker worker=new Worker(convId);
AtomicReference<ConversationInputQueueStore.QueuedInput> claimedInput=new AtomicReference<>();
try {
workers.put(goal.getId(),worker);
// Register before checking the shutdown fence so cancellation cannot miss us.
if (closing) {
worker.cancelled.set(true);
return new Result("stopped",false);
return new SegmentOutcome.Cancelled("stopped");
}
var conv=conversations.findByConversationId(convId);
if (conv==null || !Objects.equals(conv.getWorkspaceId(),goal.getWorkspaceId())
@ -102,7 +119,7 @@ public class GoalSegmentRunner {
|| (agent.getRuntimeType()!=null && !"native".equals(agent.getRuntimeType()))) {
throw new IllegalStateException("Goal requires an enabled native runtime with goal evaluation");
}
if (approvals.findPendingByConversation(convId)!=null) return new Result("",true);
if (approvals.findPendingByConversation(convId)!=null) return new SegmentOutcome.AwaitApproval("approval_required");
if (streams.isRunning(convId)) {
throw new MateClawException("err.agent.conversation_busy",409,"Conversation has pending input");
}
@ -110,48 +127,57 @@ public class GoalSegmentRunner {
+ "Inspect the workspace, progress ledger and existing async handles before acting. "
+ "Do not replay side effects whose outcome is unknown; request review if their outcome cannot be verified.\n" : "";
ChatOrigin origin=ChatOrigin.web(convId,goal.getCreatedBy(),goal.getWorkspaceId(),null).withAgent(goal.getAgentId());
Result result;
ChatStreamTracker.QueuedInput queued=streams.consumeQueuedInput(convId);
SegmentResult result;
ConversationInputQueueStore.QueuedInput queued=claimNextInput(convId,claimedRun);
do {
String input=guidance+prompt;
if (queued!=null) {
worker.interactive=true;
if (!queued.persisted()) {
var saved=conversations.saveMessage(convId,"user",queued.message(),queued.contentParts(),"queued");
if (saved!=null) origin=origin.withOriginMessageId(saved.getId());
}
claimedInput.set(queued);
if (queued.agentId()!=null && !queued.agentId().equals(goal.getAgentId())) {
inputQueue.release(queued.id(),queued.claimedByAttemptId(),LocalDateTime.now());
claimedInput.set(null);
throw new IllegalStateException("Queued input targets a different agent; user review required");
}
Long originMessageId=queued.persistedMessageId();
if (originMessageId==null) {
var saved=conversations.saveMessage(convId,"user",queued.message(),queued.contentParts(),"queued");
originMessageId=saved==null ? null : saved.getId();
if (originMessageId==null || !inputQueue.bindMessage(queued.id(),queued.claimedByAttemptId(),
originMessageId,LocalDateTime.now())) {
throw new IllegalStateException("Queued input could not be bound to its persisted message");
}
}
if (!inputQueue.consume(queued.id(),queued.claimedByAttemptId(),LocalDateTime.now())) {
throw new IllegalStateException("Queued input claim was lost before execution");
}
claimedInput.set(null);
worker.interactive=true;
origin=origin.withOriginMessageId(originMessageId);
input=queuedPrompt(queued);
streams.broadcastObject(convId,"queued_input_started",Map.of("conversationId",convId,"message",input));
} else if (goals!=null && goals.getById(goal.getId()).getStatus()!=vip.mate.goal.model.GoalStatus.ACTIVE) {
return new Result("stopped",false);
return new SegmentOutcome.Cancelled("stopped");
}
result=runSegment(goal,input,origin,permit,worker);
if (result.awaitingApproval() || "stopped".equals(result.finishReason())) return result;
queued=streams.consumeQueuedInput(convId);
result=runSegment(goal,input,origin,permit,worker,claimedRun);
if (result.awaitingApproval()) return new SegmentOutcome.AwaitApproval("approval_required");
if ("stopped".equals(result.finishReason())) return new SegmentOutcome.Cancelled("stopped");
queued=claimNextInput(convId,claimedRun);
} while (queued!=null);
return result;
if(result.evaluationUnavailable()) return new SegmentOutcome.Retry("evaluation","evaluation_unavailable");
if("error_fallback".equals(result.finishReason())) {
return new SegmentOutcome.Blocked("graph","graph_error_requires_review");
}
return new SegmentOutcome.Continue(result.finishReason()==null ? "unfinished" : result.finishReason());
} catch (RuntimeException error) {
if (Thread.interrupted()) worker.interrupted.set(true);
// Accepted user input must survive even when this goal cannot continue.
ChatStreamTracker.QueuedInput pending;
while ((pending=streams.consumeQueuedInput(convId))!=null) {
if (!pending.persisted()) conversations.saveMessage(convId,"user",pending.message(),pending.contentParts(),"queued");
}
streams.broadcastObject(convId,"warning",Map.of("message",
"Goal execution interrupted. Queued input was saved; review the execution state before resuming."));
"Goal execution interrupted. Durable queued input remains available for recovery."));
throw error;
} finally {
try {
// Cooperative cancellation returns normally, bypassing the error path.
// Accepted input must still survive process exit, including attachments.
if (worker.cancelled.get()) {
ChatStreamTracker.QueuedInput pending;
while ((pending=streams.consumeQueuedInput(convId))!=null) {
if (!pending.persisted()) conversations.saveMessage(convId,"user",pending.message(),pending.contentParts(),"queued");
}
ConversationInputQueueStore.QueuedInput claimed=claimedInput.getAndSet(null);
if (claimed!=null) {
inputQueue.release(claimed.id(),claimed.claimedByAttemptId(),LocalDateTime.now());
}
} finally {
workers.remove(goal.getId(),worker);
@ -161,7 +187,9 @@ public class GoalSegmentRunner {
}
}
private Result runSegment(GoalEntity goal, String input, ChatOrigin origin, ConversationTurnGate.Permit permit, Worker worker) {
private SegmentResult runSegment(GoalEntity goal, String input, ChatOrigin origin,
ConversationTurnGate.Permit permit, Worker worker,
GoalRunCoordinator.ClaimedRun claimedRun) {
String convId=goal.getConversationId();
var handle=streams.register(convId);
worker.handle.set(handle);
@ -177,6 +205,9 @@ public class GoalSegmentRunner {
Disposable subscription=null;
try {
if (worker.cancelled.get() || Thread.currentThread().isInterrupted()) throw new InterruptedException();
if(claimedRun!=null && !checkpoint(claimedRun,"safe","provider_started",null)) {
throw new IllegalStateException("Goal attempt lost its execution fence");
}
conversations.updateStreamStatus(convId,"running");
streams.broadcastObject(convId,"message_start",Map.of("role","assistant","trigger","goal"));
subscription=gate.withPermit(permit,() -> GoalContinuationContext.call(!worker.interactive, () ->
@ -186,6 +217,11 @@ public class GoalSegmentRunner {
convId,goal.getCreatedBy(),null,origin)
.doOnNext(delta -> {
accumulator.accept(delta,convId);
if(claimedRun!=null && "tool_call_started".equals(delta.eventType())) {
checkpoint(claimedRun,"uncertain","tool_started",null);
} else if(claimedRun!=null && "tool_call_completed".equals(delta.eventType())) {
checkpoint(claimedRun,"resolved","tool_completed",null);
}
if ("goal_evaluated".equals(delta.eventType()) && delta.eventData()!=null
&& (Boolean.TRUE.equals(delta.eventData().get("skipped"))
|| "fallback".equals(delta.eventData().get("decision")))) evaluationUnavailable.set(true);
@ -202,13 +238,17 @@ public class GoalSegmentRunner {
: accumulator.getFinishReason();
String status="stopped".equals(reason) ? "stopped" : "interrupted".equals(reason) ? "interrupted" : accumulator.isAwaitingApproval()
? "awaiting_approval" : failure.get()!=null || "error_fallback".equals(reason) ? "error" : "completed";
persist(convId,accumulator,status);
MessageEntity saved=persist(convId,accumulator,status);
if(claimedRun!=null && !checkpoint(claimedRun,"resolved","message_saved",
saved==null ? null : saved.getId())) {
throw new IllegalStateException("Goal attempt lost its checkpoint fence");
}
persisted.set(true);
streams.broadcastObject(convId,"message_complete",Map.of("status",status,"trigger","goal"));
if (failure.get()!=null && !"stopped".equals(reason)) {
throw failure.get() instanceof RuntimeException runtime ? runtime : new RuntimeException(failure.get());
}
return new Result(reason,accumulator.isAwaitingApproval(),evaluationUnavailable.get());
return new SegmentResult(reason,accumulator.isAwaitingApproval(),evaluationUnavailable.get());
} catch (InterruptedException interrupted) {
worker.interrupted.set(true);
throw new IllegalStateException("Goal worker interrupted; recover from persisted evidence",interrupted);
@ -226,14 +266,25 @@ public class GoalSegmentRunner {
}
}
private void persist(String convId,AgentStreamAccumulator accumulator,String status) {
conversations.saveMessage(convId,"assistant",accumulator.getContent(),accumulator.toAssistantParts(),status,
private MessageEntity persist(String convId,AgentStreamAccumulator accumulator,String status) {
return conversations.saveMessage(convId,"assistant",accumulator.getContent(),accumulator.toAssistantParts(),status,
accumulator.getPromptTokens(),accumulator.getCompletionTokens(),accumulator.getCacheReadTokens(),
accumulator.getCacheWriteTokens(),accumulator.getReasoningTokens(),accumulator.getRuntimeModelName(),
accumulator.getRuntimeProviderId(),accumulator.toMetadataJson());
}
private String queuedPrompt(ChatStreamTracker.QueuedInput queued) {
private boolean checkpoint(GoalRunCoordinator.ClaimedRun run,String safety,String type,Long messageId) {
if(coordinator==null) return true;
return coordinator.checkpoint(run,safety,type,messageId,LocalDateTime.now());
}
private ConversationInputQueueStore.QueuedInput claimNextInput(String conversationId,
GoalRunCoordinator.ClaimedRun claimedRun) {
String claimant=claimedRun==null ? UUID.randomUUID().toString() : claimedRun.attempt().id();
return inputQueue.claimNext(conversationId,claimant,LocalDateTime.now()).orElse(null);
}
private String queuedPrompt(ConversationInputQueueStore.QueuedInput queued) {
if (queued.contentParts()==null || queued.contentParts().isEmpty()) return queued.message();
var message=new vip.mate.workspace.conversation.model.MessageEntity();
message.setContent(queued.message());

41
mateclaw-server/src/main/resources/db/migration/h2/V189__goal_attempt_and_input_queue.sql Normal file
View File

@ -0,0 +1,41 @@
ALTER TABLE mate_goal_continuation ADD COLUMN current_attempt_id VARCHAR(36);
ALTER TABLE mate_goal_continuation ADD COLUMN revision BIGINT NOT NULL DEFAULT 0;
CREATE TABLE mate_goal_attempt (
attempt_id VARCHAR(36) PRIMARY KEY,
goal_id BIGINT NOT NULL,
conversation_id VARCHAR(160) NOT NULL,
parent_attempt_id VARCHAR(36),
trigger_type VARCHAR(32) NOT NULL,
state VARCHAR(32) NOT NULL,
lease_token VARCHAR(64) NOT NULL,
lease_until TIMESTAMP NOT NULL,
input_item_id BIGINT,
assistant_message_id BIGINT,
replay_safety VARCHAR(16) NOT NULL DEFAULT 'safe',
checkpoint_type VARCHAR(32) NOT NULL DEFAULT 'claimed',
finish_reason VARCHAR(128),
error_category VARCHAR(128),
started_at TIMESTAMP,
finished_at TIMESTAMP,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE INDEX idx_goal_attempt_goal_created ON mate_goal_attempt(goal_id, created_at);
CREATE INDEX idx_goal_attempt_expired ON mate_goal_attempt(state, lease_until);
CREATE TABLE mate_conversation_input_queue (
id BIGINT PRIMARY KEY,
conversation_id VARCHAR(160) NOT NULL,
agent_id BIGINT,
created_by VARCHAR(100) NOT NULL,
message TEXT NOT NULL,
content_parts TEXT,
state VARCHAR(16) NOT NULL,
claimed_by_attempt_id VARCHAR(36),
persisted_message_id BIGINT,
cancel_reason VARCHAR(128),
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE INDEX idx_conversation_input_due ON mate_conversation_input_queue(conversation_id, state, id);

41
mateclaw-server/src/main/resources/db/migration/kingbase/V189__goal_attempt_and_input_queue.sql Normal file
View File

@ -0,0 +1,41 @@
ALTER TABLE mate_goal_continuation ADD COLUMN current_attempt_id VARCHAR(36);
ALTER TABLE mate_goal_continuation ADD COLUMN revision BIGINT NOT NULL DEFAULT 0;
CREATE TABLE mate_goal_attempt (
attempt_id VARCHAR(36) PRIMARY KEY,
goal_id BIGINT NOT NULL,
conversation_id VARCHAR(160) NOT NULL,
parent_attempt_id VARCHAR(36),
trigger_type VARCHAR(32) NOT NULL,
state VARCHAR(32) NOT NULL,
lease_token VARCHAR(64) NOT NULL,
lease_until TIMESTAMP NOT NULL,
input_item_id BIGINT,
assistant_message_id BIGINT,
replay_safety VARCHAR(16) NOT NULL DEFAULT 'safe',
checkpoint_type VARCHAR(32) NOT NULL DEFAULT 'claimed',
finish_reason VARCHAR(128),
error_category VARCHAR(128),
started_at TIMESTAMP,
finished_at TIMESTAMP,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE INDEX idx_goal_attempt_goal_created ON mate_goal_attempt(goal_id, created_at);
CREATE INDEX idx_goal_attempt_expired ON mate_goal_attempt(state, lease_until);
CREATE TABLE mate_conversation_input_queue (
id BIGINT PRIMARY KEY,
conversation_id VARCHAR(160) NOT NULL,
agent_id BIGINT,
created_by VARCHAR(100) NOT NULL,
message TEXT NOT NULL,
content_parts TEXT,
state VARCHAR(16) NOT NULL,
claimed_by_attempt_id VARCHAR(36),
persisted_message_id BIGINT,
cancel_reason VARCHAR(128),
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE INDEX idx_conversation_input_due ON mate_conversation_input_queue(conversation_id, state, id);

41
mateclaw-server/src/main/resources/db/migration/mysql/V189__goal_attempt_and_input_queue.sql Normal file
View File

@ -0,0 +1,41 @@
ALTER TABLE mate_goal_continuation ADD COLUMN current_attempt_id VARCHAR(36);
ALTER TABLE mate_goal_continuation ADD COLUMN revision BIGINT NOT NULL DEFAULT 0;
CREATE TABLE mate_goal_attempt (
attempt_id VARCHAR(36) PRIMARY KEY,
goal_id BIGINT NOT NULL,
conversation_id VARCHAR(160) NOT NULL,
parent_attempt_id VARCHAR(36),
trigger_type VARCHAR(32) NOT NULL,
state VARCHAR(32) NOT NULL,
lease_token VARCHAR(64) NOT NULL,
lease_until DATETIME(6) NOT NULL,
input_item_id BIGINT,
assistant_message_id BIGINT,
replay_safety VARCHAR(16) NOT NULL DEFAULT 'safe',
checkpoint_type VARCHAR(32) NOT NULL DEFAULT 'claimed',
finish_reason VARCHAR(128),
error_category VARCHAR(128),
started_at DATETIME(6),
finished_at DATETIME(6),
created_at DATETIME(6) NOT NULL,
updated_at DATETIME(6) NOT NULL,
INDEX idx_goal_attempt_goal_created(goal_id, created_at),
INDEX idx_goal_attempt_expired(state, lease_until)
);
CREATE TABLE mate_conversation_input_queue (
id BIGINT PRIMARY KEY,
conversation_id VARCHAR(160) NOT NULL,
agent_id BIGINT,
created_by VARCHAR(100) NOT NULL,
message LONGTEXT NOT NULL,
content_parts LONGTEXT,
state VARCHAR(16) NOT NULL,
claimed_by_attempt_id VARCHAR(36),
persisted_message_id BIGINT,
cancel_reason VARCHAR(128),
created_at DATETIME(6) NOT NULL,
updated_at DATETIME(6) NOT NULL,
INDEX idx_conversation_input_due(conversation_id, state, id)
);

13
mateclaw-server/src/main/resources/docs/en/goals.md
View File

@ -312,6 +312,12 @@ mateclaw:
default-auto-followup: true
# Runtime master switch; when off, no goal injects a followup regardless of its per-goal flag.
allow-auto-followup: true
# Persistent goal segments that may execute concurrently in one backend instance.
max-concurrent-segments: 4
# Global minimum delay in seconds before an ordinary segment continues.
minimum-continuation-interval-seconds: 1
# Pause new claims after a retryable provider/evaluator failure (seconds; 0 disables it).
provider-failure-global-backoff-seconds: 30
# New goals run persistently; omitted budgets mean unlimited (0).
default-persistent-execution: true
supervisor-poll-ms: 5000
@ -334,6 +340,13 @@ mateclaw:
evaluator-context-messages: 8
```
The effective continuation delay is the larger of
`minimum-continuation-interval-seconds` and the goal's `followupCooldownSeconds`.
The provider-wide backoff is instance-local; restart recovery continues to use the durable
continuation `next_run_at` and per-goal failure count. For soak tests or constrained model
capacity, start with concurrency `1`, a `300` second minimum interval, and a `300` second
provider backoff, then increase load only after reviewing logs.
---
## Database

11
mateclaw-server/src/main/resources/docs/zh/goals.md
View File

@ -310,6 +310,12 @@ mateclaw:
default-auto-followup: true
# 运行期总开关；关掉则无论 per-goal 标志如何，都不注入自动延续
allow-auto-followup: true
# 单后端实例同时运行的持久化目标 Segment 上限
max-concurrent-segments: 4
# 普通 Segment 结算后再次续跑的全局最小间隔（秒）
minimum-continuation-interval-seconds: 1
# provider 或评估器发生可重试故障后，暂停认领其他目标的时间（秒；0 = 关闭）
provider-failure-global-backoff-seconds: 30
# 新目标默认持续模式；省略预算表示不限（0）
default-persistent-execution: true
supervisor-poll-ms: 5000
@ -331,6 +337,11 @@ mateclaw:
evaluator-context-messages: 8
```
`minimum-continuation-interval-seconds` 与目标自身的 `followupCooldownSeconds` 取较大值。
provider 全局退避只保存在当前后端实例内；重启恢复仍以数据库中的 continuation
`next_run_at` 和每目标失败次数为准。耐久或低配模型测试建议使用并发 `1`、最小间隔
`300` 秒、provider 全局退避 `300` 秒，再根据日志逐步放量。
---
## 数据库

63
mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerDurableQueueTest.java Normal file
View File

@ -0,0 +1,63 @@
package vip.mate.channel.web;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.springframework.security.core.Authentication;
import vip.mate.agent.AgentService;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.channel.web.ConversationInputQueueStore.QueuedInput;
import vip.mate.memory.event.ConversationCompletionPublisher;
import vip.mate.memory.identity.MemoryOwnerResolver;
import vip.mate.tool.document.preview.OfficePreviewService;
import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.core.service.ChatUploadLocationResolver;
import java.time.LocalDateTime;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
class ChatControllerDurableQueueTest {
@Test
void interruptPersistsInputBeforePublishingAcceptance() {
AgentService agents = mock(AgentService.class);
ConversationService conversations = mock(ConversationService.class);
ApprovalWorkflowService approvals = mock(ApprovalWorkflowService.class);
ChatStreamTracker streams = mock(ChatStreamTracker.class);
ConversationInputQueueStore queue = mock(ConversationInputQueueStore.class);
Authentication authentication = mock(Authentication.class);
when(authentication.getName()).thenReturn("alice");
when(conversations.isConversationOwner("conv", "alice")).thenReturn(true);
when(streams.isRunning("conv")).thenReturn(true);
when(streams.notifyQueuedInput("conv")).thenReturn(true);
QueuedInput stored = new QueuedInput(91L, "conv", 2L, "alice", "follow-up",
List.of(), "queued", null, null, null,
LocalDateTime.now(), LocalDateTime.now());
when(queue.enqueue(eq("conv"), eq(2L), eq("alice"), eq("follow-up"),
eq(List.of()), any())).thenReturn(stored);
ChatController controller = new ChatController(agents, conversations, approvals, streams,
new ObjectMapper(), mock(ConversationCompletionPublisher.class),
mock(MemoryOwnerResolver.class), mock(ChatUploadLocationResolver.class),
mock(OfficePreviewService.class), queue);
ChatController.InterruptRequest request = new ChatController.InterruptRequest();
request.setMessage("follow-up");
request.setAgentId(2L);
request.setContentParts(List.of());
var response = controller.interruptStream("conv", request, authentication);
assertThat(response.getData()).containsEntry("queued", true)
.containsEntry("queueItemId", "91");
var order = inOrder(queue, streams);
order.verify(queue).enqueue(eq("conv"), eq(2L), eq("alice"),
eq("follow-up"), eq(List.of()), any());
order.verify(streams).notifyQueuedInput("conv");
}
}

3
mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerPreviewRouteTest.java
View File

@ -62,7 +62,8 @@ class ChatControllerPreviewRouteTest {
mock(vip.mate.memory.event.ConversationCompletionPublisher.class),
mock(MemoryOwnerResolver.class),
uploadLocationResolver,
officePreviewService);
officePreviewService,
mock(ConversationInputQueueStore.class));
mockMvc = MockMvcBuilders.standaloneSetup(controller).build();
}

3
mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerWorkerReadOnlyTest.java
View File

@ -36,6 +36,7 @@ class ChatControllerWorkerReadOnlyTest {
@Mock private MemoryOwnerResolver memoryOwnerResolver;
@Mock private ChatUploadLocationResolver uploadLocationResolver;
@Mock private OfficePreviewService officePreviewService;
@Mock private ConversationInputQueueStore inputQueue;
@Mock private Authentication authentication;
private ChatController controller;
@ -45,7 +46,7 @@ class ChatControllerWorkerReadOnlyTest {
void setUp() {
controller = new ChatController(agentService, conversationService, approvalService,
streamTracker, objectMapper, completionPublisher, memoryOwnerResolver,
uploadLocationResolver, officePreviewService);
uploadLocationResolver, officePreviewService, inputQueue);
ReflectionTestUtils.setField(controller, "turnGate", gate);
}

37
mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java
View File

@ -4,7 +4,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
class ChatStreamTrackerQueueDrainTest {
@ -13,40 +14,22 @@ class ChatStreamTrackerQueueDrainTest {
}
@Test
@DisplayName("Queued inputs survive RunState replacement and drain FIFO")
void queuedInputsSurviveRunStateReplacementAndDrainFifo() {
@DisplayName("Run state tracks only a durable queue wake signal")
void runStateTracksOnlyDurableQueueWakeSignal() {
ChatStreamTracker tracker = newTracker();
String conversationId = "queue-drain";
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
assertTrue(tracker.enqueueMessage(conversationId, "q1", 101L, false));
assertTrue(tracker.enqueueMessage(conversationId, "q2", 101L, false));
assertTrue(tracker.enqueueMessage(conversationId, "q3", 101L, false));
assertTrue(tracker.notifyQueuedInput(conversationId));
assertTrue(tracker.hasQueuedInputNotification(conversationId));
ChatStreamTracker.CompletionResult first = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(first.allDone());
assertNotNull(first.queuedInput());
assertEquals("q1", first.queuedInput().message());
ChatStreamTracker.CompletionResult completed = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(completed.allDone());
assertFalse(tracker.notifyQueuedInput(conversationId));
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult second = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(second.allDone());
assertNotNull(second.queuedInput());
assertEquals("q2", second.queuedInput().message());
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult third = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(third.allDone());
assertNotNull(third.queuedInput());
assertEquals("q3", third.queuedInput().message());
tracker.register(conversationId);
tracker.incrementFlux(conversationId);
ChatStreamTracker.CompletionResult empty = tracker.completeAndConsumeIfLast(conversationId);
assertTrue(empty.allDone());
assertNull(empty.queuedInput());
assertFalse(tracker.hasQueuedInputNotification(conversationId));
}
}

71
mateclaw-server/src/test/java/vip/mate/channel/web/ConversationInputQueueStoreTest.java Normal file
View File

@ -0,0 +1,71 @@
package vip.mate.channel.web;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import vip.mate.channel.web.ConversationInputQueueStore.QueuedInput;
import vip.mate.workspace.conversation.model.MessageContentPart;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
class ConversationInputQueueStoreTest {
private JdbcTemplate jdbc;
private ObjectMapper mapper;
private ConversationInputQueueStore store;
private final LocalDateTime now = LocalDateTime.of(2026, 8, 27, 9, 0);
@BeforeEach
void setUp() {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID()
+ ";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1");
new ResourceDatabasePopulator(
new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"))
.execute(dataSource);
jdbc = new JdbcTemplate(dataSource);
mapper = new ObjectMapper();
store = new ConversationInputQueueStore(jdbc, mapper);
}
@Test
void fifoClaimSurvivesStoreReconstructionAndPreservesAttachments() {
MessageContentPart attachment = MessageContentPart.file("media-1", "report.pdf", "application/pdf");
QueuedInput first = store.enqueue("conv", 1L, "mate", "one", List.of(attachment), now);
QueuedInput second = store.enqueue("conv", 1L, "mate", "two", List.of(), now.plusNanos(1));
QueuedInput claimed = store.claimNext("conv", "attempt-a", now.plusSeconds(1)).orElseThrow();
assertThat(claimed.id()).isEqualTo(first.id());
assertThat(claimed.contentParts()).singleElement().extracting(MessageContentPart::getFileName)
.isEqualTo("report.pdf");
assertThat(store.bindMessage(first.id(), "attempt-a", 101L, now.plusSeconds(2))).isTrue();
assertThat(store.consume(first.id(), "attempt-a", now.plusSeconds(3))).isTrue();
ConversationInputQueueStore restarted = new ConversationInputQueueStore(jdbc, mapper);
assertThat(restarted.claimNext("conv", "attempt-b", now.plusSeconds(4)))
.get().extracting(QueuedInput::id).isEqualTo(second.id());
assertThat(restarted.get(first.id()).persistedMessageId()).isEqualTo(101L);
assertThat(restarted.get(first.id()).state()).isEqualTo("consumed");
}
@Test
void claimReleaseAndCancellationAreFencedByAttempt() {
QueuedInput input = store.enqueue("conv", 1L, "mate", "queued", List.of(), now);
assertThat(store.claimNext("conv", "attempt-a", now.plusSeconds(1))).isPresent();
assertThat(store.release(input.id(), "attempt-b", now.plusSeconds(2))).isFalse();
assertThat(store.release(input.id(), "attempt-a", now.plusSeconds(2))).isTrue();
assertThat(store.countQueued("conv")).isEqualTo(1);
assertThat(store.cancel(input.id(), "stream_finished", now.plusSeconds(3))).isTrue();
assertThat(store.claimNext("conv", "attempt-c", now.plusSeconds(4))).isEmpty();
}
}

6
mateclaw-server/src/test/java/vip/mate/goal/controller/GoalExecutionControllerTest.java
View File

@ -5,6 +5,7 @@ import org.springframework.security.authentication.UsernamePasswordAuthenticatio
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.service.GoalService;
import vip.mate.goal.service.GoalContinuationStore;
import vip.mate.goal.service.GoalAttemptStore;
import vip.mate.workspace.conversation.ConversationService;
import static org.mockito.Mockito.*;
import static org.junit.jupiter.api.Assertions.*;
@ -16,12 +17,15 @@ class GoalExecutionControllerTest {
var conversations=mock(ConversationService.class);
var goal=new GoalEntity();goal.setConversationId("private");
when(goals.getById(1L)).thenReturn(goal);
var controller=new GoalExecutionController(goals,store,conversations);
var attempts=mock(GoalAttemptStore.class);
var controller=new GoalExecutionController(goals,store,conversations,attempts);
var user=new UsernamePasswordAuthenticationToken("alice","ignored");
assertThrows(vip.mate.exception.MateClawException.class,()->controller.execution(1L,user));
verifyNoInteractions(store);
when(conversations.isConversationOwner("private","alice")).thenReturn(true);
controller.execution(1L,user);
verify(store).get(1L);
controller.attempts(1L,user);
verify(attempts).listRecent(1L,50);
}
}

67
mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java Normal file
View File

@ -0,0 +1,67 @@
package vip.mate.goal.service;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import vip.mate.goal.model.GoalAttempt;
import java.time.LocalDateTime;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
class GoalAttemptStoreTest {
private GoalAttemptStore store;
private final LocalDateTime now = LocalDateTime.of(2026, 8, 27, 9, 0);
@BeforeEach
void setUp() {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID()
+ ";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1");
new ResourceDatabasePopulator(
new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"))
.execute(dataSource);
store = new GoalAttemptStore(new JdbcTemplate(dataSource));
}
@Test
void lifecycleIsLeaseFencedAndTerminalRowsAreImmutable() {
GoalAttempt attempt = store.create(7L, "conv-7", null, "continuation",
"lease-a", now.plusMinutes(1), null, now);
assertThat(store.markRunning(attempt.id(), "wrong", now.plusSeconds(1))).isFalse();
assertThat(store.markRunning(attempt.id(), "lease-a", now.plusSeconds(1))).isTrue();
assertThat(store.checkpoint(attempt.id(), "lease-a", "uncertain", "tool_started", null,
now.plusSeconds(2))).isTrue();
assertThat(store.finish(attempt.id(), "wrong", "succeeded", "normal", null,
now.plusSeconds(3))).isFalse();
assertThat(store.finish(attempt.id(), "lease-a", "succeeded", "normal", null,
now.plusSeconds(3))).isTrue();
assertThat(store.markRunning(attempt.id(), "lease-a", now.plusSeconds(4))).isFalse();
GoalAttempt finished = store.get(attempt.id());
assertThat(finished.state()).isEqualTo("succeeded");
assertThat(finished.replaySafety()).isEqualTo("uncertain");
assertThat(finished.checkpointType()).isEqualTo("tool_started");
assertThat(finished.finishedAt()).isEqualTo(now.plusSeconds(3));
}
@Test
void historyPreservesRecoveryParentAndCreationOrder() {
GoalAttempt first = store.create(7L, "conv-7", null, "continuation",
"lease-a", now.plusMinutes(1), null, now);
GoalAttempt recovery = store.create(7L, "conv-7", first.id(), "recovery",
"lease-b", now.plusMinutes(2), 91L, now.plusSeconds(5));
assertThat(store.listRecent(7L, 10)).extracting(GoalAttempt::id)
.containsExactly(recovery.id(), first.id());
assertThat(store.get(recovery.id()).parentAttemptId()).isEqualTo(first.id());
assertThat(store.get(recovery.id()).inputItemId()).isEqualTo(91L);
}
}

11
mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java
View File

@ -22,7 +22,8 @@ class GoalContinuationStoreTest {
ds.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1");
new ResourceDatabasePopulator(
new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql")).execute(ds);
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds);
jdbc = new JdbcTemplate(ds);
store = new GoalContinuationStore(jdbc);
}
@ -101,6 +102,9 @@ class GoalContinuationStoreTest {
goal(1,true,"active");
var goals=org.mockito.Mockito.mock(GoalService.class);
var runner=org.mockito.Mockito.mock(GoalSegmentRunner.class);
var coordinator=new GoalRunCoordinator(new GoalContinuationStore(jdbc),new GoalAttemptStore(jdbc),goals,
new vip.mate.goal.config.GoalProperties());
var recovery=org.mockito.Mockito.mock(GoalRecoveryService.class);
var running=new vip.mate.agent.runtime.RunningConversationRegistry();
var streams=new vip.mate.channel.web.ChatStreamTracker(new com.fasterxml.jackson.databind.ObjectMapper());
var properties=new vip.mate.goal.config.GoalProperties();
@ -110,18 +114,19 @@ class GoalContinuationStoreTest {
entity.setTurnBudget(0);entity.setLlmCallBudget(0);
org.mockito.Mockito.when(goals.getById(1L)).thenReturn(entity);
var count=new java.util.concurrent.atomic.AtomicInteger();
org.mockito.Mockito.when(runner.run(org.mockito.ArgumentMatchers.any(),org.mockito.ArgumentMatchers.anyString(),org.mockito.ArgumentMatchers.anyBoolean()))
org.mockito.Mockito.when(runner.run(org.mockito.ArgumentMatchers.any(GoalRunCoordinator.ClaimedRun.class),org.mockito.ArgumentMatchers.anyString(),org.mockito.ArgumentMatchers.anyBoolean()))
.thenAnswer(inv -> {
if(count.incrementAndGet()==12) {
entity.setStatus(vip.mate.goal.model.GoalStatus.COMPLETED);
jdbc.update("UPDATE mate_agent_goal SET status='completed' WHERE id=1");
}
return new GoalSegmentRunner.Result("normal",false);
return new vip.mate.goal.model.SegmentOutcome.Continue("normal");
});
for(int i=0;i<15;i++) {
// Recreate all scheduler state between segments, as after a server restart.
var scheduler=new GoalContinuationSupervisor(new GoalContinuationStore(jdbc),goals,properties,
new GoalFollowupService(properties,new com.fasterxml.jackson.databind.ObjectMapper()),runner,running,streams,
coordinator,recovery,
java.time.Clock.fixed(now.plusSeconds(i*5L).toInstant(java.time.ZoneOffset.UTC),java.time.ZoneOffset.UTC),Runnable::run);
scheduler.tick();
}

173
mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java
View File

@ -1,15 +1,21 @@
package vip.mate.goal.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import vip.mate.agent.runtime.RunningConversationRegistry;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.goal.config.GoalProperties;
import vip.mate.goal.model.GoalAttempt;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.GoalStatus;
import com.fasterxml.jackson.databind.ObjectMapper;
import vip.mate.goal.model.SegmentOutcome;
import java.time.*;
import java.util.ArrayList;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
@ -17,96 +23,181 @@ class GoalContinuationSupervisorTest {
GoalContinuationStore store = mock(GoalContinuationStore.class);
GoalService goals = mock(GoalService.class);
GoalSegmentRunner runner = mock(GoalSegmentRunner.class);
GoalRunCoordinator coordinator = mock(GoalRunCoordinator.class);
GoalRecoveryService recovery = mock(GoalRecoveryService.class);
RunningConversationRegistry running = mock(RunningConversationRegistry.class);
ChatStreamTracker streams = mock(ChatStreamTracker.class);
GoalProperties properties = new GoalProperties();
LocalDateTime now = LocalDateTime.of(2026,8,26,12,0);
GoalEntity goal = new GoalEntity();
GoalContinuationStore.Continuation candidate;
GoalRunCoordinator.ClaimedRun claimed;
GoalContinuationSupervisor supervisor;
@BeforeEach void setup() {
goal.setId(1L); goal.setConversationId("conv"); goal.setStatus(GoalStatus.ACTIVE);
goal.setPersistentExecution(true); goal.setAutoFollowupEnabled(true);
goal.setTitle("full goal"); goal.setTurnBudget(0); goal.setLlmCallBudget(0);
candidate=new GoalContinuationStore.Continuation(1L,"conv","queued",now,null,null,0,"",null,0);
GoalAttempt attempt=new GoalAttempt("attempt",1L,"conv",null,"continuation","claimed","lease",
now.plusSeconds(60),null,null,"safe","claimed",null,null,null,null,now,now);
claimed=new GoalRunCoordinator.ClaimedRun(candidate,goal,attempt,2);
when(goals.getById(1L)).thenReturn(goal);
when(store.due(any(), anyInt())).thenReturn(List.of(new GoalContinuationStore.Continuation(
1L,"conv","queued",now,null,null,0,"")));
when(store.claim(eq(1L),anyString(),any(),any())).thenReturn(true);
when(store.settle(eq(1L),anyString(),anyString(),any(),anyInt(),anyString())).thenReturn(true);
when(runner.run(any(),anyString(),anyBoolean())).thenReturn(new GoalSegmentRunner.Result("normal",false));
when(store.due(any(), anyInt())).thenReturn(List.of(candidate));
when(coordinator.claim(candidate,goal,now)).thenReturn(claimed);
when(coordinator.markRunning(claimed,now)).thenReturn(true);
when(coordinator.settle(eq(claimed),any(),any())).thenReturn(true);
when(runner.run(eq(claimed),anyString(),anyBoolean())).thenReturn(new SegmentOutcome.Continue("normal"));
supervisor = new GoalContinuationSupervisor(store, goals, properties,
new GoalFollowupService(properties,new ObjectMapper()), runner, running, streams,
new GoalFollowupService(properties,new ObjectMapper()), runner, running, streams,coordinator,recovery,
Clock.fixed(now.toInstant(ZoneOffset.UTC),ZoneOffset.UTC), Runnable::run);
}
@Test void incompleteGoalIsRescheduledAcrossMultipleSegments() {
supervisor.tick(); supervisor.tick(); supervisor.tick();
verify(runner,times(3)).run(eq(goal),contains("full goal"),eq(false));
verify(store,times(3)).settle(eq(1L),anyString(),eq("queued"),any(),eq(0),anyString());
verify(runner,times(3)).run(eq(claimed),contains("full goal"),eq(false));
verify(coordinator,times(3)).settle(eq(claimed),isA(SegmentOutcome.Continue.class),eq(now));
}
@Test void configuredConcurrencyLimitsSubmittedSegments() {
properties.setMaxConcurrentSegments(1);
GoalEntity second = new GoalEntity();
second.setId(2L); second.setConversationId("conv-2"); second.setStatus(GoalStatus.ACTIVE);
second.setPersistentExecution(true); second.setAutoFollowupEnabled(true);
second.setTitle("second goal"); second.setTurnBudget(0); second.setLlmCallBudget(0);
var secondCandidate = new GoalContinuationStore.Continuation(
2L,"conv-2","queued",now,null,null,0,"",null,0);
GoalAttempt secondAttempt = new GoalAttempt("attempt-2",2L,"conv-2",null,"continuation",
"claimed","lease-2",now.plusSeconds(60),null,null,"safe","claimed",
null,null,null,null,now,now);
var secondClaimed = new GoalRunCoordinator.ClaimedRun(secondCandidate,second,secondAttempt,2);
when(goals.getById(2L)).thenReturn(second);
when(store.due(any(), anyInt())).thenReturn(List.of(candidate,secondCandidate));
when(coordinator.claim(secondCandidate,second,now)).thenReturn(secondClaimed);
List<Runnable> submitted = new ArrayList<>();
supervisor = new GoalContinuationSupervisor(store, goals, properties,
new GoalFollowupService(properties,new ObjectMapper()), runner, running, streams,
coordinator,recovery,Clock.fixed(now.toInstant(ZoneOffset.UTC),ZoneOffset.UTC),submitted::add);
supervisor.tick();
assertEquals(1,submitted.size());
verify(coordinator).claim(candidate,goal,now);
verify(coordinator,never()).claim(secondCandidate,second,now);
}
@Test void busyFirstCandidateDoesNotStarveAnotherDueGoalAtConcurrencyOne() {
properties.setMaxConcurrentSegments(1);
GoalEntity second = new GoalEntity();
second.setId(2L); second.setConversationId("conv-2"); second.setStatus(GoalStatus.ACTIVE);
second.setPersistentExecution(true); second.setAutoFollowupEnabled(true);
second.setTitle("second goal"); second.setTurnBudget(0); second.setLlmCallBudget(0);
var secondCandidate = new GoalContinuationStore.Continuation(
2L,"conv-2","queued",now,null,null,0,"",null,0);
GoalAttempt secondAttempt = new GoalAttempt("attempt-2",2L,"conv-2",null,"continuation",
"claimed","lease-2",now.plusSeconds(60),null,null,"safe","claimed",
null,null,null,null,now,now);
var secondClaimed = new GoalRunCoordinator.ClaimedRun(secondCandidate,second,secondAttempt,2);
when(goals.getById(2L)).thenReturn(second);
when(store.due(any(), anyInt())).thenAnswer(invocation -> {
int limit=invocation.getArgument(1);
return List.of(candidate,secondCandidate).subList(0,Math.min(limit,2));
});
when(running.isActive("conv")).thenReturn(true);
when(coordinator.claim(secondCandidate,second,now)).thenReturn(secondClaimed);
when(coordinator.markRunning(secondClaimed,now)).thenReturn(true);
when(coordinator.settle(eq(secondClaimed),any(),any())).thenReturn(true);
when(runner.run(eq(secondClaimed),anyString(),anyBoolean()))
.thenReturn(new SegmentOutcome.Continue("normal"));
supervisor.tick();
verify(coordinator).claim(secondCandidate,second,now);
}
@Test void cooldownIsDurablyDeferredWithoutCallingModel() {
goal.setFollowupCooldownSeconds(60); goal.setLastFollowupAt(now.minusSeconds(10));
supervisor.tick();
verifyNoInteractions(runner);
verify(store).settle(eq(1L),anyString(),eq("queued"),eq(now.plusSeconds(50)),eq(0),anyString());
verify(coordinator).settle(eq(claimed),argThat(outcome -> outcome instanceof SegmentOutcome.Defer defer
&& now.plusSeconds(50).equals(defer.nextRunAt())),eq(now));
}
@Test void neverStartsAlongsideUserTurnOrQueuedInput() {
when(running.isActive("conv")).thenReturn(true);
supervisor.tick();
verify(store,never()).claim(any(),any(),any(),any());
verify(coordinator,never()).claim(any(),any(),any());
verifyNoInteractions(runner);
}
@Test void completionPreventsAnotherTurn() {
when(runner.run(any(),anyString(),anyBoolean())).thenAnswer(inv -> {
goal.setStatus(GoalStatus.COMPLETED); return new GoalSegmentRunner.Result("normal",false);
@Test void completionIsSettledThroughCoordinator() {
when(runner.run(eq(claimed),anyString(),anyBoolean())).thenAnswer(inv -> {
goal.setStatus(GoalStatus.COMPLETED); return new SegmentOutcome.Continue("normal");
});
supervisor.tick(); supervisor.tick();
verify(runner).run(any(),anyString(),anyBoolean());
verify(store).settle(eq(1L),anyString(),eq("completed"),any(),eq(0),anyString());
verify(runner).run(eq(claimed),anyString(),anyBoolean());
verify(coordinator).settle(eq(claimed),isA(SegmentOutcome.Continue.class),eq(now));
}
@Test void shutdownCancellationLeavesLeaseForRestartRecovery() {
when(runner.run(any(),anyString(),anyBoolean())).thenAnswer(inv -> {
supervisor.close();
return new GoalSegmentRunner.Result("stopped",false);
when(runner.run(eq(claimed),anyString(),anyBoolean())).thenAnswer(inv -> {
supervisor.close(); return new SegmentOutcome.Cancelled("stopped");
});
supervisor.tick();
verify(runner).cancelAll();
verify(store,never()).settle(any(),anyString(),anyString(),any(),anyInt(),anyString());
verify(coordinator,never()).settle(any(),any(),any());
}
@Test void budgetReachedDuringSegmentIsReportedAsResumableBudgetLimit() {
when(runner.run(any(),anyString(),anyBoolean())).thenAnswer(inv -> {
goal.setStatus(GoalStatus.PAUSED);
when(goals.isBudgetExhausted(goal)).thenReturn(true);
when(goals.exhaustionReason(goal)).thenReturn("turn_budget");
return new GoalSegmentRunner.Result("normal",false);
});
@Test void transientFailureGetsRetryAndPermanentErrorBlocks() {
properties.setProviderFailureGlobalBackoffSeconds(0);
when(runner.run(eq(claimed),anyString(),anyBoolean()))
.thenThrow(new java.io.UncheckedIOException(new java.io.IOException("connection reset")));
supervisor.tick();
verify(store).settle(eq(1L),anyString(),eq("budget_limited"),any(),eq(0),eq("turn_budget"));
verify(coordinator).settle(eq(claimed),isA(SegmentOutcome.Retry.class),eq(now));
reset(runner,coordinator);
when(coordinator.claim(candidate,goal,now)).thenReturn(claimed);
when(coordinator.markRunning(claimed,now)).thenReturn(true);
when(coordinator.settle(eq(claimed),any(),any())).thenReturn(true);
doThrow(new IllegalArgumentException("invalid configuration")).when(runner)
.run(eq(claimed),anyString(),anyBoolean());
supervisor.tick();
verify(coordinator).settle(eq(claimed),isA(SegmentOutcome.Blocked.class),eq(now));
}
@Test void transientFailureGetsBackoffAndPermanentErrorBlocks() {
when(runner.run(any(),anyString(),anyBoolean())).thenThrow(new java.io.UncheckedIOException(new java.io.IOException("connection reset")));
@Test void retryableProviderFailureStopsNewClaimsDuringGlobalBackoff() {
properties.setProviderFailureGlobalBackoffSeconds(300);
when(runner.run(eq(claimed),anyString(),anyBoolean()))
.thenThrow(new java.io.UncheckedIOException(new java.io.IOException("provider unavailable")));
supervisor.tick();
verify(store).settle(eq(1L),anyString(),eq("retry"),eq(now.plusSeconds(5)),eq(1),anyString());
reset(store);
when(store.due(any(),anyInt())).thenReturn(List.of(new GoalContinuationStore.Continuation(1L,"conv","queued",now,null,null,0,"")));
when(store.claim(any(),any(),any(),any())).thenReturn(true);
doThrow(new IllegalArgumentException("invalid configuration")).when(runner).run(any(),anyString(),anyBoolean());
supervisor.tick();
verify(store).settle(eq(1L),anyString(),eq("blocked"),any(),eq(1),anyString());
verify(coordinator,times(1)).claim(candidate,goal,now);
verify(coordinator,times(1)).settle(eq(claimed),isA(SegmentOutcome.Retry.class),eq(now));
}
@Test void approvalAndStopNeverTurnIntoAutomaticRetry() {
when(runner.run(any(),anyString(),anyBoolean())).thenReturn(new GoalSegmentRunner.Result("normal",true));
@Test void retryOutcomeFromUnavailableEvaluationAlsoStartsGlobalBackoff() {
properties.setProviderFailureGlobalBackoffSeconds(300);
when(runner.run(eq(claimed),anyString(),anyBoolean()))
.thenReturn(new SegmentOutcome.Retry("evaluation","evaluation_unavailable"));
supervisor.tick();
verify(store).settle(eq(1L),anyString(),eq("waiting_approval"),any(),eq(0),anyString());
when(runner.run(any(),anyString(),anyBoolean())).thenReturn(new GoalSegmentRunner.Result("stopped",false));
supervisor.tick();
verify(store).settle(eq(1L),anyString(),eq("paused"),any(),eq(0),anyString());
verify(coordinator,times(1)).claim(candidate,goal,now);
}
@Test void approvalAndStopNeverBecomeAutomaticRetry() {
when(runner.run(eq(claimed),anyString(),anyBoolean()))
.thenReturn(new SegmentOutcome.AwaitApproval("approval_required"));
supervisor.tick();
verify(coordinator).settle(eq(claimed),isA(SegmentOutcome.AwaitApproval.class),eq(now));
reset(runner,coordinator);
when(coordinator.claim(candidate,goal,now)).thenReturn(claimed);
when(coordinator.markRunning(claimed,now)).thenReturn(true);
when(coordinator.settle(eq(claimed),any(),any())).thenReturn(true);
when(runner.run(eq(claimed),anyString(),anyBoolean())).thenReturn(new SegmentOutcome.Cancelled("stopped"));
supervisor.tick();
verify(coordinator).settle(eq(claimed),isA(SegmentOutcome.Cancelled.class),eq(now));
}
}

93
mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java Normal file
View File

@ -0,0 +1,93 @@
package vip.mate.goal.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import vip.mate.channel.web.ConversationInputQueueStore;
import vip.mate.goal.model.GoalAttempt;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.GoalStatus;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
class GoalRecoveryServiceTest {
JdbcTemplate jdbc;
GoalAttemptStore attempts;
GoalContinuationStore continuations;
ConversationInputQueueStore inputs;
GoalService goals=mock(GoalService.class);
GoalRunCoordinator coordinator;
GoalRecoveryService recovery;
GoalEntity goal;
LocalDateTime now=LocalDateTime.of(2026,8,27,2,0);
@BeforeEach void setup() {
JdbcDataSource ds=new JdbcDataSource();
ds.setURL("jdbc:h2:mem:"+ UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1");
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds);
jdbc=new JdbcTemplate(ds);attempts=new GoalAttemptStore(jdbc);continuations=new GoalContinuationStore(jdbc);
inputs=new ConversationInputQueueStore(jdbc,new ObjectMapper());
coordinator=new GoalRunCoordinator(continuations,attempts,goals,new vip.mate.goal.config.GoalProperties());
recovery=new GoalRecoveryService(attempts,continuations,inputs,goals);
jdbc.update("""
INSERT INTO mate_agent_goal(id,conversation_id,agent_id,workspace_id,created_by,title,
description,status,persistent_execution,auto_followup_enabled,create_time,update_time)
VALUES(1,'conv',2,3,'alice','goal','objective','active',TRUE,TRUE,?,?)
""",now,now);
goal=new GoalEntity();goal.setId(1L);goal.setConversationId("conv");goal.setAgentId(2L);
goal.setWorkspaceId(3L);goal.setCreatedBy("alice");goal.setStatus(GoalStatus.ACTIVE);
goal.setPersistentExecution(true);goal.setAutoFollowupEnabled(true);
when(goals.getById(1L)).thenReturn(goal);
continuations.discover(now);
}
@Test void classifiesCheckpointRecoveryMatrix() {
assertEquals(GoalRecoveryService.RecoveryDecision.RETRY_SAFE,recovery.classify(attempt("claimed","safe",null)));
assertEquals(GoalRecoveryService.RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT,
recovery.classify(attempt("tool_started","uncertain",null)));
assertEquals(GoalRecoveryService.RecoveryDecision.RECONCILE_MESSAGE,
recovery.classify(attempt("message_saved","resolved",42L)));
assertEquals(GoalRecoveryService.RecoveryDecision.RESUME_FROM_EVIDENCE,
recovery.classify(attempt("tool_completed","resolved",null)));
}
@Test void expiredSafeAttemptRequeuesAndReleasesClaimedInputWithParentLink() {
var old=coordinator.claim(continuations.get(1L),goal,now);
assertTrue(coordinator.markRunning(old,now));
var queued=inputs.enqueue("conv",2L,"alice","follow up",List.of(),now);
assertTrue(inputs.claimNext("conv",old.attempt().id(),now).isPresent());
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61)));
assertEquals("retryable",attempts.get(old.attempt().id()).state());
assertEquals("retry",continuations.get(1L).state());
assertEquals(1,inputs.countQueued("conv"));
var next=coordinator.claim(continuations.get(1L),goal,now.plusSeconds(61));
assertEquals(old.attempt().id(),next.attempt().parentAttemptId());
assertEquals(queued.id(),inputs.listQueued("conv").getFirst().id());
}
@Test void uncertainToolAttemptBlocksInsteadOfReplaying() {
var old=coordinator.claim(continuations.get(1L),goal,now);
assertTrue(coordinator.markRunning(old,now));
assertTrue(coordinator.checkpoint(old,"uncertain","tool_started",null,now.plusSeconds(1)));
assertEquals(1,recovery.recoverExpired(now.plusSeconds(61)));
assertEquals("blocked",attempts.get(old.attempt().id()).state());
assertEquals("blocked",continuations.get(1L).state());
verify(goals).pause(1L,"alice");
}
private GoalAttempt attempt(String checkpoint,String safety,Long messageId) {
return new GoalAttempt("a",1L,"conv",null,"continuation","running","lease",now,
null,messageId,safety,checkpoint,null,null,now,null,now,now);
}
}

90
mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java Normal file
View File

@ -0,0 +1,90 @@
package vip.mate.goal.service;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
import vip.mate.goal.config.GoalProperties;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.GoalStatus;
import vip.mate.goal.model.SegmentOutcome;
import java.time.LocalDateTime;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
class GoalRunCoordinatorTest {
JdbcTemplate jdbc;
GoalContinuationStore continuations;
GoalAttemptStore attempts;
GoalService goals=mock(GoalService.class);
GoalProperties properties=new GoalProperties();
GoalRunCoordinator coordinator;
LocalDateTime now=LocalDateTime.of(2026,8,27,1,0);
GoalEntity goal;
@BeforeEach void setup() {
JdbcDataSource ds=new JdbcDataSource();
ds.setURL("jdbc:h2:mem:"+ UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1");
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql")).execute(ds);
jdbc=new JdbcTemplate(ds);continuations=new GoalContinuationStore(jdbc);attempts=new GoalAttemptStore(jdbc);
coordinator=new GoalRunCoordinator(continuations,attempts,goals,properties);
jdbc.update("""
INSERT INTO mate_agent_goal(id,conversation_id,agent_id,workspace_id,created_by,title,
description,status,persistent_execution,auto_followup_enabled,create_time,update_time)
VALUES(1,'conv',2,3,'alice','goal','objective','active',TRUE,TRUE,?,?)
""",now,now);
goal=new GoalEntity();goal.setId(1L);goal.setConversationId("conv");goal.setAgentId(2L);
goal.setWorkspaceId(3L);goal.setCreatedBy("alice");goal.setStatus(GoalStatus.ACTIVE);
goal.setPersistentExecution(true);goal.setAutoFollowupEnabled(true);
when(goals.getById(1L)).thenReturn(goal);
continuations.discover(now);
}
@Test void claimBindsAttemptAndStaleSettlementCannotOverwriteNewProjection() {
var claim=coordinator.claim(continuations.get(1L),goal,now);
assertNotNull(claim);
assertEquals("claimed",attempts.get(claim.attempt().id()).state());
assertEquals(claim.attempt().id(),continuations.get(1L).currentAttemptId());
assertTrue(coordinator.markRunning(claim,now));
assertTrue(coordinator.settle(claim,new SegmentOutcome.Continue("unfinished"),now));
assertEquals("queued",continuations.get(1L).state());
assertEquals("succeeded",attempts.get(claim.attempt().id()).state());
assertFalse(coordinator.settle(claim,new SegmentOutcome.Complete("late"),now.plusSeconds(1)));
}
@Test void retryAndBlockedOutcomesHaveExplicitTerminalAttemptStates() {
var retry=coordinator.claim(continuations.get(1L),goal,now);
assertTrue(coordinator.markRunning(retry,now));
assertTrue(coordinator.settle(retry,new SegmentOutcome.Retry("provider","timeout"),now));
assertEquals("retryable",attempts.get(retry.attempt().id()).state());
var due=continuations.get(1L);
var blocked=coordinator.claim(due,goal,due.nextRunAt());
assertTrue(coordinator.markRunning(blocked,due.nextRunAt()));
assertTrue(coordinator.settle(blocked,new SegmentOutcome.Blocked("tool","review"),due.nextRunAt()));
assertEquals("blocked",attempts.get(blocked.attempt().id()).state());
assertEquals("blocked",continuations.get(1L).state());
}
@Test void continuationUsesTheLargerOfGlobalAndGoalCooldowns() {
properties.setMinimumContinuationIntervalSeconds(300);
goal.setFollowupCooldownSeconds(0);
var first=coordinator.claim(continuations.get(1L),goal,now);
assertTrue(coordinator.markRunning(first,now));
assertTrue(coordinator.settle(first,new SegmentOutcome.Continue("unfinished"),now));
assertEquals(now.plusSeconds(300),continuations.get(1L).nextRunAt());
LocalDateTime secondStart=now.plusSeconds(300);
goal.setFollowupCooldownSeconds(600);
var second=coordinator.claim(continuations.get(1L),goal,secondStart);
assertTrue(coordinator.markRunning(second,secondStart));
assertTrue(coordinator.settle(second,new SegmentOutcome.Continue("unfinished"),secondStart));
assertEquals(secondStart.plusSeconds(600),continuations.get(1L).nextRunAt());
}
}

62
mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java
View File

@ -10,12 +10,17 @@ import vip.mate.agent.model.AgentEntity;
import vip.mate.agent.runtime.ConversationTurnGate;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.channel.web.ConversationInputQueueStore;
import vip.mate.goal.model.GoalEntity;
import vip.mate.goal.model.SegmentOutcome;
import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.model.ConversationEntity;
import vip.mate.workspace.conversation.model.MessageContentPart;
import vip.mate.workspace.conversation.model.MessageEntity;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -28,9 +33,12 @@ class GoalSegmentRunnerTest {
ConversationService conversations=mock(ConversationService.class);
ApprovalWorkflowService approvals=mock(ApprovalWorkflowService.class);
ChatStreamTracker streams=new ChatStreamTracker(new ObjectMapper());
ConversationInputQueueStore inputQueue=mock(ConversationInputQueueStore.class);
ConcurrentLinkedQueue<ConversationInputQueueStore.QueuedInput> durableInputs=new ConcurrentLinkedQueue<>();
java.util.concurrent.atomic.AtomicLong inputIds=new java.util.concurrent.atomic.AtomicLong();
ConversationTurnGate gate=new ConversationTurnGate();
GoalEntity goal=new GoalEntity();
GoalSegmentRunner runner=new GoalSegmentRunner(agents,conversations,approvals,streams,new ObjectMapper(),gate);
GoalSegmentRunner runner=new GoalSegmentRunner(agents,conversations,approvals,streams,new ObjectMapper(),gate,inputQueue);
@BeforeEach void setup() {
goal.setId(1L);goal.setConversationId("conv");goal.setAgentId(2L);goal.setWorkspaceId(3L);goal.setCreatedBy("alice");
@ -39,6 +47,33 @@ class GoalSegmentRunnerTest {
when(conversations.findByConversationId("conv")).thenReturn(conv);
AgentEntity agent=new AgentEntity();agent.setEnabled(true);agent.setRuntimeType("native");
when(agents.getAgent(2L)).thenReturn(agent);
when(inputQueue.enqueue(anyString(),anyLong(),anyString(),anyString(),nullable(List.class),any()))
.thenAnswer(inv -> {
var now=LocalDateTime.now();
var input=new ConversationInputQueueStore.QueuedInput(inputIds.incrementAndGet(),
inv.getArgument(0),inv.getArgument(1),inv.getArgument(2),inv.getArgument(3),
inv.getArgument(4),"queued",null,null,null,now,now);
durableInputs.add(input);
return input;
});
when(inputQueue.claimNext(anyString(),anyString(),any())).thenAnswer(inv -> {
var input=durableInputs.poll();
if(input==null) return java.util.Optional.empty();
return java.util.Optional.of(new ConversationInputQueueStore.QueuedInput(input.id(),input.conversationId(),
input.agentId(),input.createdBy(),input.message(),input.contentParts(),"claimed",
inv.getArgument(1),input.persistedMessageId(),null,input.createdAt(),LocalDateTime.now()));
});
when(inputQueue.bindMessage(anyLong(),anyString(),anyLong(),any())).thenReturn(true);
when(inputQueue.consume(anyLong(),anyString(),any())).thenReturn(true);
when(inputQueue.release(anyLong(),anyString(),any())).thenReturn(true);
when(inputQueue.countQueued(anyString())).thenAnswer(inv -> durableInputs.size());
MessageEntity savedUser=new MessageEntity();savedUser.setId(77L);
when(conversations.saveMessage(eq("conv"),eq("user"),anyString(),nullable(List.class),eq("queued")))
.thenReturn(savedUser);
}
private void enqueue(String message,List<MessageContentPart> parts) {
inputQueue.enqueue("conv",2L,"alice",message,parts,LocalDateTime.now());
}
@Test void persistsStreamedResultAndUsage() {
@ -73,13 +108,13 @@ class GoalSegmentRunnerTest {
boolean autonomous = calls.incrementAndGet()==1;
assertTrue(GoalContinuationContext.active());
assertEquals(autonomous, GoalContinuationContext.explicitPrompt());
if(autonomous) streams.enqueueMessage("conv","new user instruction",2L,false);
if(autonomous) enqueue("new user instruction",null);
return Flux.just(new AgentService.StreamDelta("output",null),
AgentService.StreamDelta.event("finish_reason",Map.of("reason","normal")));
}));
runner.run(goal,"continue",false);
assertEquals(2,calls.get());
assertFalse(streams.hasQueuedMessage("conv"));
assertEquals(0,inputQueue.countQueued("conv"));
verify(agents).chatStructuredStream(eq(2L),eq("new user instruction"),eq("conv"),eq("alice"),isNull(),any());
verify(conversations).saveMessage("conv","user","new user instruction",null,"queued");
}
@ -94,7 +129,7 @@ class GoalSegmentRunnerTest {
subscribed.countDown();
})));
var failure=new java.util.concurrent.atomic.AtomicReference<Throwable>();
var result=new java.util.concurrent.atomic.AtomicReference<GoalSegmentRunner.Result>();
var result=new java.util.concurrent.atomic.AtomicReference<SegmentOutcome>();
Thread worker=Thread.ofVirtual().start(() -> {
try { result.set(runner.run(goal,"continue",false)); } catch(Throwable error) { failure.set(error); }
});
@ -145,12 +180,12 @@ class GoalSegmentRunnerTest {
.thenReturn(Flux.<AgentService.StreamDelta>never().doOnSubscribe(s -> ready.countDown()));
Thread worker = Thread.ofVirtual().start(() -> runner.run(goal,"continue",false));
assertTrue(ready.await(3,TimeUnit.SECONDS));
streams.enqueueMessage("conv","accepted steering",2L,false,parts);
enqueue("accepted steering",parts);
runner.cancelAll();
worker.join(3000);
assertFalse(worker.isAlive());
verify(conversations).saveMessage("conv","user","accepted steering",parts,"queued");
assertFalse(streams.hasQueuedMessage("conv"));
verify(conversations,never()).saveMessage("conv","user","accepted steering",parts,"queued");
assertEquals(1,inputQueue.countQueued("conv"));
}
@Test void shutdownRejectsLateWorkerAdmissionWithoutStartingModel() {
@ -190,12 +225,12 @@ class GoalSegmentRunnerTest {
@Test void permanentFailurePersistsAcceptedQueuedInput() {
when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any()))
.thenReturn(Flux.defer(() -> {
streams.enqueueMessage("conv","user instruction",2L,false);
enqueue("user instruction",null);
return Flux.error(new IllegalArgumentException("bad config"));
}));
assertThrows(IllegalArgumentException.class,()->runner.run(goal,"continue",false));
verify(conversations).saveMessage("conv","user","user instruction",null,"queued");
assertFalse(streams.hasQueuedMessage("conv"));
verify(conversations,never()).saveMessage("conv","user","user instruction",null,"queued");
assertEquals(1,inputQueue.countQueued("conv"));
}
@Test void userSteeringPreservesInterruptedStatusThenRunsQueuedInput() throws Exception {
@ -211,7 +246,8 @@ class GoalSegmentRunnerTest {
try { runner.run(goal,"continue",false); } catch(Throwable error) { failure.set(error); }
});
assertTrue(subscribed.await(3,java.util.concurrent.TimeUnit.SECONDS));
assertTrue(streams.requestInterrupt("conv","new instruction",2L,false));
enqueue("new instruction",null);
assertTrue(streams.requestInterrupt("conv","",2L,false));
worker.join(3000);
assertFalse(worker.isAlive());
assertNull(failure.get());
@ -227,7 +263,7 @@ class GoalSegmentRunnerTest {
when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any()))
.thenAnswer(inv -> {
if(calls.incrementAndGet()==1) {
streams.enqueueMessage("conv","new question",2L,false);
enqueue("new question",null);
return Flux.just(AgentService.StreamDelta.event("finish_reason",Map.of("reason","normal")));
}
return finish.asMono().flux().doOnSubscribe(s -> entered.countDown());
@ -251,7 +287,7 @@ class GoalSegmentRunnerTest {
when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any()))
.thenAnswer(inv -> {
calls.incrementAndGet();
streams.enqueueMessage("conv","new question",2L,false);
enqueue("new question",null);
return Flux.just(AgentService.StreamDelta.event("finish_reason",Map.of("reason","normal")));
});
when(conversations.saveMessage("conv","user","new question",null,"queued")).thenAnswer(inv -> {

30
mateclaw-ui/src/composables/chat/__tests__/useWorkerConversationGuard.test.ts
View File

@ -1,4 +1,4 @@
import { describe, expect, it } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { nextTick, ref } from 'vue'
import { useWorkerConversationGuard } from '../useWorkerConversationGuard'
import type { VerifiedWorkerContext } from '@/utils/conversationGovernance'
@ -18,6 +18,10 @@ const worker = (conversationId: string): VerifiedWorkerContext => ({
})
describe('useWorkerConversationGuard', () => {
afterEach(() => {
vi.useRealTimers()
})
it('fails closed while a worker-looking route is pending and after 403/500', async () => {
const conversationId = ref('worker')
const workerHint = ref(true)
@ -91,4 +95,28 @@ describe('useWorkerConversationGuard', () => {
expect(guard.state.value).toBe('error')
expect(guard.readOnly.value).toBe(true)
})
it('retries an ordinary conversation after a transient backend outage', async () => {
vi.useFakeTimers()
let attempts = 0
const guard = useWorkerConversationGuard({
conversationId: ref('ordinary'),
workerHint: ref(false),
load: async () => {
attempts += 1
if (attempts === 1) throw new Error('backend restarting')
return null
},
retryDelayMs: 100,
})
await nextTick(); await Promise.resolve()
expect(guard.state.value).toBe('error')
expect(guard.readOnly.value).toBe(true)
await vi.advanceTimersByTimeAsync(100)
expect(attempts).toBe(2)
expect(guard.state.value).toBe('nonWorker')
expect(guard.readOnly.value).toBe(false)
})
})

50
mateclaw-ui/src/composables/chat/useWorkerConversationGuard.ts
View File

@ -7,33 +7,57 @@ export function useWorkerConversationGuard(options: {
conversationId: Ref<string>
workerHint: Ref<boolean>
load: (conversationId: string) => Promise<VerifiedWorkerContext | null>
/** Initial delay for retrying an ordinary conversation while the backend restarts. */
retryDelayMs?: number
}) {
const state = ref<WorkerGuardState>('pending')
const context = ref<VerifiedWorkerContext | null>(null)
let requestVersion = 0
watch([options.conversationId, options.workerHint], async ([conversationId, workerHint]) => {
watch([options.conversationId, options.workerHint], ([conversationId, workerHint], _previous, onCleanup) => {
const version = ++requestVersion
let retryTimer: ReturnType<typeof setTimeout> | null = null
let stopped = false
onCleanup(() => {
stopped = true
if (retryTimer) clearTimeout(retryTimer)
})
state.value = 'pending'
context.value = null
if (!conversationId) {
state.value = 'nonWorker'
return
}
try {
const result = await options.load(conversationId)
if (version !== requestVersion) return
if (result?.verified && result.conversationKind === 'team_worker'
&& result.conversationId === conversationId) {
context.value = result
state.value = 'verified'
} else {
state.value = workerHint ? 'error' : 'nonWorker'
const verify = async (retryAttempt: number) => {
try {
const result = await options.load(conversationId)
if (stopped || version !== requestVersion) return
if (result?.verified && result.conversationKind === 'team_worker'
&& result.conversationId === conversationId) {
context.value = result
state.value = 'verified'
} else {
state.value = workerHint ? 'error' : 'nonWorker'
}
} catch {
if (stopped || version !== requestVersion) return
state.value = 'error'
// A normal conversation loaded while the backend is restarting must
// stay fail-closed, but it must not remain read-only forever. Worker
// routes already have an explicit hint and need no availability retry.
if (!workerHint) {
const initialDelay = Math.max(1, options.retryDelayMs ?? 1000)
const delay = Math.min(initialDelay * (2 ** retryAttempt), 10_000)
retryTimer = setTimeout(() => {
retryTimer = null
void verify(retryAttempt + 1)
}, delay)
}
}
} catch {
if (version !== requestVersion) return
state.value = 'error'
}
void verify(0)
}, { immediate: true })
return {