mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-01 14:57:38 +00:00
db88c5efe5
## 现象
关闭终端后 agent 持续收到「[终端 term_6 读取结束: read /dev/ptmx:
input/output error]」「[终端 term_6 中的进程已退出]」，而 terminal_list
显示 0 个会话——它拿不到任何证据说明这些是历史遗留，只能反复解释
"这是旧回声，不用管"，每条都唤醒一轮推理。
## 根因（不是一次回声，是每 30 秒一次）
close 工具 → session.Close() → readLoop 从 <-t.stopCh 静默 return
→ 但 **sessions 条目没删**
→ cleanupLoop 每 30s 扫到它，发现 !terminalRunning
→ 又注入一条「进程已退出」
三条退出路径（主动关闭 / 进程自己结束 / 超时）里，主动关闭那条漏了删表。
## 修法
1. readLoop 的 t.stopCh 分支补 removeSession，三条路径统一走它。
2. 新增 isClosing：所有"终端结束"通知都必须区分两种成因——
进程自己结束（是新闻，要告诉 agent）vs 我们主动关掉（是回声，不能注入）。
旧实现只看 r.err / terminalRunning，而主动关闭恰好让两者都成立。
## 测试基础设施修正
mockTerm.Kill() 原先是空操作，于是"主动关闭"在测试里从不产生 EIO，
判据永远绿（假绿）。改为模拟真实 PTY：Kill 后 Read 立刻返
`read /dev/ptmx: input/output error`。
判据：关闭后 session 必须立刻从表里消失、关闭后不得新增注入
（连开 20 次防 select 随机性）、进程自己结束仍必须通知（不过度修复）。
变异：关闭时不删表 / isClosing 恒 false，均被抓。
Co-Authored-By: Claude <noreply@anthropic.com>
1362 lines
42 KiB
Go
1362 lines
42 KiB
Go
//go:build linux || windows
|
||
|
||
package agentcli
|
||
|
||
import (
|
||
"bytes"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"os/exec"
|
||
"strings"
|
||
"sync"
|
||
"syscall"
|
||
"time"
|
||
|
||
"github.com/JianFeeeee/HologramAgent/internal/plugin"
|
||
safeclose "github.com/JianFeeeee/HologramAgent/internal/safeclose"
|
||
sdk "github.com/JianFeeeee/HologramAgent/internal/sdk"
|
||
)
|
||
|
||
const (
|
||
DefaultTimeout = 5 * time.Minute
|
||
ReadBufSize = 4096
|
||
MaxOutputBuffer = 128 * 1024
|
||
DefaultNotifyBytes = 2048 // 积累 2KB 未读输出再通知
|
||
DefaultNotifyInterval = 2 * time.Second // 同一终端两次通知的最小间隔（兜底）
|
||
)
|
||
|
||
// ptyTerm 抽象平台终端后端（Linux PTY / Windows ConPTY）。
|
||
type ptyTerm interface {
|
||
Read(buf []byte) (int, error)
|
||
WriteString(s string) (int, error)
|
||
Resize(rows, cols uint16) error
|
||
Running() bool
|
||
Kill() error
|
||
Close() error
|
||
}
|
||
|
||
// terminalRunning 判断终端后端进程是否仍在运行：优先走平台实现，
|
||
// 否则回落到 exec.Cmd 的 ProcessState（仅 Linux 传统路径）。
|
||
func terminalRunning(t *TerminalSession) bool {
|
||
if p, ok := t.session.(interface {
|
||
Running() bool
|
||
}); ok && p.Running() {
|
||
return true
|
||
}
|
||
return t.cmd != nil && (t.cmd.ProcessState == nil || !t.cmd.ProcessState.Exited())
|
||
}
|
||
|
||
// terminalWatch 终端提醒规则（由 terminal_watch 工具设置）。
|
||
type terminalWatch struct {
|
||
interval time.Duration // 固定时间反馈间隔，0 禁用
|
||
onExit bool // 命令执行结束提醒（默认 true）
|
||
bufferBytes int // 该终端专用缓冲阈值（字节），0 使用全局 notify_bytes
|
||
quiet bool // 静默模式：不随输出流通知，仅定时反馈/结束提醒/空闲汇总
|
||
}
|
||
|
||
type TerminalSession struct {
|
||
stopOnce sync.Once
|
||
doneOnce sync.Once
|
||
id string
|
||
command string
|
||
cmd *exec.Cmd
|
||
session ptyTerm
|
||
mu sync.Mutex
|
||
buf bytes.Buffer
|
||
createdAt time.Time
|
||
timeout time.Duration
|
||
closed bool
|
||
stopCh chan struct{}
|
||
done chan struct{}
|
||
|
||
// 通知节流字段
|
||
unreadBytes int // 最近一次通知后积累的未读字节数
|
||
lastNotify time.Time // 最近一次通知时间
|
||
lastData time.Time // 最近一次读到的数据时间（用于判定输出停止）
|
||
lastFeedback time.Time // 最近一次定时反馈时间
|
||
backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍
|
||
watch terminalWatch // 该终端的提醒规则
|
||
|
||
// resourcesReleased 标记 releaseResources 是否已执行（幂等保护）。
|
||
// 不与 closed 合用：closed 语义是「用户主动要求关闭」，releaseResources
|
||
// 是「后端资源已释放」，readLoop 自然退出时只后者为真。
|
||
resourcesReleased bool
|
||
|
||
// 实时画面推流（terminal_output 事件）
|
||
stream bytes.Buffer // 待推送的增量输出，由 readLoop 每 200ms flush 一次
|
||
}
|
||
|
||
func (t *TerminalSession) Write(input string) (int, error) {
|
||
return t.session.WriteString(input)
|
||
}
|
||
|
||
func (t *TerminalSession) Read(buf []byte) (int, error) {
|
||
return t.session.Read(buf)
|
||
}
|
||
|
||
func (t *TerminalSession) Resize(rows, cols uint16) error {
|
||
return t.session.Resize(rows, cols)
|
||
}
|
||
|
||
func (t *TerminalSession) Close() {
|
||
t.mu.Lock()
|
||
if t.closed {
|
||
t.mu.Unlock()
|
||
return
|
||
}
|
||
t.closed = true
|
||
t.mu.Unlock()
|
||
|
||
safeclose.Safe(&t.stopOnce, t.stopCh)
|
||
// 终止进程并释放 PTY（幂等；readLoop 自然退出时已调过就直接返回）。
|
||
t.releaseResources()
|
||
// 等 readLoop 走完退出流程（它会发最后的 output/停止事件）。
|
||
<-t.done
|
||
}
|
||
|
||
func (t *TerminalSession) ReadOutput() string {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
return t.buf.String()
|
||
}
|
||
|
||
func (t *TerminalSession) ReadAndClearOutput() string {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
s := t.buf.String()
|
||
t.buf.Reset()
|
||
return s
|
||
}
|
||
|
||
func (t *TerminalSession) appendOutput(data []byte) {
|
||
t.mu.Lock()
|
||
defer t.mu.Unlock()
|
||
if t.buf.Len()+len(data) > MaxOutputBuffer {
|
||
excess := t.buf.Len() + len(data) - MaxOutputBuffer
|
||
if t.buf.Len() > excess {
|
||
t.buf.Next(excess)
|
||
} else {
|
||
t.buf.Reset()
|
||
}
|
||
}
|
||
t.buf.Write(data)
|
||
// 同步追加到实时画面推流缓冲（最大 64KB，超出丢弃最旧部分）
|
||
const maxStream = 64 * 1024
|
||
if t.stream.Len()+len(data) > maxStream {
|
||
excess := t.stream.Len() + len(data) - maxStream
|
||
if t.stream.Len() > excess {
|
||
t.stream.Next(excess)
|
||
} else {
|
||
t.stream.Reset()
|
||
}
|
||
}
|
||
t.stream.Write(data)
|
||
}
|
||
|
||
// isClosing 报告该终端是否**已经被主动关闭**（close 工具 / 超时清理 / 停机）。
|
||
//
|
||
// 存在的理由：所有"终端结束"的通知都必须区分两种成因——
|
||
//
|
||
// 进程自己结束 → 是新闻，要告诉 agent
|
||
// 我们主动关掉 → 是回声，绝不能再注入
|
||
//
|
||
// 旧实现只看 r.err / terminalRunning，而主动关闭恰好也会让两者成立
|
||
// （Close 会 Kill 进程），于是 agent 在关闭终端后仍会收到
|
||
// 「读取结束: read /dev/ptmx: input/output error」「进程已退出」，
|
||
// 而此时 terminal_list 已是 0 个会话——这就是用户看到的
|
||
// "关闭终端后反复注入过期状态"。
|
||
// removeSession 从 sessions 表里摘掉一个终端（幂等）。
|
||
//
|
||
// 为什么单独成方法：readLoop 有三条退出路径（主动关闭 / 进程自己结束 /
|
||
// 超时），三条都必须清表。散落着写 delete 容易漏——实测漏的就是
|
||
// 「主动关闭」那条，后果是 cleanupLoop 每 30 秒把同一终端当成
|
||
// "进程已退出"再上报一次，形成持续回声。
|
||
func (p *Plugin) removeSession(id string) {
|
||
p.mu.Lock()
|
||
delete(p.sessions, id)
|
||
p.mu.Unlock()
|
||
}
|
||
|
||
func (p *Plugin) isClosing(t *TerminalSession) bool {
|
||
select {
|
||
case <-t.stopCh:
|
||
return true // Close() 已发出关闭信号
|
||
default:
|
||
}
|
||
t.mu.Lock()
|
||
closed := t.closed
|
||
t.mu.Unlock()
|
||
return closed
|
||
}
|
||
|
||
func (t *TerminalSession) IsExpired() bool {
|
||
return time.Since(t.createdAt) >= t.timeout
|
||
}
|
||
|
||
type Plugin struct {
|
||
stopChOnce sync.Once
|
||
name string
|
||
mu sync.Mutex
|
||
wg sync.WaitGroup
|
||
stopCh chan struct{}
|
||
stopOnce sync.Once
|
||
sessions map[string]*TerminalSession
|
||
nextID int
|
||
defaultTimeout time.Duration
|
||
notifyBytes int
|
||
notifyInterval time.Duration
|
||
}
|
||
|
||
func New(name string) *Plugin {
|
||
return &Plugin{
|
||
name: name,
|
||
stopCh: make(chan struct{}),
|
||
sessions: make(map[string]*TerminalSession),
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) Name() string { return p.name }
|
||
|
||
func (p *Plugin) Start(s *sdk.PluginSDK) error {
|
||
s.SetAutoRestart(true)
|
||
s.Settings().RegisterDef(sdk.ConfigDef{
|
||
Key: "default_timeout", Type: "string", DisplayName: "默认终端超时",
|
||
Description: "终端自动关闭的默认时间，例如 5m, 10m, 30m, 1h（默认 5m）",
|
||
Default: "5m",
|
||
})
|
||
s.Settings().RegisterDef(sdk.ConfigDef{
|
||
Key: "notify_bytes", Type: "int", DisplayName: "通知阈值字节数",
|
||
Description: "累积多少字节未读输出后发送通知（默认 2048）",
|
||
Default: "2048",
|
||
})
|
||
s.Settings().RegisterDef(sdk.ConfigDef{
|
||
Key: "notify_interval", Type: "string", DisplayName: "通知最小间隔",
|
||
Description: "同一终端两次通知的最小时间间隔，如 2s, 5s（默认 2s）",
|
||
Default: "2s",
|
||
})
|
||
if v, _ := s.Settings().Get("default_timeout"); v != nil {
|
||
if s, ok := v.(string); ok && s != "" {
|
||
if d, err := time.ParseDuration(s); err == nil {
|
||
p.defaultTimeout = d
|
||
}
|
||
}
|
||
}
|
||
if p.defaultTimeout <= 0 {
|
||
p.defaultTimeout = DefaultTimeout
|
||
}
|
||
if v, _ := s.Settings().Get("notify_bytes"); v != nil {
|
||
if i, ok := v.(float64); ok && i > 0 {
|
||
p.notifyBytes = int(i)
|
||
}
|
||
}
|
||
if p.notifyBytes <= 0 {
|
||
p.notifyBytes = DefaultNotifyBytes
|
||
}
|
||
if v, _ := s.Settings().Get("notify_interval"); v != nil {
|
||
if s, ok := v.(string); ok && s != "" {
|
||
if d, err := time.ParseDuration(s); err == nil {
|
||
p.notifyInterval = d
|
||
}
|
||
}
|
||
}
|
||
if p.notifyInterval <= 0 {
|
||
p.notifyInterval = DefaultNotifyInterval
|
||
}
|
||
|
||
// agentcli 通道：终端生命周期/输出事件经它注入 agent（见本文件 InjectText* 调用）。
|
||
_ = s.RegisterInputChannel("agentcli", sdk.ChannelDef{})
|
||
s.RegisterTool("terminal_create", sdk.ToolDef{
|
||
Name: "terminal_create",
|
||
Description: "创建一个新的交互式终端会话。返回终端 ID，后续通过此 ID 进行读写操作。适用于运行交互式程序如 vim、ssh、top、nano 等。" +
|
||
"通知模式通过 notify 参数选择（默认 exit）：exit=仅命令执行结束后提醒一次；interval=定时反馈（如 interval=30s 每 30 秒反馈一次状态摘要）；" +
|
||
"buffer=未读输出积累到指定字节数后提醒（如 buffer=8192）；多个模式用逗号组合（如 interval=30s,buffer=8192）。终端默认 5 分钟后自动关闭，可通过 timeout 参数调整。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"command": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "要执行的命令（默认 bash）。如需运行特定程序直接传入即可，例如：vim /tmp/test.txt",
|
||
},
|
||
"notify": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "通知模式（可选）：exit（默认，命令结束后提醒）；interval=时长（定时反馈，如 30s/1m）；buffer=字节数（缓冲阈值提醒）；可逗号组合",
|
||
},
|
||
"timeout": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端自动关闭时间，例如 5m, 10m, 30m, 1h（默认 5m）",
|
||
},
|
||
"rows": map[string]interface{}{
|
||
"type": "integer",
|
||
"description": "终端行数（默认 24）",
|
||
},
|
||
"cols": map[string]interface{}{
|
||
"type": "integer",
|
||
"description": "终端列数（默认 80）",
|
||
},
|
||
},
|
||
},
|
||
// 建会话：写共享会话表
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleCreate(s, args)
|
||
})
|
||
|
||
s.RegisterTool("terminal_write", sdk.ToolDef{
|
||
Name: "terminal_write",
|
||
Description: "向指定终端发送输入。支持普通文本和特殊键（通过 key 参数传入）。特殊键包括：enter, tab, escape, ctrl_a~ctrl_z, alt_a~alt_z, f1~f12, up, down, left, right, home, end, backspace, delete, page_up, page_down。普通文本传入 input 参数即可。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"id": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端 ID，来自 terminal_create 的返回值",
|
||
},
|
||
"input": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "要发送的文本内容（普通文字直接输入）",
|
||
},
|
||
"key": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "特殊按键：enter, tab, escape, ctrl_a~ctrl_z, alt_a~alt_z, f1~f12, up, down, left, right, home, end, backspace, delete, page_up, page_down",
|
||
},
|
||
},
|
||
"required": []string{"id"},
|
||
},
|
||
// 写终端：与会话缓冲区共享，须按序
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleWrite(s, args)
|
||
})
|
||
|
||
s.RegisterTool("terminal_read", sdk.ToolDef{
|
||
Name: "terminal_read",
|
||
Description: "读取指定终端的输出。mode=new（默认）返回自上次读取以来的新输出并清空缓冲；mode=now 返回终端当前显示的全部屏幕内容（不清空缓冲）。如需持续监控请多次调用。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"id": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端 ID",
|
||
},
|
||
"mode": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "读取模式：new（默认，新输出并清空缓冲）或 now（当前屏幕全部内容，不清理）",
|
||
},
|
||
"clear": map[string]interface{}{
|
||
"type": "boolean",
|
||
"description": "读取后是否清除缓冲区（默认与 mode 一致：new 清除，now 不清除）",
|
||
},
|
||
},
|
||
"required": []string{"id"},
|
||
},
|
||
// 读终端：与会话缓冲区共享，须按序
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleRead(args)
|
||
})
|
||
|
||
s.RegisterTool("terminal_resize", sdk.ToolDef{
|
||
Name: "terminal_resize",
|
||
Description: "调整指定终端的尺寸（行数和列数）。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"id": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端 ID",
|
||
},
|
||
"rows": map[string]interface{}{
|
||
"type": "integer",
|
||
"description": "行数",
|
||
},
|
||
"cols": map[string]interface{}{
|
||
"type": "integer",
|
||
"description": "列数",
|
||
},
|
||
},
|
||
"required": []string{"id"},
|
||
},
|
||
// 改窗口尺寸：改会话状态
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleResize(args)
|
||
})
|
||
|
||
s.RegisterTool("terminal_close", sdk.ToolDef{
|
||
Name: "terminal_close",
|
||
Description: "关闭指定终端会话。释放资源。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"id": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端 ID",
|
||
},
|
||
},
|
||
"required": []string{"id"},
|
||
},
|
||
// 关会话：改共享会话表
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleClose(s, args)
|
||
})
|
||
|
||
s.RegisterTool("terminal_list", sdk.ToolDef{
|
||
Name: "terminal_list",
|
||
Description: "列出所有活跃的终端会话及其状态。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{},
|
||
},
|
||
// 已核实只读：只列举会话
|
||
ParallelSafe: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleList()
|
||
})
|
||
|
||
s.RegisterTool("terminal_watch", sdk.ToolDef{
|
||
Name: "terminal_watch",
|
||
Description: "为指定终端设置提醒规则，避免长时间运行任务（编译/下载/构建等）的输出造成通知风暴。" +
|
||
"可选规则：interval=固定时间反馈（每隔该时长向 agent 反馈一次终端状态摘要）；" +
|
||
"on_exit=命令执行结束提醒；buffer_bytes=未读输出积累到该字节数时提醒一次；" +
|
||
"quiet=静默模式（抑制随输出流的通知，仅保留定时反馈与结束提醒，推荐长任务使用）。" +
|
||
"未提供的字段保持原值，clear=true 清除全部规则。默认 on_exit=true。",
|
||
NoMemory: true,
|
||
Parameters: map[string]interface{}{
|
||
"type": "object",
|
||
"properties": map[string]interface{}{
|
||
"id": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "终端 ID，来自 terminal_create 的返回值",
|
||
},
|
||
"interval": map[string]interface{}{
|
||
"type": "string",
|
||
"description": "固定时间反馈间隔，如 30s, 1m, 5m（可选，0 禁用）",
|
||
},
|
||
"on_exit": map[string]interface{}{
|
||
"type": "boolean",
|
||
"description": "命令执行结束时是否提醒（默认 true）",
|
||
},
|
||
"buffer_bytes": map[string]interface{}{
|
||
"type": "integer",
|
||
"description": "未读输出积累阈值（字节），达到后提醒一次（可选，默认全局 2048）",
|
||
},
|
||
"quiet": map[string]interface{}{
|
||
"type": "boolean",
|
||
"description": "静默模式：不随输出流通知，仅保留定时反馈与结束提醒（推荐编译/下载等长任务）",
|
||
},
|
||
"clear": map[string]interface{}{
|
||
"type": "boolean",
|
||
"description": "清除该终端全部提醒规则（恢复默认行为）",
|
||
},
|
||
},
|
||
"required": []string{"id"},
|
||
},
|
||
// 订阅输出：注册监听者，改共享状态
|
||
Serial: true,
|
||
}, func(args map[string]interface{}) (interface{}, error) {
|
||
return p.handleWatch(args)
|
||
})
|
||
|
||
p.wg.Add(1)
|
||
go p.cleanupLoop(s)
|
||
|
||
return nil
|
||
}
|
||
|
||
func (p *Plugin) Stop() error {
|
||
// 幂等：Stop 可能被多条路径调到（StopAndUnload 后再 StopAll、
|
||
// 停机与热重载交错）。close 一个已关闭的 channel 会 panic，
|
||
// 故用 Once 兜住。
|
||
// ★ 不能写成 p.stopOnce.Do(func(){ safeclose.Safe(&p.stopOnce, ...) })：
|
||
// sync.Once 不可重入 ⇒ 外层 Do 持锁、内层 Do 等同一把锁 ⇒ **自锁死锁**
|
||
// （实测：agentcli 测试 600s 超时，栈显示 sync.(*Once).Do 嵌套）。
|
||
// 正确做法：close 的守卫用**独立的** once。
|
||
p.stopOnce.Do(func() {
|
||
safeclose.Safe(&p.stopChOnce, p.stopCh)
|
||
p.shutdown()
|
||
})
|
||
return nil
|
||
}
|
||
|
||
func (p *Plugin) shutdown() {
|
||
// 必须**先**关掉活跃终端，再等 wg。
|
||
//
|
||
// 为什么顺序不能反：readLoop 自己也记在 p.wg 上，而它的退出条件
|
||
// 是「t.stopCh 收到信号」或「进程自行结束／超时」——它不监 p.stopCh。
|
||
// 旧实现在这里先 p.wg.Wait() 才 Close 终端：只要还有任何一个终端开着，
|
||
// readLoop 永远等不到 t.stopCh，wg.Wait() 就永返回不了。
|
||
// 后果是插件卸载 / 热重载（StopAndUnload / ReloadOne）与停机全挂在
|
||
// 这一步，且持有 registry 锁时就是全内核一起挂。
|
||
p.mu.Lock()
|
||
sessions := make([]*TerminalSession, 0, len(p.sessions))
|
||
for _, t := range p.sessions {
|
||
sessions = append(sessions, t)
|
||
}
|
||
p.sessions = nil
|
||
p.mu.Unlock()
|
||
|
||
for _, t := range sessions {
|
||
t.Close()
|
||
}
|
||
|
||
p.wg.Wait()
|
||
}
|
||
|
||
func (p *Plugin) nextIDLocked() string {
|
||
p.nextID++
|
||
return fmt.Sprintf("term_%d", p.nextID)
|
||
}
|
||
|
||
func (p *Plugin) handleCreate(s *sdk.PluginSDK, args map[string]interface{}) (interface{}, error) {
|
||
command, _ := args["command"].(string)
|
||
if command == "" {
|
||
command = defaultShell()
|
||
}
|
||
|
||
timeoutStr, _ := args["timeout"].(string)
|
||
timeout := p.defaultTimeout
|
||
if timeoutStr != "" {
|
||
if d, err := time.ParseDuration(timeoutStr); err == nil {
|
||
timeout = d
|
||
}
|
||
}
|
||
// SSH 命令自动使用更长的超时（5 分钟）
|
||
if timeoutStr == "" && (strings.HasPrefix(command, "ssh ") || strings.HasPrefix(command, "ssh -")) {
|
||
timeout = 5 * time.Minute
|
||
}
|
||
|
||
rows := uint16(24)
|
||
cols := uint16(80)
|
||
if r, ok := args["rows"].(float64); ok && r > 0 {
|
||
rows = uint16(r)
|
||
}
|
||
if c, ok := args["cols"].(float64); ok && c > 0 {
|
||
cols = uint16(c)
|
||
}
|
||
|
||
// 通知模式：默认 exit（命令执行结束后提醒一次）。
|
||
// 支持 interval=30s / buffer=8192 / quiet，可逗号组合。
|
||
watch := terminalWatch{onExit: true, quiet: true}
|
||
if notifyStr, ok := args["notify"].(string); ok && notifyStr != "" {
|
||
watch = parseNotifyMode(notifyStr, watch)
|
||
}
|
||
|
||
term, cmd, err := newCommandPty(command, rows, cols)
|
||
if err != nil {
|
||
return map[string]interface{}{"error": fmt.Sprintf("创建终端失败: %v", err)}, nil
|
||
}
|
||
|
||
session := &TerminalSession{
|
||
id: "",
|
||
command: command,
|
||
cmd: cmd,
|
||
session: term,
|
||
createdAt: time.Now(),
|
||
timeout: timeout,
|
||
stopCh: make(chan struct{}),
|
||
done: make(chan struct{}),
|
||
watch: watch,
|
||
}
|
||
|
||
p.mu.Lock()
|
||
id := p.nextIDLocked()
|
||
session.id = id
|
||
p.sessions[id] = session
|
||
p.mu.Unlock()
|
||
|
||
p.wg.Add(1)
|
||
go p.readLoop(session, s)
|
||
|
||
// 生命周期事件：终端创建即时上报（带 command），让内核权威视图与所有
|
||
// 订阅者（WebUI/CLI）即使在该终端无输出的情况下也能知道它的存在。
|
||
emitTermState(s, session, true)
|
||
|
||
log.Printf("[agentcli] created terminal %s: command=%q timeout=%v rows=%d cols=%d", id, command, timeout, rows, cols)
|
||
|
||
return map[string]interface{}{
|
||
"id": id,
|
||
"status": "created",
|
||
"command": command,
|
||
"timeout": timeout.String(),
|
||
"rows": rows,
|
||
"cols": cols,
|
||
"notify_mode": notifyModeString(watch),
|
||
}, nil
|
||
}
|
||
|
||
// notifyModeString 输出可读的通知模式描述。
|
||
func notifyModeString(w terminalWatch) string {
|
||
var parts []string
|
||
if w.onExit {
|
||
parts = append(parts, "exit")
|
||
}
|
||
if w.interval > 0 {
|
||
parts = append(parts, "interval="+w.interval.String())
|
||
}
|
||
if w.bufferBytes > 0 {
|
||
parts = append(parts, fmt.Sprintf("buffer=%d", w.bufferBytes))
|
||
}
|
||
if len(parts) == 0 {
|
||
return "quiet"
|
||
}
|
||
return strings.Join(parts, ",")
|
||
}
|
||
|
||
func (p *Plugin) handleWrite(s *sdk.PluginSDK, args map[string]interface{}) (interface{}, error) {
|
||
id, _ := args["id"].(string)
|
||
if id == "" {
|
||
return map[string]interface{}{"error": "id is required"}, nil
|
||
}
|
||
|
||
p.mu.Lock()
|
||
session, ok := p.sessions[id]
|
||
p.mu.Unlock()
|
||
if !ok {
|
||
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
|
||
}
|
||
|
||
input, _ := args["input"].(string)
|
||
key, _ := args["key"].(string)
|
||
|
||
var data []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
|
||
}
|
||
|
||
n, err := session.Write(string(data))
|
||
if err != nil {
|
||
return map[string]interface{}{"error": fmt.Sprintf("写入失败: %v", err)}, nil
|
||
}
|
||
|
||
return map[string]interface{}{
|
||
"status": "ok",
|
||
"bytes": n,
|
||
"terminal": id,
|
||
}, nil
|
||
}
|
||
|
||
// parseNotifyMode 解析 notify 参数并合并进 watch。
|
||
// 支持：exit / quiet / interval=时长 / buffer=字节数，逗号分隔组合。
|
||
func parseNotifyMode(s string, base terminalWatch) terminalWatch {
|
||
w := base
|
||
for _, part := range strings.Split(s, ",") {
|
||
part = strings.TrimSpace(part)
|
||
if part == "" {
|
||
continue
|
||
}
|
||
kv := strings.SplitN(part, "=", 2)
|
||
key := strings.TrimSpace(kv[0])
|
||
val := ""
|
||
if len(kv) == 2 {
|
||
val = strings.TrimSpace(kv[1])
|
||
}
|
||
switch key {
|
||
case "exit":
|
||
w.onExit = true
|
||
w.quiet = false
|
||
case "quiet", "silent":
|
||
w.quiet = true
|
||
case "interval":
|
||
if d, err := time.ParseDuration(val); err == nil && d > 0 {
|
||
w.interval = d
|
||
}
|
||
case "buffer":
|
||
var n int
|
||
if _, err := fmt.Sscanf(val, "%d", &n); err == nil && n > 0 {
|
||
w.bufferBytes = n
|
||
}
|
||
}
|
||
}
|
||
return w
|
||
}
|
||
|
||
func (p *Plugin) handleRead(args map[string]interface{}) (interface{}, error) {
|
||
id, _ := args["id"].(string)
|
||
if id == "" {
|
||
return map[string]interface{}{"error": "id is required"}, nil
|
||
}
|
||
|
||
mode, _ := args["mode"].(string)
|
||
clear := mode != "now"
|
||
if v, ok := args["clear"].(bool); ok {
|
||
clear = v
|
||
}
|
||
|
||
p.mu.Lock()
|
||
session, ok := p.sessions[id]
|
||
p.mu.Unlock()
|
||
if !ok {
|
||
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
|
||
}
|
||
|
||
var output string
|
||
session.mu.Lock()
|
||
if clear {
|
||
output = session.buf.String()
|
||
session.buf.Reset()
|
||
// 实时画面推流缓冲同步清空，避免 terminal_output 事件与读取结果重复
|
||
session.stream.Reset()
|
||
} else {
|
||
output = session.buf.String()
|
||
}
|
||
session.mu.Unlock()
|
||
|
||
if output == "" {
|
||
if mode == "now" {
|
||
output = "[终端当前无屏幕内容]"
|
||
} else {
|
||
output = "[终端无新输出]"
|
||
}
|
||
}
|
||
|
||
return map[string]interface{}{
|
||
"status": "ok",
|
||
"terminal": id,
|
||
"mode": mode,
|
||
"output": output,
|
||
"running": terminalRunning(session),
|
||
"uptime": time.Since(session.createdAt).String(),
|
||
}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleResize(args map[string]interface{}) (interface{}, error) {
|
||
id, _ := args["id"].(string)
|
||
if id == "" {
|
||
return map[string]interface{}{"error": "id is required"}, nil
|
||
}
|
||
|
||
rows, okRows := args["rows"].(float64)
|
||
cols, okCols := args["cols"].(float64)
|
||
if !okRows || !okCols {
|
||
return map[string]interface{}{"error": "rows 和 cols 为必填"}, nil
|
||
}
|
||
|
||
p.mu.Lock()
|
||
session, ok := p.sessions[id]
|
||
p.mu.Unlock()
|
||
if !ok {
|
||
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
|
||
}
|
||
|
||
if err := session.Resize(uint16(rows), uint16(cols)); err != nil {
|
||
return map[string]interface{}{"error": fmt.Sprintf("调整尺寸失败: %v", err)}, nil
|
||
}
|
||
|
||
return map[string]interface{}{
|
||
"status": "ok",
|
||
"terminal": id,
|
||
"rows": rows,
|
||
"cols": cols,
|
||
}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleClose(s *sdk.PluginSDK, args map[string]interface{}) (interface{}, error) {
|
||
id, _ := args["id"].(string)
|
||
if id == "" {
|
||
return map[string]interface{}{"error": "id is required"}, nil
|
||
}
|
||
|
||
p.mu.Lock()
|
||
session, ok := p.sessions[id]
|
||
if ok {
|
||
delete(p.sessions, id)
|
||
}
|
||
p.mu.Unlock()
|
||
|
||
if !ok {
|
||
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
|
||
}
|
||
|
||
session.Close()
|
||
log.Printf("[agentcli] closed terminal %s", id)
|
||
|
||
// 生命周期事件：显式上报关闭（readLoop 退出时也会发，幂等）。
|
||
emitTermState(s, session, false)
|
||
|
||
return map[string]interface{}{
|
||
"status": "closed",
|
||
"terminal": id,
|
||
}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleWatch(args map[string]interface{}) (interface{}, error) {
|
||
id, _ := args["id"].(string)
|
||
if id == "" {
|
||
return map[string]interface{}{"error": "id is required"}, nil
|
||
}
|
||
|
||
p.mu.Lock()
|
||
session, ok := p.sessions[id]
|
||
p.mu.Unlock()
|
||
if !ok {
|
||
return map[string]interface{}{"error": fmt.Sprintf("终端 %s 不存在或已关闭", id)}, nil
|
||
}
|
||
|
||
session.mu.Lock()
|
||
if v, ok := args["clear"].(bool); ok && v {
|
||
session.watch = terminalWatch{onExit: true}
|
||
} else {
|
||
if v, ok := args["interval"].(string); ok && v != "" {
|
||
if d, err := time.ParseDuration(v); err == nil && d >= 0 {
|
||
session.watch.interval = d
|
||
}
|
||
}
|
||
if v, ok := args["on_exit"].(bool); ok {
|
||
session.watch.onExit = v
|
||
}
|
||
if v, ok := args["buffer_bytes"].(float64); ok && v >= 0 {
|
||
session.watch.bufferBytes = int(v)
|
||
}
|
||
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()
|
||
|
||
log.Printf("[agentcli] watch updated for %s: %+v", id, w)
|
||
return map[string]interface{}{
|
||
"status": "ok",
|
||
"terminal": id,
|
||
"interval": w.interval.String(),
|
||
"on_exit": w.onExit,
|
||
"buffer_bytes": w.bufferBytes,
|
||
"quiet": w.quiet,
|
||
}, nil
|
||
}
|
||
|
||
func (p *Plugin) handleList() (interface{}, error) {
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
|
||
type termInfo struct {
|
||
ID string `json:"id"`
|
||
Command string `json:"command"`
|
||
Uptime string `json:"uptime"`
|
||
ExpiresIn string `json:"expires_in"`
|
||
Running bool `json:"running"`
|
||
}
|
||
|
||
var terms []termInfo
|
||
for _, t := range p.sessions {
|
||
running := terminalRunning(t)
|
||
remaining := t.timeout - time.Since(t.createdAt)
|
||
if remaining < 0 {
|
||
remaining = 0
|
||
}
|
||
terms = append(terms, termInfo{
|
||
ID: t.id,
|
||
Command: t.command,
|
||
Uptime: time.Since(t.createdAt).Round(time.Second).String(),
|
||
ExpiresIn: remaining.Round(time.Second).String(),
|
||
Running: running,
|
||
})
|
||
}
|
||
|
||
if terms == nil {
|
||
terms = []termInfo{}
|
||
}
|
||
|
||
return map[string]interface{}{
|
||
"status": "ok",
|
||
"count": len(terms),
|
||
"terminals": terms,
|
||
}, nil
|
||
}
|
||
|
||
// flushTermStream 把终端待推送的增量输出作为 terminal_output 事件发布。
|
||
//
|
||
// 必须抽成公共函数：退出路径（进程结束/超时/读取错误/stopCh）与常规 200ms
|
||
// ticker 都要走同一条推送路径。否则短命令（echo/ls 这类在首个 ticker 之前
|
||
// 就结束的）残留在 stream 里的输出永远发不出事件，只留在 buf 里——
|
||
// agent 用 terminal_read 能看到，但内核权威视图的 output 恒为空。
|
||
// 返回是否真的发布了（无残留时为 false）。
|
||
func flushTermStream(s *sdk.PluginSDK, t *TerminalSession) bool {
|
||
if s == nil || t == nil {
|
||
return false
|
||
}
|
||
var streamData string
|
||
t.mu.Lock()
|
||
if t.stream.Len() > 0 {
|
||
streamData = t.stream.String()
|
||
t.stream.Reset()
|
||
}
|
||
t.mu.Unlock()
|
||
if streamData == "" {
|
||
return false
|
||
}
|
||
s.Publish(&sdk.Event{
|
||
Type: sdk.EventTerminalOutput,
|
||
Source: "agentcli",
|
||
Payload: map[string]interface{}{"terminal_id": t.id, "command": t.command, "output": streamData, "running": terminalRunning(t)},
|
||
Timestamp: time.Now().UnixMilli(),
|
||
})
|
||
return true
|
||
}
|
||
|
||
// emitTermState 把终端存活状态作为 EventTerminalOutput 事件上报。
|
||
//
|
||
// 生命周期事件（创建/关闭/退出/超时）必须显式发：终端无输出时 ticker
|
||
// 不会发事件，而所有插件（WebUI/CLI）的终端视图都依赖这些事件
|
||
// 才能知道终端的存在与终止。payload 带 command 供无 ToolCall 事件的
|
||
// 直调路径（CLI /terminal create 经 ToolAPI.ExecuteTool）回填命令名。
|
||
// 幂等：重复发同一 running 值不会产生状态跳变。
|
||
func emitTermState(s *sdk.PluginSDK, t *TerminalSession, running bool) {
|
||
if s == nil || t == nil {
|
||
return
|
||
}
|
||
s.Publish(&sdk.Event{
|
||
Type: sdk.EventTerminalOutput,
|
||
Source: "agentcli",
|
||
Payload: map[string]interface{}{"terminal_id": t.id, "command": t.command, "running": running},
|
||
Timestamp: time.Now().UnixMilli(),
|
||
})
|
||
}
|
||
|
||
// releaseResources 幂等地释放终端后端：杀进程 + 关 PTY。
|
||
//
|
||
// 为何需要单独一个方法：readLoop 是终端自然的退出点（超时/进程结束/
|
||
// 读取错误/插件停机），但 close(t.done) 的时机意味着它**不能**调
|
||
// TerminalSession.Close()——后者会 <-t.done 等 readLoop 退出，而 readLoop
|
||
// 正在自己里面，直接死锁。所以这里只做「不再需要 readLoop 配合」的那半：
|
||
// 终止进程与释放 fd。
|
||
func (t *TerminalSession) releaseResources() {
|
||
t.mu.Lock()
|
||
if t.resourcesReleased {
|
||
t.mu.Unlock()
|
||
return
|
||
}
|
||
t.resourcesReleased = true
|
||
t.mu.Unlock()
|
||
|
||
if t.session != nil {
|
||
_ = t.session.Kill()
|
||
// 回收子进程（避免僵尸）。后端可选实现：Linux PTY 在 Kill 后
|
||
// 必须 Wait 才能把 <defunct> 清掉；不实现的后端（如 Windows
|
||
// ConPTY）跳过即可。
|
||
if reaper, ok := t.session.(interface{ Wait() error }); ok {
|
||
_ = reaper.Wait()
|
||
}
|
||
_ = t.session.Close()
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) {
|
||
defer p.wg.Done()
|
||
defer safeclose.Safe(&t.doneOnce, t.done)
|
||
|
||
// 无论从哪个分支退出，都释放进程与 PTY。
|
||
//
|
||
// 旧实现只在「超时」和「进程退出」两个分支 delete(sessions) 后直接
|
||
// return：超时分支完全不碰 session，一个 sleep 999 超时后进程、PTY fd
|
||
// 与 reader 協程全数泄漏（readLoop 已经从 sessions 里删掉了，cleanupLoop
|
||
// 也再看不到它，没人能回收）。defer 保证「只要退出就释放」。
|
||
defer t.releaseResources()
|
||
|
||
// reader 协程独享这个读缓冲：结果随 readResult 携带，
|
||
// readLoop 不再从其中做 copy（见 reader 注释，那是对共享缓冲
|
||
// 的并发读写，-race 实测触发）。
|
||
buf := make([]byte, ReadBufSize)
|
||
pollInterval := 200 * time.Millisecond
|
||
|
||
readCh := make(chan readResult, 4)
|
||
go p.reader(t, buf, readCh)
|
||
|
||
// 实时画面推流 ticker：每 200ms 批量发布一次 terminal_output 事件
|
||
flushTicker := time.NewTicker(200 * time.Millisecond)
|
||
defer flushTicker.Stop()
|
||
|
||
// 退出时先把残留输出推出去，再报停止。
|
||
//
|
||
// defer 是 LIFO：下面这两行声明顺序决定执行顺序——先 flush 后 emit。
|
||
// 若反了，停止事件会先于最后一段输出到达，内核会先把 running 置 false
|
||
// 再追加输出（状态看着对但顺序错）；更重要的是短命令的输出
|
||
// 只存在于 stream 里，不 flush 就彻底丢了。
|
||
defer emitTermState(s, t, false)
|
||
defer flushTermStream(s, t)
|
||
|
||
// 立即发送首次"终端已启动"通知，让 agent 感知存在。
|
||
// 用 NoMemory：这是状态提示，不是对话内容。不关掉的话每开一个终端都会
|
||
// 在记忆里留下一条"[终端 X 已启动]"，把真实内容挤掉。
|
||
s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 已启动]", t.id),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
now := time.Now()
|
||
t.mu.Lock()
|
||
t.lastNotify = now
|
||
t.lastData = now
|
||
t.lastFeedback = now
|
||
t.mu.Unlock()
|
||
|
||
// 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低
|
||
hardNotifyBytes := 64 * 1024
|
||
hardNotifyInterval := 10 * time.Second
|
||
// 输出停止判定：超过该时长无新数据则视为输出停止
|
||
quietLatency := 2 * time.Second
|
||
|
||
for {
|
||
// p.stopCh：插件停机。readLoop 记在 p.wg 上，若不在此退出，
|
||
// Stop() 的 wg.Wait() 就只能等终端自己超时（最长 30 分钟）。
|
||
select {
|
||
case <-p.stopCh:
|
||
return
|
||
default:
|
||
}
|
||
|
||
if t.IsExpired() {
|
||
log.Printf("[agentcli] terminal %s expired after %v", t.id, t.timeout)
|
||
s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 已超时关闭（%s）]", t.id, t.timeout),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
p.removeSession(t.id)
|
||
return
|
||
}
|
||
|
||
if !terminalRunning(t) {
|
||
// 与 r.err 分支同理：主动关闭时进程是被我们 Kill 掉的，
|
||
// "进程已退出"是**我们造成的**结果，不是需要告诉 agent 的新事件。
|
||
// 不判这一下，关闭终端会立刻收到一条"进程已退出"回声。
|
||
if !p.isClosing(t) {
|
||
if t.watch.onExit {
|
||
s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 中的命令已执行结束]", t.id),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
} else {
|
||
s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 中的进程已退出]", t.id),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
}
|
||
}
|
||
p.removeSession(t.id)
|
||
return
|
||
}
|
||
|
||
// 固定时间反馈：watch.interval > 0 时每隔该时长主动反馈一次状态摘要
|
||
t.mu.Lock()
|
||
if t.watch.interval > 0 && time.Since(t.lastFeedback) >= t.watch.interval {
|
||
t.lastFeedback = time.Now()
|
||
t.lastNotify = t.lastFeedback
|
||
unread := t.unreadBytes
|
||
t.unreadBytes = 0
|
||
preview := previewTail(t.buf.String(), 120)
|
||
t.mu.Unlock()
|
||
s.InjectText("agentcli", "agentcli",
|
||
// 刻意**不**用 NoMemory：这条带上终端真实输出（preview），
|
||
// 属于该记的内容。只有纯状态通知才关记忆。
|
||
fmt.Sprintf("[终端 %s 定时反馈: 运行中, 期间新输出约 %d 字节]\n%s", t.id, unread, preview))
|
||
continue
|
||
}
|
||
t.mu.Unlock()
|
||
|
||
select {
|
||
case <-t.stopCh:
|
||
// ★ 主动关闭：静默退出，且**必须清掉 sessions 条目**。
|
||
//
|
||
// 旧实现只 return 不删表。于是已关闭的终端仍留在 sessions 里，
|
||
// cleanupLoop 每 30s 扫到它时发现 !terminalRunning，
|
||
// 又注入一条「[终端 X 中的进程已退出]」——而 agent 早已
|
||
// 从 close 工具得到"已关闭"，terminal_list 也显示 0 个。
|
||
// 这就是"关闭终端后反复注入过期状态"的直接成因：
|
||
// 不是一次回声，而是**每 30 秒一次**的持续性回声。
|
||
p.removeSession(t.id)
|
||
return
|
||
case <-flushTicker.C:
|
||
// 批量推送终端实时画面增量（独立 ticker，避免被高密度数据饿死）
|
||
flushTermStream(s, t)
|
||
case r := <-readCh:
|
||
if r.err != nil {
|
||
// 读取错误/EOF：只在**进程自己结束**时通知。
|
||
//
|
||
// ★ 主动关闭（close 工具 / 超时清理 / 插件停机）必须静默：
|
||
// 那条路径先关 stopCh → Kill 进程 → PTY 读端立刻返
|
||
// 「read /dev/ptmx: input/output error」。若这里照样注入，
|
||
// agent 会收到一条"读取结束"的**回声**——它对应的终端早已
|
||
// 从 sessions 里删掉，terminal_list 显示 0 个，但这条消息
|
||
// 仍会被当成新事件唤醒一轮推理（用户看到的就是
|
||
// "关闭终端后反复注入过期状态"）。
|
||
//
|
||
// 判据用 closed/stopCh 而不是 r.err 的文本：错误文案随平台
|
||
// 与内核版本变化，靠匹配字符串判断"是不是正常关闭"必然腐烂。
|
||
if p.isClosing(t) || t.IsExpired() {
|
||
return // 主动关闭 / 超时清理：不注入
|
||
}
|
||
s.InjectTextOpts("agentcli", "agentcli", fmt.Sprintf("[终端 %s 读取结束: %v]", t.id, r.err),
|
||
sdk.InjectOptions{NoMemory: true})
|
||
return
|
||
}
|
||
if r.n > 0 {
|
||
t.appendOutput(r.data)
|
||
|
||
// 缓冲阈值通知（仅当 agent 显式选择 buffer 模式，或未读积累达到硬上限）。
|
||
// 默认模式（仅 exit 提醒）下不随输出流通知，杜绝通知风暴。
|
||
t.mu.Lock()
|
||
t.lastData = time.Now()
|
||
t.unreadBytes += r.n
|
||
bufThr := t.watch.bufferBytes
|
||
if bufThr <= 0 {
|
||
bufThr = p.notifyBytes
|
||
}
|
||
minInterval := p.notifyInterval
|
||
if t.watch.interval > 0 {
|
||
minInterval = t.watch.interval
|
||
}
|
||
// 风暴退避：距上次通知不足 1s 说明输出极速，通知间隔翻倍（上限 30s）
|
||
if time.Since(t.lastNotify) < time.Second && t.unreadBytes >= bufThr {
|
||
if t.backoff == 0 {
|
||
t.backoff = minInterval
|
||
} else if t.backoff < 30*time.Second {
|
||
t.backoff *= 2
|
||
if t.backoff > 30*time.Second {
|
||
t.backoff = 30 * time.Second
|
||
}
|
||
}
|
||
}
|
||
interval := t.backoff + minInterval
|
||
isHard := t.watch.bufferBytes <= 0 && t.unreadBytes >= hardNotifyBytes
|
||
if isHard && hardNotifyInterval > interval {
|
||
interval = hardNotifyInterval
|
||
}
|
||
need := t.unreadBytes >= bufThr && time.Since(t.lastNotify) >= interval
|
||
if need {
|
||
t.lastNotify = time.Now()
|
||
t.unreadBytes = 0
|
||
preview := previewTail(t.buf.String(), 200)
|
||
t.mu.Unlock()
|
||
s.InjectText("agentcli", "agentcli",
|
||
// 同样刻意保留记忆：preview 是终端新输出，是真实内容。
|
||
fmt.Sprintf("[终端 %s 有新输出]\n%s", t.id, preview))
|
||
} else {
|
||
t.mu.Unlock()
|
||
}
|
||
}
|
||
case <-time.After(pollInterval):
|
||
// 空闲轮询：输出已停止时复位退避
|
||
t.mu.Lock()
|
||
if t.backoff > 0 && time.Since(t.lastData) >= quietLatency {
|
||
t.backoff = 0
|
||
}
|
||
t.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
|
||
// previewTail 返回 s 末尾最多 n 字符，并转义控制字符保证可读。
|
||
func previewTail(s string, n int) string {
|
||
if len(s) > n {
|
||
s = s[len(s)-n:]
|
||
}
|
||
return sanitizePreview(s)
|
||
}
|
||
|
||
type readResult struct {
|
||
n int
|
||
data []byte
|
||
err error
|
||
}
|
||
|
||
// reader 从终端读取输出并通过 channel 交给 readLoop。
|
||
//
|
||
// 读到的数据**随结果一起传**而不是复用外层共享的 buf：
|
||
// reader 是唯一写 buf 的 goroutine，readLoop 又常在 reader 尚未
|
||
// 写完下一段时就从 buf[:r.n] 做 copy——同一个 shared buf 被并发
|
||
// 读写就是 data race（-race 实测触发）。改为每个结果自带切片后，
|
||
// 读与拷贝天然隔离，不再共享可变状态。
|
||
func (p *Plugin) reader(t *TerminalSession, buf []byte, ch chan<- readResult) {
|
||
for {
|
||
n, err := t.session.Read(buf)
|
||
var data []byte
|
||
if n > 0 {
|
||
data = make([]byte, n)
|
||
copy(data, buf[:n])
|
||
}
|
||
select {
|
||
case ch <- readResult{n, data, err}:
|
||
case <-t.stopCh:
|
||
return
|
||
}
|
||
if err != nil {
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
func (p *Plugin) cleanupLoop(s *sdk.PluginSDK) {
|
||
defer p.wg.Done()
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
for {
|
||
select {
|
||
case <-p.stopCh:
|
||
return
|
||
case <-ticker.C:
|
||
// 先持锁筛选出待关闭的终终并移出 map，再在锁外逐个关闭。
|
||
// 旧实现在持锁期间 go func 调 term.Close()（内部会 Kill 进程并等
|
||
// <-t.done），临界区被拉长且与 TerminalSession 的退出路径交错。
|
||
var expired []*TerminalSession
|
||
p.mu.Lock()
|
||
for id, t := range p.sessions {
|
||
switch {
|
||
case t.IsExpired():
|
||
log.Printf("[agentcli] cleanup: terminal %s expired", id)
|
||
case !terminalRunning(t):
|
||
log.Printf("[agentcli] cleanup: terminal %s process exited", id)
|
||
default:
|
||
continue
|
||
}
|
||
delete(p.sessions, id)
|
||
expired = append(expired, t)
|
||
}
|
||
p.mu.Unlock()
|
||
|
||
for _, t := range expired {
|
||
go func(term *TerminalSession) { term.Close() }(t)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func mapKey(key string) ([]byte, error) {
|
||
key = strings.ToLower(key)
|
||
switch key {
|
||
case "enter":
|
||
return []byte{0x0D}, nil
|
||
case "tab":
|
||
return []byte{0x09}, nil
|
||
case "escape", "esc":
|
||
return []byte{0x1B}, nil
|
||
case "backspace":
|
||
return []byte{0x7F}, nil
|
||
case "delete":
|
||
return []byte{0x1B, 0x5B, 0x33, 0x7E}, nil
|
||
case "home":
|
||
return []byte{0x1B, 0x5B, 0x48}, nil
|
||
case "end":
|
||
return []byte{0x1B, 0x5B, 0x46}, nil
|
||
case "page_up":
|
||
return []byte{0x1B, 0x5B, 0x35, 0x7E}, nil
|
||
case "page_down":
|
||
return []byte{0x1B, 0x5B, 0x36, 0x7E}, nil
|
||
case "up":
|
||
return []byte{0x1B, 0x5B, 0x41}, nil
|
||
case "down":
|
||
return []byte{0x1B, 0x5B, 0x42}, nil
|
||
case "right":
|
||
return []byte{0x1B, 0x5B, 0x43}, nil
|
||
case "left":
|
||
return []byte{0x1B, 0x5B, 0x44}, nil
|
||
}
|
||
|
||
if strings.HasPrefix(key, "ctrl_") && len(key) == 6 {
|
||
c := key[5]
|
||
if c >= 'a' && c <= 'z' {
|
||
return []byte{byte(c - 'a' + 1)}, nil
|
||
}
|
||
}
|
||
|
||
if strings.HasPrefix(key, "alt_") && len(key) == 5 {
|
||
c := key[4]
|
||
if c >= 'a' && c <= 'z' {
|
||
return []byte{0x1B, c}, nil
|
||
}
|
||
}
|
||
|
||
if strings.HasPrefix(key, "f") && len(key) <= 4 {
|
||
var n int
|
||
if _, err := fmt.Sscanf(key, "f%d", &n); err == nil && n >= 1 && n <= 12 {
|
||
return functionKeyCode(n), nil
|
||
}
|
||
}
|
||
|
||
return nil, fmt.Errorf("不支持的特殊按键: %s", key)
|
||
}
|
||
|
||
func functionKeyCode(n int) []byte {
|
||
switch n {
|
||
case 1:
|
||
return []byte{0x1B, 0x5B, 0x50}
|
||
case 2:
|
||
return []byte{0x1B, 0x5B, 0x51}
|
||
case 3:
|
||
return []byte{0x1B, 0x5B, 0x52}
|
||
case 4:
|
||
return []byte{0x1B, 0x5B, 0x53}
|
||
case 5:
|
||
return []byte{0x1B, 0x5B, 0x31, 0x35, 0x7E}
|
||
case 6:
|
||
return []byte{0x1B, 0x5B, 0x31, 0x37, 0x7E}
|
||
case 7:
|
||
return []byte{0x1B, 0x5B, 0x31, 0x38, 0x7E}
|
||
case 8:
|
||
return []byte{0x1B, 0x5B, 0x31, 0x39, 0x7E}
|
||
case 9:
|
||
return []byte{0x1B, 0x5B, 0x32, 0x30, 0x7E}
|
||
case 10:
|
||
return []byte{0x1B, 0x5B, 0x32, 0x31, 0x7E}
|
||
case 11:
|
||
return []byte{0x1B, 0x5B, 0x32, 0x32, 0x7E}
|
||
case 12:
|
||
return []byte{0x1B, 0x5B, 0x32, 0x34, 0x7E}
|
||
default:
|
||
return []byte{}
|
||
}
|
||
}
|
||
|
||
func isTimeoutError(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
if opErr, ok := err.(*os.PathError); ok {
|
||
err = opErr.Err
|
||
}
|
||
if syscallErr, ok := err.(syscall.Errno); ok {
|
||
return syscallErr == syscall.EAGAIN || syscallErr == syscall.EWOULDBLOCK || syscallErr == syscall.ETIMEDOUT
|
||
}
|
||
return strings.Contains(err.Error(), "timeout") || strings.Contains(err.Error(), "would block")
|
||
}
|
||
|
||
func init() {
|
||
plugin.RegisterFactory("agentcli", func(name string, config map[string]interface{}) (sdk.Plugin, error) {
|
||
return New(name), nil
|
||
})
|
||
plugin.RegisterPluginMeta("agentcli", "终端交互", "Agent CLI")
|
||
}
|
||
|
||
func sanitizePreview(s string) string {
|
||
var buf bytes.Buffer
|
||
for _, r := range s {
|
||
if r == '\n' {
|
||
buf.WriteString("\\n")
|
||
} else if r == '\r' {
|
||
buf.WriteString("\\r")
|
||
} else if r == '\t' {
|
||
buf.WriteString("\\t")
|
||
} else if r >= 32 && r <= 126 {
|
||
buf.WriteRune(r)
|
||
} else if r == 0x1B {
|
||
buf.WriteString("^[")
|
||
} else if r < 32 {
|
||
fmt.Fprintf(&buf, "^%c", r+'A'-1)
|
||
} else {
|
||
buf.WriteRune(r)
|
||
}
|
||
}
|
||
return buf.String()
|
||
}
|