diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index f189105f..a852cd14 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -743,10 +743,6 @@ public class NodeStreamingChatHelper { } // 4. 提取 token usage（通常最后一个 chunk 携带完整 usage） - // E-4 probe: log metadata keys to check for x-ratelimit-* headers - if (chatResponse.getMetadata() != null && log.isDebugEnabled()) { - log.debug("[E4-probe] metadata keys: {}", chatResponse.getMetadata().keySet()); - } if (chatResponse.getMetadata() != null && chatResponse.getMetadata().getUsage() != null) { var usage = chatResponse.getMetadata().getUsage(); if (usage.getPromptTokens() != null && usage.getPromptTokens() > 0) { diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java index a9cf1bba..ebc4b2e4 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java @@ -175,6 +175,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC // 防重保护：同 chatStructuredStream AtomicBoolean finalAnswerEmitted = new AtomicBoolean(false); AtomicBoolean finalThinkingEmitted = new AtomicBoolean(false); + AtomicReference lastEmittedStreamedContent = new AtomicReference<>(""); return compiledGraph.stream(inputs, config) .flatMapIterable(output -> { @@ -192,6 +193,14 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC boolean contentAlreadyStreamed = output.state().value(CONTENT_STREAMED, false); boolean thinkingAlreadyStreamed = output.state().value(THINKING_STREAMED, false); + // 与 chatStructuredStream 一致：把每轮 STREAMED_CONTENT 用 persistOnly 推给 Accumulator， + // 否则中间叙述（reasoning narrative + summarize）只在 SSE 上出现一次，刷新后丢失。 + String streamed = output.state().value(STREAMED_CONTENT).orElse(""); + if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) { + lastEmittedStreamedContent.set(streamed); + deltas.add(AgentService.StreamDelta.persistOnly(streamed, null)); + } + if (hasFinalAnswer(output) && finalAnswerEmitted.compareAndSet(false, true)) { String answer = extractFinalAnswer(output); if (answer != null && !answer.isEmpty()) { @@ -265,6 +274,9 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC // 用 compareAndSet 保证只取第一次，避免 content/thinking 被重复追加 AtomicBoolean finalAnswerEmitted = new AtomicBoolean(false); AtomicBoolean finalThinkingEmitted = new AtomicBoolean(false); + // STREAMED_CONTENT 是 REPLACE 策略（每轮 ReasoningNode/SummarizingNode 覆写）， + // 用 lastEmitted 跟踪已发送的值，避免在 ActionNode/ObservationNode 的 NodeOutput 上重复发送同一段内容。 + AtomicReference lastEmittedStreamedContent = new AtomicReference<>(""); return compiledGraph.stream(inputs, config) .flatMapIterable(output -> { @@ -287,6 +299,16 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC boolean thinkingAlreadyStreamed = output.state() .value(THINKING_STREAMED, false); + // 2a. 中间叙述内容持久化：每轮 ReasoningNode（带 tool_calls）和 SummarizingNode + // 都把当轮 LLM 输出写入 STREAMED_CONTENT。NodeStreamingChatHelper 已实时广播 + // 给前端，但 Accumulator 不在 SSE 订阅链路上，必须用 persistOnly StreamDelta + // 补一刀，否则刷新后正文文字全部丢失（只剩 final_answer + tool_call 卡片）。 + String streamed = output.state().value(STREAMED_CONTENT).orElse(""); + if (!streamed.isEmpty() && !streamed.equals(lastEmittedStreamedContent.get())) { + lastEmittedStreamedContent.set(streamed); + deltas.add(AgentService.StreamDelta.persistOnly(streamed, null)); + } + if (hasFinalAnswer(output) && finalAnswerEmitted.compareAndSet(false, true)) { String answer = extractFinalAnswer(output); if (answer != null && !answer.isEmpty()) { diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/SummarizingNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/SummarizingNode.java index 22fd8d6b..53b751ef 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/node/SummarizingNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/node/SummarizingNode.java @@ -183,6 +183,9 @@ public class SummarizingNode implements NodeAction { // 摘要的 content 已流式推送，但它不是最终回答，标记防重即可 .contentStreamed(true) .thinkingStreamed(!result.thinking().isEmpty()) + // 把当轮 summary 文本写入 STREAMED_CONTENT，让 StateGraphReActAgent 用 persistOnly + // StreamDelta 推给 Accumulator 持久化（用户刷新页面后能看到摘要正文，否则只剩 tool_call 卡片） + .streamedContent(summaryContent) .mergeUsage(state, result) // 不设 finishReason — summarizing 不是终止，循环继续 .events(List.of(GraphEventPublisher.phase("summarized", Map.of( diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index 10960dd2..3742fd18 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -738,6 +738,10 @@ public class ChatController { // 将 Disposable 注册到 StreamTracker，以便 stop 端点可以取消它 streamTracker.setDisposable(conversationId, disposable); + // JVM 关闭时优雅落盘：避免 mvn spring-boot:run 重启 / SIGTERM 把 + // 进行中 turn 的 assistant 消息丢失（doOnError 来不及在 Hikari 关闭前执行） + streamTracker.setEmergencySaveCallback(conversationId, + () -> emergencySaveAccumulator(conversationId, accumulator)); } catch (Exception e) { log.error("SSE setup error: {}", e.getMessage()); @@ -773,12 +777,15 @@ public class ChatController { } /** - * 中断当前流并排队一条后续消息。 + * 在执行中追加一条后续消息：仅入队，等当前 turn 自然结束后再启动。 *

- * 与 stop 的区别：interrupt 会在当前 turn 安全结束后自动启动排队消息。 - * 如果当前阶段不可中断（awaiting_approval），消息会被排队但不打断当前执行。 + * 对齐 Claude Code 行为：流式输出过程中收到新输入不会强制 dispose 当前 LLM 调用， + * 而是仅入队。当前 turn 跑到 doOnComplete/doOnError 后由 startQueuedMessage 接管。 + *

+ * 旧行为（dispose 当前 disposable + 立即重启）会导致部分 LLM 输出被丢弃 + token 浪费， + * 已废弃。如果未来需要"立即打断"语义，应该走 /stop（用户主动取消）+ 重新发起新消息的路径。 */ - @Operation(summary = "中断并排队后续消息") + @Operation(summary = "排队后续消息（不打断当前流）") @PostMapping("/{conversationId}/interrupt") public R> interruptStream( @PathVariable String conversationId, @@ -797,34 +804,20 @@ public class ChatController { Long agentId = request.getAgentId(); List contentParts = request.getContentParts(); - // 判断当前阶段是否可中断 - // awaiting_approval 阶段不直接中断，只排队 + // 判断当前阶段（仅用于 reason 字段，行为对所有阶段一致：仅入队） boolean isAwaitingApproval = approvalService.findPendingByConversation(conversationId) != null; - if (isAwaitingApproval) { - // 不可中断：排队但不打断。先持久化（含 contentParts）再入队（persisted=true） - conversationService.saveMessage(conversationId, "user", message, contentParts, "queued"); - boolean queued = streamTracker.enqueueMessage(conversationId, message, agentId, true); - log.info("Interrupt requested during approval, message queued: conversationId={}, user={}, queueSize={}", - conversationId, username, streamTracker.getQueueSize(conversationId)); - return R.ok(Map.of( - "interrupted", false, - "queued", queued, - "reason", "awaiting_approval" - )); - } - - // 可中断：先持久化（含 contentParts）再打断并入队（persisted=true） - conversationService.saveMessage(conversationId, "user", message, contentParts, "queued"); - boolean interrupted = streamTracker.requestInterrupt(conversationId, message, agentId, true); - log.info("Interrupt requested: conversationId={}, user={}, interrupted={}, queueSize={}", - conversationId, username, interrupted, streamTracker.getQueueSize(conversationId)); + // 仅入队、不 dispose。延迟持久化到 startQueuedMessage（让 Asst-N 先在 doOnComplete 落库， + // 否则 listMessages ORDER BY create_time ASC 会把 Q(N+1) 排到 Asst-N 前面） + boolean queued = streamTracker.enqueueMessage(conversationId, message, agentId, false, contentParts); + log.info("Enqueued follow-up message during running turn: conversationId={}, user={}, queueSize={}, awaitingApproval={}", + conversationId, username, streamTracker.getQueueSize(conversationId), isAwaitingApproval); return R.ok(Map.of( - "interrupted", interrupted, - "queued", true, + "interrupted", false, + "queued", queued, "queueSize", streamTracker.getQueueSize(conversationId), - "reason", interrupted ? "interrupted" : "queued" + "reason", isAwaitingApproval ? "awaiting_approval" : "queued" )); } @@ -1009,9 +1002,12 @@ public class ChatController { log.info("Starting queued message: conversationId={}, agentId={}, message={}", conversationId, agentId, queuedMessage.substring(0, Math.min(30, queuedMessage.length()))); - // 持久化排队的用户消息（幂等：如果 /interrupt 已提前持久化则跳过） + // 持久化排队的用户消息（含 contentParts；幂等：如果 /interrupt 已提前持久化则跳过）。 + // 这里持久化是为了确保 user 消息在 assistant 消息（doOnError/doOnCancel 已写入）之后落库， + // 让 listMessages ORDER BY create_time ASC 后顺序正确：Q1 → Asst1 → Q2 → Asst2。 if (queuedMessage != null && !queuedMessage.isBlank() && !preConsumedInput.persisted()) { - conversationService.saveMessage(conversationId, "user", queuedMessage); + conversationService.saveMessage(conversationId, "user", queuedMessage, + preConsumedInput.contentParts(), "queued"); } // 广播 queued_input_started 事件 @@ -1122,6 +1118,8 @@ public class ChatController { }) .subscribe(); streamTracker.setDisposable(conversationId, disposable); + streamTracker.setEmergencySaveCallback(conversationId, + () -> emergencySaveAccumulator(conversationId, accumulator)); } private void sendEvent(SseEmitter emitter, String name, Object data) throws IOException { @@ -1170,6 +1168,40 @@ public class ChatController { return payload; } + /** + * Snapshot the accumulator and persist it as an assistant message during JVM shutdown. + * Invoked from {@link ChatStreamTracker#onShutdown()} so any in-flight turn doesn't + * lose its already-streamed content + tool calls when the process exits. + *

+ * Idempotent w.r.t. the normal doOnComplete/doOnError save: if those paths already + * persisted the message, this writes a second row with status="interrupted_shutdown", + * which is rare in practice (race window is sub-second between dispose and save) and + * acceptable. Skipping save when nothing to save avoids empty rows. + */ + private void emergencySaveAccumulator(String conversationId, StreamAccumulator accumulator) { + try { + String text = accumulator.getContent(); + List parts = accumulator.toAssistantParts(); + if (text.isBlank() && parts.isEmpty()) { + return; + } + String savedText = text.isBlank() ? "[已中断 — 服务重启]" : text; + conversationService.saveMessage(conversationId, "assistant", savedText, parts, + "interrupted_shutdown", + accumulator.getPromptTokens(), + accumulator.getCompletionTokens(), + accumulator.getRuntimeModelName(), + accumulator.getRuntimeProviderId(), + accumulator.toMetadataJson()); + log.info("[ChatController] Emergency-saved in-flight assistant message: " + + "conversationId={}, textLen={}, partsCount={}", + conversationId, text.length(), parts.size()); + } catch (Exception e) { + log.error("[ChatController] Emergency save failed for {}: {}", + conversationId, e.getMessage(), e); + } + } + private List normalizeRequestParts(ChatStreamRequest request) { if (request.getContentParts() != null && !request.getContentParts().isEmpty()) { return request.getContentParts(); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java index a3fc1f1d..bf1f2b5d 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java @@ -1,10 +1,12 @@ package vip.mate.channel.web; import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import reactor.core.Disposable; +import vip.mate.workspace.conversation.model.MessageContentPart; import java.io.IOException; import java.util.ArrayList; @@ -104,6 +106,18 @@ public class ChatStreamTracker { /** 排队的用户消息队列（支持多条排队消息，按序消费） */ final java.util.Queue messageQueue = new java.util.concurrent.ConcurrentLinkedQueue<>(); + /** + * Emergency save callback registered by the SSE chain owner (ChatController). + * Invoked from {@link #onShutdown()} so the accumulated assistant content + tool_calls + * are persisted before the JVM tears down — without this, a `mvn spring-boot:run` + * restart wipes any in-flight turn and leaves only the user message in DB. + *

+ * The callback must be idempotent (will not be called twice for the same run, but + * may race with normal doOnComplete/doOnError; both paths must tolerate the other + * having saved already). + */ + volatile Runnable emergencySaveCallback; + /** 心跳定时器 */ volatile ScheduledFuture heartbeatFuture; @@ -190,6 +204,18 @@ public class ChatStreamTracker { } } + /** + * Register an emergency-save callback for this run, invoked from {@link #onShutdown()} + * before the JVM tears down. The callback should snapshot the current accumulator + * state and persist it as the assistant message (status="interrupted"). + */ + public void setEmergencySaveCallback(String conversationId, Runnable callback) { + RunState state = runs.get(conversationId); + if (state != null) { + state.emergencySaveCallback = callback; + } + } + /** * 请求停止指定会话的流。 * 取消 Flux 订阅（底层 HTTP 连接也会随之关闭），返回 true 表示确实停止了正在运行的流。 @@ -577,6 +603,11 @@ public class ChatStreamTracker { * @return true 如果成功请求了中断 */ public boolean requestInterrupt(String conversationId, String queuedMessage, Long agentId, boolean persisted) { + return requestInterrupt(conversationId, queuedMessage, agentId, persisted, null); + } + + public boolean requestInterrupt(String conversationId, String queuedMessage, Long agentId, + boolean persisted, List contentParts) { RunState state = runs.get(conversationId); if (state == null || state.done) { return false; @@ -589,7 +620,7 @@ public class ChatStreamTracker { Disposable d = state.disposable; canInterrupt = d != null && !d.isDisposed(); // 无论是否可中断，都入队（支持多条排队消息） - state.messageQueue.offer(new QueuedInput(queuedMessage, agentId, persisted)); + state.messageQueue.offer(new QueuedInput(queuedMessage, agentId, persisted, contentParts)); if (canInterrupt) { state.interruptType = InterruptType.USER_INTERRUPT_WITH_FOLLOWUP; state.stopRequested.set(true); @@ -636,11 +667,16 @@ public class ChatStreamTracker { * 将消息加入队列但不中断当前执行（用于不可中断阶段）。 */ public boolean enqueueMessage(String conversationId, String message, Long agentId, boolean persisted) { + return enqueueMessage(conversationId, message, agentId, persisted, null); + } + + public boolean enqueueMessage(String conversationId, String message, Long agentId, boolean persisted, + List contentParts) { RunState state = runs.get(conversationId); if (state == null || state.done) { return false; } - state.messageQueue.offer(new QueuedInput(message, agentId, persisted)); + state.messageQueue.offer(new QueuedInput(message, agentId, persisted, contentParts)); // broadcast 在锁外 try { String json = objectMapper.writeValueAsString(Map.of( @@ -656,9 +692,14 @@ public class ChatStreamTracker { } /** - * 排队输入的原子快照（message + agentId + persisted 一起返回，避免分离读取导致不一致） + * 排队输入的原子快照（message + agentId + persisted + contentParts 一起返回，避免分离读取导致不一致） */ - public record QueuedInput(String message, Long agentId, boolean persisted) {} + public record QueuedInput(String message, Long agentId, boolean persisted, + List contentParts) { + public QueuedInput(String message, Long agentId, boolean persisted) { + this(message, agentId, persisted, null); + } + } /** * 原子消费排队的输入（流完成/中断后调用）。 @@ -905,4 +946,64 @@ public class ChatStreamTracker { evicted, runs.size()); } } + + /** + * Flush in-flight runs before JVM shutdown. + *

+ * Spring closes singleton beans in reverse construction order; ConversationService / + * Hikari outlive ChatStreamTracker, so saveMessage from {@link #onShutdown()} still + * has a working DB connection. Without this, a {@code mvn spring-boot:run} restart or + * SIGTERM during a turn races against the Reactor cancellation: the doOnError / + * doOnComplete saveMessage may not run before HikariPool shuts down, leaving the + * conversation with only the user message and no assistant reply (the + * "对话框里除了问题外什么也没留下" symptom seen in production logs at 07:23:02). + *

+ * Behavior: + *

    + *
  1. Walk every active (not-done) RunState.
    2. + *
  2. Invoke its registered emergencySaveCallback synchronously — the callback + * (set by ChatController) snapshots the current accumulator and persists it + * as an "interrupted" assistant message.
    3. + *
  3. Dispose the Reactor disposable so the LLM stream terminates promptly.
    4. + *
+ * The callback must tolerate normal doOnError/doOnComplete having raced and saved + * already; the latest commit wins for that conversation. + */ + @PreDestroy + public void onShutdown() { + int active = (int) runs.values().stream().filter(s -> !s.done).count(); + if (active == 0) { + log.info("[ChatStreamTracker] Shutdown: no active runs to flush"); + return; + } + log.warn("[ChatStreamTracker] Shutdown: flushing {} active run(s) before JVM exit", + active); + for (Map.Entry entry : runs.entrySet()) { + RunState state = entry.getValue(); + if (state.done) continue; + String cid = entry.getKey(); + try { + Runnable callback = state.emergencySaveCallback; + if (callback != null) { + log.info("[ChatStreamTracker] Emergency-saving in-flight run: {}", cid); + callback.run(); + } else { + log.warn("[ChatStreamTracker] No emergency-save callback for active run: {} " + + "(content may be lost)", cid); + } + } catch (Exception e) { + log.error("[ChatStreamTracker] Emergency save failed for {}: {}", + cid, e.getMessage(), e); + } + try { + Disposable d = state.disposable; + if (d != null && !d.isDisposed()) { + d.dispose(); + } + } catch (Exception e) { + log.warn("[ChatStreamTracker] Disposable.dispose failed for {}: {}", + cid, e.getMessage()); + } + } + } } diff --git a/mateclaw-server/src/main/java/vip/mate/config/GraphObservationProperties.java b/mateclaw-server/src/main/java/vip/mate/config/GraphObservationProperties.java index f93d85fc..718d931d 100644 --- a/mateclaw-server/src/main/java/vip/mate/config/GraphObservationProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/config/GraphObservationProperties.java @@ -4,7 +4,13 @@ import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; /** - * Graph 观察结果处理阈值配置 + * Graph observation thresholds for triggering summarize / compaction. + *

+ * Tune values in {@code application.yml} under {@code mate.agent.graph.observation}, + * not in the Java field defaults. The yml is the source of truth and is loaded into + * a Spring bean by {@link ConfigurationProperties}; field defaults below are kept as a + * conservative fallback for tests / unit constructors and intentionally do NOT reflect + * production-tuned values. * * @author MateClaw Team */ @@ -12,27 +18,27 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "mate.agent.graph.observation") public class GraphObservationProperties { - /** 单次工具结果最大字符数 */ + /** Max chars retained per tool result (truncate above this). Tune via yml. */ private int maxSingleObservationChars = 4000; - /** 所有观察记录总字符数上限 */ + /** Total observation chars across history+current that trigger summarize. Tune via yml. */ private int maxTotalObservationChars = 12000; - /** 单次结果超过此阈值视为"大结果" */ + /** Single tool result above this is treated as "large" and triggers summarize. Tune via yml. */ private int largeResultThreshold = 3000; - /** 触发 summarize 的最小观察轮次 */ + /** Minimum observation rounds that triggers summarize as a runaway-loop safety net. Tune via yml. */ private int minRoundsForSummarize = 3; - /** 截断时保留前部占比（0-1） */ + /** Truncation: head fraction kept (0-1). */ private double headRatio = 0.4; - /** 截断省略标记（%d 会被替换为原始字符数） */ + /** Truncation marker; %d is replaced with original char count. */ private String truncationMarker = "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n"; - /** 检测到尾部错误模式时的 tail 保留比例（默认 0.8，优先保留错误信息） */ + /** Tail-keep ratio when an error pattern is detected at the tail (prefer keeping error info). */ private double errorTailRatio = 0.8; - /** 截断时最少保留字符数（避免过度截断导致信息完全丢失） */ + /** Minimum chars retained on truncation to avoid losing all information. */ private int minKeepChars = 2000; } diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index cca6a822..8859daf5 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -154,10 +154,11 @@ mate: agent: graph: observation: - max-single-observation-chars: 8000 - max-total-observation-chars: 24000 - large-result-threshold: 6000 - min-rounds-for-summarize: 3 + # 与 GraphObservationProperties.java 默认值对齐，参考 openclaw token-budget 设计 + max-single-observation-chars: 16000 + max-total-observation-chars: 200000 + large-result-threshold: 32000 + min-rounds-for-summarize: 25 head-ratio: 0.4 truncation-marker: "\n\n... [内容已截断，共 %d 字符，保留前后关键片段] ...\n\n" tool: diff --git a/mateclaw-ui/src/composables/chat/useChat.ts b/mateclaw-ui/src/composables/chat/useChat.ts index d1ed253a..f460d555 100644 --- a/mateclaw-ui/src/composables/chat/useChat.ts +++ b/mateclaw-ui/src/composables/chat/useChat.ts @@ -288,18 +288,21 @@ export function useChat(options: UseChatOptions): UseChatReturn { if (streamPhase.value !== 'summarizing_observations') { streamPhase.value = options.thinkingLevel?.value === 'off' ? 'streaming' : 'thinking' } - // Segments: all thinking deltas merge into one segment (not split by tool_call interruptions) + // Segments: per-round thinking. When tool_call_started / phase change closes the running + // thinking segment (status='completed'), a fresh thinking_delta opens a new segment instead + // of reopening the closed one. Without this split, multi-round ReAct (3 reasoning + 2 + // summarizing rounds) accumulates 9K+ chars in a single bubble. const segs = currentSegments.value - // Reuse an existing thinking segment regardless of status (running or completed) - let thinkSeg = segs.find((s: MessageSegment) => s.type === 'thinking') + let thinkSeg = segs.findLast((s: MessageSegment) => + s.type === 'thinking' && s.status === 'running' + ) if (!thinkSeg) { thinkSeg = { id: genSegId(), type: 'thinking', status: 'running', thinkingText: '', timestamp: Date.now() } - // Insert at front — thinking always appears at the top - segs.unshift(thinkSeg) + // Append in timeline order (interleaved with tool_calls) — old behavior unshift'd to top, + // but with per-round splitting that misorders rounds 2+ relative to their tool calls. + segs.push(thinkSeg) flushSegmentsToMessage() } - // Re-mark as running when new thinking content arrives - thinkSeg.status = 'running' thinkSeg.thinkingText = (thinkSeg.thinkingText || '') + (data.delta || '') } }) @@ -596,6 +599,15 @@ export function useChat(options: UseChatOptions): UseChatReturn { metadata: { ...metadata, currentPhase: data.phase } } as any) } + // Close any running thinking/content segment on phase transition so the next thinking_delta + // (e.g. summarizing → reasoning) starts a fresh round-scoped segment instead of growing the + // previous one unbounded. + const segs = currentSegments.value + for (const seg of segs) { + if (seg.status === 'running' && (seg.type === 'thinking' || seg.type === 'content')) { + seg.status = 'completed' + } + } } })