diff --git a/mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java b/mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java index 3c1a8f87..a267a264 100644 --- a/mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java @@ -1,6 +1,7 @@ package vip.mate.team.service; import cn.hutool.core.util.IdUtil; +import cn.hutool.json.JSONUtil; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; @@ -247,6 +248,49 @@ public class TeamDispatchService { return; } if (TeamTaskStatus.IN_PROGRESS.equals(current.getStatus())) { + String invalidReason = invalidResultReason(current, reply); + if (invalidReason != null) { + int attempts = current.getDispatchCount() == null ? 0 : current.getDispatchCount(); + if (attempts < TeamTaskService.MAX_DISPATCHES + && taskService.requeueUnusableResult(task.getId(), invalidReason)) { + log.warn("Team task #{} produced an unusable result on attempt {}/{}; requeued: {}", + task.getTaskNumber(), attempts, TeamTaskService.MAX_DISPATCHES, + invalidReason); + broadcast(task, "team_task_retrying", Map.of("reason", invalidReason)); + return; + } + boolean failed = taskService.failTask(task.getId(), invalidReason); + TeamTaskEntity failedTask = taskService.getTask(task.getId()); + if (failed) { + broadcast(task, "team_task_failed", Map.of("reason", invalidReason)); + announceService.announceTaskSettled(failedTask); + } + return; + } + String terminalCheckpoint = taskService.checkpointTerminalTag(current); + if (terminalCheckpoint != null) { + String terminalEvidence = "[checkpoint:" + terminalCheckpoint + "] acknowledged"; + boolean terminalAlreadyAcknowledged = taskService.listComments(current.getId()).stream() + .anyMatch(comment -> terminalEvidence.equals(comment.getContent())); + if (terminalAlreadyAcknowledged) { + List released = taskService.completeTask(task.getId(), null, + truncate(reply, MAX_RESULT_CHARS)); + TeamTaskEntity completed = taskService.getTask(task.getId()); + log.info("Team task #{} completed after deferred {} acknowledgement " + + "({} dependents released)", + task.getTaskNumber(), terminalCheckpoint, released.size()); + broadcast(task, "team_task_completed", Map.of("status", TeamTaskStatus.COMPLETED)); + announceService.announceTaskSettled(completed); + return; + } + int percent = current.getProgressPercent() == null + ? 1 : Math.max(1, current.getProgressPercent()); + taskService.updateProgress(task.getId(), null, percent, + "waiting for " + terminalCheckpoint + " checkpoint"); + log.info("Team task #{} parked as long-running checkpoint tracker until {}", + task.getTaskNumber(), terminalCheckpoint); + return; + } List released = taskService.completeTask(task.getId(), null, truncate(reply == null || reply.isBlank() ? "(no output)" : reply, MAX_RESULT_CHARS)); @@ -271,6 +315,30 @@ public class TeamDispatchService { announceService.announceTaskSettled(current); } + private String invalidResultReason(TeamTaskEntity task, String reply) { + if (reply == null || reply.isBlank()) { + return "member produced no result"; + } + String normalized = reply.strip().toLowerCase(); + if (normalized.contains("failed to generate a response, please retry") + || normalized.equals("(no output)")) { + return "member response generation failed"; + } + if (requiresDeliverable(task) && taskService.listDeliverables(task).isEmpty()) { + return "required deliverable was not attached"; + } + return null; + } + + private boolean requiresDeliverable(TeamTaskEntity task) { + try { + return task.getMetadata() != null + && JSONUtil.parseObj(task.getMetadata()).getBool("deliverableRequired", false); + } catch (Exception ignored) { + return false; + } + } + /** Per-prerequisite and whole-section caps keeping the envelope bounded. */ static final int MAX_PREREQ_RESULT_CHARS = 1500; static final int MAX_PREREQ_SECTION_CHARS = 6000; diff --git a/mateclaw-server/src/main/java/vip/mate/team/service/TeamPlanBridge.java b/mateclaw-server/src/main/java/vip/mate/team/service/TeamPlanBridge.java index cdc17d80..c82d4d7e 100644 --- a/mateclaw-server/src/main/java/vip/mate/team/service/TeamPlanBridge.java +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamPlanBridge.java @@ -55,6 +55,9 @@ public class TeamPlanBridge { /** Task subject cap; the full step text rides in the description. */ static final int SUBJECT_MAX_CHARS = 120; private static final Pattern CHECKPOINT_TAG = Pattern.compile("(?i)(?:^|\\b)(R\\d{3})(?:\\b|/)"); + private static final Pattern DELIVERABLE_REQUEST = Pattern.compile( + "(?i)(交付物|生成.{0,8}(文件|文档)|文档成稿|报告成稿|" + + "docx|xlsx|pptx|pdf|deliverable|document|spreadsheet|presentation)"); private final TeamService teamService; private final TeamTaskService taskService; @@ -197,6 +200,7 @@ public class TeamPlanBridge { .metadata(new JSONObject() .set("planId", String.valueOf(planId)) .set("stepIndex", i) + .set("deliverableRequired", DELIVERABLE_REQUEST.matcher(step).find()) .toString()) .build()); created.add(task); @@ -258,6 +262,7 @@ public class TeamPlanBridge { List tasks = taskService.listTasksByRun(latest.get().getId()); if (!tasks.isEmpty()) { recordCheckpointEvidence(latest.get().getTeamId(), tasks, checkpointTag); + tasks = taskService.listTasksByRun(latest.get().getId()); return new InFlight(buildCheckpointText(tasks, checkpointTag)); } } @@ -283,6 +288,7 @@ public class TeamPlanBridge { .toList(); if (checkpointTag != null) { recordCheckpointEvidence(teamOpt.get().getId(), tasks, checkpointTag); + tasks = taskService.listTasksByPlan(teamOpt.get().getId(), plan.getId()); return new InFlight(buildCheckpointText(tasks, checkpointTag)); } if (!allTerminal) { @@ -408,23 +414,45 @@ public class TeamPlanBridge { if ("R100".equalsIgnoreCase(checkpointTag)) { compact.append("（已完成第100轮检查点）"); } + compact.append("｜证据 [checkpoint:").append(checkpointTag).append("] acknowledged"); return compact.toString(); } private void recordCheckpointEvidence(Long teamId, List tasks, String checkpointTag) { - TeamTaskEntity tracker = taskService.findCheckpointTracker(teamId).orElseGet(() -> tasks.stream() - .filter(this::isCheckpointTracker) + TeamTaskEntity tracker = tasks.stream() + .filter(task -> taskService.checkpointTerminalTag(task) != null) .findFirst() - .orElse(tasks.get(tasks.size() - 1))); + .orElseGet(() -> tasks.stream() + .filter(this::isCheckpointTracker) + .findFirst() + .orElseGet(() -> taskService.findCheckpointTracker(teamId) + .orElse(tasks.get(tasks.size() - 1)))); String content = "[checkpoint:" + checkpointTag + "] acknowledged"; taskService.addCommentOnce(tracker.getId(), TeamTaskService.AUTHOR_SYSTEM, "team-plan-bridge", TeamTaskService.COMMENT_NOTE, content); + String terminalTag = taskService.checkpointTerminalTag(tracker); + if (terminalTag != null && TeamTaskStatus.IN_PROGRESS.equals(tracker.getStatus())) { + int current = Integer.parseInt(checkpointTag.substring(1)); + int terminal = Integer.parseInt(terminalTag.substring(1)); + int percent = terminal <= 0 ? 1 + : Math.min(99, Math.max(1, current * 100 / terminal)); + if (checkpointTag.equalsIgnoreCase(terminalTag)) { + taskService.completeTask(tracker.getId(), null, + "Checkpoint tracking completed at " + checkpointTag); + eventPublisher.publishEvent(new TeamTasksDelegatedEvent(teamId)); + } else { + taskService.updateProgress(tracker.getId(), null, percent, + checkpointTag + "/" + terminalTag + " acknowledged"); + } + } } private boolean isCheckpointTracker(TeamTaskEntity task) { - String text = (task.getSubject() == null ? "" : task.getSubject()) + " " - + (task.getDescription() == null ? "" : task.getDescription()); + if (taskService.checkpointTerminalTag(task) != null) { + return true; + } + String text = task.getSubject() == null ? "" : task.getSubject(); String lower = text.toLowerCase(); return text.contains("检查点") || text.contains("共享跟踪") || lower.contains("checkpoint") || lower.contains("r001-r100"); diff --git a/mateclaw-server/src/main/java/vip/mate/team/service/TeamTaskService.java b/mateclaw-server/src/main/java/vip/mate/team/service/TeamTaskService.java index 9001fdec..a8f9df9a 100644 --- a/mateclaw-server/src/main/java/vip/mate/team/service/TeamTaskService.java +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamTaskService.java @@ -29,6 +29,8 @@ import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; /** * Shared task board service. All status transitions are guarded conditional @@ -43,6 +45,9 @@ import java.util.Set; @RequiredArgsConstructor public class TeamTaskService { + private static final Pattern CHECKPOINT_RANGE = Pattern.compile( + "(?i)R(\\d{3})\\s*[-–—]\\s*R(\\d{3})"); + /** Execution lease length; renewed by the runner while the member works. */ static final int LOCK_MINUTES = 60; @@ -177,6 +182,7 @@ public class TeamTaskService { .eq(TeamTaskEntity::getStatus, TeamTaskStatus.PENDING) .set(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS) .set(TeamTaskEntity::getOwnerAgentId, agentId) + .set(TeamTaskEntity::getReason, null) .set(TeamTaskEntity::getLockExpiresAt, newLease())) == 1; if (assigned) { projectTask(taskId); @@ -325,6 +331,28 @@ public class TeamTaskService { return retried; } + /** + * Requeue an automatically dispatched task whose member result is unusable. + * Unlike a manual retry this deliberately preserves {@code dispatchCount}, + * so the existing dispatch circuit breaker remains the hard upper bound. + */ + public boolean requeueUnusableResult(Long taskId, String reason) { + TeamTaskEntity task = taskMapper.selectById(taskId); + boolean requeued = taskMapper.update(null, Wrappers.lambdaUpdate() + .eq(TeamTaskEntity::getId, taskId) + .eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS) + .set(TeamTaskEntity::getStatus, TeamTaskStatus.PENDING) + .set(TeamTaskEntity::getOwnerAgentId, null) + .set(TeamTaskEntity::getLockExpiresAt, null) + .set(TeamTaskEntity::getReason, reason)) == 1; + if (requeued) { + recordEvent(task == null ? null : task.getTeamId(), taskId, + TeamTaskEventEntity.RETRIED, AUTHOR_SYSTEM, null, reason); + projectTask(taskId); + } + return requeued; + } + // ==================== progress / comments ==================== /** Update progress and renew the execution lease in one shot. */ @@ -667,14 +695,33 @@ public class TeamTaskService { .eq(TeamTaskEntity::getTeamId, teamId) .and(candidate -> candidate .like(TeamTaskEntity::getSubject, "共享跟踪") - .or().like(TeamTaskEntity::getDescription, "R001-R100") - .or().like(TeamTaskEntity::getSubject, "checkpoint") - .or().like(TeamTaskEntity::getDescription, "checkpoint")) + .or().like(TeamTaskEntity::getSubject, "检查点") + .or().like(TeamTaskEntity::getSubject, "checkpoint")) .orderByDesc(TeamTaskEntity::getPriority) .orderByDesc(TeamTaskEntity::getCreateTime) .last("LIMIT 1"))); } + /** Terminal checkpoint tag declared by a long-running tracker, e.g. R300. */ + public String checkpointTerminalTag(TeamTaskEntity task) { + if (task == null) { + return null; + } + String description = task.getDescription() == null ? "" : task.getDescription(); + int contextStart = description.indexOf("[Plan context]"); + if (contextStart >= 0) { + description = description.substring(0, contextStart); + } + String text = (task.getSubject() == null ? "" : task.getSubject()) + " " + description; + String lower = text.toLowerCase(); + if (!text.contains("共享跟踪") && !text.contains("检查点") + && !lower.contains("checkpoint")) { + return null; + } + Matcher matcher = CHECKPOINT_RANGE.matcher(text); + return matcher.find() ? "R" + matcher.group(2) : null; + } + public List listTasks(Long teamId, List statuses) { return listTasks(teamId, statuses, null, null); } diff --git a/mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java b/mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java index b2d928d2..7780f580 100644 --- a/mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java +++ b/mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java @@ -7,6 +7,7 @@ import vip.mate.agent.AgentService; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.team.model.AgentTeamEntity; import vip.mate.team.model.TeamTaskEntity; +import vip.mate.team.model.TeamTaskCommentEntity; import vip.mate.team.model.TeamTaskStatus; import vip.mate.workspace.conversation.ConversationService; @@ -153,6 +154,100 @@ class TeamDispatchServiceTest { verify(announceService).announceTaskSettled(done); } + @Test + @DisplayName("a fallback final answer is requeued instead of being reported as completed") + void settleFallbackRequeues() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + running.setDispatchCount(1); + when(taskService.getTask(1L)).thenReturn(running); + when(taskService.requeueUnusableResult(1L, "member response generation failed")) + .thenReturn(true); + + service.settleOutcome(running, + "I inspected the task. Failed to generate a response, please retry."); + + verify(taskService).requeueUnusableResult(1L, "member response generation failed"); + verify(taskService, never()).completeTask(any(), any(), anyString()); + verify(announceService, never()).announceTaskSettled(any()); + verify(eventChannel).publishTaskEvent(any(), eq("team_task_retrying"), any()); + } + + @Test + @DisplayName("an unusable third result fails instead of bypassing the circuit breaker") + void settleFallbackFailsAfterDispatchBudget() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + running.setDispatchCount(TeamTaskService.MAX_DISPATCHES); + TeamTaskEntity failed = task(1L, MEMBER_A); + failed.setStatus(TeamTaskStatus.FAILED); + failed.setReason("member response generation failed"); + when(taskService.getTask(1L)).thenReturn(running, failed); + when(taskService.failTask(1L, "member response generation failed")).thenReturn(true); + + service.settleOutcome(running, "Failed to generate a response, please retry."); + + verify(taskService, never()).requeueUnusableResult(any(), anyString()); + verify(taskService).failTask(1L, "member response generation failed"); + verify(eventChannel).publishTaskEvent(any(), eq("team_task_failed"), any()); + verify(announceService).announceTaskSettled(failed); + } + + @Test + @DisplayName("a declared deliverable task without an attachment is requeued") + void settleMissingDeliverableRequeues() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + running.setDispatchCount(1); + running.setMetadata("{\"deliverableRequired\":true}"); + when(taskService.getTask(1L)).thenReturn(running); + when(taskService.listDeliverables(running)).thenReturn(List.of()); + when(taskService.requeueUnusableResult(1L, "required deliverable was not attached")) + .thenReturn(true); + + service.settleOutcome(running, "handbook completed"); + + verify(taskService).requeueUnusableResult(1L, "required deliverable was not attached"); + verify(taskService, never()).completeTask(any(), any(), anyString()); + } + + @Test + @DisplayName("a long-running checkpoint tracker stays active until its terminal round") + void settleParksCheckpointTracker() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + running.setProgressPercent(1); + when(taskService.getTask(1L)).thenReturn(running); + when(taskService.checkpointTerminalTag(running)).thenReturn("R300"); + + service.settleOutcome(running, "R001 tracker initialized"); + + verify(taskService).updateProgress(1L, null, 1, + "waiting for R300 checkpoint"); + verify(taskService, never()).completeTask(any(), any(), anyString()); + verify(announceService, never()).announceTaskSettled(any()); + } + + @Test + @DisplayName("a tracker initialized after its terminal checkpoint completes immediately") + void settleCompletesTrackerWhenTerminalEvidenceAlreadyExists() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + TeamTaskEntity completed = task(1L, MEMBER_A); + completed.setStatus(TeamTaskStatus.COMPLETED); + TeamTaskCommentEntity evidence = new TeamTaskCommentEntity(); + evidence.setContent("[checkpoint:R300] acknowledged"); + when(taskService.getTask(1L)).thenReturn(running, completed); + when(taskService.checkpointTerminalTag(running)).thenReturn("R300"); + when(taskService.listComments(1L)).thenReturn(List.of(evidence)); + when(taskService.completeTask(1L, null, "tracker initialized")).thenReturn(List.of()); + + service.settleOutcome(running, "tracker initialized"); + + verify(taskService).completeTask(1L, null, "tracker initialized"); + verify(announceService).announceTaskSettled(completed); + } + @Test @DisplayName("a task the member already failed via blocker is not completed on top") void settleRespectsMemberFailure() { diff --git a/mateclaw-server/src/test/java/vip/mate/team/service/TeamPlanBridgeTest.java b/mateclaw-server/src/test/java/vip/mate/team/service/TeamPlanBridgeTest.java index b07de0fb..b13d0e57 100644 --- a/mateclaw-server/src/test/java/vip/mate/team/service/TeamPlanBridgeTest.java +++ b/mateclaw-server/src/test/java/vip/mate/team/service/TeamPlanBridgeTest.java @@ -300,7 +300,8 @@ class TeamPlanBridgeTest { bridge.checkParkedPlan(CONV, "R100/100 最终检查点：仅用一行回复，并确认已连续完成100轮")); - assertEquals("R100｜执行中 1/2｜#47 in_progress 80%（已完成第100轮检查点）", + assertEquals("R100｜执行中 1/2｜#47 in_progress 80%（已完成第100轮检查点）" + + "｜证据 [checkpoint:R100] acknowledged", state.progressText()); assertFalse(state.progressText().contains("\n")); verify(taskService).addCommentOnce(102L, TeamTaskService.AUTHOR_SYSTEM, @@ -330,10 +331,36 @@ class TeamPlanBridgeTest { TeamPlanBridge.InFlight state = assertInstanceOf(TeamPlanBridge.InFlight.class, bridge.checkParkedPlan(CONV, "R047/100 checkpoint")); - assertEquals("R047｜已完成 2/2｜#47 completed 100%", state.progressText()); - verify(taskService).addCommentOnce(103L, TeamTaskService.AUTHOR_SYSTEM, + assertEquals("R047｜已完成 2/2｜#47 completed 100%" + + "｜证据 [checkpoint:R047] acknowledged", state.progressText()); + verify(taskService).addCommentOnce(102L, TeamTaskService.AUTHOR_SYSTEM, "team-plan-bridge", TeamTaskService.COMMENT_NOTE, "[checkpoint:R047] acknowledged"); + verify(taskService, never()).addCommentOnce(eq(103L), anyString(), anyString(), + anyString(), anyString()); + } + + @Test + @DisplayName("the terminal checkpoint completes the current run tracker and releases dispatch") + void terminalCheckpointCompletesCurrentTracker() { + parkedPlan(); + TeamTaskEntity tracker = task(102L, 59, 0, TeamTaskStatus.IN_PROGRESS); + tracker.setSubject("R001-R300 唯一共享跟踪条目"); + when(taskService.listTasksByPlan(TEAM_ID, PLAN_ID)).thenReturn(List.of(tracker)); + when(taskService.checkpointTerminalTag(tracker)).thenReturn("R300"); + when(taskService.completeTask(102L, null, + "Checkpoint tracking completed at R300")).thenReturn(List.of()); + + TeamPlanBridge.InFlight state = assertInstanceOf(TeamPlanBridge.InFlight.class, + bridge.checkParkedPlan(CONV, "最终检查点 R300")); + + assertTrue(state.progressText().contains("[checkpoint:R300] acknowledged")); + verify(taskService).addCommentOnce(102L, TeamTaskService.AUTHOR_SYSTEM, + "team-plan-bridge", TeamTaskService.COMMENT_NOTE, + "[checkpoint:R300] acknowledged"); + verify(taskService).completeTask(102L, null, + "Checkpoint tracking completed at R300"); + verify(eventPublisher).publishEvent(new TeamTasksDelegatedEvent(TEAM_ID)); } @Test diff --git a/mateclaw-server/src/test/java/vip/mate/team/service/TeamTaskServiceTest.java b/mateclaw-server/src/test/java/vip/mate/team/service/TeamTaskServiceTest.java index 163f47cc..dbef2bd9 100644 --- a/mateclaw-server/src/test/java/vip/mate/team/service/TeamTaskServiceTest.java +++ b/mateclaw-server/src/test/java/vip/mate/team/service/TeamTaskServiceTest.java @@ -600,4 +600,18 @@ class TeamTaskServiceTest { query.getValue().getSqlSegment(); assertTrue(query.getValue().getParamNameValuePairs().containsValue(RUN_ID)); } + + @Test + @DisplayName("checkpoint range detection ignores the injected overall plan context") + void checkpointTerminalTagUsesOnlyLocalTaskText() { + TeamTaskEntity tracker = task(1L, TeamTaskStatus.IN_PROGRESS); + tracker.setSubject("R001-R300 唯一共享跟踪条目"); + tracker.setDescription("每轮登记证据，R300 前保持进行中\n\n[Plan context]\nOverall request"); + assertEquals("R300", service.checkpointTerminalTag(tracker)); + + TeamTaskEntity ordinary = task(2L, TeamTaskStatus.IN_PROGRESS); + ordinary.setSubject("设计稳定性指标"); + ordinary.setDescription("产出指标清单\n\n[Plan context]\nOverall request: R001-R300 共享跟踪"); + assertNull(service.checkpointTerminalTag(ordinary)); + } }