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

feat(team): add controlled worker intervention

Browse Source
This commit is contained in:
matevip 2026-09-04 04:18:41 -04:00
parent bd41a71826
commit 26150b68e8
47 changed files with 2244 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
mateclaw-server/src/main/java/vip/mate/agent/AgentService.java
View File

@ -28,7 +28,6 @@ import vip.mate.workspace.conversation.repository.ConversationMapper;
import java.util.List;
import java.util.Locale;
import java.time.Duration;
import java.util.Map;
import java.nio.file.Path;
import java.util.concurrent.ConcurrentHashMap;
@ -524,23 +523,7 @@ public class AgentService {
* {@code _usage_final} event for token and model attribution.
*/
private ChatResult collectChatResult(Flux<StreamDelta> stream) {
StringBuilder content = new StringBuilder();
final int[] usage = {0, 0};
final String[] modelInfo = {null, null};
stream.doOnNext(delta -> {
if (delta.isEvent() && "_usage_final".equals(delta.eventType())) {
Map<String, Object> data = delta.eventData();
usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
Object model = data.get("runtimeModelName");
Object provider = data.get("runtimeProviderId");
if (model != null) modelInfo[0] = model.toString();
if (provider != null) modelInfo[1] = provider.toString();
} else if (delta.content() != null) {
content.append(delta.content());
}
}).blockLast(Duration.ofMinutes(10));
return new ChatResult(content.toString(), usage[0], usage[1], modelInfo[0], modelInfo[1]);
return ChatResultCollector.collect(stream);
}
/**
@ -970,10 +953,15 @@ public class AgentService {
* post-approval replays).
*/
public record ChatResult(String content, int promptTokens, int completionTokens,
String runtimeModel, String runtimeProvider, String finishReason) {
public ChatResult(String content, int promptTokens, int completionTokens,
String runtimeModel, String runtimeProvider) {
this(content, promptTokens, completionTokens, runtimeModel, runtimeProvider, null);
}
public static ChatResult contentOnly(String content) {
return new ChatResult(content != null ? content : "", 0, 0, null, null);
return new ChatResult(content != null ? content : "", 0, 0, null, null, null);
}
}
}

39
mateclaw-server/src/main/java/vip/mate/agent/ChatResultCollector.java Normal file
View File

@ -0,0 +1,39 @@
package vip.mate.agent;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Map;
/** Collapses a structured agent stream without discarding terminal metadata. */
final class ChatResultCollector {
private ChatResultCollector() {
}
static AgentService.ChatResult collect(Flux<AgentService.StreamDelta> stream) {
StringBuilder content = new StringBuilder();
final int[] usage = {0, 0};
final String[] modelInfo = {null, null};
final String[] finishReason = {null};
stream.doOnNext(delta -> {
if (delta.isEvent() && "_usage_final".equals(delta.eventType())) {
Map<String, Object> data = delta.eventData() != null ? delta.eventData() : Map.of();
usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
Object model = data.get("runtimeModelName");
Object provider = data.get("runtimeProviderId");
if (model != null) modelInfo[0] = model.toString();
if (provider != null) modelInfo[1] = provider.toString();
} else if (delta.isEvent() && "finish_reason".equals(delta.eventType())) {
Map<String, Object> data = delta.eventData();
Object reason = data != null ? data.get("reason") : null;
if (reason != null) finishReason[0] = reason.toString();
} else if (delta.content() != null) {
content.append(delta.content());
}
}).blockLast(Duration.ofMinutes(10));
return new AgentService.ChatResult(content.toString(), usage[0], usage[1],
modelInfo[0], modelInfo[1], finishReason[0]);
}
}

10
mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
View File

@ -26,6 +26,7 @@ import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
@ -554,6 +555,7 @@ public class NodeStreamingChatHelper {
// ("credit balance is too low") use these phrases in 402-class responses.
// Chinese provider patterns (Zhipu 1113, DashScope, general) — same hard
// failure semantics: retrying the same provider won't refill the balance.
String lowerMsg = msg.toLowerCase(Locale.ROOT);
if (msg.contains("402") || msg.contains("insufficient_quota")
|| msg.contains("credit balance is too low")
|| msg.contains("billing_error") || msg.contains("billing_hard_limit_reached")
@ -562,7 +564,13 @@ public class NodeStreamingChatHelper {
|| msg.contains("余额不足") || msg.contains("请充值")
|| msg.contains("\"code\":\"1113\"") || msg.contains("\"code\":1113")
|| msg.contains("AccountBalanceNotEnough")
|| msg.contains("balance not enough")) {
|| msg.contains("balance not enough")
|| lowerMsg.contains("invalidsubscription")
|| lowerMsg.contains("subscription has expired")
|| lowerMsg.contains("arrearage")
|| lowerMsg.contains("account is in good standing")
|| lowerMsg.contains("insufficient_balance")
|| lowerMsg.contains("insufficient balance")) {
return ErrorType.BILLING;
}
// RFC-009 P3.2: MODEL_NOT_FOUND — provider rejects the requested model id.

26
mateclaw-server/src/main/java/vip/mate/agent/graph/node/ReasoningNode.java
View File

@ -1244,6 +1244,28 @@ public class ReasoningNode implements NodeAction {
.build();
}
// Compatibility safety net for providers/adapters that return the
// runtime's reserved error placeholder as an HTTP-successful content
// response. Without this guard the long-form completion gate treats
// the placeholder as a short draft and can repeat it until the graph's
// iteration cap. Cron and other synchronous callers consume the
// resulting structured ERROR_FALLBACK; they do not need to infer from
// user-facing text.
if (isRuntimeErrorPlaceholder(result.text())) {
String errorText = result.text();
log.error("[ReasoningNode] Runtime error placeholder returned as normal content; failing turn");
return reasonOutput()
.needsToolCall(false)
.shouldSummarize(false)
.finalAnswer(errorText)
.llmCallCount(nextLlmCallCount)
.finishReason(FinishReason.ERROR_FALLBACK)
.contentStreamed(true)
.thinkingStreamed(result.thinking() != null && !result.thinking().isEmpty())
.mergeUsage(state, result)
.build();
}
if (result.partial()) {
int partialChars = result.text() != null ? result.text().length() : 0;
log.warn("[ReasoningNode] Partial LLM result ({} chars), treating as final answer", partialChars);
@ -1429,6 +1451,10 @@ public class ReasoningNode implements NodeAction {
}
}
static boolean isRuntimeErrorPlaceholder(String text) {
return text != null && text.stripLeading().startsWith("[错误]");
}
private static String evidenceWarning(List<String> unsupportedReferences) {
return "\n\n[证据不足] 以下引用未出现在本轮已读取/搜索到的工具证据中，或缺少有效来源标注："
+ String.join(", ", unsupportedReferences)

73
mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java
View File

@ -412,6 +412,24 @@ public class ApprovalWorkflowService implements ApplicationRunner {
"consumed", /* removeFromMap */ true);
}
/** Claim one exact approval before a team worker executes its guarded tool. */
@Transactional
public ResolveOutcome claimForReplay(String pendingId, String userId) {
return performResolve(pendingId, userId, "APPROVED", MetadataDecision.APPROVED,
"approved", /* removeFromMap */ false);
}
/** Consume an approval previously claimed by {@link #claimForReplay}. */
@Transactional
public ResolveOutcome consumeReplayClaim(String pendingId, String userId) {
PendingApproval target = getReplayClaim(pendingId).orElse(null);
if (target == null) {
return ResolveOutcome.alreadyResolved(pendingId);
}
return performResolveOnSnapshot(target, userId, "APPROVED", "CONSUMED",
MetadataDecision.APPROVED, "consumed", /* removeFromMap */ true);
}
/**
* Consume the earliest already-{@code approved} record for the conversation +
* tool — used when an out-of-band approval (e.g. /approve text command flow that
@ -423,7 +441,7 @@ public class ApprovalWorkflowService implements ApplicationRunner {
if (target == null) {
return ResolveOutcome.alreadyResolved(null);
}
return performResolveOnSnapshot(target, null, "CONSUMED", MetadataDecision.APPROVED,
return performResolveOnSnapshot(target, null, "APPROVED", "CONSUMED", MetadataDecision.APPROVED,
"consumed", /* removeFromMap */ true);
}
@ -447,7 +465,7 @@ public class ApprovalWorkflowService implements ApplicationRunner {
List<ResolveOutcome> outcomes = new java.util.ArrayList<>(targets.size());
for (PendingApproval target : targets) {
try {
ResolveOutcome outcome = performResolveOnSnapshot(target, userId, "DENIED",
ResolveOutcome outcome = performResolveOnSnapshot(target, userId, "PENDING", "DENIED",
MetadataDecision.DENIED, "denied", /* removeFromMap */ true);
if (outcome.dbSynced()) outcomes.add(outcome);
} catch (Exception e) {
@ -475,7 +493,7 @@ public class ApprovalWorkflowService implements ApplicationRunner {
if (targets.isEmpty()) return List.of();
List<ResolveOutcome> outcomes = new java.util.ArrayList<>(targets.size());
for (PendingApproval target : targets) {
ResolveOutcome outcome = performResolveOnSnapshot(target, null, "SUPERSEDED",
ResolveOutcome outcome = performResolveOnSnapshot(target, null, "PENDING", "SUPERSEDED",
MetadataDecision.DENIED, "superseded", /* removeFromMap */ true);
if (outcome.dbSynced()) outcomes.add(outcome);
}
@ -644,12 +662,13 @@ public class ApprovalWorkflowService implements ApplicationRunner {
pendingId, snapshot != null, snapshot != null ? snapshot.getStatus() : "n/a");
return ResolveOutcome.alreadyResolved(pendingId);
}
return performResolveOnSnapshot(snapshot, userId, dbStatus, metaDecision,
return performResolveOnSnapshot(snapshot, userId, "PENDING", dbStatus, metaDecision,
snapshotStatus, removeFromMap);
}
private ResolveOutcome performResolveOnSnapshot(PendingApproval snapshot, String userId,
String dbStatus, MetadataDecision metaDecision,
String expectedDbStatus, String dbStatus,
MetadataDecision metaDecision,
String snapshotStatus, boolean removeFromMap) {
// Phase 1 — DB UPDATE (conditional). The eq("PENDING") guard makes the call
// idempotent: if another path already won, we get rows=0 and bail without
@ -658,7 +677,7 @@ public class ApprovalWorkflowService implements ApplicationRunner {
try {
LambdaUpdateWrapper<ToolApprovalEntity> wrapper = new LambdaUpdateWrapper<ToolApprovalEntity>()
.eq(ToolApprovalEntity::getPendingId, snapshot.getPendingId())
.eq(ToolApprovalEntity::getStatus, "PENDING")
.eq(ToolApprovalEntity::getStatus, expectedDbStatus)
.set(ToolApprovalEntity::getStatus, dbStatus)
.set(ToolApprovalEntity::getResolvedAt, LocalDateTime.now());
if (userId != null) {
@ -672,8 +691,8 @@ public class ApprovalWorkflowService implements ApplicationRunner {
throw e;
}
if (rows == 0) {
log.info("[ApprovalWorkflow] resolve no-op for {}: DB row not in PENDING (concurrent resolve)",
snapshot.getPendingId());
log.info("[ApprovalWorkflow] resolve no-op for {}: DB row not in {} (concurrent resolve)",
snapshot.getPendingId(), expectedDbStatus);
return ResolveOutcome.alreadyResolved(snapshot.getPendingId());
}
@ -809,6 +828,44 @@ public class ApprovalWorkflowService implements ApplicationRunner {
return approvalService.getPending(pendingId);
}
/**
* Recover an exact APPROVED replay claim from memory or DB. APPROVED claims are
* intentionally durable so a worker replay can be finalized after a restart
* without reopening the approval to denial.
*/
public java.util.Optional<PendingApproval> getReplayClaim(String pendingId) {
PendingApproval inMemory = approvalService.getPending(pendingId)
.filter(pending -> "approved".equals(pending.getStatus()))
.orElse(null);
if (inMemory != null) {
return java.util.Optional.of(inMemory);
}
ToolApprovalEntity entity = approvalMapper.selectOne(
new LambdaQueryWrapper<ToolApprovalEntity>()
.eq(ToolApprovalEntity::getPendingId, pendingId)
.eq(ToolApprovalEntity::getStatus, "APPROVED"));
if (entity == null) {
return java.util.Optional.empty();
}
Instant createdAt = entity.getCreatedAt() == null
? Instant.now()
: entity.getCreatedAt().atZone(ZoneId.systemDefault()).toInstant();
PendingApproval snapshot = new PendingApproval(entity.getPendingId(),
entity.getConversationId(), entity.getUserId(), entity.getToolName(),
entity.getToolArguments(), entity.getSummary(), createdAt, "approved");
snapshot.setToolCallPayload(entity.getToolCallPayload());
snapshot.setSiblingToolCalls(entity.getSiblingToolCalls());
snapshot.setAgentId(entity.getAgentId());
snapshot.setChannelType(entity.getChannelType());
snapshot.setRequesterName(entity.getRequesterName());
snapshot.setReplyTarget(entity.getReplyTarget());
snapshot.setFindingsJson(entity.getFindingsJson());
snapshot.setMaxSeverity(entity.getMaxSeverity());
snapshot.setSummary(entity.getSummary());
snapshot.setChatOrigin(entity.getChatOrigin());
return java.util.Optional.of(snapshot);
}
public PendingApproval findPendingByConversation(String conversationId) {
return approvalService.findPendingByConversation(conversationId);
}

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

@ -1,5 +1,6 @@
package vip.mate.channel.web;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
@ -88,6 +89,18 @@ public class ChatStreamTracker {
@Value("${mateclaw.stream.iteration-events:true}")
private boolean iterationEventsEnabled = true;
/**
* Coalesce the tiny token fragments produced by streaming model clients
* before assigning an SSE id and touching the replay buffer. This keeps
* rendering responsive while avoiding thousands of emitter writes for a
* single long answer.
*/
@Value("${mateclaw.stream.content-batch-ms:25}")
private long contentBatchMs = 25L;
@Value("${mateclaw.stream.content-batch-chars:256}")
private int contentBatchChars = 256;
/**
* Heartbeat cadence (seconds) before the first model token arrives. Short
* because pre-token gaps strand the UI on a blank "正在生成中" placeholder
@ -127,6 +140,11 @@ public class ChatStreamTracker {
this.iterationEventsEnabled = enabled;
}
void setContentBatchingForTesting(long flushMs, int maxChars) {
this.contentBatchMs = Math.max(1L, flushMs);
this.contentBatchChars = Math.max(1, maxChars);
}
public boolean isIterationEventsEnabled() {
return iterationEventsEnabled;
}
@ -219,6 +237,11 @@ public class ChatStreamTracker {
/** 已广播的 pending approval ID 集合（用于幂等去重） */
final java.util.Set<String> broadcastedApprovalIds = java.util.concurrent.ConcurrentHashMap.newKeySet();
/** Pending visible answer text waiting for the SSE coalescing window. Guarded by lock. */
String pendingContentField;
final StringBuilder pendingContent = new StringBuilder();
ScheduledFuture<?> pendingContentFlush;
/** 创建时间（用于 stale 检测和清理） */
final long createdAt = System.currentTimeMillis();
@ -748,6 +771,99 @@ public class ChatStreamTracker {
return true;
}
private record ContentDelta(String field, String text) {}
/**
* Buffer only the two established visible-content wire shapes:
* {@code {"delta":"..."}} (workspace chat) and
* {@code {"text":"..."}} (embedded webchat). Payloads with extra
* metadata stay on the ordinary path so batching never discards fields.
*/
private boolean tryBufferContentDelta(RunState state, String eventName,
String jsonData, boolean skipBuffer) {
if (!"content_delta".equals(eventName) || skipBuffer || state == null) {
return false;
}
ContentDelta delta = parseContentDelta(jsonData);
if (delta == null) {
return false;
}
boolean flushNow = false;
synchronized (state.lock) {
if (!isCurrent(state) || state.done) {
return true;
}
// A conversation uses one wire field for a run. If a caller does
// switch shapes, flush the old batch and deliver the new payload
// unchanged rather than mixing contracts.
if (state.pendingContentField != null
&& !state.pendingContentField.equals(delta.field())) {
return false;
}
state.lastEventAt = System.currentTimeMillis();
state.pendingContentField = delta.field();
state.pendingContent.append(delta.text());
if (state.pendingContent.length() >= Math.max(1, contentBatchChars)) {
flushNow = true;
} else if (state.pendingContentFlush == null
|| state.pendingContentFlush.isDone()) {
state.pendingContentFlush = heartbeatScheduler.schedule(
() -> flushPendingContent(state),
Math.max(1L, contentBatchMs), TimeUnit.MILLISECONDS);
}
}
if (flushNow) {
flushPendingContent(state);
}
return true;
}
private ContentDelta parseContentDelta(String jsonData) {
if (jsonData == null || jsonData.isEmpty()) return null;
try {
JsonNode node = objectMapper.readTree(jsonData);
if (node == null || !node.isObject() || node.size() != 1) return null;
String field = node.has("delta") ? "delta" : node.has("text") ? "text" : null;
if (field == null || !node.path(field).isTextual()) return null;
String text = node.path(field).textValue();
return text == null || text.isEmpty() ? null : new ContentDelta(field, text);
} catch (Exception ignored) {
return null;
}
}
/** Snapshot under the run lock, then emit through the fenced raw path. */
private void flushPendingContent(RunState state) {
String field;
String text;
synchronized (state.lock) {
if (state.pendingContent.length() == 0) {
if (state.pendingContentFlush != null) {
state.pendingContentFlush.cancel(false);
state.pendingContentFlush = null;
}
state.pendingContentField = null;
return;
}
field = state.pendingContentField;
text = state.pendingContent.toString();
state.pendingContent.setLength(0);
state.pendingContentField = null;
if (state.pendingContentFlush != null) {
state.pendingContentFlush.cancel(false);
state.pendingContentFlush = null;
}
}
try {
String json = objectMapper.writeValueAsString(Map.of(field, text));
broadcastNow(new RunHandle(state), "content_delta", json, false);
} catch (Exception e) {
log.warn("Failed to flush content batch for {}: {}",
state.conversationId, e.getMessage());
}
}
/**
* 广播事件到所有订阅者并缓存到 buffer.
* <p>
@ -777,6 +893,17 @@ public class ChatStreamTracker {
public void broadcast(RunHandle handle, String eventName, String jsonData, boolean skipBuffer) {
if (handle == null) return;
RunState state = handle.state;
if (tryBufferContentDelta(state, eventName, jsonData, skipBuffer)) {
return;
}
if (!"heartbeat".equals(eventName)) {
flushPendingContent(state);
}
broadcastNow(handle, eventName, jsonData, skipBuffer);
}
private void broadcastNow(RunHandle handle, String eventName, String jsonData, boolean skipBuffer) {
RunState state = handle.state;
boolean isDone = "done".equals(eventName);
boolean isPostTurnEvent = "goal_continuation".equals(eventName)
@ -854,6 +981,18 @@ public class ChatStreamTracker {
*/
public void broadcast(String conversationId, String eventName, String jsonData, boolean skipBuffer) {
RunState state = runs.get(conversationId);
if (state == null) return;
if (tryBufferContentDelta(state, eventName, jsonData, skipBuffer)) {
return;
}
if (!"heartbeat".equals(eventName)) {
flushPendingContent(state);
}
broadcastNow(conversationId, eventName, jsonData, skipBuffer);
}
private void broadcastNow(String conversationId, String eventName, String jsonData, boolean skipBuffer) {
RunState state = runs.get(conversationId);
boolean isDone = "done".equals(eventName);
boolean isPostTurnEvent = "goal_continuation".equals(eventName)
@ -1312,6 +1451,10 @@ public class ChatStreamTracker {
private boolean complete(RunState state) {
String conversationId = state.conversationId;
// Some terminal paths do not publish a done envelope. Flush visible
// text while the run is still live so the scheduled batch cannot be
// rejected after state.done flips below.
flushPendingContent(state);
ScheduledFuture<?> oldHeartbeat;
synchronized (state.lock) {
if (!isCurrent(state)) {
@ -1354,6 +1497,7 @@ public class ChatStreamTracker {
if (state == null) {
return new CompletionResult(true);
}
flushPendingContent(state);
ScheduledFuture<?> oldHeartbeat;
synchronized (state.lock) {
if (!isCurrent(state)) {

35
mateclaw-server/src/main/java/vip/mate/cron/service/CronJobLifecycleService.java
View File

@ -138,6 +138,41 @@ public class CronJobLifecycleService {
.set(CronJobRunEntity::getErrorMessage, StrUtil.maxLength(message, 1000)));
}
/**
* T-fail — terminal graph failure that arrived as structured stream metadata
* rather than a thrown exception. Persist the diagnostic assistant message
* for conversation coherence, but never publish success, memory, or delivery
* events for an {@code error_fallback} result.
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void finishRunFailed(CronJobRunEntity run, AssistantMessage result,
String conversationId, AgentService.ChatResult chatResult) {
String convId = conversationId != null ? conversationId : run.getConversationId();
String text = result != null && result.getText() != null ? result.getText() : "";
int totalTokens = chatResult != null
? chatResult.promptTokens() + chatResult.completionTokens() : 0;
int updated = runMapper.update(null, new LambdaUpdateWrapper<CronJobRunEntity>()
.eq(CronJobRunEntity::getId, run.getId())
.eq(CronJobRunEntity::getStatus, "running")
.set(CronJobRunEntity::getStatus, "failed")
.set(CronJobRunEntity::getFinishedAt, LocalDateTime.now())
.set(CronJobRunEntity::getErrorMessage, StrUtil.maxLength(text, 1000))
.set(totalTokens > 0, CronJobRunEntity::getTokenUsage, totalTokens));
if (updated == 0) {
log.warn("[CronLifecycle] Run {} lost its running fence before graph failure; dropping late result",
run.getId());
return;
}
if (chatResult != null) {
conversationService.saveMessage(convId, "assistant", text, null, "error",
chatResult.promptTokens(), chatResult.completionTokens(),
chatResult.runtimeModel(), chatResult.runtimeProvider());
} else {
conversationService.saveMessage(convId, "assistant", text, null, "error");
}
}
/**
* Insert a {@code running} run row for a task type that does not produce
* a conversation (e.g. {@code wiki_process}). No header / user message is

5
mateclaw-server/src/main/java/vip/mate/cron/service/CronJobRunner.java
View File

@ -8,6 +8,7 @@ import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.stereotype.Component;
import vip.mate.agent.AgentService;
import vip.mate.agent.context.ChatOrigin;
import vip.mate.agent.graph.state.FinishReason;
import vip.mate.cron.CronChatOriginFactory;
import vip.mate.cron.model.CronJobEntity;
import vip.mate.dashboard.model.CronJobRunEntity;
@ -168,6 +169,10 @@ public class CronJobRunner {
// T2 — short tx
try {
if (FinishReason.ERROR_FALLBACK.getValue().equals(chatResult.finishReason())) {
lifecycle.finishRunFailed(run, result, conversationId, chatResult);
return;
}
lifecycle.finishRunAndPublish(job, run, userMessage, result, conversationId, silent, chatResult);
} catch (Exception e) {
log.error("[CronRunner] T2 finishRunAndPublish failed for job {}: {}", job.getId(), e.getMessage(), e);

8
mateclaw-server/src/main/java/vip/mate/llm/service/ModelDiscoveryService.java
View File

@ -695,7 +695,13 @@ public class ModelDiscoveryService {
requestBody.put("model", modelId);
requestBody.put("messages", List.of(Map.of("role", "user", "content", "请回复：连接正常")));
requestBody.put("max_tokens", 10);
requestBody.put("temperature", 0);
Object probeTemperature;
if (ModelFamily.detect(modelId).fixedTemperatureOne()) {
probeTemperature = 1.0d;
} else {
probeTemperature = 0;
}
requestBody.put("temperature", probeTemperature);
return requestBody;
}

82
mateclaw-server/src/main/java/vip/mate/team/controller/TeamController.java
View File

@ -24,6 +24,7 @@ import vip.mate.team.service.TeamEventChannel;
import vip.mate.team.service.TeamManualTaskService;
import vip.mate.team.service.TeamService;
import vip.mate.team.service.TeamTaskService;
import vip.mate.team.service.TeamWorkerInterventionService;
import vip.mate.workspace.core.annotation.RequireWorkspaceRole;
import java.security.Principal;
@ -56,6 +57,7 @@ public class TeamController {
private final TeamDispatchService dispatchService;
private final TeamAnnounceService announceService;
private final TeamEventChannel eventChannel;
private final TeamWorkerInterventionService workerInterventionService;
private final AgentMapper agentMapper;
// ==================== team CRUD ====================
@ -213,6 +215,60 @@ public class TeamController {
});
}
@Operation(summary = "批准 worker 工具调用并在原会话恢复执行")
@PostMapping("/{id}/tasks/{taskId}/worker/approve")
@RequireWorkspaceRole("admin")
public R<TaskVO> approveWorkerTool(@PathVariable Long id, @PathVariable Long taskId,
@RequestBody WorkerApprovalRequest req,
Principal principal) {
return workerGuarded(() -> {
requireTeam(id);
requireTask(id, taskId);
if (req == null || req.getPendingId() == null || req.getPendingId().isBlank()) {
throw new IllegalArgumentException("pending approval id is required");
}
TeamTaskEntity task = workerInterventionService.approve(id, taskId,
req.getPendingId().strip(), principalName(principal));
return R.ok(toTaskVO(task));
});
}
@Operation(summary = "拒绝 worker 工具调用")
@PostMapping("/{id}/tasks/{taskId}/worker/deny")
@RequireWorkspaceRole("admin")
public R<TaskVO> denyWorkerTool(@PathVariable Long id, @PathVariable Long taskId,
@RequestBody WorkerApprovalRequest req,
Principal principal) {
return workerGuarded(() -> {
requireTeam(id);
requireTask(id, taskId);
if (req == null || req.getPendingId() == null || req.getPendingId().isBlank()) {
throw new IllegalArgumentException("pending approval id is required");
}
TeamTaskEntity task = workerInterventionService.deny(id, taskId,
req.getPendingId().strip(), principalName(principal));
return R.ok(toTaskVO(task));
});
}
@Operation(summary = "向 worker 原会话发送任务级补充指令")
@PostMapping("/{id}/tasks/{taskId}/worker/feedback")
@RequireWorkspaceRole("admin")
public R<TaskVO> feedbackWorker(@PathVariable Long id, @PathVariable Long taskId,
@RequestBody WorkerFeedbackRequest req,
Principal principal) {
return workerGuarded(() -> {
requireTeam(id);
requireTask(id, taskId);
if (req == null || req.getMessage() == null || req.getMessage().isBlank()) {
throw new IllegalArgumentException("feedback is required");
}
TeamTaskEntity task = workerInterventionService.feedback(id, taskId,
req.getMessage(), principalName(principal));
return R.ok(toTaskVO(task));
});
}
@Operation(summary = "驳回 in_review 任务")
@PostMapping("/{id}/tasks/{taskId}/reject")
@RequireWorkspaceRole("admin")
@ -310,6 +366,10 @@ public class TeamController {
eventChannel.publishTaskEvent(taskService.getTask(taskId), event, Map.of());
}
private String principalName(Principal principal) {
return principal != null && principal.getName() != null ? principal.getName() : "admin";
}
@Operation(summary = "添加评论")
@PostMapping("/{id}/tasks/{taskId}/comments")
@RequireWorkspaceRole("admin")
@ -352,6 +412,18 @@ public class TeamController {
}
}
/** Intervention endpoints expose recoverable client states instead of generic 500s. */
private <T> R<T> workerGuarded(Supplier<R<T>> action) {
try {
return action.get();
} catch (IllegalArgumentException error) {
int code = error.getMessage() != null && error.getMessage().contains("not found") ? 404 : 400;
return R.fail(code, error.getMessage());
} catch (IllegalStateException error) {
return R.fail(409, error.getMessage());
}
}
private TeamTaskEntity requireTask(Long teamId, Long taskId) {
TeamTaskEntity task = taskService.getTask(taskId);
if (task == null || !task.getTeamId().equals(teamId)) {
@ -495,4 +567,14 @@ public class TeamController {
public static class CommentRequest {
private String content;
}
@Data
public static class WorkerApprovalRequest {
private String pendingId;
}
@Data
public static class WorkerFeedbackRequest {
private String message;
}
}

1
mateclaw-server/src/main/java/vip/mate/team/model/TeamTaskEventEntity.java
View File

@ -27,6 +27,7 @@ public class TeamTaskEventEntity {
public static final String DELIVERABLE = "deliverable";
public static final String COMPLETED = "completed";
public static final String IN_REVIEW = "in_review";
public static final String AWAITING_APPROVAL = "awaiting_approval";
public static final String FAILED = "failed";
public static final String CANCELLED = "cancelled";
public static final String APPROVED = "approved";

2
mateclaw-server/src/main/java/vip/mate/team/model/TeamTaskStatus.java
View File

@ -9,6 +9,7 @@ import java.util.Set;
* pending ──claim/assign──▶ in_progress ──complete──▶ completed
* │ │ (require_approval) ▶ in_review ──approve──▶ completed
* │ │ └──reject───▶ cancelled
* │ ├──guarded tool──▶ awaiting_approval ──approve──▶ in_progress
* │ ├──blocker/error──▶ failed ──retry──▶ pending
* │ └──lease expired──▶ stale ──retry──▶ pending
* ├──blocked_by set──▶ blocked ──all blockers released──▶ pending
@ -21,6 +22,7 @@ public final class TeamTaskStatus {
public static final String PENDING = "pending";
public static final String IN_PROGRESS = "in_progress";
public static final String AWAITING_APPROVAL = "awaiting_approval";
public static final String IN_REVIEW = "in_review";
public static final String COMPLETED = "completed";
public static final String FAILED = "failed";

28
mateclaw-server/src/main/java/vip/mate/team/service/TeamDispatchService.java
View File

@ -10,6 +10,8 @@ import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
import vip.mate.team.event.TeamTasksDelegatedEvent;
import vip.mate.agent.AgentService;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.approval.PendingApproval;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.team.model.AgentTeamEntity;
import vip.mate.team.model.TeamTaskEntity;
@ -107,6 +109,7 @@ public class TeamDispatchService {
private final ChatStreamTracker streamTracker;
private final TeamAnnounceService announceService;
private final TeamEventChannel eventChannel;
private final ApprovalWorkflowService approvalService;
/** Members with a run currently in flight in this JVM (belt-and-braces on top of hasActiveTask). */
private final Set<Long> runningMembers = ConcurrentHashMap.newKeySet();
@ -215,9 +218,7 @@ public class TeamDispatchService {
streamTracker.incrementFlux(childConvId);
// Renew the execution lease while the member works; the conditional
// UPDATE inside renewLock makes this a no-op once the task settles.
heartbeat = HEARTBEAT_SCHEDULER.scheduleAtFixedRate(
() -> taskService.renewLock(task.getId()),
HEARTBEAT_MINUTES, HEARTBEAT_MINUTES, TimeUnit.MINUTES);
heartbeat = startLeaseHeartbeat(task.getId());
broadcast(task, "team_task_dispatched", Map.of());
log.info("Team {} task #{} dispatched to agent {} (conv {})",
teamId, task.getTaskNumber(), memberId, childConvId);
@ -235,6 +236,20 @@ public class TeamDispatchService {
conversationService.saveMessage(childConvId, "assistant", reply);
}
PendingApproval pending = approvalService.findPendingByConversation(childConvId);
if (pending != null) {
String summary = pending.getSummary() == null || pending.getSummary().isBlank()
? pending.getReason() : pending.getSummary();
if (taskService.parkForToolApproval(task.getId(), pending.getPendingId(), summary)) {
TeamTaskEntity parked = taskService.getTask(task.getId());
broadcast(parked != null ? parked : task, "team_task_awaiting_approval",
Map.of("pendingId", pending.getPendingId(),
"toolName", pending.getToolName() == null ? "" : pending.getToolName(),
"summary", summary == null ? "Tool approval required" : summary));
}
return;
}
settleOutcome(task, reply);
} catch (Exception e) {
log.warn("Team {} task #{} member run ended exceptionally: {}", teamId,
@ -259,6 +274,13 @@ public class TeamDispatchService {
}
}
/** Share the same DB-backed lease heartbeat with controlled worker replays. */
ScheduledFuture<?> startLeaseHeartbeat(Long taskId) {
return HEARTBEAT_SCHEDULER.scheduleAtFixedRate(
() -> taskService.renewLock(taskId),
HEARTBEAT_MINUTES, HEARTBEAT_MINUTES, TimeUnit.MINUTES);
}
/**
* Ask the member conversation executing this task to stop at the next graph
* node boundary (cancel path). No-op when the task never dispatched or the

2
mateclaw-server/src/main/java/vip/mate/team/service/TeamRunStateMachine.java
View File

@ -16,12 +16,14 @@ public final class TeamRunStateMachine {
TeamTaskStatus.PENDING,
TeamTaskStatus.BLOCKED,
TeamTaskStatus.IN_PROGRESS,
TeamTaskStatus.AWAITING_APPROVAL,
TeamTaskStatus.STALE
);
private static final Set<String> KNOWN_TASK_STATUSES = Set.of(
TeamTaskStatus.PENDING,
TeamTaskStatus.BLOCKED,
TeamTaskStatus.IN_PROGRESS,
TeamTaskStatus.AWAITING_APPROVAL,
TeamTaskStatus.IN_REVIEW,
TeamTaskStatus.COMPLETED,
TeamTaskStatus.FAILED,

14
mateclaw-server/src/main/java/vip/mate/team/service/TeamRunViewFactory.java
View File

@ -177,6 +177,8 @@ final class TeamRunViewFactory {
List<TeamRunView.AttentionItem> items = new ArrayList<>();
for (TeamTaskEntity task : tasks) {
String type = switch (task.getStatus()) {
case TeamTaskStatus.AWAITING_APPROVAL -> replayOutcomeUncertain(task)
? "replay_uncertain" : "approval";
case TeamTaskStatus.IN_REVIEW -> "review";
case TeamTaskStatus.FAILED -> "failure";
case TeamTaskStatus.BLOCKED -> "blocked";
@ -185,7 +187,8 @@ final class TeamRunViewFactory {
};
if (type != null) {
String message = text(task.getReason());
int priority = TeamTaskStatus.IN_REVIEW.equals(task.getStatus()) ? 0 : 20;
int priority = TeamTaskStatus.IN_REVIEW.equals(task.getStatus())
|| TeamTaskStatus.AWAITING_APPROVAL.equals(task.getStatus()) ? 0 : 20;
items.add(new TeamRunView.AttentionItem("task:" + task.getId() + ":" + type,
type, priority == 0 ? "action" : "error", priority,
task.getId(), message == null ? task.getSubject() : message, task.getUpdateTime()));
@ -205,6 +208,15 @@ final class TeamRunViewFactory {
return List.copyOf(items);
}
private static boolean replayOutcomeUncertain(TeamTaskEntity task) {
try {
JSONObject approval = JSONUtil.parseObj(task.getMetadata()).getJSONObject("toolApproval");
return approval != null && approval.getBool("replayOutcomeUncertain", false);
} catch (RuntimeException invalidMetadata) {
return false;
}
}
private static TeamRunView.Liveness liveness(String status, LocalDateTime lastActivity,
List<TeamTaskEntity> tasks) {
if (TeamRunStatus.isTerminal(status)) {

303
mateclaw-server/src/main/java/vip/mate/team/service/TeamTaskService.java
View File

@ -45,6 +45,8 @@ import java.util.regex.Pattern;
@RequiredArgsConstructor
public class TeamTaskService {
private static final int MAX_STAGED_REPLAY_RESULT_CHARS = 8000;
private static final Pattern CHECKPOINT_RANGE = Pattern.compile(
"(?i)R(\\d{3})\\s*[-–—]\\s*R(\\d{3})");
@ -375,6 +377,275 @@ public class TeamTaskService {
return updated;
}
/** Park a running worker task until its guarded tool call receives a human decision. */
public boolean parkForToolApproval(Long taskId, String pendingId, String summary) {
if (pendingId == null || pendingId.isBlank()) {
throw new IllegalArgumentException("pending approval id is required");
}
TeamTaskEntity task = taskMapper.selectById(taskId);
if (task == null) {
return false;
}
JSONObject metadata;
try {
metadata = task.getMetadata() == null || task.getMetadata().isBlank()
? new JSONObject() : JSONUtil.parseObj(task.getMetadata());
} catch (RuntimeException invalid) {
metadata = new JSONObject();
}
String detail = summary == null || summary.isBlank()
? "Tool approval required" : summary.strip();
metadata.set("toolApproval", new JSONObject()
.set("pendingId", pendingId)
.set("summary", detail));
boolean parked = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.in(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS,
TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getReason, detail)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null)) == 1;
if (parked) {
recordEvent(task.getTeamId(), taskId, TeamTaskEventEntity.AWAITING_APPROVAL,
AUTHOR_SYSTEM, null, pendingId + " — " + detail);
projectTask(taskId);
}
return parked;
}
/** Resume the exact guarded tool request currently recorded on a parked task. */
public boolean resumeAfterToolApproval(Long taskId, String pendingId) {
if (pendingId == null || pendingId.isBlank()) {
throw new IllegalArgumentException("pending approval id is required");
}
TeamTaskEntity task = requireTask(taskId);
if (!TeamTaskStatus.AWAITING_APPROVAL.equals(task.getStatus())) {
throw new IllegalStateException("task #" + task.getTaskNumber()
+ " is not awaiting tool approval");
}
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
String currentPendingId = approval == null ? null : approval.getStr("pendingId");
if (!pendingId.equals(currentPendingId)) {
throw new IllegalStateException("tool approval is no longer current for task #"
+ task.getTaskNumber());
}
approval.set("replayInProgress", true);
metadata.set("toolApproval", approval);
boolean resumed = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getOwnerAgentId, task.getAssigneeAgentId())
.set(TeamTaskEntity::getReason, null)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, newLease())) == 1;
if (resumed) {
projectTask(taskId);
}
return resumed;
}
/** Settle a parked guarded tool request as denied without executing it. */
public boolean denyToolApproval(Long taskId, String pendingId, String requester) {
TeamTaskEntity task = requireTask(taskId);
if (!TeamTaskStatus.AWAITING_APPROVAL.equals(task.getStatus())) {
throw new IllegalStateException("task #" + task.getTaskNumber()
+ " is not awaiting tool approval");
}
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
if (approval == null || !Objects.equals(pendingId, approval.getStr("pendingId"))) {
throw new IllegalStateException("tool approval is no longer current for task #"
+ task.getTaskNumber());
}
metadata.remove("toolApproval");
String reason = "Tool request denied by "
+ (requester == null || requester.isBlank() ? "user" : requester);
boolean denied = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.FAILED)
.set(TeamTaskEntity::getReason, reason)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null)) == 1;
if (denied) {
recordEvent(task.getTeamId(), taskId, TeamTaskEventEntity.FAILED,
AUTHOR_USER, requester, reason);
projectTask(taskId);
}
return denied;
}
/** Durably stage a successful replay before consuming its approval. */
public boolean stageToolReplayResult(Long taskId, String pendingId, String reply) {
TeamTaskEntity task = requireTask(taskId);
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject currentApproval = metadata.getJSONObject("toolApproval");
if (!TeamTaskStatus.IN_PROGRESS.equals(task.getStatus())
|| currentApproval == null
|| !Objects.equals(pendingId, currentApproval.getStr("pendingId"))) {
throw new IllegalStateException("tool approval is no longer current for task #"
+ task.getTaskNumber());
}
JSONObject approval = new JSONObject()
.set("pendingId", pendingId)
.set("summary", "Approved tool completed; finalizing result")
.set("replayResult", truncate(reply == null ? "" : reply,
MAX_STAGED_REPLAY_RESULT_CHARS));
metadata.set("toolApproval", approval);
boolean staged = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getReason, "Approved tool completed; finalizing result")
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null)) == 1;
if (staged) {
projectTask(taskId);
}
return staged;
}
/** Park a failed replay without allowing an automatic second execution. */
public boolean parkToolReplayUncertain(Long taskId, String pendingId, String detail) {
TeamTaskEntity task = requireTask(taskId);
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
if (approval == null || !Objects.equals(pendingId, approval.getStr("pendingId"))) {
return false;
}
approval.set("replayInProgress", false);
approval.set("replayOutcomeUncertain", true);
approval.set("summary", detail);
metadata.set("toolApproval", approval);
boolean parked = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getReason, detail)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null)) == 1;
if (parked) {
recordEvent(task.getTeamId(), taskId, TeamTaskEventEntity.AWAITING_APPROVAL,
AUTHOR_SYSTEM, null, detail);
projectTask(taskId);
}
return parked;
}
public String stagedToolReplayResult(TeamTaskEntity task) {
JSONObject approval = task == null ? null : parseMetadata(task.getMetadata())
.getJSONObject("toolApproval");
return approval != null && approval.containsKey("replayResult")
? approval.getStr("replayResult", "") : null;
}
public boolean isToolReplayMessagePersisted(TeamTaskEntity task) {
JSONObject approval = task == null ? null : parseMetadata(task.getMetadata())
.getJSONObject("toolApproval");
return approval != null && approval.getBool("messagePersisted", false);
}
public boolean isToolReplayOutcomeUncertain(TeamTaskEntity task) {
JSONObject approval = task == null ? null : parseMetadata(task.getMetadata())
.getJSONObject("toolApproval");
return approval != null && approval.getBool("replayOutcomeUncertain", false);
}
public boolean markToolReplayMessagePersisted(Long taskId, String pendingId) {
TeamTaskEntity task = requireTask(taskId);
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
if (approval == null || !Objects.equals(pendingId, approval.getStr("pendingId"))) {
return false;
}
approval.set("messagePersisted", true);
metadata.set("toolApproval", approval);
return taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getMetadata, metadata.toString())) == 1;
}
/** Stop an already-claimed replay after a failed execution attempt. */
public boolean abortClaimedToolReplay(Long taskId, String pendingId, String requester) {
TeamTaskEntity task = requireTask(taskId);
if (!TeamTaskStatus.AWAITING_APPROVAL.equals(task.getStatus())) {
throw new IllegalStateException("task #" + task.getTaskNumber()
+ " is not awaiting replay recovery");
}
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
if (approval == null || !Objects.equals(pendingId, approval.getStr("pendingId"))
|| approval.containsKey("replayResult")) {
throw new IllegalStateException("tool replay is no longer abortable for task #"
+ task.getTaskNumber());
}
metadata.remove("toolApproval");
String actor = requester == null || requester.isBlank() ? "user" : requester;
String reason = "Approved tool replay aborted by " + actor
+ "; the previous execution outcome may be uncertain";
boolean aborted = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.FAILED)
.set(TeamTaskEntity::getReason, reason)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null)) == 1;
if (aborted) {
recordEvent(task.getTeamId(), taskId, TeamTaskEventEntity.FAILED,
AUTHOR_USER, requester, reason);
projectTask(taskId);
}
return aborted;
}
private static String truncate(String value, int maxChars) {
return value.length() <= maxChars ? value : value.substring(0, maxChars);
}
/** Reopen a settled worker task for one deliberate, task-scoped follow-up turn. */
public boolean resumeForWorkerFeedback(Long taskId) {
TeamTaskEntity task = requireTask(taskId);
if (TeamTaskStatus.IN_PROGRESS.equals(task.getStatus())) {
throw new IllegalStateException("worker task is already running");
}
if (TeamTaskStatus.AWAITING_APPROVAL.equals(task.getStatus())) {
throw new IllegalStateException("resolve the pending tool approval before sending feedback");
}
if (TeamTaskStatus.CANCELLED.equals(task.getStatus())
|| TeamTaskStatus.PENDING.equals(task.getStatus())
|| TeamTaskStatus.BLOCKED.equals(task.getStatus())) {
throw new IllegalStateException("task #" + task.getTaskNumber()
+ " cannot accept worker feedback while " + task.getStatus());
}
boolean resumed = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, taskId)
.in(TeamTaskEntity::getStatus, TeamTaskStatus.COMPLETED, TeamTaskStatus.FAILED,
TeamTaskStatus.STALE, TeamTaskStatus.IN_REVIEW)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getOwnerAgentId, task.getAssigneeAgentId())
.set(TeamTaskEntity::getReason, null)
.set(TeamTaskEntity::getLockExpiresAt, newLease())) == 1;
if (resumed) {
projectTask(taskId);
}
return resumed;
}
private static JSONObject parseMetadata(String raw) {
if (raw == null || raw.isBlank()) {
return new JSONObject();
}
try {
return JSONUtil.parseObj(raw);
} catch (RuntimeException invalid) {
return new JSONObject();
}
}
/** Extend the execution lease (runner heartbeat). */
public void renewLock(Long taskId) {
taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
@ -684,20 +955,48 @@ public class TeamTaskService {
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.isNotNull(TeamTaskEntity::getLockExpiresAt)
.lt(TeamTaskEntity::getLockExpiresAt, LocalDateTime.now()));
int staleCount = 0;
int uncertainReplayCount = 0;
for (TeamTaskEntity task : expired) {
JSONObject metadata = parseMetadata(task.getMetadata());
JSONObject approval = metadata.getJSONObject("toolApproval");
if (approval != null && approval.getBool("replayInProgress", false)) {
approval.set("replayInProgress", false);
approval.set("replayOutcomeUncertain", true);
approval.set("summary", "Approved tool replay was interrupted; outcome is uncertain");
metadata.set("toolApproval", approval);
String reason = "Approved tool replay lease expired; stop the replay or verify its outcome manually";
int rows = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, task.getId())
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.AWAITING_APPROVAL)
.set(TeamTaskEntity::getReason, reason)
.set(TeamTaskEntity::getMetadata, metadata.toString())
.set(TeamTaskEntity::getLockExpiresAt, null));
if (rows == 1) {
uncertainReplayCount++;
recordEvent(task.getTeamId(), task.getId(),
TeamTaskEventEntity.AWAITING_APPROVAL,
AUTHOR_SYSTEM, null, reason);
projectTask(task);
}
continue;
}
int rows = taskMapper.update(null, Wrappers.<TeamTaskEntity>lambdaUpdate()
.eq(TeamTaskEntity::getId, task.getId())
.eq(TeamTaskEntity::getStatus, TeamTaskStatus.IN_PROGRESS)
.set(TeamTaskEntity::getStatus, TeamTaskStatus.STALE)
.set(TeamTaskEntity::getReason, "execution lease expired"));
if (rows == 1) {
staleCount++;
recordEvent(task.getTeamId(), task.getId(), TeamTaskEventEntity.STALE,
AUTHOR_SYSTEM, null, "execution lease expired");
projectTask(task);
}
}
if (!expired.isEmpty()) {
log.warn("Marked {} team task(s) stale after lease expiry", expired.size());
if (staleCount > 0 || uncertainReplayCount > 0) {
log.warn("Recovered expired team task leases: stale={}, replayOutcomeUncertain={}",
staleCount, uncertainReplayCount);
}
return expired;
}

288
mateclaw-server/src/main/java/vip/mate/team/service/TeamWorkerInterventionService.java Normal file
View File

@ -0,0 +1,288 @@
package vip.mate.team.service;
import cn.hutool.json.JSONUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import vip.mate.agent.AgentService;
import vip.mate.agent.context.ChatOrigin;
import vip.mate.agent.runtime.ConversationTurnGate;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.approval.PendingApproval;
import vip.mate.approval.ResolveOutcome;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.team.model.TeamTaskEntity;
import vip.mate.workspace.conversation.ConversationService;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
/** Controlled write path for a delegated worker conversation. */
@Service
@RequiredArgsConstructor
public class TeamWorkerInterventionService {
static final String REPLAY_PROMPT = "继续执行已批准的工具调用。";
private final TeamTaskService taskService;
private final TeamWorkerConversationGovernanceService governanceService;
private final ApprovalWorkflowService approvalService;
private final AgentService agentService;
private final ConversationService conversationService;
private final ConversationTurnGate turnGate;
private final ChatStreamTracker streamTracker;
private final TeamDispatchService dispatchService;
private final TeamAnnounceService announceService;
private final TeamEventChannel eventChannel;
private final TeamWorkerReplayPersistenceService replayPersistenceService;
public TeamTaskEntity approve(Long teamId, Long taskId, String pendingId, String requester) {
Intervention intervention = requireIntervention(teamId, taskId);
if (!vip.mate.team.model.TeamTaskStatus.AWAITING_APPROVAL.equals(
intervention.task().getStatus())) {
return intervention.task();
}
if (taskService.isToolReplayOutcomeUncertain(intervention.task())) {
throw new IllegalStateException(
"tool replay outcome is uncertain; stop it or verify the side effect manually");
}
PendingApproval pending = requireReplayApproval(intervention, pendingId);
ScheduledFuture<?> heartbeat = null;
try (ConversationTurnGate.Permit permit = reserve(intervention.conversationId())) {
requireMemberIdle(intervention);
String reply = taskService.stagedToolReplayResult(intervention.task());
if (reply == null) {
PendingApproval claimedPending = claimReplay(
intervention, pendingId, requester, pending);
if (!taskService.resumeAfterToolApproval(taskId, pendingId)) {
throw new IllegalStateException(
"worker task changed while replay was being claimed");
}
heartbeat = dispatchService.startLeaseHeartbeat(taskId);
conversationService.removeApprovalPlaceholders(intervention.conversationId());
ChatOrigin origin = approvalService.restoreChatOrigin(claimedPending.getChatOrigin());
AgentService.ChatResult result;
try {
result = turnGate.withPermit(permit, () -> agentService.chatWithReplayWithUsage(
intervention.agentId(), REPLAY_PROMPT, intervention.conversationId(),
claimedPending.getToolCallPayload(), origin));
} catch (RuntimeException error) {
taskService.parkToolReplayUncertain(taskId, pendingId,
"Approved tool replay failed and its outcome is uncertain: "
+ safeMessage(error));
throw error;
}
reply = result == null ? "" : result.content();
if (!taskService.stageToolReplayResult(taskId, pendingId, reply)) {
throw new IllegalStateException("tool replay completed but its result could not be staged");
}
replayPersistenceService.persist(taskId, pendingId,
intervention.conversationId(), reply, result);
} else if (!taskService.isToolReplayMessagePersisted(intervention.task())
&& !reply.isBlank()) {
replayPersistenceService.persist(taskId, pendingId,
intervention.conversationId(), reply, null);
}
ResolveOutcome consumed = approvalService.consumeReplayClaim(pendingId, requester);
if (!consumed.isConsumed()) {
throw new IllegalStateException("approved tool replay could not be finalized");
}
if (!taskService.resumeAfterToolApproval(taskId, pendingId)) {
throw new IllegalStateException("worker task changed while replay was being finalized");
}
settleOrPark(intervention, reply);
return taskService.getTask(taskId);
} finally {
if (heartbeat != null) {
heartbeat.cancel(false);
}
}
}
@Transactional
public TeamTaskEntity deny(Long teamId, Long taskId, String pendingId, String requester) {
Intervention intervention = requireIntervention(teamId, taskId);
if (!vip.mate.team.model.TeamTaskStatus.AWAITING_APPROVAL.equals(
intervention.task().getStatus())) {
return intervention.task();
}
if (taskService.stagedToolReplayResult(intervention.task()) != null) {
throw new IllegalStateException("approved tool already executed; finalize its result instead");
}
PendingApproval approval = requireReplayApproval(intervention, pendingId);
try (ConversationTurnGate.Permit ignored = reserve(intervention.conversationId())) {
conversationService.removeApprovalPlaceholders(intervention.conversationId());
String event;
if ("approved".equals(approval.getStatus())) {
if (!taskService.abortClaimedToolReplay(taskId, pendingId, requester)) {
throw new IllegalStateException("worker task changed while replay was being stopped");
}
ResolveOutcome consumed = approvalService.consumeReplayClaim(pendingId, requester);
if (!consumed.isConsumed()) {
throw new IllegalStateException("claimed tool replay could not be stopped");
}
event = "team_task_tool_replay_aborted";
} else {
ResolveOutcome outcome = approvalService.resolve(pendingId, requester, "denied");
if (outcome.isAlreadyResolved()) {
throw new IllegalStateException("tool approval is no longer pending");
}
if (!taskService.denyToolApproval(taskId, pendingId, requester)) {
throw new IllegalStateException("worker task changed while approval was being denied");
}
event = "team_task_tool_denied";
}
TeamTaskEntity settled = taskService.getTask(taskId);
eventChannel.publishTaskEvent(settled, event, Map.of("pendingId", pendingId));
announceService.announceTaskSettled(settled);
dispatchService.requestDispatch(teamId);
return settled;
}
}
public TeamTaskEntity feedback(Long teamId, Long taskId, String message, String requester) {
String feedback = message == null ? "" : message.strip();
if (feedback.isEmpty()) {
throw new IllegalArgumentException("feedback is required");
}
if (feedback.length() > 4000) {
throw new IllegalArgumentException("feedback must be at most 4000 characters");
}
Intervention intervention = requireIntervention(teamId, taskId);
if (approvalService.findPendingByConversation(intervention.conversationId()) != null) {
throw new IllegalStateException("resolve the pending tool approval before sending feedback");
}
try (ConversationTurnGate.Permit permit = reserve(intervention.conversationId())) {
requireMemberIdle(intervention);
if (!taskService.resumeForWorkerFeedback(taskId)) {
throw new IllegalStateException("worker task changed before feedback could start");
}
conversationService.saveMessage(intervention.conversationId(), "user", feedback);
var agent = agentService.getAgent(intervention.agentId());
Long workspaceId = agent == null ? null : agent.getWorkspaceId();
ChatOrigin origin = ChatOrigin.web(
intervention.conversationId(), requester, workspaceId, null);
AgentService.ChatResult result;
try {
result = turnGate.withPermit(permit, () -> agentService.chatWithUsage(
intervention.agentId(), feedback, intervention.conversationId(), origin));
} catch (RuntimeException error) {
taskService.failTask(taskId, "worker feedback failed: " + safeMessage(error));
throw error;
}
String reply = persistAssistant(intervention.conversationId(), result);
settleOrPark(intervention, reply);
return taskService.getTask(taskId);
}
}
private Intervention requireIntervention(Long teamId, Long taskId) {
TeamTaskEntity task = taskService.getTask(taskId);
if (task == null || !teamId.equals(task.getTeamId()) || task.getRunId() == null
|| task.getConversationId() == null || task.getConversationId().isBlank()) {
throw new IllegalArgumentException("worker conversation not found for this task");
}
TeamWorkerConversationContext context = governanceService.resolve(
task.getConversationId(), task.getRunId(), taskId)
.filter(candidate -> teamId.equals(candidate.teamId())
&& Objects.equals(task.getAssigneeAgentId(), candidate.agentId()))
.orElseThrow(() -> new IllegalArgumentException(
"worker conversation not found for this task"));
return new Intervention(task, context.conversationId(), context.agentId());
}
private PendingApproval requireReplayApproval(Intervention intervention, String pendingId) {
requireCurrentPendingId(intervention, pendingId);
return approvalService.getPending(pendingId)
.filter(pending -> intervention.conversationId().equals(pending.getConversationId()))
.filter(pending -> "pending".equals(pending.getStatus())
|| "approved".equals(pending.getStatus()))
.or(() -> approvalService.getReplayClaim(pendingId)
.filter(pending -> intervention.conversationId()
.equals(pending.getConversationId())))
.orElseThrow(() -> new IllegalStateException(
"tool approval is no longer pending or claimed"));
}
private void requireCurrentPendingId(Intervention intervention, String pendingId) {
if (pendingId == null || pendingId.isBlank()) {
throw new IllegalArgumentException("pending approval id is required");
}
String currentPendingId = null;
try {
var metadata = JSONUtil.parseObj(intervention.task().getMetadata());
var approval = metadata.getJSONObject("toolApproval");
currentPendingId = approval == null ? null : approval.getStr("pendingId");
} catch (RuntimeException ignored) {
// Missing or malformed task metadata means the client cannot prove
// that this approval is the one the task is parked on.
}
if (!pendingId.equals(currentPendingId)) {
throw new IllegalStateException("tool approval is no longer current for this task");
}
}
private PendingApproval claimReplay(Intervention intervention, String pendingId,
String requester, PendingApproval pending) {
if ("pending".equals(pending.getStatus())) {
ResolveOutcome claimed = approvalService.claimForReplay(pendingId, requester);
if (claimed.isAlreadyResolved()) {
throw new IllegalStateException("tool approval was resolved concurrently");
}
}
return approvalService.getReplayClaim(pendingId)
.filter(candidate -> intervention.conversationId()
.equals(candidate.getConversationId()))
.orElseThrow(() -> new IllegalStateException(
"approved tool replay claim could not be recovered"));
}
private void requireMemberIdle(Intervention intervention) {
if (taskService.hasActiveTask(intervention.task().getTeamId(), intervention.agentId())) {
throw new IllegalStateException("worker agent is already executing another team task");
}
}
private ConversationTurnGate.Permit reserve(String conversationId) {
ConversationTurnGate.Permit permit = turnGate.tryAcquire(conversationId);
if (permit == null || streamTracker.isRunning(conversationId)) {
if (permit != null) {
permit.close();
}
throw new IllegalStateException("worker conversation is already running");
}
return permit;
}
private String persistAssistant(String conversationId, AgentService.ChatResult result) {
String reply = result == null ? "" : result.content();
if (reply != null && !reply.isBlank()) {
conversationService.saveMessage(conversationId, "assistant", reply, null, "completed",
result.promptTokens(), result.completionTokens(),
result.runtimeModel(), result.runtimeProvider());
}
return reply;
}
private void settleOrPark(Intervention intervention, String reply) {
PendingApproval next = approvalService.findPendingByConversation(intervention.conversationId());
if (next != null) {
String summary = next.getSummary() == null || next.getSummary().isBlank()
? next.getReason() : next.getSummary();
taskService.parkForToolApproval(intervention.task().getId(), next.getPendingId(), summary);
eventChannel.publishTaskEvent(taskService.getTask(intervention.task().getId()),
"team_task_awaiting_approval", Map.of("pendingId", next.getPendingId()));
return;
}
dispatchService.settleOutcome(intervention.task(), reply);
dispatchService.requestDispatch(intervention.task().getTeamId());
}
private static String safeMessage(RuntimeException error) {
return error.getMessage() == null ? error.getClass().getSimpleName() : error.getMessage();
}
private record Intervention(TeamTaskEntity task, String conversationId, Long agentId) {
}
}

34
mateclaw-server/src/main/java/vip/mate/team/service/TeamWorkerReplayPersistenceService.java Normal file
View File

@ -0,0 +1,34 @@
package vip.mate.team.service;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import vip.mate.agent.AgentService;
import vip.mate.workspace.conversation.ConversationService;
/** Atomically records a replay reply and its task-level idempotency marker. */
@Service
@RequiredArgsConstructor
public class TeamWorkerReplayPersistenceService {
private final ConversationService conversationService;
private final TeamTaskService taskService;
@Transactional
public void persist(Long taskId, String pendingId, String conversationId,
String reply, AgentService.ChatResult result) {
if (reply == null || reply.isBlank()) {
return;
}
if (result == null) {
conversationService.saveMessage(conversationId, "assistant", reply);
} else {
conversationService.saveMessage(conversationId, "assistant", reply,
null, "completed", result.promptTokens(), result.completionTokens(),
result.runtimeModel(), result.runtimeProvider());
}
if (!taskService.markToolReplayMessagePersisted(taskId, pendingId)) {
throw new IllegalStateException("tool replay message marker could not be persisted");
}
}
}

40
mateclaw-server/src/test/java/vip/mate/agent/ChatResultCollectorTest.java Normal file
View File

@ -0,0 +1,40 @@
package vip.mate.agent;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
class ChatResultCollectorTest {
@Test
void preservesStructuredFinishReasonAlongsideContentAndUsage() {
AgentService.ChatResult result = ChatResultCollector.collect(Flux.just(
new AgentService.StreamDelta("partial failure", null),
AgentService.StreamDelta.event("finish_reason",
Map.of("reason", "error_fallback")),
AgentService.StreamDelta.event("_usage_final", Map.of(
"promptTokens", 12,
"completionTokens", 3,
"runtimeModelName", "model-a",
"runtimeProviderId", "provider-a"))));
assertEquals("partial failure", result.content());
assertEquals(12, result.promptTokens());
assertEquals(3, result.completionTokens());
assertEquals("model-a", result.runtimeModel());
assertEquals("provider-a", result.runtimeProvider());
assertEquals("error_fallback", result.finishReason());
}
@Test
void lastFinishReasonWinsForReplayCompatibleStreams() {
AgentService.ChatResult result = ChatResultCollector.collect(Flux.just(
AgentService.StreamDelta.event("finish_reason", Map.of("reason", "incomplete")),
AgentService.StreamDelta.event("finish_reason", Map.of("reason", "normal"))));
assertEquals("normal", result.finishReason());
}
}

21
mateclaw-server/src/test/java/vip/mate/agent/graph/ErrorClassificationTest.java
View File

@ -246,6 +246,27 @@ class ErrorClassificationTest {
classify(new RuntimeException("AccountBalanceNotEnough: balance not enough")));
}
@Test
@DisplayName("Volcengine InvalidSubscription → BILLING instead of CLIENT_ERROR")
void volcengineExpiredCodingPlanIsBilling() throws Exception {
assertEquals(NodeStreamingChatHelper.ErrorType.BILLING,
classify(new RuntimeException("400 InvalidSubscription: CodingPlan subscription has expired")));
}
@Test
@DisplayName("DashScope Arrearage / good-standing error → BILLING")
void dashscopeArrearageIsBilling() throws Exception {
assertEquals(NodeStreamingChatHelper.ErrorType.BILLING,
classify(new RuntimeException("400 Arrearage: Access denied, make sure your account is in good standing")));
}
@Test
@DisplayName("MiniMax insufficient balance body → BILLING")
void minimaxInsufficientBalanceIsBilling() throws Exception {
assertEquals(NodeStreamingChatHelper.ErrorType.BILLING,
classify(new RuntimeException("insufficient_balance_error: insufficient balance (1008)")));
}
// ===== Infrastructure-fatal errors → AUTH_ERROR (HARD, no same-model retry) =====
//
// DNS / TLS-trust failures do not self-heal on retry. They are routed through

20
mateclaw-server/src/test/java/vip/mate/agent/graph/node/ReasoningNodeOutputTest.java
View File

@ -190,6 +190,26 @@ class ReasoningNodeOutputTest {
"Continuation prompt should ask the model to keep writing instead of ending the run");
}
@Test
@DisplayName("runtime error placeholder cannot satisfy or continue a long-form request")
void longFormTextRequest_runtimeErrorPlaceholderFailsImmediately() throws Exception {
String internalError = "[错误] Bad request: account subscription expired";
NodeStreamingChatHelper.StreamResult result = new NodeStreamingChatHelper.StreamResult(
internalError, "", new AssistantMessage(internalError),
List.of(), false, 100, 0);
when(streamingHelper.streamCall(any(), any(), anyString(), anyString())).thenReturn(result);
Map<String, Object> state = baseStateMap();
state.put(USER_MESSAGE, "请输出不少于 8000 字的技术报告");
state.put(MAX_ITERATIONS, 150);
Map<String, Object> output = createNode().apply(new OverAllState(state));
assertEquals(false, output.get(CONTINUE_REASONING));
assertEquals("error_fallback", output.get(FINISH_REASON));
assertEquals(internalError, output.get(FINAL_ANSWER));
assertNull(output.get("long_form_draft"));
}
@Test
@DisplayName("long-form continuation persists all chunks as one final answer")
void longFormTextRequest_combinesContinuationChunksInFinalAnswer() throws Exception {

44
mateclaw-server/src/test/java/vip/mate/approval/ApprovalWorkflowServiceResolveTest.java
View File

@ -208,6 +208,50 @@ class ApprovalWorkflowServiceResolveTest {
verifyNoInteractions(conversationService);
}
@Test
@DisplayName("team replay claims PENDING before execution and consumes exact APPROVED claim")
void teamReplayClaimIsDurableAndSingleShot() {
PendingApproval pending = seedPending("pid-team", "conv-team", "shell");
pending.setToolCallPayload("{\"name\":\"shell\"}");
when(approvalMapper.update(isNull(), any(Wrapper.class))).thenReturn(1);
when(conversationService.markPendingApprovalsResolved(
eq("conv-team"), eq(Set.of("pid-team")), eq(MetadataDecision.APPROVED)))
.thenReturn(1);
ResolveOutcome claimed = workflow.claimForReplay("pid-team", "alice");
ResolveOutcome consumed = workflow.consumeReplayClaim("pid-team", "alice");
assertThat(claimed.decision()).isEqualTo("approved");
assertThat(pending.getStatus()).isEqualTo("consumed");
assertThat(consumed.isConsumed()).isTrue();
assertThat(consumed.consumedSnapshot().getToolCallPayload())
.isEqualTo("{\"name\":\"shell\"}");
assertThat(approvalService.getPending("pid-team")).isEmpty();
verify(approvalMapper, times(2)).update(isNull(), any(Wrapper.class));
}
@Test
@DisplayName("an APPROVED replay claim is recoverable from DB after restart")
void replayClaimRecoversFromDatabase() {
ToolApprovalEntity entity = new ToolApprovalEntity();
entity.setPendingId("pid-restart");
entity.setConversationId("conv-restart");
entity.setUserId("alice");
entity.setAgentId("201");
entity.setToolName("shell");
entity.setToolArguments("{}");
entity.setToolCallPayload("{\"name\":\"shell\"}");
entity.setSummary("approved replay");
entity.setStatus("APPROVED");
when(approvalMapper.selectOne(any())).thenReturn(entity);
PendingApproval recovered = workflow.getReplayClaim("pid-restart").orElseThrow();
assertThat(recovered.getStatus()).isEqualTo("approved");
assertThat(recovered.getConversationId()).isEqualTo("conv-restart");
assertThat(recovered.getToolCallPayload()).isEqualTo("{\"name\":\"shell\"}");
}
@Test
@DisplayName("cancelStalePending issues a SUPERSEDED outcome per pending in the conversation")
void cancelStalePendingMultipleEntries() {

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

@ -0,0 +1,151 @@
package vip.mate.channel.web;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
class ChatStreamTrackerContentBatchTest {
private static final ObjectMapper MAPPER = new ObjectMapper();
@Test
void adjacentContentDeltasFlushAsOneTimedBatch() throws Exception {
ChatStreamTracker tracker = tracker(25, 256);
CapturingEmitter emitter = attach(tracker, "timed");
tracker.broadcast("timed", "content_delta", "{\"delta\":\"你\"}");
tracker.broadcast("timed", "content_delta", "{\"delta\":\"好\"}");
awaitEventCount(emitter, 1);
assertEquals(1, emitter.events.size());
assertEquals("content_delta", emitter.events.getFirst().name());
assertEquals("你好", text(emitter.events.getFirst().data(), "delta"));
}
@Test
void characterLimitFlushesWithoutWaitingForTimer() throws Exception {
ChatStreamTracker tracker = tracker(60_000, 4);
CapturingEmitter emitter = attach(tracker, "bounded");
tracker.broadcast("bounded", "content_delta", "{\"delta\":\"ab\"}");
tracker.broadcast("bounded", "content_delta", "{\"delta\":\"cd\"}");
assertEquals(1, emitter.events.size());
assertEquals("abcd", text(emitter.events.getFirst().data(), "delta"));
}
@Test
void lifecycleEventFlushesContentFirstAndDoneIsReplayable() throws Exception {
ChatStreamTracker tracker = tracker(60_000, 256);
CapturingEmitter live = attach(tracker, "ordered");
tracker.broadcast("ordered", "content_delta", "{\"delta\":\"answer\"}");
tracker.broadcast("ordered", "phase", "{\"phase\":\"complete\"}");
tracker.broadcast("ordered", "done", "{\"status\":\"completed\"}");
assertEquals(List.of("content_delta", "phase", "done"), live.names());
CapturingEmitter replay = new CapturingEmitter();
assertTrue(tracker.attach("ordered", replay));
assertEquals(List.of("content_delta", "phase", "done"), replay.names());
assertEquals("answer", text(replay.events.getFirst().data(), "delta"));
}
@Test
void webchatTextPayloadKeepsItsWireField() throws Exception {
ChatStreamTracker tracker = tracker(60_000, 4);
ChatStreamTracker.RunHandle handle = tracker.register("webchat");
CapturingEmitter emitter = new CapturingEmitter();
tracker.attach(handle, emitter);
tracker.broadcast(handle, "content_delta", "{\"text\":\"ab\"}");
tracker.broadcast(handle, "content_delta", "{\"text\":\"cd\"}");
assertEquals(1, emitter.events.size());
assertEquals("abcd", text(emitter.events.getFirst().data(), "text"));
}
@Test
void lifecycleCompletionFlushesPendingContentEvenWithoutDoneEnvelope() throws Exception {
ChatStreamTracker tracker = tracker(60_000, 256);
ChatStreamTracker.RunHandle handle = tracker.register("complete");
CapturingEmitter emitter = new CapturingEmitter();
tracker.attach(handle, emitter);
tracker.broadcast(handle, "content_delta", "{\"delta\":\"partial\"}");
tracker.complete(handle);
assertEquals(1, emitter.events.size());
assertEquals("partial", text(emitter.events.getFirst().data(), "delta"));
}
private static ChatStreamTracker tracker(long flushMs, int maxChars) {
ChatStreamTracker tracker = new ChatStreamTracker(MAPPER);
tracker.setContentBatchingForTesting(flushMs, maxChars);
return tracker;
}
private static CapturingEmitter attach(ChatStreamTracker tracker, String conversationId) {
tracker.register(conversationId);
CapturingEmitter emitter = new CapturingEmitter();
tracker.attach(conversationId, emitter);
return emitter;
}
private static void awaitEventCount(CapturingEmitter emitter, int expected) throws InterruptedException {
long deadline = System.currentTimeMillis() + 1_000;
while (emitter.events.size() < expected && System.currentTimeMillis() < deadline) {
Thread.sleep(5);
}
}
private static String text(String json, String field) throws Exception {
JsonNode node = MAPPER.readTree(json);
return node.path(field).asText();
}
private record Event(String name, String data) {}
private static final class CapturingEmitter extends SseEmitter {
private final List<Event> events = new CopyOnWriteArrayList<>();
CapturingEmitter() {
super(60_000L);
}
List<String> names() {
return events.stream().map(Event::name).toList();
}
@Override
public void send(SseEventBuilder builder) throws IOException {
Set<ResponseBodyEmitter.DataWithMediaType> entries = builder.build();
String name = "";
String payload = "";
boolean expectPayload = false;
for (ResponseBodyEmitter.DataWithMediaType entry : entries) {
if (!(entry.getData() instanceof String text)) continue;
if (text.contains("event:") && text.contains("data:")) {
int start = text.indexOf("event:") + 6;
int end = text.indexOf('\n', start);
name = text.substring(start, end < 0 ? text.length() : end).trim();
expectPayload = true;
} else if (expectPayload && !"\n\n".equals(text)) {
payload = text;
expectPayload = false;
}
}
events.add(new Event(name, payload));
}
}
}

6
mateclaw-server/src/test/java/vip/mate/channel/web/ChatStreamTrackerOrphanPolicyTest.java
View File

@ -430,9 +430,13 @@ class ChatStreamTrackerOrphanPolicyTest {
Map<String, ChatStreamTracker.RunState> runs =
(Map<String, ChatStreamTracker.RunState>) runsField.get(tracker);
ChatStreamTracker.RunState state = runs.get(cid);
long deadline = System.currentTimeMillis() + 1_000L;
while (state.subscribersZeroSince == null && System.currentTimeMillis() < deadline) {
Thread.sleep(5L);
}
synchronized (state.lock) {
assertNotNull(state.subscribersZeroSince,
"removing the final dead subscriber must arm the orphan clock");
"removing the final dead subscriber must arm the orphan clock within the batch window");
state.subscribersZeroSince = System.currentTimeMillis() - 3_000L;
}

20
mateclaw-server/src/test/java/vip/mate/cron/service/CronJobLifecycleFenceTest.java
View File

@ -9,6 +9,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.context.ApplicationEventPublisher;
import vip.mate.agent.AgentService;
import vip.mate.cron.model.CronJobEntity;
import vip.mate.dashboard.model.CronJobRunEntity;
import vip.mate.dashboard.repository.CronJobRunMapper;
@ -53,6 +54,25 @@ class CronJobLifecycleFenceTest {
verifyNoInteractions(fixture.conversations, fixture.completionPublisher, fixture.events);
}
@Test
void graphErrorPersistsAnErrorMessageWithoutPublishingSuccessEvents() {
Fixture fixture = new Fixture();
when(fixture.mapper.update(isNull(), any(Wrapper.class))).thenReturn(1);
AgentService.ChatResult failed = new AgentService.ChatResult(
"[错误] account expired", 12, 3, "model-a", "provider-a", "error_fallback");
fixture.service.finishRunFailed(run(), new AssistantMessage(failed.content()),
"cron-1", failed);
@SuppressWarnings("rawtypes")
ArgumentCaptor<Wrapper> captor = ArgumentCaptor.forClass(Wrapper.class);
verify(fixture.mapper).update(isNull(), captor.capture());
assertTrue(captor.getValue().getSqlSegment().contains("status"));
verify(fixture.conversations).saveMessage("cron-1", "assistant", failed.content(),
null, "error", 12, 3, "model-a", "provider-a");
verifyNoInteractions(fixture.completionPublisher, fixture.events);
}
private static CronJobRunEntity run() {
CronJobRunEntity run = new CronJobRunEntity();
run.setId(42L);

33
mateclaw-server/src/test/java/vip/mate/cron/service/CronJobOriginPropagationTest.java
View File

@ -117,6 +117,39 @@ class CronJobOriginPropagationTest {
verify(lifecycle).markRunFailed(eq(run), any(IllegalStateException.class));
}
@Test
void runnerTreatsStructuredGraphErrorAsFailedTerminalState() {
CronJobLifecycleService lifecycle = mock(CronJobLifecycleService.class);
CronRunHeartbeatService heartbeat = mock(CronRunHeartbeatService.class);
CronRunHeartbeatService.Lease lease = mock(CronRunHeartbeatService.Lease.class);
AgentService agentService = mock(AgentService.class);
CronChatOriginFactory originFactory = mock(CronChatOriginFactory.class);
CronConversationResolver resolver = mock(CronConversationResolver.class);
CronJobEntity job = job();
CronJobRunEntity run = new CronJobRunEntity();
run.setId(55L);
ChatOrigin origin = ChatOrigin.cron(CONVERSATION_ID, WORKSPACE_ID, null, null, null);
AgentService.ChatResult failed = new AgentService.ChatResult(
"[错误] account expired", 12, 3, "model-a", "provider-a", "error_fallback");
when(resolver.resolve(job)).thenReturn(CONVERSATION_ID);
when(lifecycle.startRun(job, "do work", "scheduled", CONVERSATION_ID))
.thenReturn(new CronJobLifecycleService.StartResult(run, MESSAGE_ID));
when(originFactory.from(job, CONVERSATION_ID, MESSAGE_ID)).thenReturn(origin);
when(heartbeat.begin(55L)).thenReturn(lease);
when(agentService.chatWithUsage(eq(AGENT_ID), anyString(), eq(CONVERSATION_ID), eq(origin)))
.thenReturn(failed);
CronJobRunner runner = new CronJobRunner(lifecycle, heartbeat, agentService, originFactory, resolver,
mock(WikiProcessingService.class), new ObjectMapper());
runner.executeJob(job);
verify(lifecycle).finishRunFailed(eq(run), any(org.springframework.ai.chat.messages.AssistantMessage.class),
eq(CONVERSATION_ID), eq(failed));
verify(lifecycle, never()).finishRunAndPublish(eq(job), eq(run), anyString(),
any(org.springframework.ai.chat.messages.AssistantMessage.class), eq(CONVERSATION_ID),
any(Boolean.class), eq(failed));
}
private static CronJobEntity job() {
CronJobEntity job = new CronJobEntity();
job.setId(JOB_ID);

9
mateclaw-server/src/test/java/vip/mate/llm/service/ModelDiscoveryServiceTestPromptTest.java
View File

@ -100,4 +100,13 @@ class ModelDiscoveryServiceTestPromptTest {
Map<String, Object> requestBodyFromNull = ModelDiscoveryService.buildTestPromptRequestBody("gpt-4-turbo", null);
assertEquals(requestBody, requestBodyFromNull);
}
@Test
@DisplayName("Kimi smoke test uses the same fixed temperature required by runtime")
void kimiForCoding_usesTemperatureOne() {
Map<String, Object> requestBody = ModelDiscoveryService.buildTestPromptRequestBody(
"kimi-for-coding", Map.of("temperature", 0.2));
assertEquals(1.0d, requestBody.get("temperature"));
}
}

52
mateclaw-server/src/test/java/vip/mate/team/controller/TeamControllerTest.java
View File

@ -29,6 +29,7 @@ import vip.mate.team.service.TeamManualTaskService;
import vip.mate.team.service.TeamRunService;
import vip.mate.team.service.TeamService;
import vip.mate.team.service.TeamTaskService;
import vip.mate.team.service.TeamWorkerInterventionService;
import vip.mate.workspace.core.annotation.RequireWorkspaceRole;
import vip.mate.workspace.core.service.WorkspaceService;
@ -69,6 +70,7 @@ class TeamControllerTest {
@Mock private TeamDispatchService dispatchService;
@Mock private TeamAnnounceService announceService;
@Mock private TeamEventChannel eventChannel;
@Mock private TeamWorkerInterventionService workerInterventionService;
@Mock private AgentMapper agentMapper;
@Mock private WorkspaceService workspaceService;
@Mock private AuthService authService;
@ -80,7 +82,7 @@ class TeamControllerTest {
void setUp() {
manualTaskService = new TeamManualTaskService(runService, taskService, events);
controller = new TeamController(teamService, taskService, manualTaskService, dispatchService,
announceService, eventChannel, agentMapper);
announceService, eventChannel, workerInterventionService, agentMapper);
AgentTeamEntity team = new AgentTeamEntity();
team.setId(TEAM_ID);
team.setWorkspaceId(1L);
@ -112,6 +114,54 @@ class TeamControllerTest {
return run;
}
@Test
void workerApprovalEndpointUsesTaskScopedInterventionService() {
TeamTaskEntity waiting = task(TEAM_ID, TeamTaskStatus.AWAITING_APPROVAL);
waiting.setRunId(RUN_ID);
when(taskService.getTask(TASK_ID)).thenReturn(waiting);
when(workerInterventionService.approve(TEAM_ID, TASK_ID, "pending-42", "alice"))
.thenReturn(waiting);
TeamController.WorkerApprovalRequest request = new TeamController.WorkerApprovalRequest();
request.setPendingId("pending-42");
R<TeamController.TaskVO> response = controller.approveWorkerTool(
TEAM_ID, TASK_ID, request, () -> "alice");
assertEquals(200, response.getCode());
verify(workerInterventionService).approve(TEAM_ID, TASK_ID, "pending-42", "alice");
}
@Test
void workerFeedbackEndpointRejectsBlankContentBeforeRunningAgent() {
when(taskService.getTask(TASK_ID)).thenReturn(task(TEAM_ID, TeamTaskStatus.COMPLETED));
TeamController.WorkerFeedbackRequest request = new TeamController.WorkerFeedbackRequest();
request.setMessage(" ");
R<TeamController.TaskVO> response = controller.feedbackWorker(
TEAM_ID, TASK_ID, request, () -> "alice");
assertEquals(400, response.getCode());
verify(workerInterventionService, never()).feedback(any(), any(), any(), any());
}
@Test
void workerInterventionMapsMissingLinkAndBusyConversationToActionableCodes() {
TeamTaskEntity waiting = task(TEAM_ID, TeamTaskStatus.AWAITING_APPROVAL);
waiting.setRunId(RUN_ID);
when(taskService.getTask(TASK_ID)).thenReturn(waiting);
TeamController.WorkerApprovalRequest request = new TeamController.WorkerApprovalRequest();
request.setPendingId("pending-42");
when(workerInterventionService.approve(TEAM_ID, TASK_ID, "pending-42", "alice"))
.thenThrow(new IllegalArgumentException("worker conversation not found for this task"));
when(workerInterventionService.deny(TEAM_ID, TASK_ID, "pending-42", "alice"))
.thenThrow(new IllegalStateException("worker conversation is already running"));
assertEquals(404, controller.approveWorkerTool(
TEAM_ID, TASK_ID, request, () -> "alice").getCode());
assertEquals(409, controller.denyWorkerTool(
TEAM_ID, TASK_ID, request, () -> "alice").getCode());
}
// ==================== team / membership ====================
@Test

4
mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceEventTest.java
View File

@ -12,6 +12,7 @@ import org.springframework.transaction.support.AbstractPlatformTransactionManage
import org.springframework.transaction.support.DefaultTransactionStatus;
import org.springframework.transaction.support.TransactionTemplate;
import vip.mate.agent.AgentService;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.team.event.TeamTasksDelegatedEvent;
import vip.mate.workspace.conversation.ConversationService;
@ -65,7 +66,8 @@ class TeamDispatchServiceEventTest {
return new TeamDispatchService(
mock(TeamService.class), taskService, mock(AgentService.class),
mock(ConversationService.class), mock(ChatStreamTracker.class),
mock(TeamAnnounceService.class), mock(TeamEventChannel.class));
mock(TeamAnnounceService.class), mock(TeamEventChannel.class),
mock(ApprovalWorkflowService.class));
}
}

28
mateclaw-server/src/test/java/vip/mate/team/service/TeamDispatchServiceTest.java
View File

@ -4,6 +4,8 @@ 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.approval.ApprovalWorkflowService;
import vip.mate.approval.PendingApproval;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.team.model.AgentTeamEntity;
import vip.mate.team.model.TeamTaskEntity;
@ -43,6 +45,7 @@ class TeamDispatchServiceTest {
private ChatStreamTracker streamTracker;
private TeamAnnounceService announceService;
private TeamEventChannel eventChannel;
private ApprovalWorkflowService approvalService;
private TeamDispatchService service;
@BeforeEach
@ -54,8 +57,9 @@ class TeamDispatchServiceTest {
streamTracker = mock(ChatStreamTracker.class);
announceService = mock(TeamAnnounceService.class);
eventChannel = mock(TeamEventChannel.class);
approvalService = mock(ApprovalWorkflowService.class);
service = new TeamDispatchService(teamService, taskService, agentService,
conversationService, streamTracker, announceService, eventChannel);
conversationService, streamTracker, announceService, eventChannel, approvalService);
}
private TeamTaskEntity task(Long id, Long assignee) {
@ -365,6 +369,28 @@ class TeamDispatchServiceTest {
verify(conversationService).saveMessage(startsWith("team-task-"), eq("assistant"), eq("all done"));
}
@Test
@DisplayName("a worker tool approval parks the task instead of completing or retrying it")
void runTaskParksPendingToolApproval() {
TeamTaskEntity assigned = task(1L, MEMBER_A);
assigned.setStatus(TeamTaskStatus.IN_PROGRESS);
PendingApproval pending = new PendingApproval("pending-42", "worker", "system",
"execute_shell_command", "{}", "shell command requires approval");
pending.setSummary("shell command requires approval");
when(agentService.chatWithUsage(eq(MEMBER_A), anyString(), anyString()))
.thenReturn(AgentService.ChatResult.contentOnly("I need permission first."));
when(approvalService.findPendingByConversation(startsWith("team-task-"))).thenReturn(pending);
when(taskService.parkForToolApproval(1L, "pending-42", "shell command requires approval"))
.thenReturn(true);
service.runTask(TEAM_ID, assigned);
verify(taskService).parkForToolApproval(1L, "pending-42", "shell command requires approval");
verify(taskService, never()).completeTask(any(), any(), anyString());
verify(taskService, never()).requeueUnusableResult(any(), anyString());
verify(eventChannel).publishTaskEvent(any(), eq("team_task_awaiting_approval"), any());
}
@Test
@DisplayName("member child conversation inherits the team's workspace")
void runTaskCreatesChildConversationInTeamWorkspace() {

2
mateclaw-server/src/test/java/vip/mate/team/service/TeamRunStateMachineTest.java
View File

@ -47,6 +47,8 @@ class TeamRunStateMachineTest {
TeamRunStatus.RUNNING, null, 1, 0, 0, 50),
Arguments.of("blocked tasks are active", TeamRunStatus.AWAITING_REVIEW,
tasks(TeamTaskStatus.BLOCKED), TeamRunStatus.RUNNING, null, 0, 0, 0, 0),
Arguments.of("tool approval waits are active", TeamRunStatus.FINALIZING,
tasks(TeamTaskStatus.AWAITING_APPROVAL), TeamRunStatus.RUNNING, null, 0, 0, 0, 0),
Arguments.of("review only", TeamRunStatus.RUNNING,
tasks(TeamTaskStatus.COMPLETED, TeamTaskStatus.IN_REVIEW),
TeamRunStatus.AWAITING_REVIEW, null, 1, 0, 1, 50),

33
mateclaw-server/src/test/java/vip/mate/team/service/TeamRunViewFactoryTest.java
View File

@ -81,6 +81,39 @@ class TeamRunViewFactoryTest {
assertEquals(List.of(), view.attentionItems());
}
@Test
void awaitingToolApprovalCreatesHighestPriorityAttentionItem() {
TeamTaskEntity task = task(101L, 201L,
"{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"summary\":\"shell command requires approval\"}}");
task.setStatus(TeamTaskStatus.AWAITING_APPROVAL);
task.setReason("shell command requires approval");
TeamRunView view = project(run("{}"), List.of(task));
assertEquals(1, view.attentionItems().size());
TeamRunView.AttentionItem item = view.attentionItems().getFirst();
assertEquals("approval", item.type());
assertEquals("action", item.severity());
assertEquals(0, item.priority());
assertEquals(101L, item.taskId());
assertEquals("shell command requires approval", item.message());
}
@Test
void uncertainReplayProjectsOnlyTheSafeRecoveryAttentionType() {
TeamTaskEntity task = task(101L, 201L,
"{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayOutcomeUncertain\":true}}");
task.setStatus(TeamTaskStatus.AWAITING_APPROVAL);
task.setReason("verify the tool outcome manually");
TeamRunView view = project(run("{}"), List.of(task));
assertEquals("replay_uncertain", view.attentionItems().getFirst().type());
assertEquals("action", view.attentionItems().getFirst().severity());
}
@Test
void aggregatesRunOnlyDeliverables() {
TeamRunEntity run = run("{\"deliverables\":[{\"name\":\"report.pdf\","

151
mateclaw-server/src/test/java/vip/mate/team/service/TeamTaskServiceTest.java
View File

@ -390,6 +390,157 @@ class TeamTaskServiceTest {
verify(projectionScheduler).scheduleTask(5L);
}
@Test
@DisplayName("tool approval parks an in-progress task without releasing dependents")
void toolApprovalParksTask() {
TeamTaskEntity running = runTask(5L, TeamTaskStatus.IN_PROGRESS);
running.setMetadata("{\"deliverableRequired\":true}");
when(taskMapper.selectById(5L)).thenReturn(running);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.parkForToolApproval(5L, "pending-42", "shell command requires approval"));
ArgumentCaptor<LambdaUpdateWrapper<TeamTaskEntity>> captor =
ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
verify(taskMapper).update(isNull(), captor.capture());
var values = captor.getValue().getParamNameValuePairs().values();
assertTrue(values.contains("awaiting_approval"));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("pending-42")));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("deliverableRequired")),
"parking must merge approval context into existing metadata");
verify(taskMapper, never()).selectList(any());
verify(projectionScheduler).scheduleTask(5L);
}
@Test
@DisplayName("tool approval resume requires the exact pending id and records the replay lease")
void toolApprovalResumeUsesExactPendingId() {
TeamTaskEntity waiting = runTask(5L, TeamTaskStatus.AWAITING_APPROVAL);
waiting.setAssigneeAgentId(MEMBER_ID);
waiting.setMetadata("{\"deliverableRequired\":true,\"toolApproval\":{\"pendingId\":\"pending-42\"}}");
when(taskMapper.selectById(5L)).thenReturn(waiting);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.resumeAfterToolApproval(5L, "pending-42"));
ArgumentCaptor<LambdaUpdateWrapper<TeamTaskEntity>> captor =
ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
verify(taskMapper).update(isNull(), captor.capture());
var values = captor.getValue().getParamNameValuePairs().values();
assertTrue(values.contains(TeamTaskStatus.IN_PROGRESS));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("deliverableRequired")));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("replayInProgress")));
verify(projectionScheduler).scheduleTask(5L);
}
@Test
@DisplayName("a stale pending id cannot resume a worker task")
void staleToolApprovalCannotResumeTask() {
TeamTaskEntity waiting = runTask(5L, TeamTaskStatus.AWAITING_APPROVAL);
waiting.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-new\"}}");
when(taskMapper.selectById(5L)).thenReturn(waiting);
IllegalStateException error = assertThrows(IllegalStateException.class,
() -> service.resumeAfterToolApproval(5L, "pending-old"));
assertTrue(error.getMessage().contains("no longer current"));
verify(taskMapper, never()).update(isNull(), any());
}
@Test
@DisplayName("a replay result is staged under the exact approval before it is consumed")
void stagesToolReplayResultForCrashSafeFinalization() {
TeamTaskEntity running = runTask(5L, TeamTaskStatus.IN_PROGRESS);
running.setMetadata("{\"deliverableRequired\":true,"
+ "\"toolApproval\":{\"pendingId\":\"pending-42\"}}");
when(taskMapper.selectById(5L)).thenReturn(running);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.stageToolReplayResult(5L, "pending-42", "tool completed"));
ArgumentCaptor<LambdaUpdateWrapper<TeamTaskEntity>> captor =
ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
verify(taskMapper).update(isNull(), captor.capture());
var values = captor.getValue().getParamNameValuePairs().values();
assertTrue(values.contains(TeamTaskStatus.AWAITING_APPROVAL));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("pending-42")));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("tool completed")));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value).contains("deliverableRequired")));
verify(projectionScheduler).scheduleTask(5L);
}
@Test
@DisplayName("a claimed replay can be stopped after failure but not after a result is staged")
void abortClaimedReplayHasExplicitGuard() {
TeamTaskEntity waiting = runTask(5L, TeamTaskStatus.AWAITING_APPROVAL);
waiting.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\"}}");
when(taskMapper.selectById(5L)).thenReturn(waiting);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.abortClaimedToolReplay(5L, "pending-42", "alice"));
waiting.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayResult\":\"done\"}}");
assertThrows(IllegalStateException.class,
() -> service.abortClaimedToolReplay(5L, "pending-42", "alice"));
}
@Test
@DisplayName("an expired replay lease becomes uncertain instead of an ordinary retryable stale task")
void expiredReplayLeaseRequiresManualResolution() {
TeamTaskEntity replay = runTask(5L, TeamTaskStatus.IN_PROGRESS);
replay.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayInProgress\":true}}");
when(taskMapper.selectList(any())).thenReturn(List.of(replay));
when(taskMapper.update(isNull(), any())).thenReturn(1);
service.recoverStaleTasks();
ArgumentCaptor<LambdaUpdateWrapper<TeamTaskEntity>> captor =
ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
verify(taskMapper).update(isNull(), captor.capture());
var values = captor.getValue().getParamNameValuePairs().values();
assertTrue(values.contains(TeamTaskStatus.AWAITING_APPROVAL));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value)
.contains("replayOutcomeUncertain")));
assertFalse(values.contains(TeamTaskStatus.STALE));
}
@Test
@DisplayName("a replay exception is immediately parked as outcome-uncertain")
void replayExceptionCannotBeAutomaticallyRetried() {
TeamTaskEntity replay = runTask(5L, TeamTaskStatus.IN_PROGRESS);
replay.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayInProgress\":true}}");
when(taskMapper.selectById(5L)).thenReturn(replay);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.parkToolReplayUncertain(
5L, "pending-42", "provider failed; outcome uncertain"));
ArgumentCaptor<LambdaUpdateWrapper<TeamTaskEntity>> captor =
ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
verify(taskMapper).update(isNull(), captor.capture());
var values = captor.getValue().getParamNameValuePairs().values();
assertTrue(values.contains(TeamTaskStatus.AWAITING_APPROVAL));
assertTrue(values.stream().anyMatch(value -> String.valueOf(value)
.contains("replayOutcomeUncertain")));
}
@Test
@DisplayName("feedback reopens a settled worker task but not an active or approval-blocked task")
void feedbackResumeHasExplicitStateGuard() {
TeamTaskEntity completed = runTask(5L, TeamTaskStatus.COMPLETED);
completed.setAssigneeAgentId(MEMBER_ID);
when(taskMapper.selectById(5L)).thenReturn(completed);
when(taskMapper.update(isNull(), any())).thenReturn(1);
assertTrue(service.resumeForWorkerFeedback(5L));
completed.setStatus(TeamTaskStatus.AWAITING_APPROVAL);
assertThrows(IllegalStateException.class, () -> service.resumeForWorkerFeedback(5L));
}
// ==================== blocker comment ====================
@Test

284
mateclaw-server/src/test/java/vip/mate/team/service/TeamWorkerInterventionServiceTest.java Normal file
View File

@ -0,0 +1,284 @@
package vip.mate.team.service;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import vip.mate.agent.AgentService;
import vip.mate.agent.context.ChatOrigin;
import vip.mate.agent.runtime.ConversationTurnGate;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.approval.PendingApproval;
import vip.mate.approval.ResolveOutcome;
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.Optional;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class TeamWorkerInterventionServiceTest {
private final TeamTaskService taskService = mock(TeamTaskService.class);
private final TeamWorkerConversationGovernanceService governance =
mock(TeamWorkerConversationGovernanceService.class);
private final ApprovalWorkflowService approvalService = mock(ApprovalWorkflowService.class);
private final AgentService agentService = mock(AgentService.class);
private final ConversationService conversationService = mock(ConversationService.class);
private final ChatStreamTracker streamTracker = mock(ChatStreamTracker.class);
private final TeamDispatchService dispatchService = mock(TeamDispatchService.class);
private final TeamAnnounceService announceService = mock(TeamAnnounceService.class);
private final TeamEventChannel eventChannel = mock(TeamEventChannel.class);
private final TeamWorkerReplayPersistenceService replayPersistenceService =
mock(TeamWorkerReplayPersistenceService.class);
private TeamWorkerInterventionService service;
@BeforeEach
void setUp() {
service = new TeamWorkerInterventionService(taskService, governance, approvalService,
agentService, conversationService, new ConversationTurnGate(), streamTracker,
dispatchService, announceService, eventChannel, replayPersistenceService);
}
@Test
void approvalReplaysInCanonicalConversationAndSettlesOriginalTask() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
PendingApproval pending = pending("pending-42");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(pending));
when(approvalService.claimForReplay("pending-42", "alice"))
.thenReturn(ResolveOutcome.resolved(pending, "approved", true, 1));
when(approvalService.getReplayClaim("pending-42")).thenReturn(Optional.of(pending));
when(approvalService.consumeReplayClaim("pending-42", "alice"))
.thenReturn(ResolveOutcome.consumed(pending, true, 1));
when(taskService.resumeAfterToolApproval(101L, "pending-42")).thenReturn(true);
when(taskService.stageToolReplayResult(101L, "pending-42", "tool completed")).thenReturn(true);
when(approvalService.restoreChatOrigin(null)).thenReturn(ChatOrigin.EMPTY);
when(agentService.chatWithReplayWithUsage(eq(201L), any(), eq("worker-101"),
eq("{\"name\":\"shell\"}"), eq(ChatOrigin.EMPTY)))
.thenReturn(AgentService.ChatResult.contentOnly("tool completed"));
service.approve(7L, 101L, "pending-42", "alice");
verify(conversationService).removeApprovalPlaceholders("worker-101");
verify(replayPersistenceService).persist(101L, "pending-42", "worker-101",
"tool completed", AgentService.ChatResult.contentOnly("tool completed"));
verify(dispatchService).settleOutcome(task, "tool completed");
verify(approvalService).consumeReplayClaim("pending-42", "alice");
}
@Test
void canonicalLinkMismatchRejectsBeforeApprovalMutation() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.empty());
IllegalArgumentException error = assertThrows(IllegalArgumentException.class,
() -> service.approve(7L, 101L, "pending-42", "alice"));
assertTrue(error.getMessage().contains("worker conversation"));
verify(approvalService, never()).consumeReplayClaim(any(), any());
}
@Test
void denialSettlesWithoutExecutingTheTool() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
PendingApproval pending = pending("pending-42");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(pending));
when(approvalService.resolve("pending-42", "alice", "denied"))
.thenReturn(ResolveOutcome.resolved(pending, "denied", true, 1));
when(taskService.denyToolApproval(101L, "pending-42", "alice")).thenReturn(true);
service.deny(7L, 101L, "pending-42", "alice");
verify(agentService, never()).chatWithReplayWithUsage(any(), any(), any(), any(), any());
verify(taskService).denyToolApproval(101L, "pending-42", "alice");
verify(announceService).announceTaskSettled(task);
}
@Test
void feedbackContinuesOriginalConversationAndCannotBypassPendingApproval() {
TeamTaskEntity completed = task(TeamTaskStatus.COMPLETED);
when(taskService.getTask(101L)).thenReturn(completed);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(taskService.resumeForWorkerFeedback(101L)).thenReturn(true);
when(agentService.chatWithUsage(eq(201L), eq("tighten the summary"), eq("worker-101"), any()))
.thenReturn(AgentService.ChatResult.contentOnly("revised summary"));
service.feedback(7L, 101L, "tighten the summary", "alice");
verify(conversationService).saveMessage("worker-101", "user", "tighten the summary");
verify(conversationService).saveMessage("worker-101", "assistant", "revised summary",
null, "completed", 0, 0, null, null);
verify(dispatchService).settleOutcome(completed, "revised summary");
when(approvalService.findPendingByConversation("worker-101"))
.thenReturn(pending("pending-next"));
assertThrows(IllegalStateException.class,
() -> service.feedback(7L, 101L, "run another command", "alice"));
}
@Test
void replayFailureReparksApprovedPayloadForSafeRetry() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
PendingApproval pending = pending("pending-42");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(pending));
when(approvalService.claimForReplay("pending-42", "alice"))
.thenReturn(ResolveOutcome.resolved(pending, "approved", true, 1));
when(approvalService.getReplayClaim("pending-42")).thenReturn(Optional.of(pending));
when(taskService.resumeAfterToolApproval(101L, "pending-42")).thenReturn(true);
when(approvalService.restoreChatOrigin(null)).thenReturn(ChatOrigin.EMPTY);
when(agentService.chatWithReplayWithUsage(any(), any(), any(), any(), any()))
.thenThrow(new IllegalStateException("provider timeout"));
assertThrows(IllegalStateException.class,
() -> service.approve(7L, 101L, "pending-42", "alice"));
verify(taskService).parkToolReplayUncertain(eq(101L), eq("pending-42"),
org.mockito.ArgumentMatchers.contains("provider timeout"));
verify(taskService, never()).failTask(eq(101L), any());
verify(approvalService, never()).consumeReplayClaim(any(), any());
}
@Test
void onlyTheInstanceHoldingTheTaskReplayLeaseExecutesTheTool() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
PendingApproval pending = pending("pending-42");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(pending));
when(approvalService.claimForReplay("pending-42", "alice"))
.thenReturn(ResolveOutcome.resolved(pending, "approved", true, 1));
when(approvalService.getReplayClaim("pending-42")).thenReturn(Optional.of(pending));
when(taskService.resumeAfterToolApproval(101L, "pending-42")).thenReturn(false);
assertThrows(IllegalStateException.class,
() -> service.approve(7L, 101L, "pending-42", "alice"));
verify(agentService, never()).chatWithReplayWithUsage(any(), any(), any(), any(), any());
}
@Test
void uncertainReplayOutcomeCannotBeExecutedAgainAutomatically() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(taskService.isToolReplayOutcomeUncertain(task)).thenReturn(true);
assertThrows(IllegalStateException.class,
() -> service.approve(7L, 101L, "pending-42", "alice"));
verify(approvalService, never()).claimForReplay(any(), any());
verify(agentService, never()).chatWithReplayWithUsage(any(), any(), any(), any(), any());
}
@Test
void stagedReplayFinalizationDoesNotExecuteToolAgain() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
task.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayResult\":\"tool completed\",\"messagePersisted\":true}}");
PendingApproval pending = pending("pending-42");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(pending));
when(taskService.stagedToolReplayResult(task)).thenReturn("tool completed");
when(taskService.isToolReplayMessagePersisted(task)).thenReturn(true);
when(approvalService.consumeReplayClaim("pending-42", "alice"))
.thenReturn(ResolveOutcome.consumed(pending, true, 1));
when(taskService.resumeAfterToolApproval(101L, "pending-42")).thenReturn(true);
service.approve(7L, 101L, "pending-42", "alice");
verify(agentService, never()).chatWithReplayWithUsage(any(), any(), any(), any(), any());
verify(conversationService, never()).removeApprovalPlaceholders(anyString());
verify(dispatchService).settleOutcome(task, "tool completed");
}
@Test
void stagedReplayCannotBeReportedAsDeniedAfterToolExecution() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
task.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\","
+ "\"replayResult\":\"tool completed\"}}");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(taskService.stagedToolReplayResult(task)).thenReturn("tool completed");
assertThrows(IllegalStateException.class,
() -> service.deny(7L, 101L, "pending-42", "alice"));
verify(approvalService, never()).resolve(any(), any(), anyString());
verify(taskService, never()).denyToolApproval(any(), any(), any());
}
@Test
void claimedReplayCanBeStoppedAfterExecutionFailure() {
TeamTaskEntity task = task(TeamTaskStatus.AWAITING_APPROVAL);
PendingApproval claimed = pending("pending-42");
claimed.setStatus("approved");
when(taskService.getTask(101L)).thenReturn(task);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
when(approvalService.getPending("pending-42")).thenReturn(Optional.of(claimed));
when(approvalService.consumeReplayClaim("pending-42", "alice"))
.thenReturn(ResolveOutcome.consumed(claimed, true, 1));
when(taskService.abortClaimedToolReplay(101L, "pending-42", "alice"))
.thenReturn(true);
service.deny(7L, 101L, "pending-42", "alice");
verify(taskService).abortClaimedToolReplay(101L, "pending-42", "alice");
verify(taskService, never()).denyToolApproval(any(), any(), any());
verify(eventChannel).publishTaskEvent(task, "team_task_tool_replay_aborted",
java.util.Map.of("pendingId", "pending-42"));
}
@Test
void duplicateDecisionReturnsCurrentTaskProjection() {
TeamTaskEntity completed = task(TeamTaskStatus.COMPLETED);
when(taskService.getTask(101L)).thenReturn(completed);
when(governance.resolve("worker-101", 11L, 101L)).thenReturn(Optional.of(context()));
service.approve(7L, 101L, "pending-42", "alice");
service.deny(7L, 101L, "pending-42", "alice");
verify(approvalService, never()).getPending(any());
}
private static TeamTaskEntity task(String status) {
TeamTaskEntity task = new TeamTaskEntity();
task.setId(101L);
task.setTeamId(7L);
task.setRunId(11L);
task.setTaskNumber(3);
task.setStatus(status);
task.setAssigneeAgentId(201L);
task.setConversationId("worker-101");
task.setMetadata("{\"toolApproval\":{\"pendingId\":\"pending-42\"}}");
return task;
}
private static TeamWorkerConversationContext context() {
return new TeamWorkerConversationContext(true, "team_worker", "worker-101",
11L, 101L, 7L, "lead-11", 201L);
}
private static PendingApproval pending(String id) {
PendingApproval pending = new PendingApproval(id, "worker-101", "owner",
"shell", "{}", "needs approval");
pending.setAgentId("201");
pending.setToolCallPayload("{\"name\":\"shell\"}");
return pending;
}
}

30
mateclaw-server/src/test/java/vip/mate/team/service/TeamWorkerReplayPersistenceServiceTest.java Normal file
View File

@ -0,0 +1,30 @@
package vip.mate.team.service;
import org.junit.jupiter.api.Test;
import vip.mate.agent.AgentService;
import vip.mate.workspace.conversation.ConversationService;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class TeamWorkerReplayPersistenceServiceTest {
@Test
void messageAndIdempotencyMarkerMustBothSucceed() {
ConversationService conversations = mock(ConversationService.class);
TeamTaskService tasks = mock(TeamTaskService.class);
TeamWorkerReplayPersistenceService service =
new TeamWorkerReplayPersistenceService(conversations, tasks);
AgentService.ChatResult result = AgentService.ChatResult.contentOnly("done");
when(tasks.markToolReplayMessagePersisted(101L, "pending-42")).thenReturn(false);
assertThrows(IllegalStateException.class,
() -> service.persist(101L, "pending-42", "worker-101", "done", result));
verify(conversations).saveMessage("worker-101", "assistant", "done",
null, "completed", 0, 0, null, null);
verify(tasks).markToolReplayMessagePersisted(101L, "pending-42");
}
}

6
mateclaw-ui/src/api/index.ts
View File

@ -1013,6 +1013,12 @@ export const teamApi = {
) => http.post(`/teams/${id}/tasks`, data),
listTaskEvents: (id: string, taskId: string) => http.get(`/teams/${id}/tasks/${taskId}/events`),
approveTask: (id: string, taskId: string) => http.post(`/teams/${id}/tasks/${taskId}/approve`),
approveWorkerTool: (id: string, taskId: string, pendingId: string) =>
http.post(`/teams/${id}/tasks/${taskId}/worker/approve`, { pendingId }),
denyWorkerTool: (id: string, taskId: string, pendingId: string) =>
http.post(`/teams/${id}/tasks/${taskId}/worker/deny`, { pendingId }),
feedbackWorker: (id: string, taskId: string, message: string) =>
http.post(`/teams/${id}/tasks/${taskId}/worker/feedback`, { message }),
rejectTask: (id: string, taskId: string, reason?: string) =>
http.post(`/teams/${id}/tasks/${taskId}/reject`, { reason }),
retryTask: (id: string, taskId: string) => http.post(`/teams/${id}/tasks/${taskId}/retry`),

25
mateclaw-ui/src/components/team-run/TeamRunAttention.vue
View File

@ -15,13 +15,17 @@ const emit = defineEmits<{
'view-task': [taskId: string]
'retry-task': [taskId: string]
'approve-task': [taskId: string]
'approve-tool': [taskId: string]
'deny-tool': [taskId: string]
}>()
const { t } = useI18n()
const items = computed(() => runAttention(props.run))
const retryable = (type: string) => ['failed', 'failure', 'stale'].includes(type.toLowerCase())
const reviewable = (type: string) => ['review', 'in_review'].includes(type.toLowerCase())
const isPending = (taskId: string, action: 'retry' | 'approve') => props.pendingActions.includes(`${taskId}:${action}`)
const toolApproval = (type: string) => ['approval', 'replay_uncertain'].includes(type.toLowerCase())
const toolApprovalCanExecute = (type: string) => type.toLowerCase() === 'approval'
const isPending = (taskId: string, action: 'retry' | 'approve' | 'approve-tool' | 'deny-tool') => props.pendingActions.includes(`${taskId}:${action}`)
</script>
<template>
@ -62,6 +66,24 @@ const isPending = (taskId: string, action: 'retry' | 'approve') => props.pending
:aria-busy="isPending(item.taskId, 'approve')"
@click="emit('approve-task', item.taskId)"
>{{ t('common.approve') }}</button>
<button
v-if="toolApprovalCanExecute(item.type)"
type="button"
class="is-primary"
:data-attention-approve-tool="item.taskId"
:disabled="isPending(item.taskId, 'approve-tool') || isPending(item.taskId, 'deny-tool')"
:aria-busy="isPending(item.taskId, 'approve-tool')"
@click="emit('approve-tool', item.taskId)"
>{{ t('common.approve') }}</button>
<button
v-if="toolApproval(item.type)"
type="button"
class="is-danger"
:data-attention-deny-tool="item.taskId"
:disabled="isPending(item.taskId, 'approve-tool') || isPending(item.taskId, 'deny-tool')"
:aria-busy="isPending(item.taskId, 'deny-tool')"
@click="emit('deny-tool', item.taskId)"
>{{ t('common.reject') }}</button>
</div>
</li>
</ul>
@ -79,6 +101,7 @@ li.is-error { border-left-color: #c13d3d; background: #fff5f5; }
.run-attention__actions { display: flex; min-width: 0; flex: none; flex-wrap: wrap; justify-content: flex-end; gap: 5px; }
.run-attention__actions button { min-height: 28px; padding: 4px 8px; border: 1px solid var(--mc-border, #d9e1e7); border-radius: 5px; background: #fff; color: var(--mc-text-secondary, #475569); cursor: pointer; font: inherit; white-space: nowrap; }
.run-attention__actions button.is-primary { border-color: #16835b; color: #126c4d; }
.run-attention__actions button.is-danger { border-color: #d58d8d; color: #a82929; }
.run-attention__actions button:disabled { cursor: wait; opacity: 0.6; }
.run-attention__actions button:focus-visible { outline: 2px solid #16835b; outline-offset: 2px; }
p { margin: 0; color: var(--mc-text-tertiary); font-size: 12px; }

43
mateclaw-ui/src/components/team-run/TeamRunDetail.vue
View File

@ -33,17 +33,34 @@ const emit = defineEmits<{
'view-task': [taskId: string]
'retry-task': [taskId: string]
'approve-task': [taskId: string]
'approve-tool': [taskId: string]
'deny-tool': [taskId: string]
'feedback-task': [taskId: string, message: string]
}>()
const { t } = useI18n()
const { renderMarkdown } = useMarkdownRenderer()
const localTaskId = ref<string | null>(props.selectedTaskId)
const selectedTaskSection = ref<HTMLElement | null>(null)
const workerFeedback = ref('')
watch(() => props.selectedTaskId, value => { localTaskId.value = value })
watch(localTaskId, () => { workerFeedback.value = '' })
const selectedTask = computed(() => props.run.tasks.find(task => task.id === localTaskId.value) ?? null)
const terminal = computed(() => ['completed', 'partial', 'failed', 'cancelled'].includes(props.run.status))
const renderedTaskDescription = computed(() => renderMarkdown(selectedTask.value?.description || ''))
const renderedTaskResult = computed(() => renderMarkdown(selectedTask.value?.result || ''))
const canSendFeedback = computed(() => props.managementActions
&& Boolean(selectedTask.value?.conversationId)
&& !['pending', 'blocked', 'in_progress', 'awaiting_approval', 'cancelled'].includes(selectedTask.value?.status ?? ''))
const feedbackPending = computed(() => selectedTask.value
? props.pendingActions.includes(`${selectedTask.value.id}:feedback`)
: false)
function submitFeedback() {
const message = workerFeedback.value.trim()
if (!selectedTask.value || !message || feedbackPending.value) return
emit('feedback-task', selectedTask.value.id, message)
}
function selectTask(task: TeamRunTask) {
localTaskId.value = task.id
@ -73,6 +90,8 @@ async function viewAttentionTask(taskId: string) {
@view-task="viewAttentionTask"
@retry-task="emit('retry-task', $event)"
@approve-task="emit('approve-task', $event)"
@approve-tool="emit('approve-tool', $event)"
@deny-tool="emit('deny-tool', $event)"
/>
<TeamRunOutcome :run="run" />
<TeamRunDeliverables :run="run" />
@ -120,6 +139,25 @@ async function viewAttentionTask(taskId: string) {
<dd v-else class="run-detail__result">{{ t('teamRuns.noResult') }}</dd>
</div>
</dl>
<div v-if="canSendFeedback" class="run-detail__feedback">
<label :for="`worker-feedback-${selectedTask.id}`">{{ t('teamRuns.workerFeedback') }}</label>
<textarea
:id="`worker-feedback-${selectedTask.id}`"
v-model="workerFeedback"
data-team-run-worker-feedback
maxlength="4000"
rows="3"
:placeholder="t('teamRuns.workerFeedbackPlaceholder')"
:disabled="feedbackPending"
/>
<button
type="button"
data-team-run-worker-feedback-submit
:disabled="!workerFeedback.trim() || feedbackPending"
:aria-busy="feedbackPending"
@click="submitFeedback"
>{{ t('teamRuns.sendFeedback') }}</button>
</div>
</section>
<footer v-if="canCancel && !terminal" class="run-detail__actions">
@ -152,6 +190,11 @@ async function viewAttentionTask(taskId: string) {
.run-detail__link-button { padding: 0; border: 0; background: transparent; color: #16795a; cursor: pointer; font-size: 12px; }
.run-detail__link-button:focus-visible, .run-detail__cancel:focus-visible, .run-detail__task-detail:focus-visible { outline: 2px solid #1b8f68; outline-offset: -2px; }
.run-detail__result { white-space: pre-wrap; }
.run-detail__feedback { display: grid; gap: 7px; margin-top: 14px; padding-top: 12px; border-top: 1px solid var(--mc-border-light); }
.run-detail__feedback label { color: var(--mc-text-secondary); font-size: 12px; font-weight: 600; }
.run-detail__feedback textarea { box-sizing: border-box; width: 100%; padding: 8px; border: 1px solid var(--mc-border); border-radius: 6px; resize: vertical; color: var(--mc-text-primary); font: inherit; }
.run-detail__feedback button { justify-self: end; min-height: 30px; padding: 5px 10px; border: 1px solid #16835b; border-radius: 6px; background: transparent; color: #126c4d; cursor: pointer; }
.run-detail__feedback button:disabled { cursor: wait; opacity: .55; }
.run-detail__actions { display: flex; justify-content: flex-end; padding: 10px 16px 14px; border-top: 1px solid var(--mc-border-light, #e7ebef); }
.run-detail__cancel { display: inline-flex; align-items: center; gap: 6px; min-height: 30px; padding: 5px 10px; border: 1px solid #e6b7b7; border-radius: 6px; background: transparent; color: #b53535; cursor: pointer; font-size: 12px; letter-spacing: 0; }
.run-detail__cancel:hover { background: rgba(193, 61, 61, 0.06); }

6
mateclaw-ui/src/components/team-run/TeamRunDrawer.vue
View File

@ -27,6 +27,9 @@ const emit = defineEmits<{
'view-task': [taskId: string]
'retry-task': [taskId: string]
'approve-task': [taskId: string]
'approve-tool': [taskId: string]
'deny-tool': [taskId: string]
'feedback-task': [taskId: string, message: string]
}>()
const { t } = useI18n()
const closeButton = ref<HTMLButtonElement | null>(null)
@ -97,6 +100,9 @@ onBeforeUnmount(() => document.removeEventListener('keydown', onKeydown))
@view-task="emit('view-task', $event)"
@retry-task="emit('retry-task', $event)"
@approve-task="emit('approve-task', $event)"
@approve-tool="emit('approve-tool', $event)"
@deny-tool="emit('deny-tool', $event)"
@feedback-task="(taskId, message) => emit('feedback-task', taskId, message)"
/>
</aside>
</div>

4
mateclaw-ui/src/components/team-run/TeamRunTaskList.vue
View File

@ -27,7 +27,7 @@ function iconFor(status: string) {
if (status === 'completed') return CircleCheckFilled
if (status === 'failed') return CircleCloseFilled
if (status === 'cancelled' || status === 'stale') return RemoveFilled
if (status === 'in_review') return View
if (status === 'in_review' || status === 'awaiting_approval') return View
if (status === 'in_progress') return Loading
if (status === 'blocked') return Lock
if (status === 'pending') return Clock
@ -102,7 +102,7 @@ function preview(value: string | null) {
.run-task-row:focus-visible { outline: 2px solid #1b8f68; outline-offset: -2px; }
.run-task-row__state { padding-top: 2px; color: #64748b; }
.run-task-row__state.is-completed { color: #16835b; }
.run-task-row__state.is-in_review, .run-task-row__state.is-blocked { color: #a15c05; }
.run-task-row__state.is-in_review, .run-task-row__state.is-awaiting_approval, .run-task-row__state.is-blocked { color: #a15c05; }
.run-task-row__state.is-failed { color: #c13d3d; }
.run-task-row__body { min-width: 0; display: grid; gap: 5px; }
.run-task-row__title { display: flex; justify-content: space-between; gap: 12px; font-size: 13px; font-weight: 600; }

22
mateclaw-ui/src/components/team-run/__tests__/teamRunComponents.test.ts
View File

@ -107,6 +107,28 @@ describe('TeamRunCard', () => {
})
describe('TeamRunDetail', () => {
it('submits a bounded follow-up instruction for the selected worker task', async () => {
const feedback: Array<[string, string]> = []
const task = {
id: '101', teamId: '10', runId: '20', taskNumber: 1, subject: 'Draft report', description: null,
status: 'completed', priority: 0, taskType: 'execution', assigneeAgentId: '31', ownerAgentId: null,
blockedBy: null, requireApproval: false, progressPercent: 100, progressStep: null,
result: 'Draft', reason: null, conversationId: 'worker-1', metadata: null, createTime: null, updateTime: null,
}
const host = mount(TeamRunDetail, {
run: sampleRun({ tasks: [task] }), selectedTaskId: '101', managementActions: true,
onFeedbackTask: (taskId: string, message: string) => feedback.push([taskId, message]),
})
const textarea = host.querySelector<HTMLTextAreaElement>('[data-team-run-worker-feedback]')!
textarea.value = ' tighten the conclusion '
textarea.dispatchEvent(new Event('input', { bubbles: true }))
await nextTick()
host.querySelector<HTMLButtonElement>('[data-team-run-worker-feedback-submit]')!.click()
await nextTick()
expect(feedback).toEqual([['101', 'tighten the conclusion']])
})
it('forwards attention recovery actions only in management context', async () => {
const actions: string[] = []
const attentionItems = [{ id: 'a', type: 'failed', severity: 'error', priority: 1, taskId: '101', message: 'Failed', createdAt: null }]

28
mateclaw-ui/src/components/team-run/__tests__/teamRunProjectionPrimitives.test.ts
View File

@ -8,6 +8,7 @@ import TeamRunAttention from '../TeamRunAttention.vue'
import TeamRunContributions from '../TeamRunContributions.vue'
import TeamRunRuntime from '../TeamRunRuntime.vue'
import TeamRunCard from '../TeamRunCard.vue'
import { captureTeamAttentionContext } from '../teamRunAttentionHandlers'
const messages = { teamRuns: {
outcome: 'Outcome', noSummary: 'No summary', deliverables: 'Deliverables', noDeliverables: 'No deliverables',
@ -45,6 +46,14 @@ function mount(component: Component, props: Record<string, unknown>) {
afterEach(() => { apps.splice(0).forEach(app => app.unmount()); document.body.innerHTML = '' })
describe('Team Run projection primitives', () => {
it('normalizes numeric API ids before dispatching management actions', () => {
expect(captureTeamAttentionContext(
{ id: 20, teamId: 10, tasks: [{ id: 101 }] },
'10',
'101',
)).toEqual({ teamId: '10', runId: '20', taskId: '101' })
})
it('filters unsafe canonical deliverables before rendering', () => {
const value = run({ deliverables: [
{ id: 'safe', name: 'Safe', url: '/api/v1/files/generated/safe.pdf', type: 'pdf', sourceTaskIds: [], sourceAgentIds: [], createdAt: null, verificationStatus: 'available' },
@ -104,6 +113,25 @@ describe('Team Run projection primitives', () => {
expect(actions).toEqual(['view:1', 'retry:1', 'retry:2', 'approve:3', 'view:4'])
})
it('offers approve and deny controls for worker tool approval attention', async () => {
const actions: string[] = []
const value = run({ attentionItems: [
{ id: 'approval', type: 'approval', severity: 'action', priority: 0, taskId: '9', message: 'Run shell command', createdAt: null },
] })
const host = mount(TeamRunAttention, {
run: value,
managementActions: true,
onApproveTool: (id: string) => actions.push(`approve:${id}`),
onDenyTool: (id: string) => actions.push(`deny:${id}`),
})
host.querySelector<HTMLButtonElement>('[data-attention-approve-tool="9"]')!.click()
host.querySelector<HTMLButtonElement>('[data-attention-deny-tool="9"]')!.click()
await nextTick()
expect(actions).toEqual(['approve:9', 'deny:9'])
})
it('keeps shared attention cards read-only outside Teams management context', () => {
const host = mount(TeamRunAttention, { run: run() })
expect(host.querySelector('[data-team-run-attention-actions]')).toBeNull()

15
mateclaw-ui/src/components/team-run/teamRunAttentionHandlers.ts
View File

@ -1,6 +1,6 @@
import type { WorkspaceRole } from '@/composables/capabilities'
export type TeamAttentionAction = 'approve' | 'retry'
export type TeamAttentionAction = 'approve' | 'retry' | 'approve-tool' | 'deny-tool' | 'feedback'
export interface TeamAttentionActionContext {
teamId: string
@ -8,6 +8,19 @@ export interface TeamAttentionActionContext {
taskId: string
}
export function captureTeamAttentionContext(
run: { id: string | number; teamId: string | number; tasks: Array<{ id: string | number }> } | null,
currentTeamId: string | number | null,
taskId: string | number,
): TeamAttentionActionContext | null {
if (!run || currentTeamId === null) return null
const teamId = String(currentTeamId)
const normalizedTaskId = String(taskId)
if (String(run.teamId) !== teamId
|| !run.tasks.some(task => String(task.id) === normalizedTaskId)) return null
return { teamId, runId: String(run.id), taskId: normalizedTaskId }
}
export function canManageTeamRunAttention(
role: WorkspaceRole | null,
currentWorkspaceId: string | null,

9
mateclaw-ui/src/i18n/locales/en-US.ts
View File

@ -65,6 +65,7 @@ export default {
deliverable: 'Deliverable',
completed: 'Completed',
in_review: 'In review',
awaiting_approval: 'Awaiting tool approval',
failed: 'Failed',
cancelled: 'Cancelled',
approved: 'Approved',
@ -84,6 +85,7 @@ export default {
pending: 'Pending',
blocked: 'Blocked',
in_progress: 'In Progress',
awaiting_approval: 'Awaiting Tool Approval',
in_review: 'In Review',
completed: 'Completed',
failed: 'Failed',
@ -139,6 +141,13 @@ export default {
expand: 'Expand run',
collapse: 'Collapse run',
openTask: 'Open task',
workerFeedback: 'Follow-up instruction',
workerFeedbackPlaceholder: 'Tell this worker what to revise or continue…',
sendFeedback: 'Send to worker',
approvalExpired: 'This tool approval is no longer current. The run has been refreshed.',
toolApproved: 'Tool approved; worker resumed',
toolDenied: 'Tool request denied',
feedbackSent: 'Follow-up sent to worker',
objective: 'Objective',
taskProgress: 'Task progress',
stopReason: 'Stop reason',

9
mateclaw-ui/src/i18n/locales/zh-CN.ts
View File

@ -65,6 +65,7 @@ export default {
deliverable: '交付物',
completed: '完成',
in_review: '待审核',
awaiting_approval: '等待工具审批',
failed: '失败',
cancelled: '已取消',
approved: '批准',
@ -84,6 +85,7 @@ export default {
pending: '待处理',
blocked: '等待依赖',
in_progress: '进行中',
awaiting_approval: '等待工具审批',
in_review: '待审核',
completed: '已完成',
failed: '失败',
@ -139,6 +141,13 @@ export default {
expand: '展开运行',
collapse: '收起运行',
openTask: '打开任务',
workerFeedback: '补充指令',
workerFeedbackPlaceholder: '告诉该成员需要修改或继续处理的内容…',
sendFeedback: '发送给成员',
approvalExpired: '该工具审批已失效，运行状态已刷新。',
toolApproved: '已批准工具调用，成员已恢复执行',
toolDenied: '已拒绝工具调用',
feedbackSent: '补充指令已发送',
objective: '目标',
taskProgress: '任务进度',
stopReason: '停止原因',

2
mateclaw-ui/src/stores/useTeamStore.ts
View File

@ -24,7 +24,7 @@ export const useTeamStore = defineStore('team', () => {
const taskRunId = ref<string | null>(null)
/** Statuses that mean the board is still moving and worth polling. */
const ACTIVE_STATUSES = ['pending', 'in_progress', 'in_review', 'blocked']
const ACTIVE_STATUSES = ['pending', 'in_progress', 'awaiting_approval', 'in_review', 'blocked']
const COMPLETED_STATUSES = ['completed']
const CLOSED_STATUSES = ['failed', 'cancelled', 'stale']
/** Terminal-column page size. */

79
mateclaw-ui/src/views/Teams.vue
View File

@ -243,11 +243,14 @@
:pending-actions="attentionPendingActions"
@close="closeRun"
@cancel="cancelRun"
@select-task="openRunTask"
@select-task="selectRunTask"
@navigate="router.push"
@view-task="openAttentionTask"
@retry-task="retryAttentionTask"
@approve-task="approveAttentionTask"
@approve-tool="approveWorkerTool"
@deny-tool="denyWorkerTool"
@feedback-task="feedbackWorker"
@retry-detail="runHistory.ensureSelectedRunDetail(runHistory.selectedRunId.value!, runHistory.selectedTaskId.value)"
/>
@ -601,6 +604,7 @@ import TeamRunDrawer from '@/components/team-run/TeamRunDrawer.vue'
import TeamRunsPanel from '@/components/team-run/TeamRunsPanel.vue'
import {
canManageTeamRunAttention,
captureTeamAttentionContext,
refreshAttentionTaskContext,
runAttentionTaskAction,
type TeamAttentionAction,
@ -657,7 +661,7 @@ let routeReconciliationRevision = 0
const COLUMN_DEFS = [
{ key: 'todo', statuses: ['pending', 'blocked'], terminal: false },
{ key: 'in_progress', statuses: ['in_progress'], terminal: false },
{ key: 'in_review', statuses: ['in_review'], terminal: false },
{ key: 'in_review', statuses: ['awaiting_approval', 'in_review'], terminal: false },
{ key: 'completed', statuses: ['completed'], terminal: true },
{ key: 'closed', statuses: ['failed', 'cancelled', 'stale'], terminal: true },
] as const
@ -1226,8 +1230,12 @@ async function openRunTask(
}
}
function selectRunTask(task: TeamRunTask) {
runHistory.select(String(task.runId), String(task.id))
}
function selectedRunTask(taskId: string) {
return runHistory.selectedRun.value?.tasks.find(task => task.id === taskId) ?? null
return runHistory.selectedRun.value?.tasks.find(task => String(task.id) === String(taskId)) ?? null
}
async function openAttentionTask(taskId: string) {
@ -1236,10 +1244,11 @@ async function openAttentionTask(taskId: string) {
}
function captureAttentionContext(taskId: string): TeamAttentionActionContext | null {
const run = runHistory.selectedRun.value
const teamId = String(store.currentTeam?.team.id ?? '')
if (!run || !teamId || run.teamId !== teamId || !run.tasks.some(task => task.id === taskId)) return null
return { teamId, runId: run.id, taskId }
return captureTeamAttentionContext(
runHistory.selectedRun.value,
store.currentTeam?.team.id ?? null,
taskId,
)
}
async function refreshAfterTaskAction(context: TeamAttentionActionContext) {
@ -1270,6 +1279,46 @@ async function performAttentionAction(taskId: string, action: TeamAttentionActio
})
}
function pendingApprovalId(taskId: string): string | null {
const metadata = selectedRunTask(taskId)?.metadata
if (!metadata) return null
try {
const value = JSON.parse(metadata)?.toolApproval?.pendingId
return typeof value === 'string' && value ? value : null
} catch {
return null
}
}
async function performWorkerAction(
taskId: string,
action: 'approve-tool' | 'deny-tool' | 'feedback',
message?: string,
) {
const context = captureAttentionContext(taskId)
if (!context || !canManageSelectedRun.value) return false
const pendingId = action === 'feedback' ? null : pendingApprovalId(taskId)
if (action !== 'feedback' && !pendingId) {
ElMessage.error(t('teamRuns.approvalExpired'))
await refreshAfterTaskAction(context)
return false
}
return runAttentionTaskAction({
context,
action,
pending: pendingAttentionActions,
execute: () => action === 'approve-tool'
? teamApi.approveWorkerTool(context.teamId, context.taskId, pendingId!)
: action === 'deny-tool'
? teamApi.denyWorkerTool(context.teamId, context.taskId, pendingId!)
: teamApi.feedbackWorker(context.teamId, context.taskId, message ?? ''),
refresh: () => refreshAfterTaskAction(context),
onError: cause => ElMessage.error(
cause instanceof Error && cause.message ? cause.message : t('teams.actionFailed', 'Operation failed'),
),
})
}
async function approveTaskById(taskId: string) {
if (!store.currentTeam) return
await teamApi.approveTask(store.currentTeam.team.id, taskId)
@ -1291,6 +1340,18 @@ async function retryAttentionTask(taskId: string) {
await performAttentionAction(taskId, 'retry')
}
async function approveWorkerTool(taskId: string) {
if (await performWorkerAction(taskId, 'approve-tool')) ElMessage.success(t('teamRuns.toolApproved'))
}
async function denyWorkerTool(taskId: string) {
if (await performWorkerAction(taskId, 'deny-tool')) ElMessage.success(t('teamRuns.toolDenied'))
}
async function feedbackWorker(taskId: string, message: string) {
if (await performWorkerAction(taskId, 'feedback', message)) ElMessage.success(t('teamRuns.feedbackSent'))
}
async function openTask(vo: TeamTaskVO, shouldApply: () => boolean = () => true) {
if (!store.currentTeam) return
try {
@ -1577,7 +1638,7 @@ async function cancelTask() {
}
.pill--completed { color: #16a34a; background: rgba(22, 163, 74, 0.08); border-color: rgba(22, 163, 74, 0.25); }
.pill--in_progress { color: var(--mc-primary); background: var(--mc-primary-bg); border-color: var(--mc-primary); }
.pill--in_review { color: #d97706; background: rgba(217, 119, 6, 0.08); border-color: rgba(217, 119, 6, 0.3); }
.pill--in_review, .pill--awaiting_approval { color: #d97706; background: rgba(217, 119, 6, 0.08); border-color: rgba(217, 119, 6, 0.3); }
.pill--failed, .pill--cancelled { color: #dc2626; background: rgba(220, 38, 38, 0.07); border-color: rgba(220, 38, 38, 0.25); }
/* ==================== detail header ==================== */
@ -1731,7 +1792,7 @@ async function cancelTask() {
}
.dot--todo { background: var(--mc-text-tertiary); }
.dot--in_progress { background: var(--mc-primary); }
.dot--in_review { background: #d97706; }
.dot--in_review, .dot--awaiting_approval { background: #d97706; }
.dot--completed { background: #16a34a; }
.dot--closed { background: #dc2626; }