diff --git a/internal/plugins/agentcli/plugin.go b/internal/plugins/agentcli/plugin.go index a2556224..b6ec9a76 100644 --- a/internal/plugins/agentcli/plugin.go +++ b/internal/plugins/agentcli/plugin.go @@ -71,6 +71,9 @@ type TerminalSession struct { done chan struct{} // 通知节流字段 + // writeMu 串行化本终端的写入:保证一次 terminal_write 的 + // input 与 key 不会被其它写入插进中间。 + writeMu sync.Mutex unreadBytes int // 最近一次通知后积累的未读字节数 lastNotify time.Time // 最近一次通知时间 lastData time.Time // 最近一次读到的数据时间(用于判定输出停止) @@ -265,11 +268,17 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { } // agentcli 通道:终端生命周期/输出事件经它注入 agent(见本文件 InjectText* 调用)。 - _ = s.RegisterInputChannel("agentcli", sdk.ChannelDef{}) + // + // ★ 不吞错误(2026-11-03):注册失败时这条通道根本不存在,之后所有 + // InjectText("agentcli", ...) 都会投进虚空 —— 通知静默消失,日志里 + // 一点痕迹都没有。宁可让插件启动失败,也不要静默降级。 + if err := s.RegisterInputChannel("agentcli", sdk.ChannelDef{}); err != nil { + return fmt.Errorf("agentcli: register input channel failed: %w", err) + } s.RegisterTool("terminal_create", sdk.ToolDef{ Name: "terminal_create", Description: "创建一个新的交互式终端会话。返回终端 ID,后续通过此 ID 进行读写操作。适用于运行交互式程序如 vim、ssh、top、nano 等。" + - "通知模式通过 notify 参数选择(默认 exit):exit=仅命令执行结束后提醒一次;interval=定时反馈(如 interval=30s 每 30 秒反馈一次状态摘要);" + + "默认**不主动推送**(终端像人开的窗口:输出由你 terminal_read 主动读)。需要推送时用 notify 显式声明:exit=命令执行结束后提醒一次;interval=定时反馈(如 interval=30s 每 30 秒反馈一次状态摘要);" + "buffer=未读输出积累到指定字节数后提醒(如 buffer=8192);多个模式用逗号组合(如 interval=30s,buffer=8192)。终端默认 5 分钟后自动关闭,可通过 timeout 参数调整。", NoMemory: true, Parameters: map[string]interface{}{ @@ -281,7 +290,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { }, "notify": map[string]interface{}{ "type": "string", - "description": "通知模式(可选):exit(默认,命令结束后提醒);interval=时长(定时反馈,如 30s/1m);buffer=字节数(缓冲阈值提醒);可逗号组合", + "description": "通知模式(可选,默认不推送)。不传 = 完全静默,输出靠 terminal_read 主动读。显式声明才有推送:exit(命令结束后提醒一次);interval=时长(定时反馈,如 30s/1m);buffer=字节数(未读输出达阈值提醒);quiet(显式静默);可逗号组合", }, "timeout": map[string]interface{}{ "type": "string", @@ -542,9 +551,20 @@ func (p *Plugin) handleCreate(s *sdk.PluginSDK, args map[string]interface{}) (in cols = uint16(c) } - // 通知模式:默认 exit(命令执行结束后提醒一次)。 - // 支持 interval=30s / buffer=8192 / quiet,可逗号组合。 - watch := terminalWatch{onExit: true, quiet: true} + // 通知模式:**默认不推送**(用户裁定 2026-11-03)。 + // + // 「agentcli 没必要进行直接推送,因为本质它就类似人类开启的终端窗口」 + // + // 终端窗口的语义是:人敲命令 → 自己去看输出。输出不该主动跳到眼前。 + // 主动注入虽然走 inputCh(排队、不抢占),但每次都会唤醒 agent 跑一轮 + // 推理;`tail -f` / `npm install` 这类长输出命令会让它被反复唤醒。 + // + // 所以默认 onExit=false:结束、退出、超时一律静默,输出只由 + // terminal_read 主动读取。需要推送时必须**显式**声明: + // notify=exit → 命令结束后提醒一次 + // notify=interval=30s → 定时反馈状态摘要 + // notify=buffer=8192 → 未读输出达阈值提醒 + watch := terminalWatch{onExit: false, quiet: true} if notifyStr, ok := args["notify"].(string); ok && notifyStr != "" { watch = parseNotifyMode(notifyStr, watch) } @@ -610,6 +630,19 @@ func notifyModeString(w terminalWatch) string { return strings.Join(parts, ",") } +// terminalGoneResult 是「终端已不存在」的**结构化**答复。 +// +// 为什么不能只回一句文本(2026-11-03):模型拿到普通失败会继续用同一个 id +// 重试(实测 browser 插件同形态:连续 3 次失败仍用旧 id)。给出可执行的 +// 下一步,它才会停下来重建。 +func (p *Plugin) terminalGoneResult(id string) map[string]interface{} { + return map[string]interface{}{ + "error": "terminal_gone", + "terminal": id, + "next_action": "该终端已关闭或超时清理。需要继续执行请调用 terminal_create 新建终端,并使用返回的新 id;terminal_list 可查看现存终端。", + } +} + func (p *Plugin) handleWrite(s *sdk.PluginSDK, args map[string]interface{}) (interface{}, error) { id, _ := args["id"].(string) if id == "" { @@ -620,34 +653,62 @@ func (p *Plugin) handleWrite(s *sdk.PluginSDK, args map[string]interface{}) (int session, ok := p.sessions[id] p.mu.Unlock() if !ok { - return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil + return p.terminalGoneResult(id), nil } input, _ := args["input"].(string) key, _ := args["key"].(string) - var data []byte + // ★ input 与 key **可以同时提供**(2026-11-03 修复)。 + // + // 旧实现是 `if key != "" { ... } else if input != "" { ... }` 的互斥结构: + // 同时传两者时 input 被**静默丢弃**,只发出 key。于是 + // + // {"input":"ls","key":"enter"} ⇒ 只发了 \r + // + // 命令从未输入,模型却收到 status=ok / bytes=1,以为成功、去读输出又 + // 什么都读不到,然后反复重试 —— 这正是"终端没反应"的直接成因。 + // + // 现在按**声明顺序**发送:先 input,再 key。"输入命令并回车"是最高频 + // 的表达,不该要求模型拆成两次调用。 + if input == "" && key == "" { + return map[string]interface{}{"error": "需要提供 input 或 key 参数"}, nil + } + var keyBytes []byte if key != "" { b, err := mapKey(key) if err != nil { return map[string]interface{}{"error": err.Error()}, nil } - data = b - } else if input != "" { - data = []byte(input) - } else { - return map[string]interface{}{"error": "需要提供 input 或 key 参数"}, nil + keyBytes = b } - n, err := session.Write(string(data)) - if err != nil { - return map[string]interface{}{"error": fmt.Sprintf("写入失败: %v", err)}, nil + // ★ 整对写入必须在同一把会话锁里完成:否则同终端的并发写可能插进 + // input 与 key 之间,把命令和回车拆坏(terminal_write 声明了 Serial, + // 但那只约束内核同批调度,不约束插件内部多路写入)。 + session.writeMu.Lock() + defer session.writeMu.Unlock() + + written := 0 + if input != "" { + n, err := session.Write(input) + if err != nil { + return map[string]interface{}{"error": fmt.Sprintf("写入失败: %v", err)}, nil + } + written += n + } + if len(keyBytes) > 0 { + n, err := session.Write(string(keyBytes)) + if err != nil { + return map[string]interface{}{"error": fmt.Sprintf("写入按键失败: %v", err)}, nil + } + written += n } return map[string]interface{}{ "status": "ok", - "bytes": n, + "bytes": written, "terminal": id, }, nil } @@ -703,7 +764,7 @@ func (p *Plugin) handleRead(args map[string]interface{}) (interface{}, error) { session, ok := p.sessions[id] p.mu.Unlock() if !ok { - return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil + return p.terminalGoneResult(id), nil } var output string @@ -752,7 +813,7 @@ func (p *Plugin) handleResize(args map[string]interface{}) (interface{}, error) session, ok := p.sessions[id] p.mu.Unlock() if !ok { - return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil + return p.terminalGoneResult(id), nil } if err := session.Resize(uint16(rows), uint16(cols)); err != nil { @@ -781,7 +842,7 @@ func (p *Plugin) handleClose(s *sdk.PluginSDK, args map[string]interface{}) (int p.mu.Unlock() if !ok { - return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil + return p.terminalGoneResult(id), nil } session.Close() @@ -806,12 +867,13 @@ func (p *Plugin) handleWatch(args map[string]interface{}) (interface{}, error) { session, ok := p.sessions[id] p.mu.Unlock() if !ok { - return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil + return p.terminalGoneResult(id), nil } session.mu.Lock() if v, ok := args["clear"].(bool); ok && v { - session.watch = terminalWatch{onExit: true} + // clear = 回到默认(不推送),不是"回到 exit 提醒"。 + session.watch = terminalWatch{onExit: false, quiet: true} } else { if v, ok := args["interval"].(string); ok && v != "" { if d, err := time.ParseDuration(v); err == nil && d >= 0 { @@ -827,9 +889,6 @@ func (p *Plugin) handleWatch(args map[string]interface{}) (interface{}, error) { if v, ok := args["quiet"].(bool); ok { session.watch.quiet = v } - if session.watch.interval == 0 && session.watch.bufferBytes == 0 && !session.watch.quiet { - session.watch.onExit = true - } } w := session.watch session.mu.Unlock() diff --git a/internal/plugins/agentcli/plugin_test.go b/internal/plugins/agentcli/plugin_test.go index c47b95ba..3de69e56 100644 --- a/internal/plugins/agentcli/plugin_test.go +++ b/internal/plugins/agentcli/plugin_test.go @@ -5,6 +5,7 @@ package agentcli import ( "encoding/json" "fmt" + "os" "strings" "sync" "testing" @@ -480,6 +481,8 @@ type mockTerm struct { data chan []byte running bool err error + // writes 记录每次 WriteString 的原文,供判据断言"到底发出了什么"。 + writes []string // reads 记录 Read 被调用次数,供 waitReadLoopReady 判断 readLoop 已起来。 // 不用固定 sleep:机器慢时 sleep 不够、快时白等。 reads int @@ -523,8 +526,22 @@ func (m *mockTerm) Read(buf []byte) (int, error) { } } -func (m *mockTerm) WriteString(s string) (int, error) { return len(s), nil } -func (m *mockTerm) Resize(rows, cols uint16) error { return nil } +func (m *mockTerm) WriteString(s string) (int, error) { + m.mu.Lock() + m.writes = append(m.writes, s) + m.mu.Unlock() + return len(s), nil +} + +// writeLog 返回已写入内容(并发安全)。 +func (m *mockTerm) writeLog() []string { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]string, len(m.writes)) + copy(out, m.writes) + return out +} +func (m *mockTerm) Resize(rows, cols uint16) error { return nil } func (m *mockTerm) Running() bool { m.mu.Lock() defer m.mu.Unlock() @@ -1073,3 +1090,187 @@ func TestNoExpiredNotificationInjected(t *testing.T) { } } } + +// ── 2026-11-03 修复的判据 ── + +// TestWriteSendsInputThenKey 钉住 input 与 key 可以同时提交。 +// +// 旧实现的 if key / else if input 是**互斥**结构: +// +// {"input":"ls","key":"enter"} ⇒ 只发 \r,input 被静默丢弃 +// +// 模型收到 status=ok / bytes=1 以为成功,读输出却是空的,于是反复重试。 +// 判据必须断言**两段都真的写进了终端**,只看返回值是测不出来的。 +func TestWriteSendsInputThenKey(t *testing.T) { + p, tc, err := setupPlugin() + if err != nil { + t.Fatal(err) + } + + term := newMockTerm() + ts := newTestSession(term) + ts.id = "t-write" + p.mu.Lock() + p.sessions["t-write"] = ts + p.mu.Unlock() + + handler, ok := tc.handlers["terminal_write"] + if !ok { + t.Fatal("terminal_write 未注册") + } + if _, err := handler(map[string]interface{}{ + "id": "t-write", + "input": "ls", + "key": "enter", + }); err != nil { + t.Fatal(err) + } + + writes := term.writeLog() + if len(writes) != 2 { + t.Fatalf("应分两次写(input 一次、key 一次),实际 %d 次: %q", len(writes), writes) + } + if writes[0] != "ls" { + t.Fatalf("第一段必须是 input,实际 %q", writes[0]) + } + if writes[1] != "\r" { + t.Fatalf("第二段必须是 enter 键(\\r),实际 %q", writes[1]) + } +} + +// TestWriteInputOnlyUnchanged 保证只传 input 时不额外发键。 +func TestWriteInputOnlyUnchanged(t *testing.T) { + p, tc, err := setupPlugin() + if err != nil { + t.Fatal(err) + } + term := newMockTerm() + ts := newTestSession(term) + ts.id = "t-in" + p.mu.Lock() + p.sessions["t-in"] = ts + p.mu.Unlock() + + if _, err := tc.handlers["terminal_write"](map[string]interface{}{ + "id": "t-in", + "input": "echo hi", + }); err != nil { + t.Fatal(err) + } + writes := term.writeLog() + if len(writes) != 1 || writes[0] != "echo hi" { + t.Fatalf("只传 input 时应只写一次,实际 %q", writes) + } +} + +// TestTerminalCreateDefaultsToNoPush 钉住默认不推送。 +// +// 用户裁定(2026-11-03): +// +// 「agentcli 没必要进行直接推送,因为本质它就类似人类开启的终端窗口」 +// +// 终端窗口不会把输出推到眼前;唤醒 agent 跑一轮推理的代价必须由调用方 +// 显式声明(notify=exit / interval / buffer)才付出。 +// +// ★ 为什么不 spawn 真终端来断言:这条默认值在 handleCreate 的局部变量里, +// +// 没有可观测的返回字段。用源码约束 + parseNotifyMode 的默认入参一起钉: +// 删掉那行默认值、或把它改回 onExit:true,这条会红。 +func TestTerminalCreateDefaultsToNoPush(t *testing.T) { + // ① 生产代码里 handleCreate 的初始 watch 必须是「不推送」 + src, err := os.ReadFile("plugin.go") + if err != nil { + t.Fatal(err) + } + text := string(src) + if !strings.Contains(text, "watch := terminalWatch{onExit: false, quiet: true}") { + t.Fatal("handleCreate 的默认 watch 必须是 {onExit:false, quiet:true}(默认不推送)") + } + + // ② 不传 notify 时,该初始值不会被任何解析改写 + w := terminalWatch{onExit: false, quiet: true} + if w.onExit || w.interval != 0 || w.bufferBytes != 0 { + t.Fatalf("默认 watch 不应开启任何推送: %+v", w) + } +} + +// TestExplicitExitNotifyStillWorks 显式声明时推送仍然生效(A 方案:默认关,可显式开)。 +func TestExplicitExitNotifyStillWorks(t *testing.T) { + w := parseNotifyMode("exit", terminalWatch{onExit: false, quiet: true}) + if !w.onExit { + t.Fatal("显式 notify=exit 必须开启结束提醒") + } + if w.quiet { + t.Fatal("显式 notify=exit 应关闭 quiet(否则退出通知被静默)") + } +} + +// TestWatchClearRestoresNoPush 钉住 clear 的语义:回到默认(不推送), +// 而不是回到 exit 提醒。 +func TestWatchClearRestoresNoPush(t *testing.T) { + p, tc, err := setupPlugin() + if err != nil { + t.Fatal(err) + } + term := newMockTerm() + ts := newTestSession(term) + ts.id = "t-clear" + ts.watch = terminalWatch{onExit: true, interval: time.Minute} + p.mu.Lock() + p.sessions["t-clear"] = ts + p.mu.Unlock() + + if _, err := tc.handlers["terminal_watch"](map[string]interface{}{ + "id": "t-clear", + "clear": true, + }); err != nil { + t.Fatal(err) + } + + ts.mu.Lock() + got := ts.watch + ts.mu.Unlock() + if got.onExit { + t.Fatal("clear 后不应恢复 exit 提醒(应回到默认不推送)") + } + if got.interval != 0 { + t.Fatalf("clear 后 interval 应清零,实际 %v", got.interval) + } +} + +// TestTerminalGoneIsStructured 钉住已关闭终端的答复是**可执行**的, +// 不是一句没有出路的文本 —— 否则模型会拿同一个 id 反复重试。 +func TestTerminalGoneIsStructured(t *testing.T) { + _, tc, err := setupPlugin() + if err != nil { + t.Fatal(err) + } + for _, tool := range []string{"terminal_write", "terminal_read", "terminal_resize", "terminal_close", "terminal_watch"} { + h, ok := tc.handlers[tool] + if !ok { + t.Fatalf("%s 未注册", tool) + } + args := map[string]interface{}{"id": "does-not-exist"} + if tool == "terminal_resize" { + args["rows"] = float64(24) + args["cols"] = float64(80) + } + if tool == "terminal_write" { + args["input"] = "x" + } + res, err := h(args) + if err != nil { + t.Fatalf("%s: %v", tool, err) + } + m, ok := res.(map[string]interface{}) + if !ok { + t.Fatalf("%s: 返回类型不是 map: %T", tool, res) + } + if m["error"] != "terminal_gone" { + t.Fatalf("%s: error 应为 terminal_gone,实际 %v", tool, m["error"]) + } + if _, ok := m["next_action"]; !ok { + t.Fatalf("%s: 缺少 next_action(模型没有可执行的下一步)", tool) + } + } +}