From cc0661c07c17f08d6d899f8f70d491af98e1d0f9 Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Sun, 16 Aug 2026 14:42:39 +0800 Subject: [PATCH] fix(team): harden checkpoint loops and board recovery --- .../agent/graph/NodeStreamingChatHelper.java | 18 ++++- .../graph/plan/node/PlanGenerationNode.java | 17 ++++- .../vip/mate/team/service/TeamPlanBridge.java | 76 ++++++++++++++----- .../vip/mate/team/service/TeamRunService.java | 13 ++++ .../mate/team/service/TeamTaskService.java | 33 ++++++++ .../src/main/resources/application.yml | 4 +- .../src/main/resources/logback-spring.xml | 7 +- .../mate/team/service/TeamPlanBridgeTest.java | 30 ++++++++ .../team/service/TeamTaskServiceTest.java | 14 ++++ .../src/api/__tests__/teamRuns.test.ts | 5 +- mateclaw-ui/src/api/index.ts | 19 ++++- .../__tests__/teamStoreGeneration.test.ts | 29 +++++++ mateclaw-ui/src/stores/useTeamStore.ts | 19 ++++- mateclaw-ui/src/views/Teams.vue | 21 +++++ 14 files changed, 270 insertions(+), 35 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index 52ae444d..ada8e499 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -408,6 +408,7 @@ public class NodeStreamingChatHelper { // retry (e.g., proxy timeout returns HTTP 200 with empty body). Keep // the cap low — if it truly takes 4+ attempts, the provider is sick. static final int MAX_RETRIES_EMPTY_RESPONSE = 3; + static final long EMPTY_RESPONSE_BACKOFF_MS = 250; // UNKNOWN: conservative retry cap. Defensive: retry what we can't // classify, but with a smaller budget than SERVER_ERROR (5 vs 10) to // avoid masking truly fatal errors. MAX_TOTAL_DURATION_MS is the @@ -888,6 +889,7 @@ public class NodeStreamingChatHelper { if (errType == ErrorType.EMPTY_RESPONSE && attempt < errType.retryBudget()) { log.warn("[{}] Primary returned empty response (attempt {}/{}), retrying same model...", phase, attempt + 1, errType.retryBudget() + 1); + retryType.set(ErrorType.EMPTY_RESPONSE); continue; } // Generic routing — driven entirely by the ErrorType policy @@ -1108,6 +1110,7 @@ public class NodeStreamingChatHelper { AtomicReference retryHintRef) { if (attempt > 0) { boolean overloaded = retryTypeRef.get() == ErrorType.OVERLOADED; + boolean emptyResponse = retryTypeRef.get() == ErrorType.EMPTY_RESPONSE; Long hintedMs = retryHintRef.get(); long delay; if (hintedMs != null && hintedMs > 0) { @@ -1118,6 +1121,8 @@ public class NodeStreamingChatHelper { // in lockstep at the stated instant. delay = Math.min(hintedMs, HINTED_BACKOFF_CAP_MS) + ThreadLocalRandom.current().nextLong(0, 1_000); + } else if (emptyResponse) { + delay = EMPTY_RESPONSE_BACKOFF_MS; } else if (overloaded) { // Saturated provider: recovery periods run tens of seconds, so // the generic 3s-based exponential would burn attempts before @@ -1135,7 +1140,7 @@ public class NodeStreamingChatHelper { log.warn("[{}] Retry attempt {}/{} after {}ms (prev type={}) for conversation {}", phase, attempt, MAX_RETRIES, delay, retryTypeRef.get(), conversationId); // 广播给前端：用户可见的重试倒计时 - if (broadcast) { + if (broadcast && !emptyResponse) { String cause = overloaded ? "模型服务繁忙" : "请求频率受限"; broadcastDelta(conversationId, "warning", buildDeltaJson("⏱️ " + cause + "，等待 " + (delay / 1000) + " 秒后重试（第 " + attempt + "/" + MAX_RETRIES + " 次）...")); @@ -1588,7 +1593,10 @@ public class NodeStreamingChatHelper { && thinkingAccum.length() == 0 && toolCallAccumulators.isEmpty()) { log.warn("[{}] LLM returned empty response (no content, no thinking, no tool calls) — marking as EMPTY_RESPONSE for fallback", phase); - return buildErrorResultWithType("LLM 返回空响应", conversationId, phase, ErrorType.EMPTY_RESPONSE); + // The outer policy owns retry/failover. Keep transient empty + // attempts out of the user-visible error stream, and leave text + // blank so callers can apply a deterministic final fallback. + return buildEmptyResponseResult(); } String truncationReason = truncatedByThinkingCap ? "thinking_only_no_content" @@ -1914,6 +1922,12 @@ public class NodeStreamingChatHelper { List.of(), false, 0, 0, false, errorMsg, errorType); } + private StreamResult buildEmptyResponseResult() { + return new StreamResult("", "", new AssistantMessage(""), + List.of(), false, 0, 0, false, + "LLM 返回空响应", ErrorType.EMPTY_RESPONSE); + } + /** 构建 error 事件的 JSON payload */ private static String buildErrorEventJson(String message, String conversationId, ErrorType errorType) { StringBuilder sb = new StringBuilder("{"); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java index b2941f87..44bd5a3d 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/PlanGenerationNode.java @@ -520,7 +520,7 @@ public class PlanGenerationNode implements NodeAction { .build(); } if (parked instanceof TeamPlanBridge.InFlight inFlight) { - log.info("[PlanGeneration] Delegated plan still in flight — answering with progress"); + log.info("[PlanGeneration] Answering from delegated team state without triage LLM"); if (streamingHelper != null) { streamingHelper.broadcastContent(conversationId, inFlight.progressText()); } @@ -686,7 +686,8 @@ public class PlanGenerationNode implements NodeAction { String llmResponse = result.text(); log.info("[PlanGeneration] Triage completed in {}ms", triageMs); - log.debug("[PlanGeneration] LLM response: {}", llmResponse); + log.debug("[PlanGeneration] LLM response received ({} chars)", + llmResponse == null ? 0 : llmResponse.length()); // D-6: emit triage perf summary events.add(GraphEventPublisher.perfSummary("triage", Map.of( @@ -695,7 +696,17 @@ public class PlanGenerationNode implements NodeAction { "completion_tokens", result.completionTokens() ))); - TriageResult triage = converter.convert(llmResponse); + TriageResult triage; + if (!StringUtils.hasText(llmResponse)) { + // An upstream model can occasionally finish without content. + // Treat it as a recoverable single-step route, not a parser + // exception (and therefore not a false backend ERROR). + log.warn("[PlanGeneration] Triage returned empty content; using single-step fallback"); + triage = new TriageResult(true, null, "single_step", + List.of(persistGoal), null, null); + } else { + triage = converter.convert(llmResponse); + } boolean needsPlanning = triage != null && triage.needsPlanning(); if (!needsPlanning) { 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 4891c14d..cdc17d80 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 @@ -249,8 +249,19 @@ public class TeamPlanBridge { * Normal status questions retain the detailed board snapshot. */ public ParkedPlanState checkParkedPlan(String conversationId, String currentMessage) { + String checkpointTag = checkpointTagOf(currentMessage); PlanEntity plan = planningService.findDelegatedPlan(conversationId); if (plan == null) { + if (checkpointTag != null) { + Optional latest = runService.findLatestConversationRun(conversationId); + if (latest.isPresent()) { + List tasks = taskService.listTasksByRun(latest.get().getId()); + if (!tasks.isEmpty()) { + recordCheckpointEvidence(latest.get().getTeamId(), tasks, checkpointTag); + return new InFlight(buildCheckpointText(tasks, checkpointTag)); + } + } + } return new None(); } Optional teamOpt = leadTeam(parseAgentId(plan.getAgentId())); @@ -270,6 +281,10 @@ public class TeamPlanBridge { List steps = planningService.getSubPlans(plan.getId()).stream() .map(sub -> sub.getDescription()) .toList(); + if (checkpointTag != null) { + recordCheckpointEvidence(teamOpt.get().getId(), tasks, checkpointTag); + return new InFlight(buildCheckpointText(tasks, checkpointTag)); + } if (!allTerminal) { return new InFlight(buildProgressText(tasks, currentMessage)); } @@ -352,24 +367,7 @@ public class TeamPlanBridge { private String buildProgressText(List tasks, String currentMessage) { String checkpointTag = checkpointTagOf(currentMessage); if (checkpointTag != null) { - long completed = tasks.stream() - .filter(task -> TeamTaskStatus.COMPLETED.equals(task.getStatus())) - .count(); - TeamTaskEntity active = tasks.stream() - .filter(task -> !TeamTaskStatus.isTerminal(task.getStatus())) - .findFirst() - .orElse(tasks.get(tasks.size() - 1)); - StringBuilder compact = new StringBuilder(checkpointTag) - .append("｜执行中 ").append(completed).append('/').append(tasks.size()) - .append("｜#").append(active.getTaskNumber()).append(' ') - .append(active.getStatus()); - if (active.getProgressPercent() != null) { - compact.append(' ').append(active.getProgressPercent()).append('%'); - } - if ("R100".equalsIgnoreCase(checkpointTag)) { - compact.append("（已完成第100轮检查点）"); - } - return compact.toString(); + return buildCheckpointText(tasks, checkpointTag); } StringBuilder sb = new StringBuilder("计划仍在团队任务板上执行中：\n"); for (TeamTaskEntity task : tasks) { @@ -390,6 +388,48 @@ public class TeamPlanBridge { return sb.toString(); } + private String buildCheckpointText(List tasks, String checkpointTag) { + long completed = tasks.stream() + .filter(task -> TeamTaskStatus.COMPLETED.equals(task.getStatus())) + .count(); + boolean allTerminal = tasks.stream().allMatch(task -> TeamTaskStatus.isTerminal(task.getStatus())); + TeamTaskEntity focus = tasks.stream() + .filter(task -> !TeamTaskStatus.isTerminal(task.getStatus())) + .findFirst() + .orElse(tasks.get(tasks.size() - 1)); + StringBuilder compact = new StringBuilder(checkpointTag) + .append(allTerminal ? "｜已完成 " : "｜执行中 ") + .append(completed).append('/').append(tasks.size()) + .append("｜#").append(focus.getTaskNumber()).append(' ') + .append(focus.getStatus()); + if (focus.getProgressPercent() != null) { + compact.append(' ').append(focus.getProgressPercent()).append('%'); + } + if ("R100".equalsIgnoreCase(checkpointTag)) { + compact.append("（已完成第100轮检查点）"); + } + return compact.toString(); + } + + private void recordCheckpointEvidence(Long teamId, List tasks, + String checkpointTag) { + TeamTaskEntity tracker = taskService.findCheckpointTracker(teamId).orElseGet(() -> tasks.stream() + .filter(this::isCheckpointTracker) + .findFirst() + .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); + } + + private boolean isCheckpointTracker(TeamTaskEntity task) { + String text = (task.getSubject() == null ? "" : task.getSubject()) + " " + + (task.getDescription() == null ? "" : task.getDescription()); + String lower = text.toLowerCase(); + return text.contains("检查点") || text.contains("共享跟踪") + || lower.contains("checkpoint") || lower.contains("r001-r100"); + } + static String checkpointTagOf(String message) { if (message == null || message.isBlank()) { return null; diff --git a/mateclaw-server/src/main/java/vip/mate/team/service/TeamRunService.java b/mateclaw-server/src/main/java/vip/mate/team/service/TeamRunService.java index de00227c..28d02e2a 100644 --- a/mateclaw-server/src/main/java/vip/mate/team/service/TeamRunService.java +++ b/mateclaw-server/src/main/java/vip/mate/team/service/TeamRunService.java @@ -27,6 +27,7 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.List; import java.util.Set; +import java.util.Optional; import java.util.stream.Collectors; /** Owns team run creation, lifecycle transitions, authorization, and reads. */ @@ -166,6 +167,18 @@ public class TeamRunService { return listRuns(null, conversationId, workspaceId, false); } + /** Latest run linked to an internal lead conversation, including terminal runs. */ + public Optional findLatestConversationRun(String conversationId) { + if (conversationId == null || conversationId.isBlank()) { + return Optional.empty(); + } + return Optional.ofNullable(runMapper.selectOne(Wrappers.lambdaQuery() + .eq(TeamRunEntity::getLeadConversationId, conversationId) + .orderByDesc(TeamRunEntity::getCreateTime) + .orderByDesc(TeamRunEntity::getId) + .last("LIMIT 1"))); + } + private List listRuns(Long teamId, String conversationId, Long workspaceId, boolean activeOnly) { var query = Wrappers.lambdaQuery() 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 316a539d..9001fdec 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 @@ -396,6 +396,20 @@ public class TeamTaskService { .orderByAsc(TeamTaskCommentEntity::getCreateTime)); } + /** Persist a note once, keyed by an exact stable content value. */ + @Transactional + public synchronized boolean addCommentOnce(Long taskId, String authorType, String authorId, + String commentType, String content) { + Long existing = commentMapper.selectCount(Wrappers.lambdaQuery() + .eq(TeamTaskCommentEntity::getTaskId, taskId) + .eq(TeamTaskCommentEntity::getContent, content)); + if (existing != null && existing > 0) { + return false; + } + addComment(taskId, authorType, authorId, commentType, content); + return true; + } + // ==================== timeline events ==================== /** Timeline detail cap, matching the column width. */ @@ -642,6 +656,25 @@ public class TeamTaskService { .orderByAsc(TeamTaskEntity::getCreateTime)); } + /** + * Locate the team's dedicated long-running checkpoint tracker, even when + * the current checkpoint belongs to a later run. Highest priority and + * newest creation win when historical tests left more than one candidate. + */ + public java.util.Optional findCheckpointTracker(Long teamId) { + return java.util.Optional.ofNullable(taskMapper.selectOne( + Wrappers.lambdaQuery() + .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")) + .orderByDesc(TeamTaskEntity::getPriority) + .orderByDesc(TeamTaskEntity::getCreateTime) + .last("LIMIT 1"))); + } + public List listTasks(Long teamId, List statuses) { return listTasks(teamId, statuses, null, null); } diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index b2ff731c..e5fe0b8e 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -113,7 +113,9 @@ spring: mybatis-plus: configuration: map-underscore-to-camel-case: true - log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl + # SQL parameter values can contain prompts/model thinking. Keep them out of + # logs by default; operators may explicitly opt in for short-lived diagnosis. + log-impl: ${MATECLAW_MYBATIS_LOG_IMPL:org.apache.ibatis.logging.nologging.NoLoggingImpl} # SpringDoc OpenAPI # 注意：/swagger-ui*、/v3/api-docs*、/webjars/** 落在 SecurityConfig 的 diff --git a/mateclaw-server/src/main/resources/logback-spring.xml b/mateclaw-server/src/main/resources/logback-spring.xml index 48ebe834..e915772e 100644 --- a/mateclaw-server/src/main/resources/logback-spring.xml +++ b/mateclaw-server/src/main/resources/logback-spring.xml @@ -104,9 +104,6 @@ - - - @@ -117,6 +114,10 @@ + + + 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 881ddb87..b07de0fb 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 @@ -294,6 +294,7 @@ class TeamPlanBridgeTest { TeamTaskEntity active = task(102L, 47, 1, TeamTaskStatus.IN_PROGRESS); active.setProgressPercent(80); when(taskService.listTasksByPlan(TEAM_ID, PLAN_ID)).thenReturn(List.of(completed, active)); + when(taskService.findCheckpointTracker(TEAM_ID)).thenReturn(Optional.of(active)); TeamPlanBridge.InFlight state = assertInstanceOf(TeamPlanBridge.InFlight.class, bridge.checkParkedPlan(CONV, @@ -302,10 +303,39 @@ class TeamPlanBridgeTest { assertEquals("R100｜执行中 1/2｜#47 in_progress 80%（已完成第100轮检查点）", state.progressText()); assertFalse(state.progressText().contains("\n")); + verify(taskService).addCommentOnce(102L, TeamTaskService.AUTHOR_SYSTEM, + "team-plan-bridge", TeamTaskService.COMMENT_NOTE, + "[checkpoint:R100] acknowledged"); assertEquals("R002", TeamPlanBridge.checkpointTagOf("R002/100 checkpoint")); assertNull(TeamPlanBridge.checkpointTagOf("R002 ordinary status")); } + @Test + @DisplayName("checkpoint fast path survives after the delegated plan has settled") + void compactCheckpointUsesLatestTerminalRun() { + when(planningService.findDelegatedPlan(CONV)).thenReturn(null); + TeamRunEntity latest = new TeamRunEntity(); + latest.setId(RUN_ID); + when(runService.findLatestConversationRun(CONV)).thenReturn(Optional.of(latest)); + TeamTaskEntity first = task(101L, 46, 0, TeamTaskStatus.COMPLETED); + TeamTaskEntity tracker = task(102L, 47, 1, TeamTaskStatus.COMPLETED); + tracker.setSubject("R001-R100 共享跟踪检查点"); + tracker.setProgressPercent(100); + when(taskService.listTasksByRun(RUN_ID)).thenReturn(List.of(first, tracker)); + latest.setTeamId(TEAM_ID); + TeamTaskEntity crossRunTracker = task(103L, 54, 2, TeamTaskStatus.COMPLETED); + crossRunTracker.setSubject("R001-R100 共享跟踪检查点"); + when(taskService.findCheckpointTracker(TEAM_ID)).thenReturn(Optional.of(crossRunTracker)); + + 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, + "team-plan-bridge", TeamTaskService.COMMENT_NOTE, + "[checkpoint:R047] acknowledged"); + } + @Test @DisplayName("a settled board syncs the sub-plan mirror and returns summary-ready results") void gateSettled() { 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 d2fcf84f..163f47cc 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 @@ -421,6 +421,20 @@ class TeamTaskServiceTest { verify(taskMapper, never()).update(isNull(), any()); } + @Test + @DisplayName("checkpoint evidence note is inserted only when its stable key is absent") + void addCommentOnceIsIdempotent() { + when(commentMapper.selectCount(any(LambdaQueryWrapper.class))).thenReturn(1L, 0L); + when(taskMapper.selectById(5L)).thenReturn(task(5L, TeamTaskStatus.COMPLETED)); + + assertFalse(service.addCommentOnce(5L, TeamTaskService.AUTHOR_SYSTEM, "bridge", + TeamTaskService.COMMENT_NOTE, "[checkpoint:R001] acknowledged")); + assertTrue(service.addCommentOnce(5L, TeamTaskService.AUTHOR_SYSTEM, "bridge", + TeamTaskService.COMMENT_NOTE, "[checkpoint:R002] acknowledged")); + + verify(commentMapper, times(1)).insert(any(TeamTaskCommentEntity.class)); + } + // ==================== circuit breaker ==================== @Test diff --git a/mateclaw-ui/src/api/__tests__/teamRuns.test.ts b/mateclaw-ui/src/api/__tests__/teamRuns.test.ts index 7d3df40c..42ed388c 100644 --- a/mateclaw-ui/src/api/__tests__/teamRuns.test.ts +++ b/mateclaw-ui/src/api/__tests__/teamRuns.test.ts @@ -24,7 +24,7 @@ describe('teamRunApi', () => { assigneeAgentId: '2', }) - expect(get).toHaveBeenNthCalledWith(1, `/team-runs/${runId}`) + expect(get).toHaveBeenNthCalledWith(1, `/team-runs/${runId}`, { timeout: 15_000 }) expect(get).toHaveBeenNthCalledWith(2, `/teams/${teamId}/runs`) expect(get).toHaveBeenNthCalledWith( 3, @@ -48,11 +48,12 @@ describe('teamRunApi', () => { expect(get).toHaveBeenNthCalledWith(1, `/teams/${teamId}/runs/page`, { params: { activeOnly: true, cursor: 'team-cursor', limit: 17 }, + timeout: 15_000, }) expect(get).toHaveBeenNthCalledWith( 2, `/conversations/${encodeURIComponent(conversationId)}/team-runs/page`, - { params: { cursor: 'conversation-cursor', limit: 19 } }, + { params: { cursor: 'conversation-cursor', limit: 19 }, timeout: 15_000 }, ) }) diff --git a/mateclaw-ui/src/api/index.ts b/mateclaw-ui/src/api/index.ts index 58984ccf..b3c9a68d 100644 --- a/mateclaw-ui/src/api/index.ts +++ b/mateclaw-ui/src/api/index.ts @@ -951,6 +951,8 @@ export interface TeamTaskEvent { createTime?: string } +const TEAM_TASK_READ_TIMEOUT_MS = 15_000 + export const teamApi = { list: () => http.get('/teams'), get: (id: string) => http.get(`/teams/${id}`), @@ -968,6 +970,7 @@ export const teamApi = { removeMember: (id: string, agentId: string) => http.delete(`/teams/${id}/members/${agentId}`), listTasks: (id: string, status?: string[], opts?: { limit?: number; offset?: number; runId?: string }) => http.get(`/teams/${id}/tasks`, { + timeout: TEAM_TASK_READ_TIMEOUT_MS, params: { ...(status?.length ? { status: status.join(',') } : {}), ...(opts?.limit != null ? { limit: opts.limit } : {}), @@ -976,9 +979,12 @@ export const teamApi = { }, }), taskStats: (id: string, runId?: string) => http.get(`/teams/${id}/tasks/stats`, { + timeout: TEAM_TASK_READ_TIMEOUT_MS, params: runId ? { runId } : {}, }), - getTask: (id: string, taskId: string) => http.get(`/teams/${id}/tasks/${taskId}`), + getTask: (id: string, taskId: string) => http.get(`/teams/${id}/tasks/${taskId}`, { + timeout: TEAM_TASK_READ_TIMEOUT_MS, + }), createTask: ( id: string, data: { @@ -1081,8 +1087,10 @@ export interface TeamRun { tasks: TeamRunTask[] } +const TEAM_RUN_READ_TIMEOUT_MS = 15_000 + export const teamRunApi = { - get: (runId: string) => http.get(`/team-runs/${runId}`), + get: (runId: string) => http.get(`/team-runs/${runId}`, { timeout: TEAM_RUN_READ_TIMEOUT_MS }), listByTeamPage: ( teamId: string, options: { activeOnly?: boolean; cursor?: string; limit?: number } = {}, @@ -1092,7 +1100,7 @@ export const teamRunApi = { } if (options.activeOnly) params.activeOnly = true if (options.cursor) params.cursor = options.cursor - return http.get(`/teams/${teamId}/runs/page`, { params }) + return http.get(`/teams/${teamId}/runs/page`, { params, timeout: TEAM_RUN_READ_TIMEOUT_MS }) }, listByConversationPage: ( conversationId: string, @@ -1100,7 +1108,10 @@ export const teamRunApi = { ) => { const params: { cursor?: string; limit: number } = { limit: options.limit ?? 20 } if (options.cursor) params.cursor = options.cursor - return http.get(`/conversations/${encId(conversationId)}/team-runs/page`, { params }) + return http.get(`/conversations/${encId(conversationId)}/team-runs/page`, { + params, + timeout: TEAM_RUN_READ_TIMEOUT_MS, + }) }, listByTeam: (teamId: string, activeOnly = false, cursor?: string, limit = 30) => { const query = new URLSearchParams() diff --git a/mateclaw-ui/src/stores/__tests__/teamStoreGeneration.test.ts b/mateclaw-ui/src/stores/__tests__/teamStoreGeneration.test.ts index 225a1b9e..64ff1069 100644 --- a/mateclaw-ui/src/stores/__tests__/teamStoreGeneration.test.ts +++ b/mateclaw-ui/src/stores/__tests__/teamStoreGeneration.test.ts @@ -104,6 +104,35 @@ describe('useTeamStore request generation', () => { for (const call of api.listTasks.mock.calls) { expect(call[2]).toMatchObject({ runId: '9007199254740993' }) } + + const callsBeforeAggregateRefresh = api.listTasks.mock.calls.length + await store.setTaskRunId('A', null) + const aggregateCalls = api.listTasks.mock.calls.slice(callsBeforeAggregateRefresh) + expect(aggregateCalls).toHaveLength(3) + for (const call of aggregateCalls) { + expect(call[2]).toMatchObject({ runId: undefined }) + } expect(api.taskStats).toHaveBeenCalledWith('A', '9007199254740993') }) + + it('retries one transient board timeout and publishes the recovered snapshot', async () => { + api.get.mockResolvedValue(detail('A')) + const timeout = Object.assign(new Error('timeout of 15000ms exceeded'), { + code: 'ECONNABORTED', + }) + api.listTasks + .mockRejectedValueOnce(timeout) + .mockResolvedValueOnce({ data: [] }) + .mockResolvedValueOnce({ data: [] }) + .mockImplementation((_teamId: string, statuses: string[]) => + Promise.resolve({ data: statuses.includes('pending') ? [task('recovered')] : [] })) + api.taskStats.mockResolvedValue({ data: { pending: 1 } }) + const store = useTeamStore() + + await store.openTeam('A') + + expect(api.listTasks).toHaveBeenCalledTimes(6) + expect(api.taskStats).toHaveBeenCalledTimes(2) + expect(store.tasks.map(item => item.task.id)).toEqual(['recovered']) + }) }) diff --git a/mateclaw-ui/src/stores/useTeamStore.ts b/mateclaw-ui/src/stores/useTeamStore.ts index 2ade5303..01bf29ff 100644 --- a/mateclaw-ui/src/stores/useTeamStore.ts +++ b/mateclaw-ui/src/stores/useTeamStore.ts @@ -100,7 +100,8 @@ export const useTeamStore = defineStore('team', () => { * loaded size, so a poll/event refresh never collapses a column the user * has extended with load-more. */ - async function fetchTasks(teamId: string, expectedGeneration = teamGeneration) { + async function fetchTasks(teamId: string, expectedGeneration = teamGeneration, + retryOnTimeout = true) { const requestSequence = ++boardRequestSequence boardLoading.value = true try { @@ -120,6 +121,17 @@ export const useTeamStore = defineStore('team', () => { closedTasks.value = closed.data || [] taskStats.value = stats.data || {} } catch (e) { + const message = e instanceof Error ? e.message : String(e) + const timedOut = (e as { code?: string } | null)?.code === 'ECONNABORTED' + || /timeout/i.test(message) + if (retryOnTimeout + && timedOut + && expectedGeneration === teamGeneration + && requestSequence === boardRequestSequence + && String(currentTeam.value?.team.id ?? '') === teamId) { + await fetchTasks(teamId, expectedGeneration, false) + return + } if (expectedGeneration === teamGeneration && requestSequence === boardRequestSequence) { console.error('Failed to fetch team tasks', e) } @@ -163,7 +175,10 @@ export const useTeamStore = defineStore('team', () => { } async function setTaskRunId(teamId: string, runId: string | null) { - if (taskRunId.value === runId) return + if (taskRunId.value === runId) { + await fetchTasks(teamId) + return + } taskRunId.value = runId activeTasks.value = [] completedTasks.value = [] diff --git a/mateclaw-ui/src/views/Teams.vue b/mateclaw-ui/src/views/Teams.vue index 97b3e0fb..40033fb5 100644 --- a/mateclaw-ui/src/views/Teams.vue +++ b/mateclaw-ui/src/views/Teams.vue @@ -777,6 +777,7 @@ function onBoardEvent(e: { event: string; data: Record }) { function startEventSubscription(teamId: string) { stopEventSubscription() + if (document.hidden) return unsubscribeEvents = subscribeTeamEvents(teamId, onBoardEvent) } @@ -795,6 +796,7 @@ let pollTimer: ReturnType | null = null function startPolling() { stopPolling() + if (document.hidden) return pollTimer = setInterval(() => { const team = store.currentTeam if (team && store.hasActiveTasks) { @@ -803,6 +805,23 @@ function startPolling() { }, 3000) } +/** + * A team SSE stream occupies one HTTP/1.1 connection for the life of the + * page. Pause background tabs so several open team pages cannot consume all + * same-origin connection slots and starve ordinary board API requests. + */ +function handleVisibilityChange() { + const team = store.currentTeam + if (document.hidden || !team) { + stopPolling() + stopEventSubscription() + return + } + startPolling() + startEventSubscription(String(team.team.id)) + void store.fetchTasks(String(team.team.id)) +} + function stopPolling() { if (pollTimer) { clearInterval(pollTimer) @@ -895,6 +914,7 @@ watch( ) onMounted(() => { + document.addEventListener('visibilitychange', handleVisibilityChange) store.fetchTeams() if (agentStore.agents.length === 0) { agentStore.fetchAgents() @@ -902,6 +922,7 @@ onMounted(() => { }) onBeforeUnmount(() => { + document.removeEventListener('visibilitychange', handleVisibilityChange) stopPolling() stopEventSubscription() })