diff --git a/mateclaw-server/src/main/java/vip/mate/team/service/TeamAnnounceService.java b/mateclaw-server/src/main/java/vip/mate/team/service/TeamAnnounceService.java new file mode 100644 index 00000000..33f8c340 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamAnnounceService.java @@ -0,0 +1,211 @@ +package vip.mate.team.service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import vip.mate.agent.AgentService; +import vip.mate.agent.model.AgentEntity; +import vip.mate.agent.repository.AgentMapper; +import vip.mate.agent.runtime.RunningConversationRegistry; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.team.model.AgentTeamEntity; +import vip.mate.team.model.TeamTaskEntity; +import vip.mate.team.model.TeamTaskStatus; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * Delivers settled task results back to the team lead. Results arriving close + * together are debounced per lead conversation and merged into ONE combined + * announcement, so parallel members finishing near-simultaneously wake the + * lead once instead of once per task. + * + * Delivery is guaranteed, not opportunistic: when the lead is mid-turn the + * announcement is NOT injected into the running turn (an in-turn notification + * is dropped if the turn ends before the next reasoning round — silent result + * loss). Instead delivery re-arms itself until the lead is idle, then starts a + * fresh lead turn in the originating conversation; the lead's synthesized + * reply is persisted there and pushed over SSE. + * + * @author MateClaw Team + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class TeamAnnounceService { + + /** Collect window: results arriving within it join the same announcement. */ + static final long DEBOUNCE_MILLIS = 2000; + + /** A batch drains immediately once it reaches this size. */ + static final int MAX_BATCH = 20; + + /** Re-check interval while waiting for a busy lead to go idle. */ + static final long BUSY_RETRY_MILLIS = 2000; + + /** Give up waiting and wake the lead anyway after this many busy retries. */ + static final int MAX_BUSY_RETRIES = 900; // ~30 minutes + + private static final ScheduledExecutorService DEBOUNCE_SCHEDULER = + Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "team-announce-debounce"); + t.setDaemon(true); + return t; + }); + + /** One JDK 21 virtual thread per lead wake-up run. */ + private static final ExecutorService ANNOUNCE_EXECUTOR = + Executors.newVirtualThreadPerTaskExecutor(); + + private final TeamService teamService; + private final AgentService agentService; + private final AgentMapper agentMapper; + private final RunningConversationRegistry runningConversations; + private final ChatStreamTracker streamTracker; + + /** Pending items per lead conversation; the first item arms the drain timer. */ + private final Map> pending = new ConcurrentHashMap<>(); + + record AnnounceItem(Long teamId, Integer taskNumber, String subject, String status, + String memberName, String detail) { + } + + /** + * Queue a settled task for announcement to its lead. Safe to call from any + * thread; no-op when the task has no originating lead conversation. + */ + public void announceTaskSettled(TeamTaskEntity task) { + if (task == null || task.getLeadConversationId() == null) { + return; + } + String detail = TeamTaskStatus.COMPLETED.equals(task.getStatus()) + || TeamTaskStatus.IN_REVIEW.equals(task.getStatus()) + ? task.getResult() : task.getReason(); + AnnounceItem item = new AnnounceItem(task.getTeamId(), task.getTaskNumber(), + task.getSubject(), task.getStatus(), + agentName(task.getAssigneeAgentId()), + detail == null ? "" : detail); + + String key = task.getLeadConversationId(); + List drainNow = null; + synchronized (pending) { + List queue = pending.computeIfAbsent(key, k -> new ArrayList<>()); + queue.add(item); + if (queue.size() >= MAX_BATCH) { + drainNow = pending.remove(key); + } else if (queue.size() == 1) { + DEBOUNCE_SCHEDULER.schedule(() -> drain(key), DEBOUNCE_MILLIS, TimeUnit.MILLISECONDS); + } + } + if (drainNow != null) { + deliver(key, drainNow); + } + } + + /** Timer callback: take whatever accumulated and deliver it. */ + void drain(String leadConversationId) { + List items; + synchronized (pending) { + items = pending.remove(leadConversationId); + } + if (items != null && !items.isEmpty()) { + deliver(leadConversationId, items); + } + } + + void deliver(String leadConversationId, List items) { + deliver(leadConversationId, items, 0); + } + + private void deliver(String leadConversationId, List items, int busyRetries) { + Long teamId = items.get(0).teamId(); + AgentTeamEntity team = teamService.getTeam(teamId); + if (team == null) { + log.warn("Announce dropped: team {} vanished", teamId); + return; + } + if (runningConversations.isActive(leadConversationId) && busyRetries < MAX_BUSY_RETRIES) { + // Lead is mid-turn. Late tasks settling meanwhile join this batch + // via the pending map, so re-queue and re-arm instead of injecting + // into the running turn (which can drop the message on turn end). + List merged = items; + synchronized (pending) { + List late = pending.remove(leadConversationId); + if (late != null) { + merged = new ArrayList<>(items); + merged.addAll(late); + } + } + List retryItems = merged; + DEBOUNCE_SCHEDULER.schedule(() -> deliver(leadConversationId, retryItems, busyRetries + 1), + BUSY_RETRY_MILLIS, TimeUnit.MILLISECONDS); + return; + } + String message = buildAnnouncement(items); + ANNOUNCE_EXECUTOR.submit(() -> wakeLead(team, leadConversationId, message, items.size())); + } + + /** Start a fresh lead turn carrying the merged results; its reply reaches the user. */ + private void wakeLead(AgentTeamEntity team, String leadConversationId, + String message, int taskCount) { + try { + streamTracker.broadcastObject(leadConversationId, "team_announce_start", + Map.of("teamId", String.valueOf(team.getId()), "tasks", taskCount)); + AgentService.ChatResult result = agentService.chatWithUsage( + team.getLeadAgentId(), message, leadConversationId); + streamTracker.broadcastObject(leadConversationId, "team_announce_reply", + Map.of("teamId", String.valueOf(team.getId()), + "content", result == null || result.content() == null + ? "" : result.content())); + log.info("Team {} lead woken with {} task result(s)", team.getId(), taskCount); + } catch (Exception e) { + log.warn("Team {} lead wake-up failed: {}", team.getId(), e.getMessage()); + } + } + + /** Merged announcement text; single- and multi-result variants. */ + static String buildAnnouncement(List items) { + StringBuilder sb = new StringBuilder(512); + long failed = items.stream().filter(i -> TeamTaskStatus.FAILED.equals(i.status())).count(); + if (items.size() == 1) { + sb.append("[System Message] A delegated team task has settled.\n"); + } else { + sb.append("[System Message] ").append(items.size()) + .append(" delegated team tasks have settled"); + if (failed > 0) { + sb.append(" (").append(failed).append(" failed)"); + } + sb.append(".\n"); + } + for (AnnounceItem item : items) { + sb.append("\n--- Task #").append(item.taskNumber()) + .append(" \"").append(item.subject()).append("\" — ") + .append(item.status()) + .append(" (member: ").append(item.memberName()).append(") ---\n"); + if (!item.detail().isBlank()) { + sb.append(item.detail()).append('\n'); + } + } + sb.append(""" + + Review these results against the original request, then reply to the user with ONE synthesized answer. \ + For failed tasks, fix the missing input and re-dispatch with team_tasks(action="retry", taskId=...), or cancel them. \ + Tasks in in_review await human approval — mention that instead of treating them as done."""); + return sb.toString(); + } + + private String agentName(Long agentId) { + if (agentId == null) { + return "-"; + } + AgentEntity agent = agentMapper.selectById(agentId); + return agent != null && agent.getName() != null ? agent.getName() : String.valueOf(agentId); + } +} 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 new file mode 100644 index 00000000..93dfaf41 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java @@ -0,0 +1,234 @@ +package vip.mate.team.service; + +import cn.hutool.core.util.IdUtil; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +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.TeamTaskStatus; +import vip.mate.workspace.conversation.ConversationService; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * Dispatches board tasks to their assigned member agents and closes the + * execution loop: run the member in an isolated child conversation, complete + * (or fail) the task from the run outcome, then re-sweep so released + * dependents and the now-idle member pick up follow-up work. + * + * Concurrency model: the sweep itself takes no locks — {@code assignTask}'s + * conditional UPDATE (pending → in_progress) is the single arbiter, so + * overlapping sweeps can never double-dispatch a task. One member executes at + * most one task at a time. A scheduled sweep self-heals anything a + * notification-path dispatch missed (releases, retries, restarts). + * + * @author MateClaw Team + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class TeamDispatchService { + + /** Result summaries are capped before persisting to keep the board readable. */ + static final int MAX_RESULT_CHARS = 8000; + + /** One JDK 21 virtual thread per member-agent run. */ + private static final ExecutorService DISPATCH_EXECUTOR = + Executors.newVirtualThreadPerTaskExecutor(); + + private final TeamService teamService; + private final TeamTaskService taskService; + private final AgentService agentService; + private final ConversationService conversationService; + private final ChatStreamTracker streamTracker; + private final TeamAnnounceService announceService; + + /** Members with a run currently in flight in this JVM (belt-and-braces on top of hasActiveTask). */ + private final Set runningMembers = ConcurrentHashMap.newKeySet(); + + /** Asynchronously sweep the team's board and dispatch whatever is eligible. */ + public void requestDispatch(Long teamId) { + DISPATCH_EXECUTOR.submit(() -> { + try { + sweep(teamId); + } catch (Exception e) { + log.warn("Team {} dispatch sweep failed: {}", teamId, e.getMessage()); + } + }); + } + + /** + * Periodic self-heal: mark expired leases stale, then sweep every active + * team so released dependents, manual retries and work orphaned by a + * restart are dispatched even when no tool-path notification fired. + */ + @Scheduled(fixedDelay = 30_000, initialDelay = 30_000) + public void scheduledSweep() { + taskService.recoverStaleTasks(); + for (AgentTeamEntity team : teamService.listTeams()) { + if (TeamService.STATUS_ACTIVE.equals(team.getStatus())) { + try { + sweep(team.getId()); + } catch (Exception e) { + log.warn("Scheduled sweep failed for team {}: {}", team.getId(), e.getMessage()); + } + } + } + } + + /** + * Dispatch at most one eligible pending task per assignee. Priority order + * comes from the query; the conditional assign makes the winner unique. + */ + void sweep(Long teamId) { + List candidates = taskService.findDispatchable(teamId); + if (candidates.isEmpty()) { + return; + } + Set dispatchedThisRound = new HashSet<>(); + for (TeamTaskEntity task : candidates) { + Long assignee = task.getAssigneeAgentId(); + if (dispatchedThisRound.contains(assignee) + || runningMembers.contains(assignee) + || taskService.hasActiveTask(teamId, assignee)) { + continue; + } + if (!taskService.assignTask(task.getId(), assignee)) { + continue; // another sweep won the race, or status moved on + } + if (!taskService.tryAcquireDispatch(task.getId())) { + // Circuit breaker tripped; the task was auto-failed — the lead + // must hear about it or the work silently disappears. + announceService.announceTaskSettled(taskService.getTask(task.getId())); + continue; + } + dispatchedThisRound.add(assignee); + TeamTaskEntity assigned = taskService.getTask(task.getId()); + DISPATCH_EXECUTOR.submit(() -> runTask(teamId, assigned)); + } + } + + /** Execute one dispatched task on its member agent, then settle the outcome. */ + private void runTask(Long teamId, TeamTaskEntity task) { + Long memberId = task.getAssigneeAgentId(); + if (!runningMembers.add(memberId)) { + // Same member picked up concurrently in this JVM; put the task back. + taskService.retryTask(task.getId()); + return; + } + String childConvId = "team-task-" + IdUtil.fastSimpleUUID(); + try { + conversationService.createChildConversation(childConvId, memberId, "system", + null, task.getLeadConversationId()); + taskService.attachConversation(task.getId(), childConvId); + broadcast(task, "team_task_dispatched", Map.of()); + log.info("Team {} task #{} dispatched to agent {} (conv {})", + teamId, task.getTaskNumber(), memberId, childConvId); + + AgentService.ChatResult result = agentService.chatWithUsage( + memberId, buildDispatchContent(task), childConvId); + + settleOutcome(task, result == null ? null : result.content()); + } catch (Exception e) { + log.warn("Team {} task #{} member run failed: {}", teamId, task.getTaskNumber(), + e.getMessage()); + taskService.failTask(task.getId(), truncate("member run error: " + e.getMessage(), 1000)); + broadcast(task, "team_task_failed", Map.of("reason", String.valueOf(e.getMessage()))); + announceService.announceTaskSettled(taskService.getTask(task.getId())); + } finally { + runningMembers.remove(memberId); + // Chain: dispatch released dependents and the member's next task. + requestDispatch(teamId); + } + } + + /** + * Settle a finished member run. If the member already moved the task + * (explicit complete, blocker fail) the run outcome is not applied on top; + * otherwise the final reply becomes the task result (auto-completion). + */ + void settleOutcome(TeamTaskEntity task, String reply) { + TeamTaskEntity current = taskService.getTask(task.getId()); + if (current == null) { + return; + } + if (TeamTaskStatus.IN_PROGRESS.equals(current.getStatus())) { + List released = taskService.completeTask(task.getId(), null, + truncate(reply == null || reply.isBlank() ? "(no output)" : reply, + MAX_RESULT_CHARS)); + current = taskService.getTask(task.getId()); + log.info("Team task #{} auto-completed ({} dependents released)", + task.getTaskNumber(), released.size()); + } + String event = switch (current.getStatus()) { + case TeamTaskStatus.FAILED -> "team_task_failed"; + case TeamTaskStatus.IN_REVIEW -> "team_task_in_review"; + default -> "team_task_completed"; + }; + Map payload = new HashMap<>(); + payload.put("status", current.getStatus()); + if (current.getResult() != null) { + payload.put("resultPreview", truncate(current.getResult(), 200)); + } + if (current.getReason() != null) { + payload.put("reason", current.getReason()); + } + broadcast(task, event, payload); + announceService.announceTaskSettled(current); + } + + /** The full instruction envelope the member receives; it cannot see the lead's conversation. */ + private String buildDispatchContent(TeamTaskEntity task) { + StringBuilder sb = new StringBuilder(1024); + sb.append("[Assigned team task #").append(task.getTaskNumber()) + .append(" (taskId: ").append(task.getId()).append(")]\n") + .append("Subject: ").append(task.getSubject()).append('\n'); + if (task.getDescription() != null && !task.getDescription().isBlank()) { + sb.append("\n").append(task.getDescription()).append('\n'); + } + sb.append(""" + + [Instructions] + - Execute this task now. Your final reply becomes the task result reported to the team lead, so end with a complete, self-contained summary of what you produced. + - Report milestones with team_tasks(action="progress", taskId=%s, percent=..., step=...). + - If you are missing an input you cannot obtain yourself, call team_tasks(action="comment", taskId=%s, type="blocker", text="what you need") and stop. + """.formatted(task.getId(), task.getId())); + return sb.toString(); + } + + /** Push a task event onto the lead conversation's SSE stream (UI + observability). */ + private void broadcast(TeamTaskEntity task, String event, Map extra) { + if (task.getLeadConversationId() == null) { + return; + } + try { + Map payload = new HashMap<>(extra); + payload.put("taskId", String.valueOf(task.getId())); + payload.put("taskNumber", task.getTaskNumber()); + payload.put("subject", task.getSubject()); + payload.put("teamId", String.valueOf(task.getTeamId())); + payload.put("assigneeAgentId", String.valueOf(task.getAssigneeAgentId())); + streamTracker.broadcastObject(task.getLeadConversationId(), event, payload); + } catch (Exception e) { + log.debug("Team task event broadcast skipped: {}", e.getMessage()); + } + } + + private static String truncate(String s, int max) { + if (s == null || s.length() <= max) { + return s; + } + return s.substring(0, max) + "\n...(truncated)"; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/team/service/TeamAnnounceServiceTest.java b/mateclaw-server/src/test/java/vip/mate/team/service/TeamAnnounceServiceTest.java new file mode 100644 index 00000000..b4d1be82 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/team/service/TeamAnnounceServiceTest.java @@ -0,0 +1,152 @@ +package vip.mate.team.service; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import vip.mate.agent.AgentService; +import vip.mate.agent.model.AgentEntity; +import vip.mate.agent.repository.AgentMapper; +import vip.mate.agent.runtime.RunningConversationRegistry; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.team.model.AgentTeamEntity; +import vip.mate.team.model.TeamTaskEntity; +import vip.mate.team.model.TeamTaskStatus; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +/** + * Pins the announce contract: settled results are batched per lead + * conversation and delivered as ONE merged wake-up message; a busy lead defers + * delivery instead of risking an in-turn drop; the merged text carries the + * synthesis / retry instructions the lead acts on. + */ +class TeamAnnounceServiceTest { + + private static final Long TEAM_ID = 10L; + private static final Long LEAD_ID = 1L; + private static final String LEAD_CONV = "lead-conv"; + + private TeamService teamService; + private AgentService agentService; + private AgentMapper agentMapper; + private RunningConversationRegistry runningConversations; + private ChatStreamTracker streamTracker; + private TeamAnnounceService service; + + @BeforeEach + void setUp() { + teamService = mock(TeamService.class); + agentService = mock(AgentService.class); + agentMapper = mock(AgentMapper.class); + runningConversations = mock(RunningConversationRegistry.class); + streamTracker = mock(ChatStreamTracker.class); + service = new TeamAnnounceService(teamService, agentService, agentMapper, + runningConversations, streamTracker); + + AgentTeamEntity team = new AgentTeamEntity(); + team.setId(TEAM_ID); + team.setLeadAgentId(LEAD_ID); + when(teamService.getTeam(TEAM_ID)).thenReturn(team); + + AgentEntity member = new AgentEntity(); + member.setName("写手"); + when(agentMapper.selectById(any())).thenReturn(member); + } + + private TeamTaskEntity settled(Long id, String status, String detail) { + TeamTaskEntity t = new TeamTaskEntity(); + t.setId(id); + t.setTeamId(TEAM_ID); + t.setTaskNumber(id.intValue()); + t.setSubject("task " + id); + t.setStatus(status); + t.setAssigneeAgentId(2L); + t.setLeadConversationId(LEAD_CONV); + if (TeamTaskStatus.FAILED.equals(status)) { + t.setReason(detail); + } else { + t.setResult(detail); + } + return t; + } + + @Test + @DisplayName("results settling together wake the lead ONCE with a merged message") + void batchedResultsSingleWakeUp() { + when(runningConversations.isActive(LEAD_CONV)).thenReturn(false); + service.announceTaskSettled(settled(1L, TeamTaskStatus.COMPLETED, "report done")); + service.announceTaskSettled(settled(2L, TeamTaskStatus.FAILED, "blocked: no docs")); + + service.drain(LEAD_CONV); + + ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); + verify(agentService, timeout(3000)).chatWithUsage(eq(LEAD_ID), captor.capture(), eq(LEAD_CONV)); + String message = captor.getValue(); + assertTrue(message.contains("2 delegated team tasks have settled (1 failed)")); + assertTrue(message.contains("Task #1")); + assertTrue(message.contains("report done")); + assertTrue(message.contains("Task #2")); + assertTrue(message.contains("blocked: no docs")); + // Drained means a later timer fire must not wake the lead again. + service.drain(LEAD_CONV); + verify(agentService, after(300).times(1)).chatWithUsage(any(), anyString(), anyString()); + } + + @Test + @DisplayName("a busy lead defers delivery — no concurrent turn is started") + void busyLeadDefers() { + when(runningConversations.isActive(LEAD_CONV)).thenReturn(true); + service.announceTaskSettled(settled(1L, TeamTaskStatus.COMPLETED, "done")); + + service.drain(LEAD_CONV); + + verify(agentService, after(500).never()).chatWithUsage(any(), anyString(), anyString()); + } + + @Test + @DisplayName("a task without a lead conversation is silently skipped") + void noLeadConversationNoop() { + TeamTaskEntity orphan = settled(1L, TeamTaskStatus.COMPLETED, "done"); + orphan.setLeadConversationId(null); + + service.announceTaskSettled(orphan); + service.drain(LEAD_CONV); + + verify(agentService, after(300).never()).chatWithUsage(any(), anyString(), anyString()); + } + + @Test + @DisplayName("the wake-up run emits start and reply SSE events") + void wakeUpEmitsSseEvents() { + when(runningConversations.isActive(LEAD_CONV)).thenReturn(false); + when(agentService.chatWithUsage(eq(LEAD_ID), anyString(), eq(LEAD_CONV))) + .thenReturn(AgentService.ChatResult.contentOnly("综合汇报")); + + service.announceTaskSettled(settled(1L, TeamTaskStatus.COMPLETED, "done")); + service.drain(LEAD_CONV); + + verify(streamTracker, timeout(3000)) + .broadcastObject(eq(LEAD_CONV), eq("team_announce_start"), any()); + verify(streamTracker, timeout(3000)) + .broadcastObject(eq(LEAD_CONV), eq("team_announce_reply"), any()); + } + + @Test + @DisplayName("announcement text: single result keeps the singular form and the playbook") + void announcementText() { + String single = TeamAnnounceService.buildAnnouncement(List.of( + new TeamAnnounceService.AnnounceItem(TEAM_ID, 1, "collect", TeamTaskStatus.COMPLETED, + "写手", "all collected"))); + assertTrue(single.contains("A delegated team task has settled")); + assertTrue(single.contains("member: 写手")); + assertTrue(single.contains("ONE synthesized answer")); + assertTrue(single.contains("action=\"retry\"")); + } +} 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 new file mode 100644 index 00000000..4651d0b4 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java @@ -0,0 +1,179 @@ +package vip.mate.team.service; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import vip.mate.agent.AgentService; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.team.model.TeamTaskEntity; +import vip.mate.team.model.TeamTaskStatus; +import vip.mate.workspace.conversation.ConversationService; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.*; + +/** + * Pins the dispatch loop's arbitration rules: one task per member per sweep, + * busy members skipped, the conditional assign as the only winner gate, the + * circuit breaker short-circuit, and outcome settling (auto-complete vs + * respecting a state the member already set). + */ +class TeamDispatchServiceTest { + + private static final Long TEAM_ID = 10L; + private static final Long MEMBER_A = 2L; + private static final Long MEMBER_B = 3L; + + private TeamService teamService; + private TeamTaskService taskService; + private AgentService agentService; + private ConversationService conversationService; + private ChatStreamTracker streamTracker; + private TeamAnnounceService announceService; + private TeamDispatchService service; + + @BeforeEach + void setUp() { + teamService = mock(TeamService.class); + taskService = mock(TeamTaskService.class); + agentService = mock(AgentService.class); + conversationService = mock(ConversationService.class); + streamTracker = mock(ChatStreamTracker.class); + announceService = mock(TeamAnnounceService.class); + service = new TeamDispatchService(teamService, taskService, agentService, + conversationService, streamTracker, announceService); + } + + private TeamTaskEntity task(Long id, Long assignee) { + TeamTaskEntity t = new TeamTaskEntity(); + t.setId(id); + t.setTeamId(TEAM_ID); + t.setTaskNumber(id.intValue()); + t.setSubject("task " + id); + t.setStatus(TeamTaskStatus.PENDING); + t.setAssigneeAgentId(assignee); + t.setLeadConversationId("lead-conv"); + return t; + } + + // ==================== sweep arbitration ==================== + + @Test + @DisplayName("one task per assignee per sweep — the second task for the same member waits") + void onePerAssigneePerSweep() { + TeamTaskEntity first = task(1L, MEMBER_A); + TeamTaskEntity second = task(2L, MEMBER_A); + TeamTaskEntity other = task(3L, MEMBER_B); + // Second stub is empty so the async post-run re-sweep chain terminates. + when(taskService.findDispatchable(TEAM_ID)) + .thenReturn(List.of(first, second, other)) + .thenReturn(List.of()); + when(taskService.hasActiveTask(eq(TEAM_ID), any())).thenReturn(false); + when(taskService.assignTask(any(), any())).thenReturn(true); + when(taskService.tryAcquireDispatch(any())).thenReturn(true); + when(taskService.getTask(any())).thenAnswer(inv -> + task(inv.getArgument(0), MEMBER_A)); + + service.sweep(TEAM_ID); + + verify(taskService).assignTask(1L, MEMBER_A); + verify(taskService, never()).assignTask(eq(2L), any()); + verify(taskService).assignTask(3L, MEMBER_B); + } + + @Test + @DisplayName("a member already executing a task is skipped entirely") + void busyMemberSkipped() { + when(taskService.findDispatchable(TEAM_ID)).thenReturn(List.of(task(1L, MEMBER_A))); + when(taskService.hasActiveTask(TEAM_ID, MEMBER_A)).thenReturn(true); + + service.sweep(TEAM_ID); + + verify(taskService, never()).assignTask(any(), any()); + } + + @Test + @DisplayName("losing the conditional assign means another sweep won — no dispatch") + void assignRaceLostSkips() { + when(taskService.findDispatchable(TEAM_ID)).thenReturn(List.of(task(1L, MEMBER_A))); + when(taskService.hasActiveTask(TEAM_ID, MEMBER_A)).thenReturn(false); + when(taskService.assignTask(1L, MEMBER_A)).thenReturn(false); + + service.sweep(TEAM_ID); + + verify(taskService, never()).tryAcquireDispatch(any()); + } + + @Test + @DisplayName("a tripped circuit breaker skips the run but announces the auto-fail to the lead") + void breakerStopsDispatch() { + when(taskService.findDispatchable(TEAM_ID)).thenReturn(List.of(task(1L, MEMBER_A))); + when(taskService.hasActiveTask(TEAM_ID, MEMBER_A)).thenReturn(false); + when(taskService.assignTask(1L, MEMBER_A)).thenReturn(true); + when(taskService.tryAcquireDispatch(1L)).thenReturn(false); + TeamTaskEntity failed = task(1L, MEMBER_A); + failed.setStatus(TeamTaskStatus.FAILED); + when(taskService.getTask(1L)).thenReturn(failed); + + service.sweep(TEAM_ID); + + // The work must not vanish silently: the lead hears about the auto-fail. + verify(announceService).announceTaskSettled(failed); + verifyNoInteractions(agentService); + } + + // ==================== outcome settling ==================== + + @Test + @DisplayName("an in_progress task is auto-completed with the member's final reply") + void settleAutoCompletes() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + TeamTaskEntity done = task(1L, MEMBER_A); + done.setStatus(TeamTaskStatus.COMPLETED); + done.setResult("analysis finished"); + when(taskService.getTask(1L)).thenReturn(running, done); + when(taskService.completeTask(eq(1L), isNull(), anyString())).thenReturn(List.of()); + + service.settleOutcome(running, "analysis finished"); + + verify(taskService).completeTask(1L, null, "analysis finished"); + verify(streamTracker).broadcastObject(eq("lead-conv"), eq("team_task_completed"), any()); + verify(announceService).announceTaskSettled(done); + } + + @Test + @DisplayName("a task the member already failed via blocker is not completed on top") + void settleRespectsMemberFailure() { + TeamTaskEntity failed = task(1L, MEMBER_A); + failed.setStatus(TeamTaskStatus.FAILED); + failed.setReason("blocked: missing docs"); + when(taskService.getTask(1L)).thenReturn(failed); + + service.settleOutcome(failed, "irrelevant reply"); + + verify(taskService, never()).completeTask(any(), any(), anyString()); + verify(streamTracker).broadcastObject(eq("lead-conv"), eq("team_task_failed"), any()); + } + + @Test + @DisplayName("an oversized member reply is truncated before persisting") + void settleTruncatesLongReply() { + TeamTaskEntity running = task(1L, MEMBER_A); + running.setStatus(TeamTaskStatus.IN_PROGRESS); + when(taskService.getTask(1L)).thenReturn(running, running); + when(taskService.completeTask(eq(1L), isNull(), anyString())).thenReturn(List.of()); + + service.settleOutcome(running, "x".repeat(TeamDispatchService.MAX_RESULT_CHARS + 500)); + + verify(taskService).completeTask(eq(1L), isNull(), argThat(r -> + r.length() <= TeamDispatchService.MAX_RESULT_CHARS + 20 + && r.endsWith("...(truncated)"))); + } +} diff --git a/mateclaw-ui/src/types/components.d.ts b/mateclaw-ui/src/types/components.d.ts index 1dc64a8d..14909b9f 100644 --- a/mateclaw-ui/src/types/components.d.ts +++ b/mateclaw-ui/src/types/components.d.ts @@ -11,9 +11,7 @@ export {} /* prettier-ignore */ declare module 'vue' { export interface GlobalComponents { - ElAlert: typeof import('element-plus/es/components/alert/index')['ElAlert'] ElButton: typeof import('element-plus/es/components/button/index')['ElButton'] - ElCard: typeof import('element-plus/es/components/card/index')['ElCard'] ElConfigProvider: typeof import('element-plus/es/components/config-provider/index')['ElConfigProvider'] ElDatePicker: typeof import('element-plus/es/components/date-picker/index')['ElDatePicker'] ElDialog: typeof import('element-plus/es/components/dialog/index')['ElDialog'] @@ -22,22 +20,17 @@ declare module 'vue' { ElDropdownItem: typeof import('element-plus/es/components/dropdown/index')['ElDropdownItem'] ElDropdownMenu: typeof import('element-plus/es/components/dropdown/index')['ElDropdownMenu'] ElEmpty: typeof import('element-plus/es/components/empty/index')['ElEmpty'] - ElForm: typeof import('element-plus/es/components/form/index')['ElForm'] - ElFormItem: typeof import('element-plus/es/components/form/index')['ElFormItem'] ElIcon: typeof import('element-plus/es/components/icon/index')['ElIcon'] ElImageViewer: typeof import('element-plus/es/components/image-viewer/index')['ElImageViewer'] - ElInput: typeof import('element-plus/es/components/input/index')['ElInput'] ElOption: typeof import('element-plus/es/components/select/index')['ElOption'] ElPagination: typeof import('element-plus/es/components/pagination/index')['ElPagination'] ElPopover: typeof import('element-plus/es/components/popover/index')['ElPopover'] - ElProgress: typeof import('element-plus/es/components/progress/index')['ElProgress'] ElSelect: typeof import('element-plus/es/components/select/index')['ElSelect'] ElSkeleton: typeof import('element-plus/es/components/skeleton/index')['ElSkeleton'] ElTable: typeof import('element-plus/es/components/table/index')['ElTable'] ElTableColumn: typeof import('element-plus/es/components/table/index')['ElTableColumn'] ElTabPane: typeof import('element-plus/es/components/tabs/index')['ElTabPane'] ElTabs: typeof import('element-plus/es/components/tabs/index')['ElTabs'] - ElTag: typeof import('element-plus/es/components/tag/index')['ElTag'] ElTooltip: typeof import('element-plus/es/components/tooltip/index')['ElTooltip'] RouterLink: typeof import('vue-router')['RouterLink'] RouterView: typeof import('vue-router')['RouterView']