diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index e57e5704..e43dfd3f 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -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 diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index d8c1b90c..122f30e0 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -190,8 +190,8 @@ public class ChatStreamTracker { /** 等待原因（审批等待时有值） */ volatile String waitingReason; - /** 排队的用户消息队列（支持多条排队消息，按序消费） */ - final java.util.Queue 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, diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ConversationInputQueueStore.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ConversationInputQueueStore.java new file mode 100644 index 00000000..7dfd820e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ConversationInputQueueStore.java @@ -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> 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 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 claimNext(String conversationId, String attemptId, + LocalDateTime now) { + for (int tries = 0; tries < 8; tries++) { + List 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 rows = jdbc.query(""" + SELECT * FROM mate_conversation_input_queue WHERE id=? + """, (rs, row) -> read(rs), id); + return rows.isEmpty() ? null : rows.getFirst(); + } + + public List 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 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 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 contentParts, + String state, + String claimedByAttemptId, + Long persistedMessageId, + String cancelReason, + LocalDateTime createdAt, + LocalDateTime updatedAt) {} +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/config/GoalProperties.java b/mateclaw-server/src/main/java/vip/mate/goal/config/GoalProperties.java index 4e075a98..564487ac 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/config/GoalProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/config/GoalProperties.java @@ -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 diff --git a/mateclaw-server/src/main/java/vip/mate/goal/controller/GoalExecutionController.java b/mateclaw-server/src/main/java/vip/mate/goal/controller/GoalExecutionController.java index c56e3cb0..712106ea 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/controller/GoalExecutionController.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/controller/GoalExecutionController.java @@ -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 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> 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()); + } + } } diff --git a/mateclaw-server/src/main/java/vip/mate/goal/model/GoalAttempt.java b/mateclaw-server/src/main/java/vip/mate/goal/model/GoalAttempt.java new file mode 100644 index 00000000..31198535 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/model/GoalAttempt.java @@ -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 TERMINAL_STATES = + Set.of("succeeded", "retryable", "blocked", "cancelled"); + + public boolean terminal() { + return TERMINAL_STATES.contains(state); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/model/SegmentOutcome.java b/mateclaw-server/src/main/java/vip/mate/goal/model/SegmentOutcome.java new file mode 100644 index 00000000..e860f8fe --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/model/SegmentOutcome.java @@ -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 {} +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java new file mode 100644 index 00000000..b0289a37 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalAttemptStore.java @@ -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 rows = jdbc.query(""" + SELECT * FROM mate_goal_attempt WHERE attempt_id=? + """, (rs, row) -> read(rs), id); + return rows.isEmpty() ? null : rows.getFirst(); + } + + public List 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 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); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java index 2fd12dc5..034e32e2 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationStore.java @@ -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) { diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java index 469eb04e..a7f7b993 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java @@ -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 active = new ConcurrentHashMap<>(); + private final GoalRunCoordinator coordinator; + private final GoalRecoveryService recovery; + private final ConcurrentHashMap active = new ConcurrentHashMap<>(); + private final AtomicReference 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())); } } diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java new file mode 100644 index 00000000..5d75487c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java @@ -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; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java new file mode 100644 index 00000000..9bbbebd6 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRunCoordinator.java @@ -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) {} +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java index 7bc9707c..a87f0561 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java @@ -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 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 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()); diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V189__goal_attempt_and_input_queue.sql b/mateclaw-server/src/main/resources/db/migration/h2/V189__goal_attempt_and_input_queue.sql new file mode 100644 index 00000000..6c7fbe9e --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V189__goal_attempt_and_input_queue.sql @@ -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); diff --git a/mateclaw-server/src/main/resources/db/migration/kingbase/V189__goal_attempt_and_input_queue.sql b/mateclaw-server/src/main/resources/db/migration/kingbase/V189__goal_attempt_and_input_queue.sql new file mode 100644 index 00000000..6c7fbe9e --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/kingbase/V189__goal_attempt_and_input_queue.sql @@ -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); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V189__goal_attempt_and_input_queue.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V189__goal_attempt_and_input_queue.sql new file mode 100644 index 00000000..2103b46d --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V189__goal_attempt_and_input_queue.sql @@ -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) +); diff --git a/mateclaw-server/src/main/resources/docs/en/goals.md b/mateclaw-server/src/main/resources/docs/en/goals.md index 23b9f45b..86bb5802 100644 --- a/mateclaw-server/src/main/resources/docs/en/goals.md +++ b/mateclaw-server/src/main/resources/docs/en/goals.md @@ -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 diff --git a/mateclaw-server/src/main/resources/docs/zh/goals.md b/mateclaw-server/src/main/resources/docs/zh/goals.md index dafcf614..654020f3 100644 --- a/mateclaw-server/src/main/resources/docs/zh/goals.md +++ b/mateclaw-server/src/main/resources/docs/zh/goals.md @@ -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` 秒，再根据日志逐步放量。 + --- ## 数据库 diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerDurableQueueTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerDurableQueueTest.java new file mode 100644 index 00000000..b30c43e5 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerDurableQueueTest.java @@ -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"); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerPreviewRouteTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerPreviewRouteTest.java index 88db9f93..10a3d1bb 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerPreviewRouteTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerPreviewRouteTest.java @@ -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(); } diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerWorkerReadOnlyTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerWorkerReadOnlyTest.java index 0e648e77..bfd53de3 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerWorkerReadOnlyTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatControllerWorkerReadOnlyTest.java @@ -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); } diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java index 16456c3a..86b3547e 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerQueueDrainTest.java @@ -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)); } } diff --git a/mateclaw-server/src/test/java/vip/mate/channel/web/ConversationInputQueueStoreTest.java b/mateclaw-server/src/test/java/vip/mate/channel/web/ConversationInputQueueStoreTest.java new file mode 100644 index 00000000..1c49539f --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/web/ConversationInputQueueStoreTest.java @@ -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(); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/goal/controller/GoalExecutionControllerTest.java b/mateclaw-server/src/test/java/vip/mate/goal/controller/GoalExecutionControllerTest.java index 1fee7f92..9ff0b341 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/controller/GoalExecutionControllerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/controller/GoalExecutionControllerTest.java @@ -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); } } diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java new file mode 100644 index 00000000..853e6e0c --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalAttemptStoreTest.java @@ -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); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java index e74383d0..75d92dca 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationStoreTest.java @@ -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(); } diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java index d7c59731..2b56121f 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java @@ -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 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)); } } diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java new file mode 100644 index 00000000..95af80c4 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRecoveryServiceTest.java @@ -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); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java new file mode 100644 index 00000000..f8d2b212 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalRunCoordinatorTest.java @@ -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()); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java index 55771169..31f8d4f6 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java @@ -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 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 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(); - var result=new java.util.concurrent.atomic.AtomicReference(); + var result=new java.util.concurrent.atomic.AtomicReference(); 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.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 -> { diff --git a/mateclaw-ui/src/composables/chat/__tests__/useWorkerConversationGuard.test.ts b/mateclaw-ui/src/composables/chat/__tests__/useWorkerConversationGuard.test.ts index d9000019..3bf2180b 100644 --- a/mateclaw-ui/src/composables/chat/__tests__/useWorkerConversationGuard.test.ts +++ b/mateclaw-ui/src/composables/chat/__tests__/useWorkerConversationGuard.test.ts @@ -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) + }) }) diff --git a/mateclaw-ui/src/composables/chat/useWorkerConversationGuard.ts b/mateclaw-ui/src/composables/chat/useWorkerConversationGuard.ts index 4ca7f5fd..dfc98917 100644 --- a/mateclaw-ui/src/composables/chat/useWorkerConversationGuard.ts +++ b/mateclaw-ui/src/composables/chat/useWorkerConversationGuard.ts @@ -7,33 +7,57 @@ export function useWorkerConversationGuard(options: { conversationId: Ref workerHint: Ref load: (conversationId: string) => Promise + /** Initial delay for retrying an ordinary conversation while the backend restarts. */ + retryDelayMs?: number }) { const state = ref('pending') const context = ref(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 | 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 {