From 9740d46fbcd03c27ab95ab2b6ab0862bb928e193 Mon Sep 17 00:00:00 2001 From: matevip Date: Thu, 23 Apr 2026 08:09:48 +0800 Subject: [PATCH] fix(delegate): distinguish outcome/blank/rawLength in parallel delegation, translate all comments to English --- .../mate/tool/builtin/DelegateAgentTool.java | 273 ++++++++++--- mateclaw-ui/src/composables/chat/useChat.ts | 383 +++++++++--------- 2 files changed, 414 insertions(+), 242 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java index 4b660307..8d7a8626 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java @@ -19,16 +19,17 @@ import java.util.concurrent.*; import java.util.stream.Collectors; /** - * 内置工具：Agent 委派（多 Agent 协作） + * Built-in tool: Agent delegation (multi-agent collaboration). *

- * 支持两种模式： + * Two modes: *

- * 被委派的 Agent 在独立子会话中运行（记录父子关系）， - * 执行期间通过 SSE 事件 relay 向父会话实时推送进度。 - * 子 Agent 的工具集自动收窄——禁止递归委派和 Agent 发现工具。 + * Each delegated agent runs in an isolated child conversation (parent-child relationship is + * persisted). Progress is relayed to the parent session via SSE events in real time. + * Child agents have a narrowed tool set — recursive delegation and agent-discovery tools are + * blocked. * * @author MateClaw Team */ @@ -48,14 +49,14 @@ public class DelegateAgentTool { */ private static final int PARALLEL_TIMEOUT_SECONDS = 120; - /** 子 Agent 禁用的工具：防递归 + 防副作用 */ + /** Tools blocked for child agents — prevents recursion and side effects. */ private static final Set CHILD_DENIED_TOOLS = Set.of( - "delegateToAgent", // 禁止递归委派 - "delegateParallel", // 禁止并行递归 - "listAvailableAgents" // 子 Agent 不需要发现其他 Agent + "delegateToAgent", // no recursive serial delegation + "delegateParallel", // no recursive parallel delegation + "listAvailableAgents" // child agents do not need to discover other agents ); - /** 并行委派执行器：JDK 21 虚拟线程，每个子 Agent 一个轻量级虚拟线程 */ + /** Executor for parallel delegation — one JDK 21 virtual thread per child agent. */ private static final ExecutorService DELEGATION_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); @@ -65,7 +66,7 @@ public class DelegateAgentTool { private final ConversationService conversationService; private final ObjectMapper objectMapper; - // ==================== 单任务委派 ==================== + // ==================== Single-task delegation ==================== @vip.mate.tool.ConcurrencyUnsafe("spawns a child agent session and writes to mate_conversation; serialize to keep session graph deterministic") @Tool(description = """ @@ -97,10 +98,10 @@ public class DelegateAgentTool { String parentConversationId = resolveParentConversationId(); String childConversationId = createChildConv(target, parentConversationId); - log.info("Agent 委派: depth={}, target={}({}), childConv={}, parentConv={}", + log.info("Agent delegation: depth={}, target={}({}), childConv={}, parentConv={}", depth + 1, target.getName(), target.getId(), childConversationId, parentConversationId); - // SSE 广播 + relay + // Broadcast delegation_start + register event relay to parent session boolean hasParent = parentConversationId != null && streamTracker.isRunning(parentConversationId); if (hasParent) { streamTracker.broadcastObject(parentConversationId, "delegation_start", Map.of( @@ -110,10 +111,10 @@ public class DelegateAgentTool { } Runnable stopRelay = hasParent ? registerRelay(childConversationId, parentConversationId, target.getName()) : null; - // 执行 + // Execute child agent ChildResult result = runSingleChild(0, target, task, parentConversationId, childConversationId); - // 清理 + 广播结果 + // Cleanup relay, then broadcast final result if (stopRelay != null) stopRelay.run(); if (hasParent) { broadcastEnd(parentConversationId, childConversationId, target.getName(), result); @@ -122,7 +123,7 @@ public class DelegateAgentTool { return result.toToolResponse(target.getName()); } - // ==================== 并行委派 ==================== + // ==================== Parallel delegation ==================== @vip.mate.tool.ConcurrencyUnsafe("internally fans out to its own thread pool; outer executor must not double-parallelize") @Tool(description = """ @@ -134,7 +135,7 @@ public class DelegateAgentTool { @ToolParam(description = "JSON array of tasks: [{\"agentName\":\"X\",\"task\":\"Y\"}, ...]") String tasksJson) { - // 1. 解析任务列表 + // 1. Parse task list List> tasks; try { tasks = objectMapper.readValue(tasksJson, new TypeReference<>() {}); @@ -157,7 +158,7 @@ public class DelegateAgentTool { String parentConversationId = resolveParentConversationId(); boolean hasParent = parentConversationId != null && streamTracker.isRunning(parentConversationId); - // 2. 主线程：校验所有 Agent + 创建子会话 + 注册 relay + // 2. Main thread: validate agents, create child conversations, register relays record PreparedChild(int index, AgentEntity agent, String task, String childConvId, Runnable stopRelay) {} List prepared = new ArrayList<>(); List errors = new ArrayList<>(); @@ -187,9 +188,9 @@ public class DelegateAgentTool { return "[错误] 所有任务校验失败：\n" + String.join("\n", errors); } - log.info("并行委派: {} 个任务, parentConv={}", prepared.size(), parentConversationId); + log.info("Parallel delegation: {} tasks, parentConv={}", prepared.size(), parentConversationId); - // 3. 广播 delegation_start（并行模式） + // 3. Broadcast delegation_start (parallel mode) if (hasParent) { List> childrenInfo = prepared.stream().map(p -> Map.of( "childConversationId", p.childConvId, @@ -201,7 +202,7 @@ public class DelegateAgentTool { "children", childrenInfo)); } - // 4. 并行执行 + // 4. Fan out — execute children in parallel long startTime = System.currentTimeMillis(); Map> futures = new LinkedHashMap<>(); @@ -212,38 +213,50 @@ public class DelegateAgentTool { // Broadcast per-child completion as soon as each child finishes // — frontend can update that child's status without waiting for all children. + // Guard: skip CancellationException (fired when the timeout loop calls cancel(true)) + // because the timeout result is already handled in the collection loop below and + // emitting here first would race-replace the correct "timeout" error before delegation_end + // has a chance to patch remaining running segments. if (hasParent) { final String parentConvIdFinal = parentConversationId; future.whenComplete((result, ex) -> { + if (ex instanceof java.util.concurrent.CancellationException) return; if (!streamTracker.isRunning(parentConvIdFinal)) return; ChildResult r = (result != null) ? result - : ChildResult.error(p.index, p.agent.getName(), ex != null ? ex.getMessage() : "Unknown error"); - streamTracker.broadcastObject(parentConvIdFinal, "delegation_child_complete", Map.of( - "taskIndex", r.taskIndex, - "childConversationId", p.childConvId, - "childAgentName", r.agentName, - "success", r.success, - "durationMs", r.durationMs, - "resultPreview", r.success ? truncate(r.result, 150) - : (r.error != null ? r.error : "error"))); + : ChildResult.ofError(p.index, p.agent.getName(), + ex != null ? ex.getMessage() : "Unknown error"); + Map payload = new java.util.LinkedHashMap<>(); + payload.put("taskIndex", r.taskIndex); + payload.put("childConversationId", p.childConvId); + payload.put("childAgentName", r.agentName); + payload.put("success", r.success); + payload.put("outcome", r.outcome); + payload.put("rawLength", r.rawLength); + payload.put("trimmedLength", r.trimmedLength); + payload.put("blank", r.isBlank()); + payload.put("durationMs", r.durationMs); + payload.put("resultPreview", r.success + ? truncate(r.result, 400) + : (r.error != null ? r.error : "error")); + streamTracker.broadcastObject(parentConvIdFinal, "delegation_child_complete", payload); }); } futures.put(p.index, future); } - // 5. 等待全部完成（带超时） + // 5. Wait for all children (with timeout) List results = new ArrayList<>(); try { CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])) .get(PARALLEL_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (TimeoutException e) { - log.warn("并行委派超时 ({}s)，收集已完成的结果", PARALLEL_TIMEOUT_SECONDS); + log.warn("Parallel delegation timed out ({}s), collecting completed results", PARALLEL_TIMEOUT_SECONDS); } catch (Exception e) { - log.error("并行委派异常: {}", e.getMessage()); + log.error("Parallel delegation error: {}", e.getMessage()); } - // 收集结果（已完成的 + 超时的） + // Collect results — completed futures get their value; unfinished ones are cancelled and recorded as timeout for (var entry : futures.entrySet()) { int idx = entry.getKey(); CompletableFuture f = entry.getValue(); @@ -254,29 +267,39 @@ public class DelegateAgentTool { try { results.add(f.get()); } catch (Exception ex) { - results.add(ChildResult.error(idx, agentName, ex.getMessage())); + results.add(ChildResult.ofError(idx, agentName, ex.getMessage())); } } else { f.cancel(true); - results.add(ChildResult.error(idx, agentName, "超时 (" + PARALLEL_TIMEOUT_SECONDS + "s)")); + // Use ofTimeout so outcome="timeout" is explicit and distinct from "error". + results.add(ChildResult.ofTimeout(idx, agentName, PARALLEL_TIMEOUT_SECONDS)); } } long totalDurationMs = System.currentTimeMillis() - startTime; - // 6. 清理 relay + // 6. Stop all relays for (PreparedChild p : prepared) { if (p.stopRelay != null) p.stopRelay.run(); } - // 7. 广播 delegation_end（含每个子任务的摘要，前端可用于展示分项结果） + // 7. Broadcast delegation_end with per-child structured summary if (hasParent) { List> childResults = results.stream().map(r -> { Map m = new java.util.LinkedHashMap<>(); m.put("taskIndex", r.taskIndex); m.put("agentName", r.agentName); m.put("success", r.success); + m.put("outcome", r.outcome); // "success"|"blank_success"|"timeout"|"error" + m.put("rawLength", r.rawLength); // chars before truncation + m.put("trimmedLength", r.trimmedLength); + m.put("blank", r.isBlank()); m.put("durationMs", r.durationMs); + // childConversationId for stable frontend segment lookup + prepared.stream() + .filter(p -> p.index == r.taskIndex) + .findFirst() + .ifPresent(p -> m.put("childConversationId", p.childConvId)); if (!r.success && r.error != null) m.put("error", r.error); return m; }).toList(); @@ -285,66 +308,185 @@ public class DelegateAgentTool { "totalDurationMs", totalDurationMs, "success", results.stream().allMatch(r -> r.success), "completedCount", results.stream().filter(r -> r.success).count(), + "blankCount", results.stream().filter(ChildResult::isBlank).count(), "totalCount", results.size(), "childResults", childResults)); } - // 8. 构建返回结果 + // 8. Build return text — structured so the parent LLM cannot misread current results + // using memory of past timeouts. The machine-readable header line is the source of truth. results.sort(Comparator.comparingInt(r -> r.taskIndex)); + long successCount = results.stream().filter(r -> r.success && !r.isBlank()).count(); + long blankCount = results.stream().filter(ChildResult::isBlank).count(); + long timeoutCount = results.stream().filter(r -> "timeout".equals(r.outcome)).count(); + long errorCount = results.stream().filter(r -> "error".equals(r.outcome)).count(); + StringBuilder sb = new StringBuilder(); + + // Machine-readable summary line (highest priority, appears first). + // Explicit blank/timeout/error counts prevent the parent agent from misreading a + // successful run as a timeout even when historical memory says "this agent often times out". + sb.append("[PARALLEL_DELEGATION_RESULT]") + .append(" total=").append(results.size()) + .append(" success=").append(successCount) + .append(" blank_success=").append(blankCount) + .append(" timeout=").append(timeoutCount) + .append(" error=").append(errorCount) + .append(" durationMs=").append(totalDurationMs) + .append("\n\n"); + + // Important: this result is from the current execution. Any timeout entries in the + // conversation history were from previous runs and must not be applied to this result. + sb.append("⚠ 注意：本次结果基于当前执行，与历史对话中出现的超时记录无关。\n\n"); + if (!errors.isEmpty()) { - sb.append("⚠️ 部分任务未执行：\n"); + sb.append("⚠️ 部分任务未执行（Agent 未找到或参数错误）：\n"); errors.forEach(e -> sb.append(" ").append(e).append("\n")); sb.append("\n"); } - sb.append("并行执行 ").append(results.size()).append(" 个任务（总耗时 ") - .append(totalDurationMs / 1000).append("s）：\n\n"); + + sb.append("## 各子任务执行结果\n\n"); for (ChildResult r : results) { - sb.append("---\n### [任务 ").append(r.taskIndex + 1).append("] Agent「").append(r.agentName).append("」"); - sb.append(r.success ? " ✓" : " ✗").append(" (").append(r.durationMs / 1000).append("s)\n\n"); - sb.append(r.success ? r.result : "[错误] " + r.error).append("\n\n"); + sb.append("### [任务 ").append(r.taskIndex + 1).append("] ").append(r.agentName).append("\n"); + // Per-row machine-readable status — impossible to confuse with a different outcome + sb.append("outcome=").append(r.outcome) + .append(" | contentLength=").append(r.trimmedLength).append("chars") + .append(" | rawLength=").append(r.rawLength).append("chars") + .append(" | duration=").append(r.durationMs / 1000).append("s") + .append("\n\n"); + + switch (r.outcome) { + case "success" -> { + sb.append("✅ 执行成功，有实质内容（").append(r.trimmedLength).append(" 字符）\n\n"); + sb.append(r.result); + } + case "blank_success" -> { + sb.append("⚠ 执行成功，但返回内容为空（rawLength=").append(r.rawLength) + .append("，trim 后 0 字符）。请勿将此误报为超时或失败——子 Agent 已正常完成，只是本次无输出。\n"); + } + case "timeout" -> + sb.append("❌ 超时（").append(PARALLEL_TIMEOUT_SECONDS).append("s 内未返回）\n"); + default -> + sb.append("❌ 失败：").append(r.error).append("\n"); + } + sb.append("\n"); } return truncate(sb.toString(), MAX_RESULT_LENGTH * 2); // 并行结果允许更长 } - // ==================== 子 Agent 执行（单/并行共用） ==================== + // ==================== Child agent execution (shared by single and parallel paths) ==================== /** - * 执行单个子 Agent。在子线程内独立设置 DelegationContext，解决 ThreadLocal 并行问题。 + * Runs a single child agent. Sets up {@link DelegationContext} independently per virtual thread + * so that parallel children do not share ThreadLocal state. + *

+ * Raw result length must be measured before calling {@code truncate()}, otherwise + * {@link ChildResult#rawLength} and {@link ChildResult#trimmedLength} would always reflect the + * truncated length, making "blank_success" detection unreliable. */ private ChildResult runSingleChild(int taskIndex, AgentEntity target, String task, String parentConversationId, String childConversationId) { DelegationContext.enter(parentConversationId, CHILD_DENIED_TOOLS); try { long startTime = System.currentTimeMillis(); - String result = agentService.chat(target.getId(), task, childConversationId); + String rawResult = agentService.chat(target.getId(), task, childConversationId); long durationMs = System.currentTimeMillis() - startTime; - return ChildResult.success(taskIndex, target.getName(), truncate(result, MAX_RESULT_LENGTH), durationMs); + // Measure lengths before truncation so ChildResult carries accurate metadata. + return ChildResult.ofSuccess(taskIndex, target.getName(), rawResult, durationMs, + MAX_RESULT_LENGTH); } catch (Exception e) { - log.error("子 Agent 执行失败: taskIndex={}, agent={}, error={}", + log.error("Child agent failed: taskIndex={}, agent={}, error={}", taskIndex, target.getName(), e.getMessage()); - return ChildResult.error(taskIndex, target.getName(), e.getMessage()); + return ChildResult.ofError(taskIndex, target.getName(), e.getMessage()); } finally { DelegationContext.exit(); } } - /** 子 Agent 执行结果 */ - private record ChildResult(int taskIndex, String agentName, boolean success, - String result, String error, long durationMs) { + /** + * Result carrier for a single child agent execution. + * + *

{@code outcome} values: + *

    + *
  • {@code "success"} — completed successfully with non-empty content (trimmedLength > 0)
    • + *
  • {@code "blank_success"} — completed successfully but returned empty content (trimmedLength == 0)
    • + *
  • {@code "timeout"} — did not complete within the parallel wait window
    • + *
  • {@code "error"} — threw an exception during execution
    • + *
+ * + *

{@code rawLength} and {@code trimmedLength} are measured before truncation and reflect the + * true content length. + */ + private record ChildResult( + int taskIndex, String agentName, boolean success, + String result, String error, long durationMs, + /** "success" | "blank_success" | "timeout" | "error" */ + String outcome, + int rawLength, int trimmedLength) { + + /** Whether the child returned no usable content (blank_success). */ + boolean isBlank() { return "blank_success".equals(outcome); } + + /** + * Factory for a successful child execution. + * Measures lengths from the raw result before applying the truncation limit. + */ + static ChildResult ofSuccess(int idx, String name, String rawResult, long ms, int maxLen) { + String safe = rawResult != null ? rawResult : ""; + String trimmed = safe.trim(); + boolean blank = trimmed.isEmpty(); + return new ChildResult( + idx, name, true, + truncate(safe, maxLen), + null, ms, + blank ? "blank_success" : "success", + safe.length(), trimmed.length()); + } + + /** + * Factory for a child that failed (exception or timeout). + * Detects timeout by inspecting the error message so callers don't need to branch. + */ + static ChildResult ofError(int idx, String name, String err) { + String msg = err != null ? err : "Unknown error"; + boolean isTimeout = msg.contains("超时") || msg.toLowerCase().contains("timeout"); + return new ChildResult(idx, name, false, null, msg, 0, + isTimeout ? "timeout" : "error", 0, 0); + } + + /** Factory for an explicit timeout (parallel window exceeded). */ + static ChildResult ofTimeout(int idx, String name, int timeoutSec) { + String msg = "超时 (" + timeoutSec + "s)"; + return new ChildResult(idx, name, false, null, msg, (long) timeoutSec * 1000L, + "timeout", 0, 0); + } + + // Legacy shims — kept for callers that pre-date the factory methods static ChildResult success(int idx, String name, String result, long ms) { - return new ChildResult(idx, name, true, result, null, ms); + // result may already be truncated at call site — lengths will be approximate + String safe = result != null ? result : ""; + String trimmed = safe.trim(); + boolean blank = trimmed.isEmpty(); + return new ChildResult(idx, name, true, safe, null, ms, + blank ? "blank_success" : "success", safe.length(), trimmed.length()); } static ChildResult error(int idx, String name, String err) { - return new ChildResult(idx, name, false, null, err != null ? err : "Unknown error", 0); + return ofError(idx, name, err); } + String toToolResponse(String agentName) { - if (success) return "[Agent「" + agentName + "」的回复]\n\n" + result; + if (success) return "[Agent「" + agentName + "」的回复]\n\n" + (result != null ? result : ""); return "[错误] Agent「" + agentName + "」执行失败: " + error; } + + private static String truncate(String text, int maxLength) { + if (text == null) return ""; + if (text.length() <= maxLength) return text; + return text.substring(0, maxLength) + "\n... [截断，原文 " + text.length() + " 字符]"; + } } - // ==================== 辅助方法 ==================== + // ==================== Helper methods ==================== @Tool(description = "List all available Agents (enabled), including name, type, and description.") public String listAvailableAgents() { @@ -386,11 +528,20 @@ public class DelegateAgentTool { return streamTracker.addEventRelay(childConvId, (eventName, jsonData) -> { if ("tool_call_started".equals(eventName) || "tool_call_completed".equals(eventName) || "phase".equals(eventName)) { try { + // Parse jsonData into a plain Object so the frontend receives a proper + // JSON object under "data", not a string containing serialized JSON. + // If parsing fails (e.g. plain text payload), fall back to the raw string. + Object parsedData; + try { + parsedData = objectMapper.readValue(jsonData, Object.class); + } catch (Exception ignored) { + parsedData = jsonData; + } streamTracker.broadcastObject(parentConvId, "delegation_progress", Map.of( "childConversationId", childConvId, "childAgentName", childAgentName, "originalEvent", eventName, - "data", jsonData)); + "data", parsedData)); } catch (Exception e) { log.debug("Relay error: {}", e.getMessage()); } diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index 653d07b8..d1ed253a 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -1,12 +1,13 @@ /** - * 聊天功能统一 Composable - * 整合 useMessages、useStream、useMessageQueue，提供完整的聊天功能 + * Unified chat composable. + * Integrates useMessages, useStream, and useMessageQueue into a complete chat feature. * - * 核心机制（参考 claude-code-haha 的 Interrupt + Queue + Resume 模型）： - * - 运行中允许继续输入新消息 - * - 可中断阶段：发送 interrupt 请求，中断后自动续跑排队消息 - * - 不可中断阶段：消息排队，等当前步骤结束后自动继续 - * - 审批中：消息排队，不打断审批流程 + * Core mechanism (Interrupt + Queue + Resume model): + * - New messages can be sent while a response is already generating. + * - Interruptible phases (thinking/streaming/executing_tool): sends an interrupt request; + * the queued message resumes automatically after interruption. + * - Non-interruptible phases: message is queued and auto-resumed when the current step ends. + * - During approval: message is queued, approval flow is not interrupted. */ import { ref, computed } from 'vue' import { useMessages } from './useMessages' @@ -17,78 +18,78 @@ import { classifyBackendError, type ChatErrorInfo } from '@/types/chatError' import { http } from '@/api' export interface UseChatOptions { - /** API 基础 URL */ + /** Base API URL */ baseUrl: string - /** 认证 Token */ + /** Auth token */ token?: string - /** 当前思考深度（响应式 ref），off 时抑制 thinking 展示 */ + /** Current thinking depth (reactive ref); when "off", thinking segments are suppressed */ thinkingLevel?: import('vue').Ref /** - * 统一回调：流结束后（done/error/stopped 都会触发）。 - * 前端应在此回调中做持久化历史收口（reconcile）。 + * Unified callback fired when the stream ends (done/error/stopped all trigger this). + * The caller should perform history reconcile / persistence in this callback. */ onStreamEnd?: (meta: StreamEndMeta) => void } -/** 流结束元信息 */ +/** Metadata emitted when a stream ends */ export interface StreamEndMeta { conversationId: string reason: 'completed' | 'stopped' | 'interrupted' | 'failed' | 'error' | 'awaiting_approval' - /** 后端持久化的 assistant 消息 ID（若有） */ + /** Backend-persisted assistant message ID, if available */ assistantMessageId?: number - /** 后端是否已持久化 */ + /** Whether the backend has already persisted the message */ persisted?: boolean - /** 后端当前消息总数 */ + /** Total message count reported by the backend */ messageCount?: number } export interface UseChatReturn { - /** 消息列表 */ + /** Message list */ messages: import('vue').Ref - /** 是否正在生成 */ + /** Whether the assistant is currently generating */ isGenerating: import('vue').ComputedRef - /** 当前流阶段 */ + /** Current stream phase */ streamPhase: import('vue').Ref - /** 最近一次阶段事件 */ + /** Most recent phase event */ phaseInfo: import('vue').Ref - /** 当前错误 */ + /** Current error */ error: import('vue').Ref - /** 排队的消息 */ + /** Queued message waiting to be sent */ queuedMessage: import('vue').Ref - /** 是否有排队消息 */ + /** Whether there is a queued message */ hasQueued: import('vue').ComputedRef - /** 排队消息数量 */ + /** Number of queued messages */ queueSize: import('vue').ComputedRef - /** 心跳数据 */ + /** Latest heartbeat data */ heartbeat: import('vue').Ref - /** 发送消息（运行中也可调用，自动走 interrupt/queue） */ + /** Send a message (can be called while generating — automatically routes to interrupt/queue) */ sendMessage: (content: string, options: SendMessageOptions) => Promise - /** 停止生成（用户主动停止，不自动续跑） */ + /** Stop generation (user-initiated stop; does not auto-resume queued messages) */ stopGeneration: () => void - /** 取消排队消息 */ + /** Cancel the queued message */ cancelQueued: () => void - /** 重新生成 */ + /** Regenerate a message */ regenerate: (messageId: string | number) => Promise - /** 添加消息 */ + /** Add a message */ addMessage: (message: Omit & { id?: string | number }) => Message - /** 清空消息 */ + /** Clear all messages */ clearMessages: () => void - /** 重连到运行中的流 */ + /** Reconnect to a stream that is already running on the backend */ reconnectStream: (conversationId: string) => Promise - /** 彻底重置流上下文 — 切换/新建会话时调用 */ + /** Fully reset stream context — call when switching or creating a conversation */ resetForNewConversation: () => void } export interface SendMessageOptions { - /** 会话 ID */ + /** Conversation ID */ conversationId: string /** Agent ID */ agentId: string | number - /** 附件列表 */ + /** Attachment list */ attachments?: MessageContentPart[] - /** 消息内容 */ + /** Message content parts */ contentParts?: MessageContentPart[] - /** 思考深度：off / low / medium / high / max */ + /** Thinking depth: off / low / medium / high / max */ thinkingLevel?: string } @@ -97,7 +98,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { const thinkingLevelRef = options.thinkingLevel /** - * 带认证的 fetch 封装 — 从 localStorage 读取 token（与 useStream / http.ts 一致） + * Authenticated fetch wrapper — reads the token from localStorage (consistent with useStream / http.ts). */ const fetchWithAuth = (url: string, init: RequestInit = {}): Promise => { const headers: Record = { @@ -112,32 +113,32 @@ export function useChat(options: UseChatOptions): UseChatReturn { const error = ref(null) const currentAssistantId = ref(null) - /** stopGeneration 的 fallback timer，新流开始时必须清除，防止误杀新连接 */ + /** Fallback timer for stopGeneration — must be cleared when a new stream starts to avoid killing the new connection */ let stopFallbackTimer: ReturnType | null = null const streamPhase = ref('idle') const phaseInfo = ref(null) - /** 分段式展示数据：当前助手消息的所有分段 */ + /** All segments of the current assistant message (for segmented display) */ const currentSegments = ref([]) const segIdCounter = { value: 0 } const genSegId = () => `seg-${Date.now()}-${segIdCounter.value++}` - /** 当前 turn 的唯一标识 — 确保 flushSegmentsToMessage 不会把旧 turn 的 segments 写到新消息 */ + /** Unique ID for the current turn — prevents flushSegmentsToMessage from writing stale segments to a new message */ let activeTurnId = '' - /** 重置当前 turn 的流式状态 — 必须在每次创建新 assistant placeholder 之前调用 */ + /** Reset streaming state for the current turn — must be called before creating a new assistant placeholder */ function resetCurrentTurnState() { currentSegments.value = [] segIdCounter.value = 0 activeTurnId = `turn-${Date.now()}-${Math.random().toString(36).slice(2, 6)}` } - /** 将当前 segments 同步到助手消息的 metadata 中（实时渲染用） */ + /** Sync current segments into the assistant message metadata (used for real-time rendering) */ const flushSegmentsToMessage = () => { if (!currentAssistantId.value || currentSegments.value.length === 0) return const msg = getMessage(currentAssistantId.value) if (!msg) return - // 保护：只写入当前 turn 创建的消息，避免旧 turn segments 污染新消息 + // Guard: only write to the message created in the current turn to avoid stale segment pollution if ((msg as any)._turnId && (msg as any)._turnId !== activeTurnId) return const metadata = parseMetadata((msg as any).metadata) updateMessage(currentAssistantId.value, { @@ -148,7 +149,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { const heartbeat = ref(null) /** Track which conversation the current stream belongs to */ let streamConversationId = '' - /** 判断事件是否属于已过期的对话（防止旧流事件污染新会话） */ + /** Returns true if the event belongs to an expired conversation (prevents stale stream events from polluting a new session) */ function isStaleEvent(data: any): boolean { const eventConvId = data?.conversationId if (eventConvId && streamConversationId && eventConvId !== streamConversationId) { @@ -156,18 +157,18 @@ export function useChat(options: UseChatOptions): UseChatReturn { } return false } - /** 已处理的 approval pendingId 集合（幂等去重） */ + /** Set of already-processed approval pendingIds (idempotency dedup) */ const processedApprovalIds = new Set() /** - * 解析 metadata - 处理从数据库加载的 JSON 字符串 + * Parse metadata — handles JSON strings loaded from the database. */ const parseMetadata = (metadata: any): any => { if (!metadata) return {} if (typeof metadata === 'string') { try { let parsed = JSON.parse(metadata) - // 处理双重 JSON 编码（DB metadata 是字符串，Jackson 可能再次转义） + // Handle double-encoded JSON (DB metadata is a string; Jackson may escape it again) if (typeof parsed === 'string') { try { parsed = JSON.parse(parsed) } catch { /* ignore */ } } @@ -181,8 +182,9 @@ export function useChat(options: UseChatOptions): UseChatReturn { } /** - * SSE 以 error/done 结束但审批实际上已失效时，收口残留的 awaiting_approval UI 状态。 - * 必须用 updateMessage 触发响应式更新，不能只改嵌套字段。 + * Expire stale awaiting_approval UI state when the stream ends with error/done but the approval + * is no longer active. Must use updateMessage to trigger Vue reactivity — mutating nested fields + * alone is not sufficient. */ const expirePendingApprovals = (finalStatus: 'completed' | 'failed' | 'stopped') => { for (const m of messages.value) { @@ -218,7 +220,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // 消息管理 + // Message management const { messages, isGenerating, @@ -232,14 +234,14 @@ export function useChat(options: UseChatOptions): UseChatReturn { getMessage, } = useMessages({ onComplete: () => { - // 不在 onComplete 里清 currentAssistantId — 让 done 事件来清 + // Do not clear currentAssistantId here — the 'done' event handles cleanup }, }) - // 消息队列 + // Message queue const messageQueue = useMessageQueue() - // 流连接（注入 auth + workspace header，与 axios interceptor 保持一致） + // Stream connection (inject auth + workspace headers, consistent with the axios interceptor) const streamHeaders: Record = {} if (token) { streamHeaders['Authorization'] = `Bearer ${token}` @@ -253,7 +255,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { headers: streamHeaders, }) - // ===== SSE 事件处理器 ===== + // ===== SSE event handlers ===== stream.on('content_delta', (data) => { if (isStaleEvent(data)) return @@ -262,16 +264,16 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (['thinking', 'reasoning', 'drafting_answer', 'preparing_context'].includes(streamPhase.value)) { streamPhase.value = 'streaming' } - // 分段：追加到当前 content segment 或创建新的 + // Segments: append to the current running content segment, or create a new one const segs = currentSegments.value let contentSeg = segs.findLast((s: MessageSegment) => s.type === 'content' && s.status === 'running') if (!contentSeg) { - // 关闭之前的 thinking segment + // Close any running thinking segment first const thinkingSeg = segs.findLast((s: MessageSegment) => s.type === 'thinking' && s.status === 'running') if (thinkingSeg) thinkingSeg.status = 'completed' contentSeg = { id: genSegId(), type: 'content', status: 'running', text: '', timestamp: Date.now() } segs.push(contentSeg) - flushSegmentsToMessage() // 新 content segment 创建时同步一次 + flushSegmentsToMessage() // sync once when a new content segment is created } contentSeg.text = (contentSeg.text || '') + (data.delta || '') } @@ -279,24 +281,24 @@ export function useChat(options: UseChatOptions): UseChatReturn { stream.on('thinking_delta', (data) => { if (isStaleEvent(data)) return - // thinkingLevel=off 时抑制 thinking 展示 + // Suppress thinking display when thinkingLevel=off if (options.thinkingLevel?.value === 'off') return if (currentAssistantId.value) { appendMessageContent(currentAssistantId.value, data.delta || '', 'thinking') if (streamPhase.value !== 'summarizing_observations') { streamPhase.value = options.thinkingLevel?.value === 'off' ? 'streaming' : 'thinking' } - // 分段：所有 thinking 合并到一个 segment（不因 tool_call 中断而创建多个） + // Segments: all thinking deltas merge into one segment (not split by tool_call interruptions) const segs = currentSegments.value - // 优先复用已有的 thinking segment（无论 running 还是 completed） + // Reuse an existing thinking segment regardless of status (running or completed) let thinkSeg = segs.find((s: MessageSegment) => s.type === 'thinking') if (!thinkSeg) { thinkSeg = { id: genSegId(), type: 'thinking', status: 'running', thinkingText: '', timestamp: Date.now() } - // 插入到开头（thinking 始终在最上方） + // Insert at front — thinking always appears at the top segs.unshift(thinkSeg) flushSegmentsToMessage() } - // 新的 thinking 到来，重新设为 running + // Re-mark as running when new thinking content arrives thinkSeg.status = 'running' thinkSeg.thinkingText = (thinkSeg.thinkingText || '') + (data.delta || '') } @@ -313,7 +315,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { return } - // 没有 placeholder 时才创建（正常路径 placeholder 已在 sendMessage 中创建） + // Only create a placeholder here if one does not already exist (the normal path creates it in sendMessage) resetCurrentTurnState() const assistantMessage = createAssistantMessage('', streamConversationId) ;(assistantMessage as any)._turnId = activeTurnId @@ -342,7 +344,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (currentAssistantId.value) { const msg = getMessage(currentAssistantId.value) if (msg?.status === 'failed') { - // 不清除 currentAssistantId — 让 done 事件来做 + // Do not clear currentAssistantId — the 'done' event handles cleanup return } if (msg) { @@ -362,16 +364,16 @@ export function useChat(options: UseChatOptions): UseChatReturn { status: data.status || 'completed', metadata: { ...metadata, toolCalls } } as any) - // 关键修复：不在这里清除 currentAssistantId，让 done 来清 + // Do not clear currentAssistantId here — the 'done' event does it return } } } setMessageStatus(currentAssistantId.value, data.status || 'completed') - // 关键修复：不在这里清除 currentAssistantId + // Do not clear currentAssistantId here — the 'done' event does it } - // 分段：标记所有 running segments 为 completed，并持久化到 message metadata + // Segments: mark all running segments as completed and persist to message metadata if (currentAssistantId.value && currentSegments.value.length > 0) { currentSegments.value.forEach((s: MessageSegment) => { if (s.status === 'running') s.status = 'completed' }) const msg = getMessage(currentAssistantId.value) @@ -384,7 +386,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // === 自动 TTS：message_complete 且 status=completed 时触发 === + // Auto TTS: trigger when message_complete arrives with status=completed if (data.status === 'completed' && data.hasContent && currentAssistantId.value) { const msg = getMessage(currentAssistantId.value) if (msg?.content && streamConversationId) { @@ -402,13 +404,13 @@ export function useChat(options: UseChatOptions): UseChatReturn { setMessageStatus(currentAssistantId.value, data.status || 'completed') } - // 更新 token 信息 + 用持久化 id 替换本地临时 id（关键：让 reconcile 能匹配） + // Update token counts + replace the local temp ID with the backend-persisted ID (critical: enables reconcile by ID) const msgIndex = messages.value.findIndex(m => m.id === currentAssistantId.value) if (msgIndex >= 0) { const msg = messages.value[msgIndex] if (data.promptTokens !== undefined) msg.promptTokens = data.promptTokens if (data.completionTokens !== undefined) msg.completionTokens = data.completionTokens - // 用后端持久化 id 替换本地临时 id，使 reconcile 时能按 id 匹配 + // Replace the local temp ID with the backend-persisted ID so reconcile can match by ID if (data.assistantMessageId) { msg.id = data.assistantMessageId } @@ -424,11 +426,11 @@ export function useChat(options: UseChatOptions): UseChatReturn { expirePendingApprovals(data.status === 'stopped' ? 'stopped' : 'completed') } - // 兜底清理排队状态（如果 queued_input_started 已经处理了则这里是 no-op） + // Safety cleanup for queue state (no-op if queued_input_started already handled it) if (!messageQueue.hasQueued.value) { - // 队列已空，确保 phase 不残留 queued + // Queue already empty — phase cannot linger at 'queued' } else if (data.status === 'stopped') { - // 用户主动停止，清除排队 + // User-initiated stop — discard queued message messageQueue.clear() } @@ -467,7 +469,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { error.value = new Error(data.message || '请求失败') streamPhase.value = 'idle' phaseInfo.value = null - // 错误时清理排队状态，避免脏残留 + // Clear queue on error to avoid stale state messageQueue.clear() expirePendingApprovals('failed') @@ -482,7 +484,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { }) }) - // ===== Agent 事件处理 ===== + // ===== Agent event handlers ===== stream.on('tool_call_started', (data) => { if (isStaleEvent(data)) return @@ -503,7 +505,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { metadata: { ...metadata, toolCalls, currentPhase: 'executing_tool', runningToolName: data.toolName } } as any) } - // 分段：关闭之前的 thinking/content segment，创建新的 tool_call segment + // Segments: close any running thinking/content segment, then push a new tool_call segment const segs = currentSegments.value const runningSeg = segs.findLast((s: MessageSegment) => s.status === 'running' && (s.type === 'thinking' || s.type === 'content')) if (runningSeg) runningSeg.status = 'completed' @@ -537,7 +539,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { metadata: { ...metadata, toolCalls, runningToolName: undefined } } as any) } - // 分段：找到对应的 running tool_call segment 并标记完成 + // Segments: find the matching running tool_call segment and mark it complete const segs = currentSegments.value const toolSeg = segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName === data.toolName) @@ -550,7 +552,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) - // ===== Browser 执行事件 ===== + // ===== Browser action events ===== stream.on('browser_action', (data) => { if (isStaleEvent(data)) return @@ -587,7 +589,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { const msg = getMessage(currentAssistantId.value) if (msg) { const metadata = parseMetadata((msg as any).metadata) - // 去重：相同 phase 不触发 updateMessage，避免不必要的 Vue 响应式更新 + // Dedup: skip updateMessage if the phase hasn't changed, to avoid unnecessary Vue reactivity if (metadata.currentPhase === data.phase) return updateMessage(currentAssistantId.value, { ...msg, @@ -597,21 +599,24 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) - // ===== Agent 委派事件 ===== + // ===== Agent delegation events ===== stream.on('delegation_start', (data) => { if (isStaleEvent(data)) return streamPhase.value = 'executing_tool' if (currentAssistantId.value) { const segs = currentSegments.value - // 关闭之前的 thinking/content segment + // Close any running thinking/content segment const runningSeg = segs.findLast((s: MessageSegment) => s.status === 'running') if (runningSeg) runningSeg.status = 'completed' if (data.parallel && Array.isArray(data.children)) { - // 并行模式：为每个子任务创建一个 delegation segment + // Parallel mode: one segment per child. Use childConversationId as the segment ID + // so downstream events (delegation_child_complete, delegation_progress) can look up + // the correct row by stable ID instead of agent name — which is not unique when + // two concurrent tasks go to the same agent. for (const child of data.children) { segs.push({ - id: genSegId(), + id: child.childConversationId || genSegId(), type: 'tool_call', status: 'running', toolName: `→ ${child.childAgentName || 'Agent'}`, @@ -620,9 +625,9 @@ export function useChat(options: UseChatOptions): UseChatReturn { }) } } else { - // 单任务模式 + // Single-task mode: same stable ID approach segs.push({ - id: genSegId(), + id: data.childConversationId || genSegId(), type: 'tool_call', status: 'running', toolName: `→ ${data.childAgentName || 'Agent'}`, @@ -638,35 +643,40 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (isStaleEvent(data)) return if (!currentAssistantId.value) return const segs = currentSegments.value - const childName = data.childAgentName || '' - // Find the running delegation segment for this child (or fall back to any running delegation) - const delegSeg = segs.findLast((s: MessageSegment) => - s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`) - || segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) + + // Primary lookup: by stable childConversationId (set as the segment ID at creation time). + // Fallback: any running delegation segment (for older backends that don't send the field). + const delegSeg = (data.childConversationId + ? segs.find((s: MessageSegment) => s.id === data.childConversationId) + : undefined) + || segs.findLast((s: MessageSegment) => + s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) if (!delegSeg) return + // Normalize data.data: the backend relays the child event's JSON payload. + // After the P2 fix it arrives as an object; be defensive for older backends. + const rawPayload = data.data + const childData: Record = rawPayload && typeof rawPayload === 'object' + ? rawPayload + : (() => { try { return JSON.parse(String(rawPayload || '{}')) } catch { return {} } })() + if (data.originalEvent === 'tool_call_started') { - // Child started a sub-tool — append activity hint so the user sees the child is working - const childData = data.data - const toolName = typeof childData === 'object' ? childData?.toolName : String(childData || '') + const toolName = childData?.toolName || '' if (toolName) { delegSeg.toolArgs = (delegSeg.toolArgs || '') + `\n → ${toolName}` } } else if (data.originalEvent === 'tool_call_completed') { - // Child finished a sub-tool call — update the running hint - const childData = data.data - const toolName = typeof childData === 'object' ? childData?.toolName : String(childData || '') - const success = typeof childData === 'object' ? childData?.success !== false : true + const toolName = childData?.toolName || '' + const success = childData?.success !== false if (toolName) { - // Replace last appended "→ toolName" with "✓/✗ toolName" + // Replace the matching "→ toolName" hint with "✓/✗ toolName" delegSeg.toolArgs = (delegSeg.toolArgs || '').replace( new RegExp(`\\n → ${toolName.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}\\s*$`), `\n ${success ? '✓' : '✗'} ${toolName}`) } } else if (data.originalEvent === 'phase') { - // Child entered a new phase (reasoning, executing_tool, etc.) - const phase = typeof data.data === 'object' ? data.data?.phase : String(data.data || '') + const phase = childData?.phase || String(rawPayload || '') const phaseHints: Record = { reasoning: '…', executing_tool: '→', @@ -688,20 +698,22 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (isStaleEvent(data)) return if (!currentAssistantId.value) return const segs = currentSegments.value - const childName = data.childAgentName || '' - const delegSeg = segs.findLast((s: MessageSegment) => - s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`) + // Prefer childConversationId (stable) over agent name (non-unique) + const delegSeg = (data.childConversationId + ? segs.find((s: MessageSegment) => s.id === data.childConversationId) + : undefined) || segs.findLast((s: MessageSegment) => - s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) + s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→')) if (delegSeg) { delegSeg.status = data.success ? 'completed' : 'error' delegSeg.toolSuccess = data.success - // Append duration to args so the user sees how long each child took if (data.durationMs) { const durSec = Math.round(data.durationMs / 1000) delegSeg.toolArgs = (delegSeg.toolArgs || '').trimEnd() + ` (${durSec}s)` } - if (!data.success && data.resultPreview) { + // Write resultPreview for both success and failure so ToolCallSegment can show + // an expand arrow with the child agent's actual output, not just a green/red dot. + if (data.resultPreview) { delegSeg.toolResult = data.resultPreview } } @@ -717,19 +729,26 @@ export function useChat(options: UseChatOptions): UseChatReturn { // fall back to aggregate success flag for older backends. if (Array.isArray(data.childResults) && data.childResults.length > 0) { for (const cr of data.childResults) { - const agentName = cr.agentName || '' - const seg = segs.findLast((s: MessageSegment) => - s.type === 'tool_call' && - (s.status === 'running' || s.status === 'completed') && - s.toolName?.includes(agentName)) + // Primary: stable childConversationId lookup. Fallback: agent name substring. + const seg = (cr.childConversationId + ? segs.find((s: MessageSegment) => s.id === cr.childConversationId) + : undefined) + || segs.findLast((s: MessageSegment) => + s.type === 'tool_call' && s.toolName?.includes(cr.agentName || '')) if (seg && seg.status === 'running') { - // Segment not yet closed by delegation_child_complete (e.g. timed out child) + // Segment not yet closed by delegation_child_complete (e.g. timed-out child). + // Write whatever result info is available so ToolCallSegment can show content. seg.status = cr.success ? 'completed' : 'error' seg.toolSuccess = cr.success if (cr.durationMs) { const durSec = Math.round(cr.durationMs / 1000) seg.toolArgs = (seg.toolArgs || '').trimEnd() + ` (${durSec}s)` } + // Show error reason for failures; for successes leave toolResult empty here + // (delegation_child_complete already wrote the preview before we get to delegation_end). + if (cr.error) { + seg.toolResult = cr.error + } } } } else { @@ -814,11 +833,11 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) - // ===== 工具审批事件（带幂等去重） ===== + // ===== Tool approval events (with idempotency dedup) ===== stream.on('tool_approval_requested', (data) => { if (isStaleEvent(data)) return - // 幂等去重：同一 pendingId 只处理一次 + // Idempotency: process each pendingId only once if (data.pendingId && processedApprovalIds.has(data.pendingId)) { // duplicate approval ignored return @@ -911,12 +930,12 @@ export function useChat(options: UseChatOptions): UseChatReturn { streamPhase.value = data.decision === 'approved' ? 'streaming' : 'completed' }) - // ===== Heartbeat 事件 ===== + // ===== Heartbeat events ===== stream.on('heartbeat', (data: HeartbeatData) => { heartbeat.value = data - // heartbeat 到达意味着连接活跃，useStream 的 timeout 已由 resetStreamTimeout 自动重置 - // 从 heartbeat 中更新 phase（如果前端还没有更精确的 phase） + // Heartbeat arrival means the connection is alive; useStream resets the timeout automatically. + // Update phase from heartbeat only when the frontend doesn't have a more precise phase yet. if (data.currentPhase && streamPhase.value !== 'interrupting') { const phaseMap: Record = { 'preparing_context': 'preparing_context', @@ -934,16 +953,16 @@ export function useChat(options: UseChatOptions): UseChatReturn { const mapped = phaseMap[data.currentPhase] if (mapped) streamPhase.value = mapped } - // 利用 heartbeat 的 queueLength 校准本地排队状态 - // 仅在消息已被后端确认（status=sending）后才以 heartbeat 兜底清理， - // 避免在 interrupt 请求仍在途中时误清尚未到达后端的消息 + // Use heartbeat queueLength to reconcile local queue state. + // Only clear when the message has been acknowledged by the backend (status=sending), + // to avoid discarding a message whose interrupt request is still in flight. if (data.queueLength === 0 && messageQueue.hasQueued.value && messageQueue.queuedMessage.value?.status === 'sending') { messageQueue.clear() } }) - // ===== Interrupt + Queue 事件 ===== + // ===== Interrupt + Queue events ===== stream.on('turn_interrupt_requested', () => { streamPhase.value = 'interrupting' @@ -951,31 +970,31 @@ export function useChat(options: UseChatOptions): UseChatReturn { stream.on('turn_interrupted', (data) => { if (isStaleEvent(data)) return - // 当前 turn 已中断，等待后端自动启动排队消息 - // 如果后端会自动续跑，前端不需要做额外操作 - // 如果后端没有排队消息但前端有（应该不会发生），则前端发送 + // Current turn has been interrupted. Wait for the backend to resume with the queued message. + // If the backend will auto-resume, the frontend does nothing extra. + // Edge case: backend has no queued message but frontend does (should not happen in practice). if (data.hasQueuedMessage) { streamPhase.value = 'queued' } }) stream.on('queued_input_accepted', (data) => { - // 后端已确认接收排队消息，标记为 sending（允许 heartbeat 兜底清理） + // Backend confirmed receipt of the queued message — mark as 'sending' to allow heartbeat cleanup messageQueue.markSending() streamPhase.value = 'queued' }) stream.on('queued_input_started', (data) => { if (isStaleEvent(data)) return - // 后端已开始处理排队消息 - // 1. 先用排队的内容创建用户消息（此时上一轮回答已完成，顺序正确） + // Backend has started processing the queued message. + // 1. Create the user message first (previous turn is now complete so ordering is correct) const queued = messageQueue.dequeue() const messageContent = data.message || queued?.content || '' if (messageContent) { const convId = data.conversationId || streamConversationId createUserMessage(messageContent, queued?.contentParts, convId) } - // 2. 再创建 assistant 占位消息 + // 2. Create the assistant placeholder message resetCurrentTurnState() const convId2 = data.conversationId || streamConversationId const assistantMessage = createAssistantMessage('', convId2) @@ -985,7 +1004,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { phaseInfo.value = null }) - // ===== 异步任务完成事件（视频生成、图片生成等） ===== + // ===== Async task completion events (video generation, image generation, etc.) ===== stream.on('async_task_completed', (data) => { if (isStaleEvent(data)) return if (data.success && streamConversationId) { @@ -1008,7 +1027,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (!mediaPart) return - // 优先附加到当前 assistant 消息（避免图片跑到文字回复上方） + // Prefer appending to the current assistant message (avoids image appearing above the text reply) if (currentAssistantId.value) { const msg = getMessage(currentAssistantId.value) if (msg) { @@ -1020,7 +1039,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // 回退：Agent 已结束，新建独立消息 + // Fallback: agent already finished — create a standalone message addMessage({ role: 'assistant', content: '', @@ -1031,13 +1050,13 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) - // ===== TTS 自动朗读 ===== + // ===== Auto TTS ===== let ttsAutoModeCache: string | null = null let ttsCacheExpiry = 0 async function triggerAutoTts(conversationId: string, text: string) { try { - // 缓存 settings 5 分钟，避免每条消息都请求 + // Cache settings for 5 minutes to avoid a request on every message const now = Date.now() if (!ttsAutoModeCache || now > ttsCacheExpiry) { const res: any = await http.get('/system-settings') @@ -1045,14 +1064,14 @@ export function useChat(options: UseChatOptions): UseChatReturn { ttsCacheExpiry = now + 5 * 60 * 1000 } if (ttsAutoModeCache !== 'always') return - // 调用后端合成，后端会通过 SSE tts_ready 广播 + // Kick off backend synthesis; the backend broadcasts tts_ready via SSE when done http.post('/tts/synthesize', { conversationId, text }).catch(() => {}) } catch { - // 静默失败 + // Silently ignore TTS errors — it's a best-effort feature } } - // ===== TTS 自动朗读：监听 tts_ready 事件 ===== + // ===== Auto TTS: listen for tts_ready events ===== stream.on('tts_ready', (data) => { if (data.audioUrl) { const token = localStorage.getItem('token') || '' @@ -1068,26 +1087,26 @@ export function useChat(options: UseChatOptions): UseChatReturn { } }) - // ===== 发送消息（支持运行中继续发送） ===== + // ===== Send message (supports sending while generating) ===== const sendMessage = async (content: string, options: SendMessageOptions) => { const { conversationId, agentId, attachments = [], contentParts = [] } = options - // 审批命令不走 interrupt 逻辑 + // Approval commands bypass the interrupt logic const isApprovalCommand = /^\/(approve|deny)$/i.test(content.trim()) - // ===== 运行中发送新消息：走 interrupt / queue 路径 ===== + // ===== Sending while generating: route to interrupt / queue path ===== if (isGenerating.value && !isApprovalCommand) { return await handleInterruptOrQueue(content, options) } - // ===== 正常发送路径 ===== - // 清除上一次 stop 的 fallback timer，防止误杀新连接 + // ===== Normal send path ===== + // Clear the previous stop fallback timer to avoid killing the new connection if (stopFallbackTimer) { clearTimeout(stopFallbackTimer) stopFallbackTimer = null } - // 切换会话时断开旧流，防止旧事件污染新会话 + // Disconnect the old stream when switching conversations to prevent event pollution if (streamConversationId && streamConversationId !== conversationId) { stream.disconnect() currentAssistantId.value = null @@ -1108,7 +1127,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { ;(assistantMessage as any)._turnId = activeTurnId currentAssistantId.value = assistantMessage.id as string - // contentParts 已由 buildOutgoingParts 包含 file entries，不要重复合并 attachments + // contentParts already includes file entries from buildOutgoingParts — do not re-merge attachments const body: Record = { agentId, message: content, @@ -1127,16 +1146,16 @@ export function useChat(options: UseChatOptions): UseChatReturn { } /** - * 运行中发送新消息：判断当前阶段是否可中断 - * - 可中断（thinking/streaming/executing_tool）：发送 interrupt 请求 - * - 不可中断（awaiting_approval）：排队 + * Send a new message while one is already generating. + * - Interruptible phases (thinking/streaming/executing_tool): send an interrupt request. + * - Non-interruptible phases (awaiting_approval): queue the message. */ const handleInterruptOrQueue = async (content: string, options: SendMessageOptions) => { const { conversationId, agentId } = options - // 不立即创建用户消息 —— 等 queued_input_started 再插入， - // 这样用户消息会出现在上一轮回答之后，保证正确的消息顺序。 - // 加入本地队列（保存 contentParts 以便延迟创建时使用） + // Do not create the user message immediately — wait for queued_input_started so the + // user message appears after the previous turn's reply, preserving correct ordering. + // Add to the local queue now (saves contentParts for delayed creation). messageQueue.enqueue(content, options.contentParts, conversationId) try { @@ -1151,14 +1170,14 @@ export function useChat(options: UseChatOptions): UseChatReturn { const result = await res.json() if (result.data?.interrupted) { - // 可中断：后端已发起中断，排队消息会被后端自动续跑 + // Interruptible: backend initiated the interrupt; queued message will auto-resume streamPhase.value = 'interrupting' messageQueue.markSending() } else if (result.data?.queued) { - // 不可中断但已排队：等当前步骤结束后自动续跑 + // Non-interruptible but queued: will auto-resume when the current step ends streamPhase.value = 'queued' } else { - // 没有活跃的流，直接发送 + // No active stream — send directly messageQueue.clear() createUserMessage(content, options.contentParts, conversationId) resetCurrentTurnState() @@ -1176,8 +1195,9 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } catch (e) { console.error('[useChat] Interrupt request failed:', e) - // interrupt 失败：后端从未收到这条消息，不能指望 heartbeat/queue 机制。 - // 回退为本地可见消息 + 清队列，避免消息静默丢失。 + // Interrupt failed: the backend never received the message, so the heartbeat/queue mechanism + // cannot be relied upon. Fall back to making the message locally visible + clear the queue + // to prevent silent message loss. const failedQueued = messageQueue.dequeue() if (failedQueued) { createUserMessage(failedQueued.content, failedQueued.contentParts, conversationId) @@ -1186,41 +1206,42 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // 停止生成（用户主动停止，不自动续跑） + // Stop generation (user-initiated; does not auto-resume queued messages). // - // 设计参考 claude-code-haha 的 useCancelRequest： - // 不立即断开 SSE，而是先发 stop 信号，等后端通过 SSE 返回 done 事件后再清理。 - // 这样 done 事件能正常到达，onStreamEnd 被触发，消息状态和会话列表都能正确更新。 - // 加一个 fallback timeout（3 秒），防止 done 事件因网络问题永远不到达。 + // Design: do not disconnect the SSE immediately — send a stop signal first and wait for the + // backend to return a 'done' event. This ensures onStreamEnd fires and message/conversation + // state is updated correctly. + // A 3-second fallback timeout guards against 'done' never arriving due to network issues. const stopGeneration = async () => { - // 在任何 await 之前冻结标识符 + 安装 fallback timer，防止 resetForNewConversation 并发清空后丢失上下文 + // Freeze identifiers and install the fallback timer before any await, so a concurrent + // resetForNewConversation cannot clear context out from under us. const convId = streamConversationId const assistantId = currentAssistantId.value - // 仅在前端确实在生成时才触发停止（含 SSE 接收中 / reconnect 中 / 审批等待中）。 - // 否则只是"旁观者"身份，不能把对方（渠道用户）的 agent run 也一起杀掉。 + // Only stop when the frontend is actively involved in the stream (receiving SSE / + // reconnecting / awaiting approval). As a bystander we must not kill another user's run. const activelyStreaming = isGenerating.value || streamPhase.value === 'reconnecting' || streamPhase.value === 'awaiting_approval' if (!activelyStreaming) { - // 没有真正在流 → 什么都不做，让调用方直接走 resetForNewConversation + // Not actually streaming — let the caller go straight to resetForNewConversation return } - // 先取消排队消息 + // Cancel queued message first messageQueue.clear() - // 标记为停止中（让 UI 立即反馈） + // Mark as stopped immediately so the UI gives instant feedback streamPhase.value = 'stopped' phaseInfo.value = null - // 在 await 之前安装 fallback timer，确保即使 resetForNewConversation 并发执行也不会遗漏 + // Install fallback timer before any await so it is not missed by a concurrent resetForNewConversation if (stopFallbackTimer) clearTimeout(stopFallbackTimer) stopFallbackTimer = setTimeout(() => { stopFallbackTimer = null console.warn('[useChat] Stop fallback: done event not received within 3s, force cleanup') - // 只有当 stream 仍属于旧会话时才 disconnect，防止误杀新会话的流 + // Only disconnect if the stream still belongs to the old conversation — avoids killing a new session's stream if (streamConversationId === convId || !streamConversationId) { stream.disconnect() } @@ -1234,7 +1255,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { }) }, 3000) - // 当 done 事件到达时，取消 fallback timer + // Cancel the fallback timer when the done/error event arrives const unsubscribe = stream.on('done', () => { if (stopFallbackTimer) { clearTimeout(stopFallbackTimer); stopFallbackTimer = null } unsubscribe() @@ -1244,7 +1265,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { unsubscribeError() }) - // 发送后端 stop 请求（fire-and-forget，不阻塞 resetForNewConversation） + // Send the backend stop request (fire-and-forget, does not block resetForNewConversation) if (convId) { fetchWithAuth(`${baseUrl}/api/v1/chat/${convId}/stop`, { method: 'POST', @@ -1254,25 +1275,25 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // 取消排队消息 + // Cancel the queued message const cancelQueued = () => { messageQueue.cancel() - // 通知后端清除排队消息（fire-and-forget） + // Notify backend (fire-and-forget) if (streamConversationId) { const headers: Record = { 'Content-Type': 'application/json' } if (token) headers.Authorization = `Bearer ${token}` - // 后端没有专门的取消排队 API，用 stop 的语义来处理 + // No dedicated cancel-queue API — the stop semantic covers this case } if (streamPhase.value === 'queued') { streamPhase.value = isGenerating.value ? 'streaming' : 'idle' } } - // 重连到运行中的流 + // Reconnect to a stream that is already running on the backend const reconnectStream = async (conversationId: string) => { if (isGenerating.value) return - // 清除残留的 stop fallback timer + // Clear any leftover stop fallback timer if (stopFallbackTimer) { clearTimeout(stopFallbackTimer); stopFallbackTimer = null } streamPhase.value = 'reconnecting' streamConversationId = conversationId @@ -1282,8 +1303,8 @@ export function useChat(options: UseChatOptions): UseChatReturn { resetCurrentTurnState() - // 清理尾部的空 assistant 消息（来自上一轮被误杀的 run 留下的空壳，或 placeholder 遗留）， - // 避免与即将重连产生的 streaming 气泡共存形成"重复两条"假象。 + // Remove trailing empty assistant messages left over from a killed run or a stale placeholder. + // Prevents "two bubbles" appearing when the reconnect creates a new streaming placeholder. while (messages.value.length > 0) { const tail = messages.value[messages.value.length - 1] if (tail && tail.role === 'assistant' @@ -1307,11 +1328,11 @@ export function useChat(options: UseChatOptions): UseChatReturn { }) } catch (e) { console.error('[useChat] Reconnect failed:', e) - // 重连失败：清理占位消息 + // Reconnect failed — clean up the placeholder message const msgIndex = messages.value.findIndex(m => m.id === currentAssistantId.value) if (msgIndex >= 0) { const msg = messages.value[msgIndex] - // 如果占位消息没有内容，移除它 + // Remove the placeholder if it has no content if (!msg.content && (!msg.contentParts || msg.contentParts.length === 0)) { messages.value.splice(msgIndex, 1) } else { @@ -1324,7 +1345,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { } } - // 重新生成 + // Regenerate a message const regenerate = async (messageId: string | number) => { const message = getMessage(messageId) if (!message) return @@ -1348,7 +1369,7 @@ export function useChat(options: UseChatOptions): UseChatReturn { }) } - /** 彻底重置流上下文 — 切换/新建会话时调用，确保旧流状态不污染新会话 */ + /** Fully reset stream context — call when switching or creating a conversation to prevent state pollution */ const resetForNewConversation = () => { stream.disconnect() streamConversationId = ''