diff --git a/cmd/gui/renderer/app.js b/cmd/gui/renderer/app.js
index 357cf1d..c4dc67d 100644
--- a/cmd/gui/renderer/app.js
+++ b/cmd/gui/renderer/app.js
@@ -1166,6 +1166,9 @@ function buildChatLayout() {
'' +
+ '" +
'" +
@@ -2036,6 +2039,7 @@ function buildChatStarmapGraph() {
async function sendChat() {
var inp = document.getElementById("chat-input");
var btn = document.getElementById("chat-send-btn");
+ var stopBtn = document.getElementById("chat-stop-btn");
var text = inp.value.trim();
if (!text || state.chatLoading) return;
if (state.currentConn && state.currentConn.type === "cli") {
@@ -2057,6 +2061,7 @@ async function sendChat() {
state.chatStage = __("等待AI回复...", "Waiting for AI...");
btn.disabled = true;
btn.textContent = "";
+ if (stopBtn) stopBtn.style.display = ""; // 生成期间可停止
rerenderChat();
try {
// webui 连接：同步 POST 等完整回复（服务端 X-Trigger-Only 也返回 response；SSE 公网不稳时靠同步兜底）
@@ -2124,10 +2129,26 @@ async function sendChat() {
state.chatStage = "";
btn.disabled = false;
btn.textContent = __("发送", "Send");
+ var sb2 = document.getElementById("chat-stop-btn");
+ if (sb2) sb2.style.display = "none"; // 回复完成/失败，隐藏停止按钮
rerenderChat();
}
}
+// 停止生成 / 发送中断消息。核心拦截语义：有 LLM 在跑则取消当前请求并以
+// [中断消息] 重启轮次；无则在跑则作为普通消息处理。
+async function interruptChat() {
+ try {
+ await api("/chat/interrupt", {
+ method: "POST",
+ body: JSON.stringify({}),
+ });
+ toast(__("已发送中断信号", "Interrupt signal sent"));
+ } catch (e) {
+ toast(__("中断失败: ", "Interrupt failed: ") + e.message, true);
+ }
+}
+
async function queryMemoryChat() {
var q = document.getElementById("mem-query")?.value;
var r = document.getElementById("mem-result-chat");
diff --git a/cmd/waiter/builtin.go b/cmd/waiter/builtin.go
index 9272d2e..7a9de6c 100644
--- a/cmd/waiter/builtin.go
+++ b/cmd/waiter/builtin.go
@@ -12,6 +12,7 @@ func handleBuiltin(cmd string, cfg *Config, state *State, reconnect func(), out
case cmd == "/help":
fmt.Fprintln(out, `Built-in commands:
/help show this help
+ /stop [msg] stop generation / send interrupt (alias /interrupt)
/exit, /quit exit waiter
/clear clear screen
/reconnect force reconnection
diff --git a/cmd/waiter/main.go b/cmd/waiter/main.go
index bb86801..c430687 100644
--- a/cmd/waiter/main.go
+++ b/cmd/waiter/main.go
@@ -202,10 +202,15 @@ func runInteractive(state *State, cfg *Config) {
}
func oneshot(state *State, msg string) {
+ var sr streamRender
stop := startSpinner("thinking...")
resp, err := state.SendChatStream(msg, func(rl respLine) {
// 第一个过程帧到达即停转，后续帧直接渲染
stop()
+ if sr.handleDelta(rl) {
+ return // delta 已增量渲染
+ }
+ sr.reset() // 聚合帧/工具帧：结束 delta 流，换行输出
printServerEvent(rl)
})
stop()
@@ -347,6 +352,56 @@ loop:
}
}
+// streamRender 累积 token 级 delta 帧并增量重绘当前行。
+// 聚合帧（reasoning/tool_call/response）到达时清空累积状态（该轮已结束）。
+// 旧服务器不发 delta，此结构始终为空，行为与原来完全一致。
+type streamRender struct {
+ reasoning strings.Builder
+ content strings.Builder
+}
+
+// handleDelta 处理 delta 帧；返回是否消费了该帧。
+// reset=true 的空帧表示服务端轮次作废（用户中断）：清空累积并定格已显示内容。
+func (sr *streamRender) handleDelta(rl respLine) bool {
+ switch rl.Type {
+ case "reasoning_delta":
+ if rl.Reset {
+ sr.reasoning.Reset()
+ fmt.Print(clearLine)
+ return true
+ }
+ sr.reasoning.WriteString(rl.Content)
+ if colors {
+ fmt.Printf("%s%s· %s%s", clearLine, colorDim, sr.reasoning.String(), colorReset)
+ } else {
+ fmt.Printf("%s[思考] %s", clearLine, sr.reasoning.String())
+ }
+ return true
+ case "content_delta":
+ if rl.Reset {
+ sr.content.Reset()
+ fmt.Print(clearLine + "\n") // 定格已显示的部分内容，换行
+ return true
+ }
+ sr.content.WriteString(rl.Content)
+ if colors {
+ fmt.Printf("%s%s%s%s", clearLine, colorGreen, sr.content.String(), colorReset)
+ } else {
+ fmt.Printf("%s%s", clearLine, sr.content.String())
+ }
+ return true
+ }
+ return false
+}
+
+// reset 在收到聚合帧/工具帧时调用：delta 流被打断或结束，
+// 下一行输出不再覆盖 delta 内容。
+func (sr *streamRender) reset() {
+ sr.reasoning.Reset()
+ sr.content.Reset()
+ fmt.Print(clearLine + "\n")
+}
+
// printServerOutput 渲染一行服务器输出（JSON 帧）。
func printServerOutput(content string) {
rl := parseRespLineStruct(content)
@@ -369,8 +424,10 @@ func printServerEvent(rl respLine) {
// renderPlain 无色模式下的纯文本渲染。
func renderPlain(rl respLine, raw string) string {
switch rl.Type {
- case "reasoning":
+ case "reasoning", "reasoning_delta":
return "[思考] " + rl.Content
+ case "content_delta":
+ return rl.Content
case "tool_call":
return fmt.Sprintf("[工具] %s (%s) %s", rl.Tool, rl.Status, rl.Result)
case "response":
diff --git a/cmd/waiter/state.go b/cmd/waiter/state.go
index 7bf67cd..03579cf 100644
--- a/cmd/waiter/state.go
+++ b/cmd/waiter/state.go
@@ -126,6 +126,7 @@ type respLine struct {
Tool string `json:"tool"`
Status string `json:"status"`
Result string `json:"result"`
+ Reset bool `json:"reset,omitempty"` // delta 帧：服务端轮次作废，清空累积
}
// parseRespLineStruct 解析一行 JSON 响应帧，解析失败时将原文放入 Content。
diff --git a/cmd/waiter/tui.go b/cmd/waiter/tui.go
index 7ce7e7e..051b2ba 100644
--- a/cmd/waiter/tui.go
+++ b/cmd/waiter/tui.go
@@ -69,6 +69,7 @@ type chatMsg struct {
tool string // msgTool: 工具名
status string // msgTool: ok/denied/interrupted/error/running
result string // msgTool: 结果预览
+ final bool // msgAgent: 流式消息已完成（后续 delta 不再追加）
}
// ---- tea.Msg ----
@@ -375,18 +376,61 @@ func (m *tuiModel) handleServerLine(line string) {
} else {
m.append(cm)
}
+ case "reasoning_delta":
+ // token 级增量：与 reasoning 同样合并到最后一条 reasoning 消息
+ if rl.Reset {
+ m.messages = []chatMsg{}
+ break
+ }
+ if n := len(m.messages); n > 0 && m.messages[n-1].kind == msgReasoning {
+ m.messages[n-1].text += rl.Content
+ } else if rl.Content != "" {
+ m.append(chatMsg{kind: msgReasoning, text: rl.Content})
+ }
+ case "content_delta":
+ // token 级增量：追加到最后一条 agent 消息（流式生成中的回复）
+ if rl.Reset {
+ m.sealLastAgent()
+ break
+ }
+ if n := len(m.messages); n > 0 && m.messages[n-1].kind == msgAgent && !m.messages[n-1].final {
+ m.messages[n-1].text += rl.Content
+ } else if rl.Content != "" {
+ m.append(chatMsg{kind: msgAgent, text: rl.Content})
+ }
case "tool_call":
+ // 工具调用打断内容流：置 final 防止后续 delta 误追加到旧消息
+ m.sealLastAgent()
m.append(chatMsg{kind: msgTool, tool: rl.Tool, status: rl.Status, result: rl.Result})
case "response":
+ // 聚合最终响应：覆盖/替换 delta 累积的最后一条 agent 消息（内容相同），
+ // 或在无 delta 时新建。置 final 标记本轮完成。
m.busy = false
- m.append(chatMsg{kind: msgAgent, text: rl.Content})
+ if n := len(m.messages); n > 0 && m.messages[n-1].kind == msgAgent && !m.messages[n-1].final {
+ if rl.Content != "" {
+ m.messages[n-1].text = rl.Content // 以聚合为准（含 stage 插件改写后的最终文本）
+ }
+ m.messages[n-1].final = true
+ } else {
+ cm := chatMsg{kind: msgAgent, text: rl.Content, final: true}
+ m.append(cm)
+ }
case "error":
m.busy = false
+ m.sealLastAgent()
m.append(chatMsg{kind: msgError, text: rl.Error})
default:
// 非 JSON 旧行（旧服务器）：当最终输出
m.busy = false
- m.append(chatMsg{kind: msgAgent, text: line})
+ m.append(chatMsg{kind: msgAgent, text: line, final: true})
+ }
+ m.refreshViewport()
+}
+
+// sealLastAgent 将最后一条未完成的 agent 流式消息标记为完成。
+func (m *tuiModel) sealLastAgent() {
+ if n := len(m.messages); n > 0 && m.messages[n-1].kind == msgAgent {
+ m.messages[n-1].final = true
}
}
diff --git a/internal/agent/core/process.go b/internal/agent/core/process.go
index c43a5d5..e746681 100644
--- a/internal/agent/core/process.go
+++ b/internal/agent/core/process.go
@@ -315,11 +315,28 @@ func chatStreamWithFallback(ctx context.Context, p agentAPI.Provider, req *agent
}
resp, accErr := accumulateStream(ctx, ch, a)
+
+ // 中断/超时取消必须保持取消语义传给调用方（与原 Chat() 行为一致：
+ // 被 cancel 时丢弃已收内容返回 err），让 process() 的 continue 分支
+ // 重启轮次并以 [中断消息] 注入打断内容。绝不能把部分内容当成功返回，
+ // 否则用户打断会被无视、继续执行工具/输出。
+ if errors.Is(accErr, context.Canceled) || errors.Is(accErr, context.DeadlineExceeded) {
+ // 通知客户端：本轮流式作废，清空 delta 累积并定格已显示内容
+ if a != nil {
+ a.publishEvent(events.EventContentDelta, map[string]interface{}{
+ "content": "",
+ "channel": a.currentOutputChannel,
+ "reset": true,
+ })
+ }
+ return resp, accErr
+ }
+
if accErr == nil {
return resp, nil
}
- // 流中途错误：若已累积到内容则返回部分结果，否则回退非流式
+ // 其他错误（网络中断等）：已累积到实质内容则返回部分结果，否则回退非流式
if resp != nil && (resp.Content != "" || len(resp.ToolCalls) > 0) {
log.Printf("[agent] stream interrupted mid-way (%v), returning partial result", accErr)
return resp, nil
diff --git a/internal/agent/core/stream_accumulate_test.go b/internal/agent/core/stream_accumulate_test.go
new file mode 100644
index 0000000..ba6838c
--- /dev/null
+++ b/internal/agent/core/stream_accumulate_test.go
@@ -0,0 +1,69 @@
+package core
+
+import (
+ "context"
+ "testing"
+
+ agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+)
+
+// 验证流式 tool call 分片累积：模拟 llmsproxy/big-pickle 的分片序列
+func TestAccumulateStreamToolCalls(t *testing.T) {
+ ch := make(chan agentAPI.StreamChunk, 10)
+ go func() {
+ // 分片1: name + id + arguments 开头
+ ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
+ {ID: "call_1", Name: "cmd_run", RawArguments: "{\""},
+ }}
+ // 分片2-3: 只有 arguments 分片
+ ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
+ {RawArguments: "command\""},
+ }}
+ ch <- agentAPI.StreamChunk{ToolCalls: []agentAPI.ToolCall{
+ {RawArguments: ":\"date\"}"},
+ }}
+ ch <- agentAPI.StreamChunk{Done: true, FinishReason: "tool_calls"}
+ close(ch)
+ }()
+
+ resp, err := accumulateStream(context.Background(), ch, nil)
+ if err != nil {
+ t.Fatalf("accumulateStream: %v", err)
+ }
+ if len(resp.ToolCalls) != 1 {
+ t.Fatalf("want 1 tool call, got %d", len(resp.ToolCalls))
+ }
+ tc := resp.ToolCalls[0]
+ if tc.Name != "cmd_run" || tc.ID != "call_1" {
+ t.Fatalf("bad name/id: %s/%s", tc.ID, tc.Name)
+ }
+ cmd, _ := tc.Arguments["command"].(string)
+ if cmd != "date" {
+ t.Fatalf("arguments not merged, got: %v", tc.Arguments)
+ }
+ if resp.FinishReason != "tool_calls" {
+ t.Fatalf("finish reason: %q", resp.FinishReason)
+ }
+}
+
+// 验证 content/reasoning 增量累积
+func TestAccumulateStreamContent(t *testing.T) {
+ ch := make(chan agentAPI.StreamChunk, 5)
+ go func() {
+ ch <- agentAPI.StreamChunk{ReasoningContent: "think "}
+ ch <- agentAPI.StreamChunk{Content: "你"}
+ ch <- agentAPI.StreamChunk{Content: "好"}
+ ch <- agentAPI.StreamChunk{Done: true, FinishReason: "stop"}
+ close(ch)
+ }()
+ resp, err := accumulateStream(context.Background(), ch, nil)
+ if err != nil {
+ t.Fatalf("accumulateStream: %v", err)
+ }
+ if resp.Content != "你好" {
+ t.Fatalf("content: %q", resp.Content)
+ }
+ if resp.ReasoningContent != "think " {
+ t.Fatalf("reasoning: %q", resp.ReasoningContent)
+ }
+}
diff --git a/internal/plugins/cli/plugin.go b/internal/plugins/cli/plugin.go
index 0f705bc..5e1a5f9 100644
--- a/internal/plugins/cli/plugin.go
+++ b/internal/plugins/cli/plugin.go
@@ -211,8 +211,33 @@ func (p *Plugin) handleChat(w *connWriter, line string, s *sdk.PluginSDK) {
"result": truncateOneLine(result, 160),
})
})
+ // token 级流式增量帧：客户端可选订做逐 token 渲染。
+ // 旧客户端收到未知 type 会忽略；聚合 reasoning/response 帧仍照常发送，
+ // 保证旧/新客户端最终都能看到完整文本。
+ unsubReasoningDelta := s.Subscribe(sdk.EventReasoningDelta, func(evt *sdk.Event) {
+ if ch, _ := evt.Payload["channel"].(string); ch != cliChannel {
+ return
+ }
+ content, _ := evt.Payload["content"].(string)
+ if content == "" {
+ return
+ }
+ w.writeLine(map[string]interface{}{"type": "reasoning_delta", "content": content})
+ })
+ unsubContentDelta := s.Subscribe(sdk.EventContentDelta, func(evt *sdk.Event) {
+ if ch, _ := evt.Payload["channel"].(string); ch != cliChannel {
+ return
+ }
+ content, _ := evt.Payload["content"].(string)
+ if content == "" {
+ return
+ }
+ w.writeLine(map[string]interface{}{"type": "content_delta", "content": content})
+ })
defer unsubReasoning()
defer unsubToolCall()
+ defer unsubReasoningDelta()
+ defer unsubContentDelta()
resp := s.InjectTextSync(cliSource, cliChannel, line)
if resp != nil {
@@ -260,6 +285,8 @@ func (p *Plugin) handleBuiltin(conn net.Conn, line string, s *sdk.PluginSDK) boo
switch parts[0] {
case "/help":
p.cmdHelp(conn)
+ case "/stop", "/interrupt":
+ p.cmdInterrupt(conn, parts, s)
case "/status":
p.cmdStatus(conn, s)
case "/kernel":
@@ -285,6 +312,7 @@ func (p *Plugin) cmdHelp(conn net.Conn) {
"type": "response",
"content": `内置命令（直接对话内核，不依赖网络）:
/help 显示此帮助
+ /stop [消息] 停止当前生成/发送中断消息（别名 /interrupt）
/status 系统运行状态
/kernel 内核状态（插件、工具、LLM、记忆）
/settings 列出所有配置
@@ -304,6 +332,36 @@ func (p *Plugin) cmdHelp(conn net.Conn) {
})
}
+// ======== /stop ========
+
+// cmdInterrupt 注入用户中断。核心拦截语义（interceptLoop）：
+// - 有 LLM 在跑：cancelLLM 取消当前流式请求，中断入队，process() 以
+// [中断消息] 重启轮次（模型看到被打断的上下文 + 用户新输入）；
+// - 无 LLM 在跑：作为普通输入处理（等同发了一条消息）。
+//
+// 可选附带消息：/stop 换个话题（空参数 = 纯取消）。
+func (p *Plugin) cmdInterrupt(conn net.Conn, parts []string, s *sdk.PluginSDK) {
+ msg := strings.TrimSpace(strings.TrimPrefix(line2(parts), "/stop"))
+ if alias := strings.TrimSpace(strings.TrimPrefix(line2(parts), "/interrupt")); alias != "" {
+ msg = alias
+ }
+ s.InjectInterrupt(cliSource, cliChannel, "text", map[string]interface{}{
+ "content": msg,
+ })
+ writeLine(conn, map[string]interface{}{
+ "type": "response",
+ "content": "已发送中断信号",
+ })
+}
+
+// line2 将命令行参数重组为原始字符串（保留词间空格，去掉首 token）。
+func line2(parts []string) string {
+ if len(parts) < 2 {
+ return ""
+ }
+ return strings.Join(parts[1:], " ")
+}
+
// ======== /status ========
func (p *Plugin) cmdStatus(conn net.Conn, s *sdk.PluginSDK) {
diff --git a/internal/plugins/webui/dashboard.html b/internal/plugins/webui/dashboard.html
index 7c671e4..632eea9 100644
--- a/internal/plugins/webui/dashboard.html
+++ b/internal/plugins/webui/dashboard.html
@@ -2774,6 +2774,9 @@ background:
'' +
+ '" +
'" +
@@ -3633,6 +3636,7 @@ background:
async function sendChat() {
var inp = document.getElementById("chat-input");
var btn = document.getElementById("chat-send-btn");
+ var stopBtn = document.getElementById("chat-stop-btn");
var text = inp.value.trim();
if (!text || state.chatLoading) return;
state.chatStick = true;
@@ -3644,6 +3648,7 @@ background:
state.chatStage = __("等待AI回复...", "Waiting for AI...");
btn.disabled = true;
btn.textContent = "";
+ if (stopBtn) stopBtn.style.display = ""; // 生成期间可停止
rerenderChat(true);
// 触发式 POST：短超时仅确认受理；回复靠 SSE 流式渲染（对齐 GUI 行为）。
try {
@@ -3715,10 +3720,26 @@ background:
state.chatStage = "";
btn.disabled = false;
btn.textContent = __("发送", "Send");
+ var sb2 = document.getElementById("chat-stop-btn");
+ if (sb2) sb2.style.display = "none"; // 回复完成/失败，隐藏停止按钮
rerenderChat(true);
}
}
+ // 停止生成 / 发送中断消息。核心拦截语义：有 LLM 在跑则取消当前
+ // 请求并以 [中断消息] 重启轮次；无则在跑则作为普通消息处理。
+ async function interruptChat() {
+ try {
+ await api("/chat/interrupt", {
+ method: "POST",
+ body: JSON.stringify({}),
+ });
+ toast(__("已发送中断信号", "Interrupt signal sent"));
+ } catch (e) {
+ toast(__("中断失败: ", "Interrupt failed: ") + e.message, true);
+ }
+ }
+
async function queryMemoryChat() {
var q = document.getElementById("mem-query")?.value;
var r = document.getElementById("mem-result-chat");
@@ -4119,8 +4140,11 @@ background:
lastM2.role === "assistant" &&
!lastM2._final
) {
+ // 聚合最终响应：覆盖 delta 累积的中间内容（以聚合为准，
+ // 含 stage 插件改写后的最终文本），并置 final 结束本轮流式。
lastM2._grow = true;
- lastM2.content += p.content;
+ lastM2.content = p.content;
+ lastM2._final = true;
rerenderChat();
return;
}
@@ -4137,12 +4161,50 @@ background:
content: p.content,
_streaming: true,
_grow: true,
+ _final: true,
});
rerenderChat();
} catch (ex) {
console.error("[SSE] agent_output error", ex);
}
});
+ // token 级流式增量：逐块追加到当前回复内容（流式生成中）；
+ // reset 帧表示轮次作废（用户中断）：定格已显示的部分内容，置 final。
+ es.addEventListener("content_delta", function (e) {
+ try {
+ var ev = JSON.parse(e.data);
+ var p = ev.payload || {};
+ if (p.channel === "_consolidation_") return;
+ if (p.reset) {
+ var lm = state.messages.length
+ ? state.messages[state.messages.length - 1]
+ : null;
+ if (lm && lm.role === "assistant" && !lm._final) {
+ lm._final = true;
+ rerenderChat();
+ }
+ return;
+ }
+ if (!p.content) return;
+ state.chatStage = __("AI 回复中...", "AI replying...");
+ var last =
+ state.messages.length > 0
+ ? state.messages[state.messages.length - 1]
+ : null;
+ if (!last || last.role !== "assistant" || last._final) {
+ state.messages.push({
+ role: "assistant",
+ content: "",
+ tool_calls: [],
+ _streaming: true,
+ _grow: true,
+ });
+ last = state.messages[state.messages.length - 1];
+ }
+ last.content += p.content;
+ rerenderChat();
+ } catch (ex) {}
+ });
es.addEventListener("terminal_output", function (e) {
try {
var ev = JSON.parse(e.data);
@@ -4170,6 +4232,7 @@ background:
try {
var ev = JSON.parse(e.data);
var p = ev.payload || {};
+ if (p.channel === "_consolidation_") return;
if (p.content) {
state.chatStage = __("AI 思考中...", "AI thinking...");
var last =
@@ -4186,12 +4249,39 @@ background:
});
last = state.messages[state.messages.length - 1];
}
- last.reasoning_content =
- (last.reasoning_content || "") + p.content;
+ // 聚合 reasoning 帧携带全文：直接覆盖（若已有 delta 累积则等价）
+ last.reasoning_content = p.content;
rerenderChat();
}
} catch (ex) {}
});
+ // token 级流式增量：逐块追加到当前思考内容；reset 帧表示轮次作废
+ es.addEventListener("reasoning_delta", function (e) {
+ try {
+ var ev = JSON.parse(e.data);
+ var p = ev.payload || {};
+ if (p.channel === "_consolidation_") return;
+ if (p.reset) return; // 轮次作废（用户中断）：清空累积中的思考
+ if (!p.content) return;
+ state.chatStage = __("AI 思考中...", "AI thinking...");
+ var last =
+ state.messages.length > 0
+ ? state.messages[state.messages.length - 1]
+ : null;
+ if (!last || last.role !== "assistant" || last._final) {
+ state.messages.push({
+ role: "assistant",
+ content: "",
+ reasoning_content: "",
+ tool_calls: [],
+ _streaming: true,
+ });
+ last = state.messages[state.messages.length - 1];
+ }
+ last.reasoning_content = (last.reasoning_content || "") + p.content;
+ rerenderChat();
+ } catch (ex) {}
+ });
es.addEventListener("tool_call", function (e) {
try {
var ev = JSON.parse(e.data);
diff --git a/internal/plugins/webui/handler.go b/internal/plugins/webui/handler.go
index 33a9a01..c7ab3ed 100644
--- a/internal/plugins/webui/handler.go
+++ b/internal/plugins/webui/handler.go
@@ -694,6 +694,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
mux.HandleFunc("/api/v1/tracker/", h.requireAPI(h.handleTracker))
mux.HandleFunc("/api/v1/chat", h.requireAPI(h.handleChat))
mux.HandleFunc("/api/v1/chat/history", h.requireAPI(h.handleChatHistory))
+ mux.HandleFunc("/api/v1/chat/interrupt", h.requireAPI(h.handleChatInterrupt))
mux.HandleFunc("/api/v1/chat/events", h.requireAPI(h.handleChatEvents))
mux.HandleFunc("/api/v1/terminals", h.requireAPI(h.handleTerminals))
mux.HandleFunc("/api/v1/cmd/history", h.requireAPI(h.handleCmdHistory))
@@ -1221,6 +1222,39 @@ func (h *Handler) handleChatHistory(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]interface{}{"messages": result})
}
+// handleChatInterrupt 注入用户中断：取消正在进行的 LLM 生成并/或发送打断消息。
+// 核心拦截语义（interceptLoop）：
+// - 有 LLM 在跑：cancelLLM 取消当前请求 + 中断入队，process() 以
+// [中断消息] 重启轮次，模型看到被打断的上下文和用户新输入；
+// - 无 LLM 在跑：作为普通输入处理（等同发了一条消息）。
+// message 可选：空则纯取消（仍会注入空内容中断触发取消）。
+func (h *Handler) handleChatInterrupt(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPost {
+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+ return
+ }
+ if h.sdk == nil {
+ writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "agent unavailable"})
+ return
+ }
+ var body struct {
+ Message string `json:"message"`
+ DeviceID string `json:"device_id"`
+ }
+ if r.Body != nil {
+ _ = json.NewDecoder(r.Body).Decode(&body) // body 可选
+ }
+
+ source := "webui"
+ if body.DeviceID != "" {
+ source = "webui/" + body.DeviceID
+ }
+ h.sdk.InjectInterrupt(source, "webui", "text", map[string]interface{}{
+ "content": body.Message,
+ })
+ writeJSON(w, http.StatusOK, map[string]string{"status": "interrupted"})
+}
+
func (h *Handler) handleTerminals(w http.ResponseWriter, r *http.Request) {
h.termMu.Lock()
terms := make([]*termState, 0, len(h.termStates))
@@ -1413,8 +1447,27 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
}
subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output"}
+ // token 级流式增量事件：实时转发给浏览器做逐 token 渲染。
+ // 不进 sseEventRing —— 断线重连只重放聚合事件（最终真相），
+ // 避免重放 delta 与聚合内容重复追加。
var unsubs []func()
var seq int64
+ appendDeltaSub := func(evtType sdk.EventType) {
+ unsub := h.sdk.Subscribe(evtType, func(evt *sdk.Event) {
+ data, _ := json.Marshal(evt)
+ seq++
+ id := fmt.Sprintf("%d-%d", evt.Timestamp, seq)
+ select {
+ case writeCh <- fmt.Sprintf("id: %s\nevent: %s\ndata: %s\n", id, evt.Type, string(data)):
+ default:
+ log.Printf("[SSE] DROPPED %s (writeCh full, len=%d)", evt.Type, len(writeCh))
+ }
+ })
+ unsubs = append(unsubs, unsub)
+ }
+ appendDeltaSub(sdk.EventReasoningDelta)
+ appendDeltaSub(sdk.EventContentDelta)
+
for _, t := range subTypes {
t2 := t
unsub := h.sdk.Subscribe(sdk.EventType(t2), func(evt *sdk.Event) {
diff --git a/internal/sdk/events.go b/internal/sdk/events.go
index f1b9eb2..0283a3c 100644
--- a/internal/sdk/events.go
+++ b/internal/sdk/events.go
@@ -18,4 +18,9 @@ const (
EventSystem = events.EventSystem
EventTerminalOutput = events.EventTerminalOutput
EventAll = events.EventAll
+
+ // 流式增量事件（token 级）：核心 process() 流式化后每收到一个增量块发布。
+ // 客户端可选订做真逐 token 渲染；聚合事件仍照常发布，旧订阅者不受影响。
+ EventReasoningDelta = events.EventReasoningDelta
+ EventContentDelta = events.EventContentDelta
)