mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-24 02:48:04 +00:00
Expose the LLM token-level streaming deltas (EventReasoningDelta /
EventContentDelta) to every client channel and add user-initiated
interrupt (cancel generation / send interrupt message) to all three
frontends, preserving the existing interrupt-injection semantics.
SDK/events:
- EventReasoningDelta, EventContentDelta constants exported in the
public/internal SDK event alias tables.
CLI plugin:
- handleChat subscribes to both delta events and forwards
reasoning_delta / content_delta JSON frames (channel-filtered);
aggregated reasoning/tool_call/response frames still fire as before.
- New /stop (alias /interrupt) builtin injects an interrupt via
InjectInterrupt(cliSource, cliChannel) - matches interceptLoop
semantics: cancels an active stream and re-injects the message as
a [中断消息] for a restarted turn; with no active LLM it behaves
as a plain input.
Waiter client (line mode + TUI):
- streamRender accumulates delta chunks and redraws the current line;
a reset frame (stream abandoned, e.g. user interrupt) flushes the
partial buffer so the next turn does not concatenate onto stale
content. Aggregated frames terminate the delta line and render the
final text (old servers without deltas behave exactly as before).
- TUI merges content_delta into the in-flight agent message and seals
it (final flag) on response/tool_call/error so subsequent deltas
never append to a finished message.
WebUI:
- SSE handler subscribes to the two delta events but does NOT record
them into the replay ring - reconnection replays only aggregated
events (the final truth), avoiding duplicate delta accumulation.
- POST /api/v1/chat/interrupt calls InjectInterrupt(webui, webui)
with optional message; fronted by a Stop button shown only while
a generation is in flight.
dashboard.html / GUI app.js:
- Stop button next to Send (hidden until chatLoading); interruptChat
POSTs /chat/interrupt. Delta listeners append incrementally;
agent_output (aggregated) now replaces (not appends) the in-flight
content and marks _final; reset frames finalize the partial message.
process.go:
- chatStreamWithFallback preserves the context.Canceled/
DeadlineExceeded contract: a user interrupt returns the canceled
error (never a partial-content success) so the existing continue
branch restarts the turn with the [中断消息]. A reset
EventContentDelta is published so connected clients drop stale
partial renderings before the new turn begins.
Verified: /stop 'msg' via waiter triggers 'interrupt from cli/cli' in
interceptLoop; unit TestChatStreamCancelPreservesInterrupt confirms the
canceled error propagates instead of being swallowed.
475 lines
12 KiB
Go
475 lines
12 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"flag"
|
||
"fmt"
|
||
"os"
|
||
"os/signal"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"syscall"
|
||
"time"
|
||
)
|
||
|
||
const (
|
||
colorReset = "\033[0m"
|
||
colorGreen = "\033[32m"
|
||
colorRed = "\033[31m"
|
||
colorYellow = "\033[33m"
|
||
colorBold = "\033[1m"
|
||
colorDim = "\033[2m"
|
||
)
|
||
|
||
const clearLine = "\033[2K\r"
|
||
|
||
var colors = true
|
||
|
||
var spinnerFrames = []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"}
|
||
|
||
// isTTYFile 判断文件是否为字符终端(非终端时禁用 spinner 转圈)。
|
||
func isTTYFile(f *os.File) bool {
|
||
fi, err := f.Stat()
|
||
if err != nil {
|
||
return false
|
||
}
|
||
return fi.Mode()&os.ModeCharDevice != 0
|
||
}
|
||
|
||
// startSpinner 启动 npm 风格的加载动画,返回停止函数。
|
||
// stop() 幂等:终止动画并清除当前行。非终端环境直接空操作。
|
||
func startSpinner(label string) func() {
|
||
if !colors || !isTTYFile(os.Stdout) {
|
||
return func() {}
|
||
}
|
||
done := make(chan struct{})
|
||
var once sync.Once
|
||
go func() {
|
||
ticker := time.NewTicker(80 * time.Millisecond)
|
||
defer ticker.Stop()
|
||
i := 0
|
||
for {
|
||
select {
|
||
case <-done:
|
||
return
|
||
case <-ticker.C:
|
||
fmt.Printf("%s%s %s%s\n", clearLine, colorDim, spinnerFrames[i%len(spinnerFrames)]+" "+label, colorReset)
|
||
i++
|
||
}
|
||
}
|
||
}()
|
||
stop := func() {
|
||
once.Do(func() {
|
||
close(done)
|
||
fmt.Print(clearLine)
|
||
})
|
||
}
|
||
return stop
|
||
}
|
||
|
||
func init() {
|
||
if os.Getenv("NO_COLOR") != "" {
|
||
colors = false
|
||
}
|
||
}
|
||
|
||
func printlnC(color, msg string) {
|
||
if !colors {
|
||
fmt.Println(msg)
|
||
return
|
||
}
|
||
fmt.Printf("%s%s%s\n", color, msg, colorReset)
|
||
}
|
||
|
||
func historyPath() string {
|
||
home, _ := os.UserHomeDir()
|
||
xdgData := os.Getenv("XDG_DATA_HOME")
|
||
if xdgData == "" {
|
||
xdgData = filepath.Join(home, ".local", "share")
|
||
}
|
||
dir := filepath.Join(xdgData, "homeagent")
|
||
os.MkdirAll(dir, 0755)
|
||
return filepath.Join(dir, "cli_history")
|
||
}
|
||
|
||
func discoverSocket(configured string) string {
|
||
if configured != "" {
|
||
return configured
|
||
}
|
||
if s := os.Getenv("HOMEAGENT_SOCKET"); s != "" {
|
||
return s
|
||
}
|
||
home, _ := os.UserHomeDir()
|
||
candidates := []string{
|
||
filepath.Join(home, ".homeagent", "cli.sock"),
|
||
"/var/lib/homeagent/cli.sock",
|
||
}
|
||
if xdg := os.Getenv("XDG_RUNTIME_DIR"); xdg != "" {
|
||
candidates = append([]string{filepath.Join(xdg, "homeagent", "cli.sock")}, candidates...)
|
||
}
|
||
for _, c := range candidates {
|
||
if _, err := os.Stat(c); err == nil {
|
||
return c
|
||
}
|
||
}
|
||
return candidates[0]
|
||
}
|
||
|
||
func main() {
|
||
socket := flag.String("socket", "", "unix socket path")
|
||
remote := flag.String("remote", "", "remote webui URL")
|
||
apiKey := flag.String("api-key", "", "API key for remote mode")
|
||
configPath := flag.String("config", "", "config file path")
|
||
chat := flag.String("chat", "", "send a message and print final text (one-shot)")
|
||
say := flag.String("say", "", "deprecated alias of -chat")
|
||
deviceGateway := flag.String("device", "", "remotedevice 网关地址(如 127.0.0.1:9890),启动设备桥")
|
||
deviceToken := flag.String("device-token", "", "设备接入 token")
|
||
deviceAuthorized := flag.Bool("device-authorized", false, "客户端本地授权(允许远程操控本机;也可在 waiter.yaml 配 device_authorized: true)")
|
||
testCap := flag.String("test-cap", "", "测试本地能力(screensue/speakeruse/screensee/clipboardsee/clipboardsue/computeruse/camerasue),如 --test-cap screensue")
|
||
testCapArgs := flag.String("test-cap-args", "", "测试能力的参数")
|
||
flag.Parse()
|
||
|
||
// 本地能力测试模式(无需连接服务器)
|
||
if *testCap != "" {
|
||
runCapTest(*testCap, *testCapArgs)
|
||
return
|
||
}
|
||
|
||
cfg := discoverConfig(*configPath)
|
||
cfg.MergeCLI(*socket, *remote, *apiKey)
|
||
cfg.ApplyDefault()
|
||
|
||
if cfg.Socket == "" && cfg.Remote == "" {
|
||
cfg.Socket = discoverSocket("")
|
||
}
|
||
|
||
oneShotMsg := *chat
|
||
if oneShotMsg == "" {
|
||
oneShotMsg = *say
|
||
}
|
||
|
||
state := &State{}
|
||
if err := state.Connect(cfg); err != nil {
|
||
printlnC(colorRed, fmt.Sprintf("connect: %v", err))
|
||
os.Exit(1)
|
||
}
|
||
defer state.Disconnect()
|
||
|
||
// 设备桥:--device 或配置 device_gateway 时,waiter 作为被控设备接入 remotedevice
|
||
dg := *deviceGateway
|
||
if dg == "" {
|
||
dg = cfg.DeviceGateway
|
||
}
|
||
dt := *deviceToken
|
||
if dt == "" {
|
||
dt = cfg.DeviceToken
|
||
}
|
||
if dg != "" && dt != "" {
|
||
if err := startDeviceBridge(dg, dt); err != nil {
|
||
printlnC(colorYellow, fmt.Sprintf("device bridge: %v (continue without)", err))
|
||
} else {
|
||
// 客户端本地授权:命令行 --device-authorized 或 waiter.yaml device_authorized
|
||
auth := *deviceAuthorized || cfg.DeviceAuthorized
|
||
deviceBridge.SetAuthorized(auth)
|
||
printlnC(colorGreen, "device bridge active: "+deviceBridgeID+" authorized="+fmt.Sprint(auth))
|
||
defer stopDeviceBridge()
|
||
}
|
||
}
|
||
|
||
if oneShotMsg != "" {
|
||
oneshot(state, oneShotMsg)
|
||
return
|
||
}
|
||
|
||
runInteractive(state, cfg)
|
||
}
|
||
|
||
// runInteractive 交互入口:TTY 下走 Bubble Tea 全屏 TUI,非 TTY 回退行式 REPL。
|
||
func runInteractive(state *State, cfg *Config) {
|
||
history := newHistory(historyPath(), 1000)
|
||
history.load()
|
||
|
||
if isTTYFile(os.Stdin) && isTTYFile(os.Stdout) && colors {
|
||
if err := runTUI(state, cfg, history); err != nil {
|
||
printlnC(colorRed, fmt.Sprintf("tui: %v", err))
|
||
printlnC(colorYellow, "falling back to line mode")
|
||
runLineMode(state, cfg, history)
|
||
}
|
||
return
|
||
}
|
||
runLineMode(state, cfg, history)
|
||
}
|
||
|
||
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()
|
||
if err != nil {
|
||
printlnC(colorRed, fmt.Sprintf("error: %v", err))
|
||
os.Exit(1)
|
||
}
|
||
printlnC(colorGreen, resp)
|
||
}
|
||
|
||
// runCapTest 本地能力测试(无需连接服务器)
|
||
func runCapTest(capName, args string) {
|
||
fmt.Printf("=== 测试能力: %s ===\n", capName)
|
||
fmt.Printf("参数: %s\n", args)
|
||
fmt.Println("===========================")
|
||
|
||
// 覆盖 sendBridgeResult 为本地打印
|
||
sendBridgeResult = func(reqID, status, output, errMsg string) {
|
||
fmt.Printf("结果状态: %s\n", status)
|
||
if output != "" {
|
||
fmt.Printf("输出: %s\n", output)
|
||
}
|
||
if errMsg != "" {
|
||
fmt.Printf("错误: %s\n", errMsg)
|
||
}
|
||
}
|
||
|
||
handleHomeagentCmd("test-001", "homeagent-"+capName+" "+args)
|
||
fmt.Println("===========================")
|
||
fmt.Println("测试完成")
|
||
}
|
||
|
||
// runLineMode 传统行式 REPL(非 TTY 回退 / TUI 启动失败时使用)。
|
||
func runLineMode(state *State, cfg *Config, history *History) {
|
||
sigCh := make(chan os.Signal, 1)
|
||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||
|
||
line := newLineEditor(history)
|
||
|
||
restore, err := setRawMode(0)
|
||
if err != nil {
|
||
restore = func() {}
|
||
}
|
||
defer restore()
|
||
|
||
modeLabel := "local"
|
||
addrLabel := cfg.Socket
|
||
if cfg.Remote != "" {
|
||
modeLabel = "remote"
|
||
addrLabel = cfg.Remote
|
||
}
|
||
if colors {
|
||
fmt.Printf("%sHomeAgent CLI%s %s(%s://%s)%s\n", colorBold, colorReset, colorDim, modeLabel, addrLabel, colorReset)
|
||
} else {
|
||
fmt.Printf("HomeAgent CLI (%s://%s)\n", modeLabel, addrLabel)
|
||
}
|
||
fmt.Println("Type /help for commands.")
|
||
|
||
var readerCancel func()
|
||
var spinnerStopMu sync.Mutex
|
||
var spinnerStop = func() {}
|
||
startReader := func() {
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
readerCancel = cancel
|
||
go state.ReadLoop(ctx, func(line string) {
|
||
spinnerStopMu.Lock()
|
||
stop := spinnerStop
|
||
spinnerStopMu.Unlock()
|
||
stop()
|
||
printServerOutput(line)
|
||
})
|
||
}
|
||
startReader()
|
||
|
||
reconnect := func() {
|
||
if readerCancel != nil {
|
||
readerCancel()
|
||
}
|
||
state.Disconnect()
|
||
for i := 0; i < 30; i++ {
|
||
if err := state.Connect(cfg); err != nil {
|
||
printlnC(colorYellow, fmt.Sprintf("reconnecting (%d/30): %v", i+1, err))
|
||
time.Sleep(2 * time.Second)
|
||
continue
|
||
}
|
||
printlnC(colorGreen, "reconnected")
|
||
startReader()
|
||
return
|
||
}
|
||
printlnC(colorRed, "giving up after 30 attempts")
|
||
}
|
||
|
||
loop:
|
||
for {
|
||
text, err := line.read()
|
||
if err != nil {
|
||
break
|
||
}
|
||
line.clear()
|
||
|
||
cmd := strings.TrimSpace(text)
|
||
if cmd == "" {
|
||
continue
|
||
}
|
||
|
||
if cmd[0] == '/' {
|
||
if handleBuiltin(cmd, cfg, state, reconnect, os.Stdout) {
|
||
if cmd == "/exit" || cmd == "/quit" {
|
||
break loop
|
||
}
|
||
continue
|
||
}
|
||
}
|
||
|
||
history.add(cmd)
|
||
history.save()
|
||
|
||
if err := state.Send(cmd); err != nil {
|
||
printlnC(colorYellow, "connection lost, reconnecting...")
|
||
line.redrawPending(cmd)
|
||
reconnect()
|
||
state.Send(cmd)
|
||
}
|
||
|
||
// 发送成功后启动加载动画,收到第一帧服务器输出时自动停止
|
||
spinnerStopMu.Lock()
|
||
spinnerStop = startSpinner("thinking...")
|
||
spinnerStopMu.Unlock()
|
||
|
||
select {
|
||
case <-sigCh:
|
||
break loop
|
||
default:
|
||
}
|
||
}
|
||
|
||
if readerCancel != nil {
|
||
readerCancel()
|
||
}
|
||
}
|
||
|
||
// 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)
|
||
if !colors {
|
||
fmt.Printf("%s%s\n", clearLine, renderPlain(rl, content))
|
||
return
|
||
}
|
||
printServerEventColored(rl, content)
|
||
}
|
||
|
||
// printServerEvent 渲染一个已解析的过程/终结事件。
|
||
func printServerEvent(rl respLine) {
|
||
if !colors {
|
||
fmt.Printf("%s%s\n", clearLine, renderPlain(rl, ""))
|
||
return
|
||
}
|
||
printServerEventColored(rl, "")
|
||
}
|
||
|
||
// renderPlain 无色模式下的纯文本渲染。
|
||
func renderPlain(rl respLine, raw string) string {
|
||
switch rl.Type {
|
||
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":
|
||
return rl.Content
|
||
case "error":
|
||
return "[错误] " + rl.Error
|
||
default:
|
||
if raw != "" {
|
||
return raw
|
||
}
|
||
return rl.Content
|
||
}
|
||
}
|
||
|
||
// printServerEventColored 彩色模式下的帧渲染。
|
||
func printServerEventColored(rl respLine, raw string) {
|
||
switch rl.Type {
|
||
case "reasoning":
|
||
fmt.Printf("%s%s· %s%s\n", clearLine, colorDim, rl.Content, colorReset)
|
||
case "tool_call":
|
||
mark, markColor := "⚙", colorYellow
|
||
switch rl.Status {
|
||
case "ok":
|
||
mark, markColor = "✔", colorGreen
|
||
case "denied", "interrupted", "error":
|
||
mark, markColor = "✘", colorRed
|
||
}
|
||
preview := rl.Result
|
||
if preview != "" {
|
||
preview = " " + preview
|
||
}
|
||
fmt.Printf("%s%s%s %s [%s]%s%s\n", clearLine, markColor, mark, rl.Tool, rl.Status, preview, colorReset)
|
||
case "response":
|
||
fmt.Printf("%s%s%s%s\n", clearLine, colorGreen, rl.Content, colorReset)
|
||
case "error":
|
||
fmt.Printf("%s%s%s%s\n", clearLine, colorRed, rl.Error, colorReset)
|
||
default:
|
||
text := raw
|
||
if text == "" {
|
||
text = rl.Content
|
||
}
|
||
fmt.Printf("%s%s%s\n", clearLine, text, colorReset)
|
||
}
|
||
}
|