refactor(terminal): 内核开终端/命令历史权威视图,WebUI 与 CLI 都改接内核

按「内核开,两个插件接」重构终端与命令历史的数据归属。

背景:此前 WebUI 与 CLI 各订 EventToolCall/EventTerminalOutput 攅一份状态,
同一件事两份推导,还各自踩过同一个坑——工具 result 是 Go 的 map 文本
(map[cols:80 ... id:term_2 ...]),断言成 map[string]interface{} 永远失败,
terminal_create 的 id 回填不生效,/terminals 因此恒空(WebUI 也一样)。
实测确认:WebUI 自己的 /api/v1/terminals 与 /api/v1/cmd/history 同样是空的。

内核开(权威唯一真相):
- internal/agent/core/terminal_registry.go:TerminalRegistry 归并两类事件——
  EventToolCall(terminal_create/close、cmd_run,id/command 从 args 或 Go map
  文本回填)与 EventTerminalOutput(agentcli 生命周期 + 输出,含 64KB 缓冲上限、
  100 条命令历史、50 个终端上限)。
- internal/sdk/terminal.go:新增 TerminalAPI(ListTerminals/CmdHistory)与
  TerminalStatus/CmdExecStatus DTO。**不塞进 KernelStatus**:那是全量快照,
  前端每 3 秒轮询 /kernel,背上每终端最多 64KB 输出会让轮询成本爆炸;
  终端输出是按需拉取的明细,另开接口。
- Agent 订阅自己的事件总线(subscribeTerminalRegistry),且**只根 agent 建**
  (驻留子共用同一总线,每个子都建会 N+1 份重复记账)。
- SDKConfig/Registry/bootstrap 接线:pluginReg.SetTerminalAPI(agent)。

生产者补全(agentcli):终端无输出时 ticker 不发事件,内核就无从知道终端
存在。新增 emitTermState,在 handleCreate/handleClose/readLoop 退出(超时/
进程结束/读取错误/stopCh)显式上报 running 状态,并给输出事件补 command 字段。
handleClose 改为接收 *sdk.PluginSDK 以便上报。

两个插件接(消费方):
- WebUI:删掉本地 termStates/cmdHistory/subscribeTerminalStream/handleToolEvent
  及不再使用的 getStr;/terminals 与 /cmd/history 直接读 s.Terminal()。
- CLI:删掉上一轮刚加的 subscribeToolEvents 与 cliTermState/cliCmdExec;
  /terminals 与 /cmd/history 直接读 s.Terminal()。两条路(local/remote)都通。

测试:新增 terminal_registry_test.go,锁死 Go map 文本解析(旧缺陷根因)、
生命周期、CLI 直调路径(无 EventToolCall 仅凭 output 事件建条目)、历史与
终端数量上限。全量 go test ./internal/... ./cmd/... 通过。
This commit is contained in:
JianFeeeee
2026-09-17 18:56:15 +08:00
parent 166c0a4eaf
commit dbc493a545
14 changed files with 595 additions and 330 deletions

View File

@ -0,0 +1,107 @@
package core
import "testing"
// 这条用例锁死的是曾经的线上缺陷根因:EventToolCall 的 result 是 Go 的
// map 文本(map[cols:80 command:sleep 120 id:term_2 ...]),不是
// map[string]interface{}。旧代码断言成后者永远失败 → /terminals 恒空。
func TestMapFieldFromMapText(t *testing.T) {
res := "map[cols:80 command:sleep 120 id:term_2 notify_mode:exit rows:24 status:created timeout:5m0s]"
if got := mapFieldFromMapText(res, "id"); got != "term_2" {
t.Fatalf("id = %q, want term_2", got)
}
// command 含空格:按空格切字段会把它切断,这里只要求拿到首段(与事件
// 负载同源,command 的真实值另有 args 路径可拿,不靠 map 文本)。
if got := mapFieldFromMapText(res, "cols"); got != "80" {
t.Fatalf("cols = %q, want 80", got)
}
if got := mapFieldFromMapText(res, "notify_mode"); got != "exit" {
t.Fatalf("notify_mode = %q, want exit", got)
}
if got := mapFieldFromMapText(res, "nope"); got != "" {
t.Fatalf("missing key = %q, want empty", got)
}
if got := terminalIDFromMapText("not a map at all"); got != "" {
t.Fatalf("garbage = %q, want empty", got)
}
}
func TestTerminalRegistryLifecycle(t *testing.T) {
r := NewTerminalRegistry()
// agent 路径:terminal_create 的 result 是 Go map 文本,要能回填 id。
r.OnToolCall(map[string]interface{}{
"tool": "terminal_create",
"args": map[string]interface{}{"command": "sleep 120"},
"result": "map[cols:80 command:sleep 120 id:term_7 status:created]",
"status": "ok",
})
terms, _ := r.Snapshot()
if len(terms) != 1 || terms[0].ID != "term_7" || !terms[0].Running {
t.Fatalf("after create: %+v", terms)
}
// agentcli 输出事件:追加 output。
r.OnTerminalOutput(map[string]interface{}{
"terminal_id": "term_7", "output": "hello", "running": true,
})
terms, _ = r.Snapshot()
if terms[0].Output != "hello" {
t.Fatalf("output = %q", terms[0].Output)
}
// 生命周期事件:退出置 running=false。
r.OnTerminalOutput(map[string]interface{}{
"terminal_id": "term_7", "running": false,
})
terms, _ = r.Snapshot()
if terms[0].Running {
t.Fatalf("should be stopped: %+v", terms)
}
// CLI 直调路径:没有 EventToolCall,只有 terminal_output 生命周期事件,
// 依然要能凭 command 字段建出条目(设备名不丢)。
r.OnTerminalOutput(map[string]interface{}{
"terminal_id": "term_9", "command": "top", "running": true,
})
terms, _ = r.Snapshot()
var found bool
for _, tm := range terms {
if tm.ID == "term_9" && tm.Command == "top" && tm.Running {
found = true
}
}
if !found {
t.Fatalf("direct-path terminal missing: %+v", terms)
}
// cmd_run 历史:只保留最近 maxCmdHistory 条。
for i := 0; i < maxCmdHistory+10; i++ {
r.OnToolCall(map[string]interface{}{
"tool": "cmd_run",
"args": map[string]interface{}{"command": "echo hi"},
"status": "ok",
})
}
_, cmds := r.Snapshot()
if len(cmds) != maxCmdHistory {
t.Fatalf("cmd history len = %d, want %d", len(cmds), maxCmdHistory)
}
if cmds[0].Command != "echo hi" || cmds[0].Status != "ok" {
t.Fatalf("cmd exec = %+v", cmds[0])
}
}
// 终端数超上限时要淘汰,不能无界增长。
func TestTerminalRegistryCap(t *testing.T) {
r := NewTerminalRegistry()
for i := 0; i < maxTerminals+20; i++ {
r.OnTerminalOutput(map[string]interface{}{
"terminal_id": string(rune('a'+i%26)) + "-x", "running": true,
})
}
terms, _ := r.Snapshot()
if len(terms) > maxTerminals {
t.Fatalf("terminals = %d, want <= %d", len(terms), maxTerminals)
}
}