diff --git a/internal/plugins/agentcli/plugin.go b/internal/plugins/agentcli/plugin.go index e9e133e..3b38520 100644 --- a/internal/plugins/agentcli/plugin.go +++ b/internal/plugins/agentcli/plugin.go @@ -18,10 +18,10 @@ import ( ) const ( - DefaultTimeout = 5 * time.Minute - ReadBufSize = 4096 - MaxOutputBuffer = 128 * 1024 - DefaultNotifyBytes = 2048 // 积累 2KB 未读输出再通知 + DefaultTimeout = 5 * time.Minute + ReadBufSize = 4096 + MaxOutputBuffer = 128 * 1024 + DefaultNotifyBytes = 2048 // 积累 2KB 未读输出再通知 DefaultNotifyInterval = 2 * time.Second // 同一终端两次通知的最小间隔（兜底） ) @@ -68,12 +68,12 @@ type TerminalSession struct { done chan struct{} // 通知节流字段 - unreadBytes int // 最近一次通知后积累的未读字节数 - lastNotify time.Time // 最近一次通知时间 - lastData time.Time // 最近一次读到的数据时间（用于判定输出停止） - lastFeedback time.Time // 最近一次定时反馈时间 - backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍 - watch terminalWatch // 该终端的提醒规则 + unreadBytes int // 最近一次通知后积累的未读字节数 + lastNotify time.Time // 最近一次通知时间 + lastData time.Time // 最近一次读到的数据时间（用于判定输出停止） + lastFeedback time.Time // 最近一次定时反馈时间 + backoff time.Duration // 输出风暴退避：持续高速输出时通知间隔翻倍 + watch terminalWatch // 该终端的提醒规则 // 实时画面推流（terminal_output 事件） stream bytes.Buffer // 待推送的增量输出，由 readLoop 每 200ms flush 一次 @@ -226,7 +226,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { 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, + NoMemory: true, Parameters: map[string]interface{}{ "type": "object", "properties": map[string]interface{}{ @@ -283,7 +283,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { }) s.RegisterTool("terminal_read", sdk.ToolDef{ - Name: "terminal_read", + Name: "terminal_read", Description: "读取指定终端的输出。mode=new（默认）返回自上次读取以来的新输出并清空缓冲；mode=now 返回终端当前显示的全部屏幕内容（不清空缓冲）。如需持续监控请多次调用。", NoMemory: true, Parameters: map[string]interface{}{ @@ -759,11 +759,11 @@ func (p *Plugin) handleList() (interface{}, error) { defer p.mu.Unlock() type termInfo struct { - ID string `json:"id"` - Command string `json:"command"` - Uptime string `json:"uptime"` + ID string `json:"id"` + Command string `json:"command"` + Uptime string `json:"uptime"` ExpiresIn string `json:"expires_in"` - Running bool `json:"running"` + Running bool `json:"running"` } var terms []termInfo @@ -787,8 +787,8 @@ func (p *Plugin) handleList() (interface{}, error) { } return map[string]interface{}{ - "status": "ok", - "count": len(terms), + "status": "ok", + "count": len(terms), "terminals": terms, }, nil } @@ -797,6 +797,9 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { defer p.wg.Done() defer close(t.done) + // reader 协程独享这个读缓冲：结果随 readResult 携带， + // readLoop 不再从其中做 copy（见 reader 注释，那是对共享缓冲 + // 的并发读写，-race 实测触发）。 buf := make([]byte, ReadBufSize) pollInterval := 200 * time.Millisecond @@ -816,7 +819,7 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { t.lastFeedback = now t.mu.Unlock() -// 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低 + // 硬上限：未读输出积累达到该值也通知一次（防大输出静默丢失），频率极低 hardNotifyBytes := 64 * 1024 hardNotifyInterval := 10 * time.Second // 输出停止判定：超过该时长无新数据则视为输出停止 @@ -886,9 +889,7 @@ func (p *Plugin) readLoop(t *TerminalSession, s *sdk.PluginSDK) { return } if r.n > 0 { - data := make([]byte, r.n) - copy(data, buf[:r.n]) - t.appendOutput(data) + t.appendOutput(r.data) // 缓冲阈值通知（仅当 agent 显式选择 buffer 模式，或未读积累达到硬上限）。 // 默认模式（仅 exit 提醒）下不随输出流通知，杜绝通知风暴。 @@ -951,15 +952,28 @@ func previewTail(s string, n int) string { } type readResult struct { - n int - err error + 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, err}: + case ch <- readResult{n, data, err}: case <-t.stopCh: return } diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go index bc8db01..6a0fb05 100644 --- a/internal/plugins/remotedevice/registry.go +++ b/internal/plugins/remotedevice/registry.go @@ -37,6 +37,24 @@ type DeviceMeta struct { type wconn struct { deviceID string w *bufio.Writer + // wmu 序列化对该连接 bufio.Writer 的所有写。 + // + // 两个角色会并发写同一连接：handleWS 主循环（读设备帧后的 hello_ack/ + // bind_ack/pong 回写）与 PushJSON/PushData（agent→设备的下发路径，可能 + // 来自任意 goroutine）。bufio.Writer 不是线程安全的，不加锁会在 + // WriteByte/Flush 上产生 data race（生产实测触发）。 + wmu sync.Mutex +} + +// lockWrite 对 wconn 加写锁并返回 writer；调用方必须 defer unlockWrite。 +// 单独写成方法而不是直接暴露字段，避免调用方绕过锁。 +func (c *wconn) lockWrite() *bufio.Writer { + c.wmu.Lock() + return c.w +} + +func (c *wconn) unlockWrite() { + c.wmu.Unlock() } // Registry 是设备接入网关的注册表：管理在线连接、设备元数据。线程安全。 @@ -319,7 +337,9 @@ func (r *Registry) PushJSON(deviceID string, payload map[string]interface{}) err if !ok { return fmt.Errorf("device %s not online", deviceID) } - return writeText(c.w, mustJSON(payload)) + w := c.lockWrite() + defer c.unlockWrite() + return writeText(w, mustJSON(payload)) } // PushCmd 向设备发送命令执行请求。 @@ -348,7 +368,11 @@ func (r *Registry) PushData(deviceID, reqID, kind, mime string, data []byte) err if !ok { return fmt.Errorf("device %s not online", deviceID) } - if err := writeText(c.w, mustJSON(map[string]interface{}{ + // 整条下发（start + N 个 chunk + end）持锁：设备侧按协议串行聚合， + // 若中途被 handleWS 的 hello/pong 插帧会破坏协议顺序。 + w := c.lockWrite() + defer c.unlockWrite() + if err := writeText(w, mustJSON(map[string]interface{}{ "op": "cmd_speech_start", "req_id": reqID, "kind": kind, @@ -363,11 +387,11 @@ func (r *Registry) PushData(deviceID, reqID, kind, mime string, data []byte) err if end > len(data) { end = len(data) } - if err := writeBinary(c.w, data[off:end]); err != nil { + if err := writeBinary(w, data[off:end]); err != nil { return fmt.Errorf("push data chunk: %w", err) } } - if err := writeText(c.w, mustJSON(map[string]interface{}{ + if err := writeText(w, mustJSON(map[string]interface{}{ "op": "cmd_speech_end", "req_id": reqID, })); err != nil { @@ -618,6 +642,26 @@ func (r *Registry) ServeWS(w http.ResponseWriter, req *http.Request) { go r.handleWS(conn, rw) } +// wsWriteLocked 在指定设备连接的写锁保护下执行写回调。 +// +// handleWS 主循环与 Push* 是两条并发写同一 bufio.Writer 的路径， +// 必须共用同一把锁。handleWS 里拿到的是 rw.Writer（与 conns 存储的是 +// 同一个对象），回写前必须经此函数取锁，否则跟 Push* 依然会撞。 +// +// 注意设备已离线（conns 中已删除）时直接报错——设备断开后仍尝试 +// 回写没有意义，还可能在已关闭的 bufio 上写入。 +func (r *Registry) wsWriteLocked(deviceID string, fn func(w *bufio.Writer) error) error { + r.mu.RLock() + c, ok := r.conns[deviceID] + r.mu.RUnlock() + if !ok { + return fmt.Errorf("device %s not online", deviceID) + } + w := c.lockWrite() + defer c.unlockWrite() + return fn(w) +} + func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { defer conn.Close() var curID string @@ -635,7 +679,9 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { payload, isClose, opcode, err := readFrame(rw.Reader) if err != nil { if err == errPing { - if werr := writePong(rw.Writer); werr != nil { + // pong 也走写锁：它可能在 Push* 持锁推送大块数据时到达。 + err := r.wsWriteLocked(curID, writePong) + if err != nil { return } continue @@ -679,11 +725,13 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { r.mu.Lock() r.conns[meta.DeviceID] = &wconn{deviceID: meta.DeviceID, w: rw.Writer} r.mu.Unlock() - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{ - "op": "hello_ack", - "device": meta.DeviceID, - "online": true, - })); err != nil { + if err := r.wsWriteLocked(meta.DeviceID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{ + "op": "hello_ack", + "device": meta.DeviceID, + "online": true, + })) + }); err != nil { return } case "bind": @@ -694,11 +742,17 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { // 默认不授权：bind 仅验证 token + 登记设备；授权完全由用户手动 // （GUI 设备页 / REST /api/v1/device/auth）控制，绝不自动授权。 } - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": true})); err != nil { + err := r.wsWriteLocked(curID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": true})) + }) + if err != nil { return } } else { - if err := writeText(rw.Writer, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": false, "error": "bad token"})); err != nil { + err := r.wsWriteLocked(curID, func(w *bufio.Writer) error { + return writeText(w, mustJSON(map[string]interface{}{"op": "bind_ack", "ok": false, "error": "bad token"})) + }) + if err != nil { return } }