This commit is contained in:
2026-06-18 22:56:41 +08:00
parent 3d831650b7
commit db3d030169
15 changed files with 1123 additions and 0 deletions
+1
View File
@@ -6,6 +6,7 @@ require (
github.com/alexedwards/argon2id v1.0.0
github.com/anthropics/anthropic-sdk-go v1.38.0
github.com/coder/websocket v1.8.14
github.com/creack/pty v1.1.24
github.com/felixge/httpsnoop v1.0.4
github.com/go-chi/chi/v5 v5.2.5
github.com/golang-migrate/migrate/v4 v4.19.1
+27
View File
@@ -47,6 +47,7 @@ import (
mcp "github.com/yan1h/agent-coding-workflow/internal/mcp"
"github.com/yan1h/agent-coding-workflow/internal/project"
runmod "github.com/yan1h/agent-coding-workflow/internal/run"
"github.com/yan1h/agent-coding-workflow/internal/terminal"
httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http"
"github.com/yan1h/agent-coding-workflow/internal/user"
"github.com/yan1h/agent-coding-workflow/internal/workspace"
@@ -63,6 +64,7 @@ type App struct {
jobsCancel context.CancelFunc
acpSup *acp.Supervisor // ACP supervisor — ShutdownAll on Run shutdown
runSup *runmod.RunSupervisor // run profiles supervisor — ShutdownAll on Run shutdown
termSup *terminal.Supervisor // admin terminal supervisor — CloseAll on shutdown
}
// New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误,不留半成品资源。
@@ -406,6 +408,23 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
}).Mount(r)
}
// ===== admin terminal module wiring =====
// 仅限管理员的网页端交互式终端:在宿主进程(部署时即 runtime 容器)内开 shell,
// 主要用于运维(容器内 npm i -g 安装 ACP 客户端 CLI)。会话纯内存、不落库。
// WS 心跳沿用 acp 全局默认值(同类参数,避免新增配置面)。
termSup := terminal.NewSupervisor(terminal.SupervisorConfig{
Shell: cfg.Terminal.Shell,
MaxSessions: cfg.Terminal.MaxSessions,
KillGrace: cfg.Terminal.KillGrace,
EnvWhitelist: cfg.Terminal.EnvWhitelist,
}, log)
if cfg.Terminal.Enabled {
terminal.NewHandler(termSup, userSvc, userAdminAdapter{svc: userSvc}, auditRec, cfg.Terminal.Shell, terminal.WSConfig{
PingInterval: cfg.Acp.WSPingInterval,
PongTimeout: cfg.Acp.WSPongTimeout,
}).Mount(r)
}
// ===== MCP 模块装配(第二段:Server 本体 + transports + admin CRUD)=====
mcpDeps := mcp.ServerDeps{
Caller: mcp.NewCallerResolver(userSvc),
@@ -612,6 +631,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) {
jobsCancel: jobsCancel,
acpSup: acpSup,
runSup: runSup,
termSup: termSup,
}, nil
}
@@ -645,6 +665,10 @@ func (a *App) Run(ctx context.Context) error {
if a.runSup != nil {
a.runSup.ShutdownAll(shutCtx)
}
// Close admin terminal sessions (整树 kill shell + 子孙) on shutdown.
if a.termSup != nil {
a.termSup.CloseAll(shutCtx)
}
// Stop jobs module first so no in-flight worker can dispatch new
// notifications after we begin to drain the dispatcher.
if a.jobs != nil && a.cfg.Jobs.Enabled {
@@ -673,6 +697,9 @@ func (a *App) Run(ctx context.Context) error {
if a.runSup != nil {
a.runSup.ShutdownAll(drainCtx)
}
if a.termSup != nil {
a.termSup.CloseAll(drainCtx)
}
if a.jobs != nil && a.cfg.Jobs.Enabled {
a.jobsCancel()
stopCtx, stopCancel := context.WithTimeout(context.Background(), a.cfg.Jobs.ShutdownGrace+5*time.Second)
+25
View File
@@ -44,6 +44,7 @@ type Config struct {
Acp AcpConfig `mapstructure:"acp"`
MCP MCPConfig `mapstructure:"mcp"`
Run RunConfig `mapstructure:"run"`
Terminal TerminalConfig `mapstructure:"terminal"`
}
// HTTPConfig 描述 HTTP 服务监听参数。
@@ -215,6 +216,18 @@ type RunConfig struct {
EnvWhitelist []string `mapstructure:"env_whitelist"`
}
// TerminalConfig 控制仅限管理员的网页端终端模块:开关、启动的 shell、并发会话上限、
// 整组终止宽限,以及透传给 shell 的宿主环境变量白名单(沿用 RunConfig 的安全取舍)。
type TerminalConfig struct {
Enabled bool `mapstructure:"enabled"`
Shell string `mapstructure:"shell"`
MaxSessions int `mapstructure:"max_sessions"`
KillGrace time.Duration `mapstructure:"kill_grace"`
// EnvWhitelist 是允许透传给交互式 shell 的环境变量名白名单。
// APP_MASTER_KEY、DB DSN、git 凭据等敏感变量绝不应出现在此名单中。
EnvWhitelist []string `mapstructure:"env_whitelist"`
}
// MCPConfig 控制 MCP 模块的开关、对外 URL(注入 ACP 子进程 env)、system token TTL 与速率限制。
type MCPConfig struct {
Enabled bool `mapstructure:"enabled"`
@@ -311,6 +324,18 @@ func Load(configFile string) (*Config, error) {
"SystemRoot", "TEMP", "TMP", "USERPROFILE", "APPDATA", "LOCALAPPDATA",
"PATHEXT", "ComSpec", "NUMBER_OF_PROCESSORS",
})
v.SetDefault("terminal.enabled", true)
v.SetDefault("terminal.shell", "/bin/bash") // Windows 开发环境经 APP_TERMINAL_SHELL 覆盖
v.SetDefault("terminal.max_sessions", 5)
v.SetDefault("terminal.kill_grace", "5s")
// 透传 PATH/HOME/TERM + npm 全局安装所需(NPM_CONFIG_PREFIX 指向持久卷),
// 以便容器内 `npm i -g <acp-cli>` 可用。敏感变量不在名单内。
v.SetDefault("terminal.env_whitelist", []string{
"PATH", "HOME", "LANG", "LC_ALL", "TZ", "USER", "SHELL", "TERM",
"NPM_CONFIG_PREFIX", "NODE_PATH", "NODE_ENV",
"SystemRoot", "TEMP", "TMP", "USERPROFILE", "APPDATA", "LOCALAPPDATA",
"PATHEXT", "ComSpec", "NUMBER_OF_PROCESSORS",
})
if configFile != "" {
v.SetConfigFile(configFile)
+220
View File
@@ -0,0 +1,220 @@
package terminal
import (
"context"
"encoding/json"
"net/http"
"strconv"
"time"
"github.com/go-chi/chi/v5"
"github.com/google/uuid"
ws "github.com/coder/websocket"
"github.com/yan1h/agent-coding-workflow/internal/audit"
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http"
"github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware"
)
// readBufSize 是从 shell 读输出的缓冲块大小。
const readBufSize = 32 * 1024
// wsReadLimit 放宽单条消息上限以容纳较大的粘贴输入(默认 32KB 对粘贴偏小)。
const wsReadLimit = 1 << 20
// AdminLookup 解析用户是否为管理员(与 run/acp 模块同一模式)。
type AdminLookup interface {
IsAdmin(ctx context.Context, userID uuid.UUID) (bool, error)
}
// WSConfig 控制 WebSocket 心跳。
type WSConfig struct {
PingInterval time.Duration
PongTimeout time.Duration
}
// Handler 暴露仅限管理员的终端 WS 端点。
type Handler struct {
sup *Supervisor
resolver middleware.SessionResolver
adminLookup AdminLookup
audit audit.Recorder
shell string
cfg WSConfig
}
// NewHandler 构造 terminal.Handler。
func NewHandler(sup *Supervisor, resolver middleware.SessionResolver, al AdminLookup, rec audit.Recorder, shell string, cfg WSConfig) *Handler {
return &Handler{sup: sup, resolver: resolver, adminLookup: al, audit: rec, shell: shell, cfg: cfg}
}
// Mount 注册终端路由(仅 admin)。WS 鉴权经 middleware.Auth(浏览器原生 WebSocket
// 无法带自定义 header,token 走 ?token= 查询参数,沿用 run/acp 的 WS 约定)。
func (h *Handler) Mount(r chi.Router) {
r.Route("/api/v1/admin/terminal", func(r chi.Router) {
r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{}))
r.Get("/stream", h.stream)
})
}
// requireAdmin 从请求解析用户并要求其为管理员,否则返回 forbidden。
func (h *Handler) requireAdmin(r *http.Request) (uuid.UUID, error) {
uid, ok := middleware.UserIDFromContext(r.Context())
if !ok {
return uuid.Nil, errs.New(errs.CodeUnauthorized, "not authenticated")
}
admin, err := h.adminLookup.IsAdmin(r.Context(), uid)
if err != nil {
return uuid.Nil, err
}
if !admin {
return uuid.Nil, errs.New(errs.CodeForbidden, "admin only")
}
return uid, nil
}
// resizeMsg 是客户端经文本帧上报的窗口尺寸控制消息。
type resizeMsg struct {
Type string `json:"type"`
Rows uint16 `json:"rows"`
Cols uint16 `json:"cols"`
}
// stream 升级为 WebSocket,开一个 shell 会话并双向桥接:
// - server→client:shell 输出作为二进制帧推送
// - client→server:二进制帧 = stdin 原始字节;文本帧 = JSON 控制(resize)
//
// 二进制/文本帧天然区分 stdin 与控制消息,无需额外前缀字节。
func (h *Handler) stream(w http.ResponseWriter, r *http.Request) {
uid, err := h.requireAdmin(r)
if err != nil {
writeErr(w, r, err)
return
}
rows := uint16(parseUint(r.URL.Query().Get("rows"), uint64(defaultRows)))
cols := uint16(parseUint(r.URL.Query().Get("cols"), uint64(defaultCols)))
conn, aerr := ws.Accept(w, r, &ws.AcceptOptions{})
if aerr != nil {
return
}
defer conn.Close(ws.StatusInternalError, "internal")
conn.SetReadLimit(wsReadLimit)
sess, oerr := h.sup.Open(OpenParams{Rows: rows, Cols: cols})
if oerr != nil {
conn.Close(ws.StatusTryAgainLater, "cannot open terminal session")
return
}
h.record(r.Context(), r.RemoteAddr, uid, "admin.terminal.opened", sess.ID)
defer func() {
// 连接已断时 r.Context() 多半已取消,审计与终止改用独立 context。
closeCtx, cancel := context.WithTimeout(context.Background(), h.cfg.PongTimeout)
defer cancel()
h.sup.Close(closeCtx, sess.ID)
h.record(closeCtx, r.RemoteAddr, uid, "admin.terminal.closed", sess.ID)
}()
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
// shell 进程自行退出(如用户输入 exit)→ 取消,触发主循环退出。
go func() {
select {
case <-ctx.Done():
case <-sess.Done():
cancel()
}
}()
// read pump:shell 输出 → WS 二进制帧。
go func() {
buf := make([]byte, readBufSize)
for {
n, rerr := sess.Read(buf)
if n > 0 {
if werr := conn.Write(ctx, ws.MessageBinary, buf[:n]); werr != nil {
cancel()
return
}
}
if rerr != nil {
cancel()
return
}
}
}()
// 心跳。
pingTicker := time.NewTicker(h.cfg.PingInterval)
defer pingTicker.Stop()
go func() {
for {
select {
case <-ctx.Done():
return
case <-pingTicker.C:
pingCtx, pcancel := context.WithTimeout(ctx, h.cfg.PongTimeout)
perr := conn.Ping(pingCtx)
pcancel()
if perr != nil {
cancel()
return
}
}
}
}()
// write pump(主循环):WS → shell stdin / resize。
for {
typ, data, rerr := conn.Read(ctx)
if rerr != nil {
return
}
switch typ {
case ws.MessageBinary:
if _, werr := sess.Write(data); werr != nil {
return
}
case ws.MessageText:
var m resizeMsg
if json.Unmarshal(data, &m) == nil && m.Type == "resize" {
_ = sess.Resize(m.Rows, m.Cols)
}
}
}
}
// record 写一条审计日志,吞错(审计失败不应影响终端会话)。
func (h *Handler) record(ctx context.Context, remoteAddr string, uid uuid.UUID, action string, sessID uuid.UUID) {
if h.audit == nil {
return
}
u := uid
_ = h.audit.Record(ctx, audit.Entry{
UserID: &u,
Action: action,
TargetType: "terminal_session",
TargetID: sessID.String(),
IP: remoteAddr,
Metadata: map[string]any{"shell": h.shell},
})
}
func parseUint(s string, def uint64) uint64 {
if s == "" {
return def
}
n, err := strconv.ParseUint(s, 10, 16)
if err != nil {
return def
}
return n
}
func writeErr(w http.ResponseWriter, r *http.Request, err error) {
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
}
+73
View File
@@ -0,0 +1,73 @@
package terminal
import (
"io"
"os"
"os/exec"
"strings"
procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc"
)
// 终端窗口的兜底尺寸(前端尚未上报 resize 前使用)。
const (
defaultRows uint16 = 24
defaultCols uint16 = 80
)
// ptySession 持有平台相关的 shell 进程与 IO。resizeFn 由 startPTY 注入:
// Unix 调 pty.Setsize,Windows 为 nil(降级实现无 resize)。这样 ptySession
// 类型与生命周期方法可保持平台无关,仅 startPTY 按 build tag 特化。
type ptySession struct {
cmd *exec.Cmd
group procgrp.Group
rw io.ReadWriteCloser // PTY master(Unix)或 stdio 桥接(Windows)
resizeFn func(rows, cols uint16) error
}
func (p *ptySession) resize(rows, cols uint16) error {
if p.resizeFn == nil {
return nil
}
return p.resizeFn(rows, cols)
}
// terminate 是 Open 失败路径的尽力清理(kill 进程、释放句柄、关闭 IO)。
// 正常关闭走 Supervisor.Close → group.TerminateGroup 的整树终止。
func (p *ptySession) terminate() {
if p.cmd.Process != nil {
_ = p.cmd.Process.Kill()
}
p.group.Close()
_ = p.rw.Close()
}
// buildEnv 从宿主环境按白名单过滤出传给 shell 的环境变量,并补上 TERM 以便
// 终端程序正确渲染。白名单确保 APP_MASTER_KEY、DB DSN、git 凭据等敏感变量
// 不会透传给交互式 shell。
func buildEnv(whitelist []string) []string {
allow := make(map[string]bool, len(whitelist))
for _, k := range whitelist {
allow[k] = true
}
out := make([]string, 0, len(whitelist)+1)
hasTerm := false
for _, kv := range os.Environ() {
i := strings.IndexByte(kv, '=')
if i < 0 {
continue
}
name := kv[:i]
if !allow[name] {
continue
}
if name == "TERM" {
hasTerm = true
}
out = append(out, kv)
}
if !hasTerm {
out = append(out, "TERM=xterm-256color")
}
return out
}
+46
View File
@@ -0,0 +1,46 @@
//go:build !windows
package terminal
import (
"os/exec"
"github.com/creack/pty"
procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc"
)
// startPTY 在真 PTY 中拉起 shell(Unix)。
//
// 关于进程组:pty.StartWithSize 内部会设置 SysProcAttr.Setsid,使 shell 成为新
// 会话的首进程,其进程组 pgid == pid。因此这里**不**调用 group.Prepare(它设的
// Setpgid 与 Setsid 同时设置在 Go exec 下是非法/冗余的);group.Adopt 直接记录
// pgid = cmd.Process.Pid,TerminateGroup 向 -pgid 发信号即可整树终止 shell 及其
// 派生的全部子孙进程。
func startPTY(shell string, env []string, rows, cols uint16) (*ptySession, error) {
cmd := exec.Command(shell)
cmd.Env = env
ptmx, err := pty.StartWithSize(cmd, &pty.Winsize{Rows: rows, Cols: cols})
if err != nil {
return nil, err
}
group := procgrp.NewGroup()
if err := group.Adopt(cmd); err != nil {
_ = ptmx.Close()
if cmd.Process != nil {
_ = cmd.Process.Kill()
}
return nil, err
}
return &ptySession{
cmd: cmd,
group: group,
rw: ptmx,
resizeFn: func(r, c uint16) error {
return pty.Setsize(ptmx, &pty.Winsize{Rows: r, Cols: c})
},
}, nil
}
+75
View File
@@ -0,0 +1,75 @@
//go:build windows
package terminal
import (
"io"
"os"
"os/exec"
procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc"
)
// stdioBridge 把 shell 的 stdin(写端)与合并后的 stdout/stderr(读端)合成一个
// io.ReadWriteCloser,供降级实现使用。
type stdioBridge struct {
stdin io.WriteCloser
stdout io.ReadCloser
}
func (b *stdioBridge) Read(p []byte) (int, error) { return b.stdout.Read(p) }
func (b *stdioBridge) Write(p []byte) (int, error) { return b.stdin.Write(p) }
func (b *stdioBridge) Close() error {
_ = b.stdin.Close()
return b.stdout.Close()
}
// startPTY 是 Windows 降级实现:用普通管道桥接 stdin/stdout,无 PTY、无 resize,
// 仅供本地开发调试,不追求生产可用(无行编辑/光标控制/作业控制)。
//
// stdout/stderr 合并:手动建 os.Pipe,把写端同时交给子进程的 Stdout/Stderr,父进程
// Start 后立即关闭自己持有的写端;子进程退出后读端 pr 收到 EOF,read pump 据此感知结束。
func startPTY(shell string, env []string, rows, cols uint16) (*ptySession, error) {
cmd := exec.Command(shell)
cmd.Env = env
pr, pw, err := os.Pipe()
if err != nil {
return nil, err
}
cmd.Stdout = pw
cmd.Stderr = pw
stdin, err := cmd.StdinPipe()
if err != nil {
_ = pr.Close()
_ = pw.Close()
return nil, err
}
group := procgrp.NewGroup()
group.Prepare(cmd) // Windows: no-op
if err := cmd.Start(); err != nil {
_ = pr.Close()
_ = pw.Close()
_ = stdin.Close()
return nil, err
}
_ = pw.Close() // 父进程放掉写端;子进程仍持有,退出后 pr 读到 EOF
if err := group.Adopt(cmd); err != nil {
if cmd.Process != nil {
_ = cmd.Process.Kill()
}
_ = pr.Close()
_ = stdin.Close()
return nil, err
}
return &ptySession{
cmd: cmd,
group: group,
rw: &stdioBridge{stdin: stdin, stdout: pr},
resizeFn: nil, // 降级实现不支持 resize
}, nil
}
+185
View File
@@ -0,0 +1,185 @@
// Package terminal 提供一个仅限管理员的网页端交互式终端:在 server 宿主进程
// (部署时即 runtime 容器)内拉起一个 shell,经 WebSocket 双向桥接 stdin/stdout,
// 主要用途是运维(例如容器内 npm i -g 安装 ACP 客户端 CLI)。
//
// 会话纯内存、不落库:开即用、断即清,server 重启即全部销毁(沿用 run profiles
// 「运行态不持久化」的取舍)。进程树终止复用 internal/infra/proc,断连时整树清理,
// 避免 shell 派生的子孙进程(npm→node…)残留为孤儿。
//
// 跨平台:Unix 走真 PTY(github.com/creack/pty);Windows 走降级的 stdio 管道
// 实现(无 PTY、无 resize),仅供本地开发调试,不追求生产可用。
package terminal
import (
"context"
"log/slog"
"sync"
"time"
"github.com/google/uuid"
)
// SupervisorConfig 是从 config.TerminalConfig 派生的 supervisor 子集。
type SupervisorConfig struct {
Shell string // 启动的 shell,如 /bin/bash
MaxSessions int // 同时存在的会话上限
KillGrace time.Duration // 终止整组时 SIGTERM→SIGKILL 的宽限
// EnvWhitelist 是允许从宿主进程透传给 shell 的环境变量名白名单。
// APP_MASTER_KEY、DB DSN、git 凭据等敏感变量绝不应出现在此名单中。
EnvWhitelist []string
}
// ErrTooManySessions 在达到 MaxSessions 上限时返回,由 Handler 翻译为 errs code。
var ErrTooManySessions = errSentinel("terminal: too many active sessions")
type errSentinel string
func (e errSentinel) Error() string { return string(e) }
// Session 是单个终端会话的运行时句柄。pty 持有平台相关的 shell 进程与 IO,
// done 在进程退出(monitor)时关闭。
type Session struct {
ID uuid.UUID
pty *ptySession
done chan struct{} // monitor 在 shell 退出后 close
}
// Read 从 shell 读输出(PTY master / 管道读端)。
func (s *Session) Read(p []byte) (int, error) { return s.pty.rw.Read(p) }
// Write 向 shell 写输入(PTY master / stdin 管道)。
func (s *Session) Write(p []byte) (int, error) { return s.pty.rw.Write(p) }
// Resize 调整终端窗口尺寸(Windows 降级实现为 no-op)。
func (s *Session) Resize(rows, cols uint16) error { return s.pty.resize(rows, cols) }
// Done 在 shell 进程退出后被关闭,供调用方感知进程自行结束(如用户输入 exit)。
func (s *Session) Done() <-chan struct{} { return s.done }
// Supervisor 是终端会话总管。被动结构:Open 时起 monitor goroutine,无主循环。
// 并发骨架照搬 run.RunSupervisor 已被测试验证的不变量(sessions map 写锁仅在
// Open 插入 / monitor 删除时持有;done 严格 close-once,monitor 唯一持有)。
type Supervisor struct {
mu sync.RWMutex
sessions map[uuid.UUID]*Session
cfg SupervisorConfig
log *slog.Logger
}
// NewSupervisor 构造空 Supervisor。
func NewSupervisor(cfg SupervisorConfig, log *slog.Logger) *Supervisor {
if log == nil {
log = slog.Default()
}
return &Supervisor{
sessions: map[uuid.UUID]*Session{},
cfg: cfg,
log: log,
}
}
// OpenParams 描述一次终端会话的初始参数。
type OpenParams struct {
Rows uint16
Cols uint16
}
// Open 启动一个 shell 会话并托管。达到 MaxSessions 上限时返回 ErrTooManySessions。
func (s *Supervisor) Open(p OpenParams) (*Session, error) {
s.mu.RLock()
full := len(s.sessions) >= s.cfg.MaxSessions
s.mu.RUnlock()
if full {
return nil, ErrTooManySessions
}
rows, cols := p.Rows, p.Cols
if rows == 0 {
rows = defaultRows
}
if cols == 0 {
cols = defaultCols
}
ps, err := startPTY(s.cfg.Shell, buildEnv(s.cfg.EnvWhitelist), rows, cols)
if err != nil {
return nil, err
}
id, err := uuid.NewRandom()
if err != nil {
ps.terminate()
return nil, err
}
sess := &Session{ID: id, pty: ps, done: make(chan struct{})}
s.mu.Lock()
s.sessions[id] = sess
s.mu.Unlock()
go s.monitor(sess)
return sess, nil
}
// monitor 阻塞等待 shell 退出后回收:关闭 done、移除映射、释放平台句柄。
func (s *Supervisor) monitor(sess *Session) {
_ = sess.pty.cmd.Wait() // reap,避免僵尸进程
close(sess.done)
s.mu.Lock()
delete(s.sessions, sess.ID)
s.mu.Unlock()
sess.pty.group.Close()
}
// Close 终止指定会话(整树)。幂等:会话已不存在时直接返回。阻塞直到 done 关闭
// 或宽限耗尽。
func (s *Supervisor) Close(ctx context.Context, id uuid.UUID) {
s.mu.RLock()
sess, ok := s.sessions[id]
s.mu.RUnlock()
if !ok {
return
}
sess.pty.group.TerminateGroup(sess.done, s.cfg.KillGrace)
select {
case <-sess.done:
case <-ctx.Done():
}
}
// Count 返回当前活跃会话数。
func (s *Supervisor) Count() int {
s.mu.RLock()
defer s.mu.RUnlock()
return len(s.sessions)
}
// CloseAll 在服务关停时终止全部会话。
func (s *Supervisor) CloseAll(ctx context.Context) {
s.mu.RLock()
ids := make([]uuid.UUID, 0, len(s.sessions))
for id := range s.sessions {
ids = append(ids, id)
}
s.mu.RUnlock()
var wg sync.WaitGroup
for _, id := range ids {
wg.Add(1)
go func(sid uuid.UUID) {
defer wg.Done()
s.Close(ctx, sid)
}(id)
}
done := make(chan struct{})
go func() {
wg.Wait()
close(done)
}()
select {
case <-done:
case <-ctx.Done():
}
}
+178
View File
@@ -0,0 +1,178 @@
package terminal
import (
"context"
"errors"
"runtime"
"strings"
"testing"
"time"
)
// testShell 返回各平台一个可交互读 stdin 的 shell。
func testShell() string {
if runtime.GOOS == "windows" {
return "cmd"
}
return "/bin/sh"
}
func newTestSup(t *testing.T, max int) *Supervisor {
t.Helper()
sup := NewSupervisor(SupervisorConfig{
Shell: testShell(),
MaxSessions: max,
KillGrace: 2 * time.Second,
EnvWhitelist: []string{
"PATH", "HOME", "TERM",
"SystemRoot", "TEMP", "TMP", "USERPROFILE", "ComSpec", "PATHEXT",
},
}, nil)
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
sup.CloseAll(ctx)
})
return sup
}
// readUntil 在独立 goroutine 中持续读会话输出,直到累计内容包含 marker 或超时。
// 阻塞的 Read goroutine 会在会话被关闭(defer/Cleanup)后随 Read 报错而退出。
func readUntil(sess *Session, marker string, d time.Duration) string {
out := make(chan []byte, 16)
go func() {
buf := make([]byte, 4096)
for {
n, err := sess.Read(buf)
if n > 0 {
b := make([]byte, n)
copy(b, buf[:n])
out <- b
}
if err != nil {
close(out)
return
}
}
}()
deadline := time.After(d)
var acc strings.Builder
for {
select {
case b, ok := <-out:
if !ok {
return acc.String()
}
acc.Write(b)
if strings.Contains(acc.String(), marker) {
return acc.String()
}
case <-deadline:
return acc.String()
}
}
}
// TestOpen_ReadsShellOutput 验证开会话后向 shell 写命令能读回其输出。
// Unix 走真 PTY;Windows 走降级管道实现。
func TestOpen_ReadsShellOutput(t *testing.T) {
sup := newTestSup(t, 5)
sess, err := sup.Open(OpenParams{})
if err != nil {
t.Fatalf("open: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
defer sup.Close(ctx, sess.ID)
const marker = "ACWMARK42"
cmd := "echo " + marker + "\n"
if runtime.GOOS == "windows" {
cmd = "echo " + marker + "\r\n"
}
if _, err := sess.Write([]byte(cmd)); err != nil {
t.Fatalf("write: %v", err)
}
got := readUntil(sess, marker, 5*time.Second)
if !strings.Contains(got, marker) {
t.Fatalf("marker %q not found in shell output; got %q", marker, got)
}
}
// TestOpen_MaxSessions 验证达到上限后再开会话返回 ErrTooManySessions。
func TestOpen_MaxSessions(t *testing.T) {
sup := newTestSup(t, 1)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
s1, err := sup.Open(OpenParams{})
if err != nil {
t.Fatalf("first open: %v", err)
}
defer sup.Close(ctx, s1.ID)
if _, err := sup.Open(OpenParams{}); !errors.Is(err, ErrTooManySessions) {
t.Fatalf("expected ErrTooManySessions, got %v", err)
}
if got := sup.Count(); got != 1 {
t.Fatalf("expected 1 active session, got %d", got)
}
}
// TestClose_TerminatesSession 验证 Close 终止 shell(done 关闭)并从映射移除。
func TestClose_TerminatesSession(t *testing.T) {
sup := newTestSup(t, 5)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
sess, err := sup.Open(OpenParams{})
if err != nil {
t.Fatalf("open: %v", err)
}
if got := sup.Count(); got != 1 {
t.Fatalf("expected 1 active session, got %d", got)
}
sup.Close(ctx, sess.ID)
select {
case <-sess.Done():
case <-time.After(8 * time.Second):
t.Fatal("session not terminated within 8s after Close")
}
// monitor 在 done 关闭后异步移除映射,轮询直至归零。
deadline := time.Now().Add(3 * time.Second)
for sup.Count() != 0 {
if time.Now().After(deadline) {
t.Fatalf("session not removed from supervisor; count=%d", sup.Count())
}
time.Sleep(20 * time.Millisecond)
}
}
// TestCloseAll_ClosesEverySession 验证 CloseAll 终止全部会话。
func TestCloseAll_ClosesEverySession(t *testing.T) {
sup := newTestSup(t, 5)
for i := 0; i < 3; i++ {
if _, err := sup.Open(OpenParams{}); err != nil {
t.Fatalf("open %d: %v", i, err)
}
}
if got := sup.Count(); got != 3 {
t.Fatalf("expected 3 active sessions, got %d", got)
}
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
defer cancel()
sup.CloseAll(ctx)
deadline := time.Now().Add(3 * time.Second)
for sup.Count() != 0 {
if time.Now().After(deadline) {
t.Fatalf("sessions not all closed; count=%d", sup.Count())
}
time.Sleep(20 * time.Millisecond)
}
}
+2
View File
@@ -21,6 +21,8 @@
"packageManager": "pnpm@10.32.1",
"dependencies": {
"@vicons/ionicons5": "^0.13.0",
"@xterm/addon-fit": "^0.11.0",
"@xterm/xterm": "^6.0.0",
"markdown-it": "^14.1.1",
"naive-ui": "^2.44.1",
"pinia": "^3.0.4",
+16
View File
@@ -11,6 +11,12 @@ importers:
'@vicons/ionicons5':
specifier: ^0.13.0
version: 0.13.0
'@xterm/addon-fit':
specifier: ^0.11.0
version: 0.11.0
'@xterm/xterm':
specifier: ^6.0.0
version: 6.0.0
markdown-it:
specifier: ^14.1.1
version: 14.1.1
@@ -644,6 +650,12 @@ packages:
'@vue/server-renderer':
optional: true
'@xterm/addon-fit@0.11.0':
resolution: {integrity: sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==}
'@xterm/xterm@6.0.0':
resolution: {integrity: sha512-TQwDdQGtwwDt+2cgKDLn0IRaSxYu1tSUjgKarSDkUM0ZNiSRXFpjxEsvc/Zgc5kq5omJ+V0a8/kIM2WD3sMOYg==}
abbrev@2.0.0:
resolution: {integrity: sha512-6/mh1E2u2YgEsCHdY0Yx5oW+61gZU+1vXaoiHHrpKeuRNNgFvS+/jrwHiQhB5apAf5oB7UB7E19ol2R2LKH8hQ==}
engines: {node: ^14.17.0 || ^16.13.0 || >=18.0.0}
@@ -2290,6 +2302,10 @@ snapshots:
optionalDependencies:
'@vue/server-renderer': 3.5.33(vue@3.5.33(typescript@6.0.3))
'@xterm/addon-fit@0.11.0': {}
'@xterm/xterm@6.0.0': {}
abbrev@2.0.0: {}
acorn-jsx@5.3.2(acorn@8.16.0):
+3
View File
@@ -32,6 +32,9 @@
<RouterLink :to="{ name: 'admin-acp-agent-kinds' }" class="text-sm hover:underline">
ACP 客户端
</RouterLink>
<RouterLink :to="{ name: 'admin-terminal' }" class="text-sm hover:underline">
终端
</RouterLink>
</template>
</nav>
<span v-else class="flex-1" />
+125
View File
@@ -0,0 +1,125 @@
// useTerminalSocket 封装管理员终端的双向 WebSocket,结构照搬 useRunStream:
// - connect + 自动重连(指数退避 1s/3s,最多 2 次)
// - server→client:二进制帧 = shell 输出,回调喂给 xterm
// - client→server:二进制帧 = stdin 原始字节;文本帧 JSON = resize 控制
// 浏览器原生 WS 不支持自定义 header,token 经 ?token= 传递(同 run/acp WS 约定)。
import { ref, onScopeDispose, type Ref } from 'vue'
import { useAuthStore } from '@/stores/auth'
export type TerminalStatus = 'idle' | 'connecting' | 'open' | 'closed' | 'error'
export interface UseTerminalSocketOptions {
// onData 接收 shell 输出(已解码为字符串)。
onData: (text: string) => void
// 首次连接的初始窗口尺寸。
initialRows?: number
initialCols?: number
}
export interface UseTerminalSocketReturn {
status: Ref<TerminalStatus>
sendInput: (data: string) => void
sendResize: (rows: number, cols: number) => void
reconnect: () => void
close: () => void
}
function wsURL(rows: number, cols: number, token: string): string {
const proto = window.location.protocol === 'https:' ? 'wss' : 'ws'
const host = window.location.host
return (
`${proto}://${host}/api/v1/admin/terminal/stream` +
`?rows=${rows}&cols=${cols}&token=${encodeURIComponent(token)}`
)
}
export function useTerminalSocket(opts: UseTerminalSocketOptions): UseTerminalSocketReturn {
const status = ref<TerminalStatus>('idle')
const rows = ref(opts.initialRows ?? 24)
const cols = ref(opts.initialCols ?? 80)
let ws: WebSocket | null = null
let retried = 0
let manualClose = false
let reconnectTimer: ReturnType<typeof setTimeout> | null = null
const auth = useAuthStore()
const decoder = new TextDecoder()
const encoder = new TextEncoder()
function connect() {
if (ws || manualClose) return
status.value = 'connecting'
const token = auth.token ?? ''
const sock = new WebSocket(wsURL(rows.value, cols.value, token))
sock.binaryType = 'arraybuffer'
ws = sock
sock.onopen = () => {
status.value = 'open'
retried = 0
}
sock.onmessage = (ev: MessageEvent) => {
if (ev.data instanceof ArrayBuffer) {
opts.onData(decoder.decode(new Uint8Array(ev.data)))
} else if (typeof ev.data === 'string') {
// 兼容性兜底:服务端目前只发二进制帧。
opts.onData(ev.data)
}
}
sock.onerror = () => {
status.value = 'error'
}
sock.onclose = () => {
ws = null
if (manualClose) {
status.value = 'closed'
return
}
if (retried < 2) {
const delay = retried === 0 ? 1000 : 3000
retried += 1
reconnectTimer = setTimeout(connect, delay)
} else {
status.value = 'closed'
}
}
}
function sendInput(data: string) {
if (!ws || ws.readyState !== WebSocket.OPEN) return
ws.send(encoder.encode(data))
}
function sendResize(r: number, c: number) {
rows.value = r
cols.value = c
if (!ws || ws.readyState !== WebSocket.OPEN) return
ws.send(JSON.stringify({ type: 'resize', rows: r, cols: c }))
}
function reconnect() {
if (ws) ws.close()
retried = 0
manualClose = false
connect()
}
function close() {
manualClose = true
if (reconnectTimer) {
clearTimeout(reconnectTimer)
reconnectTimer = null
}
if (ws) ws.close()
status.value = 'closed'
}
connect()
onScopeDispose(close)
return { status, sendInput, sendResize, reconnect, close }
}
+6
View File
@@ -175,6 +175,12 @@ const routes: RouteRecordRaw[] = [
component: () => import('@/views/admin/AgentKindEditView.vue'),
meta: { requiresAuth: true, requiresAdmin: true },
},
{
path: '/admin/terminal',
name: 'admin-terminal',
component: () => import('@/views/admin/AdminTerminalView.vue'),
meta: { requiresAuth: true, requiresAdmin: true },
},
{
// 兜底:匹配不到的路径(如空参数的 /p/x/w/、手输错误 URL)统一回首页,避免白屏
path: '/:pathMatch(.*)*',
+141
View File
@@ -0,0 +1,141 @@
<template>
<AppShell>
<div class="space-y-3">
<div class="flex justify-between items-center">
<h2 class="text-lg font-semibold">
终端
</h2>
<div class="flex items-center gap-3">
<NTag
:type="statusType"
size="small"
round
>
{{ statusLabel }}
</NTag>
<NButton
size="small"
:disabled="sock.status.value === 'connecting' || sock.status.value === 'open'"
@click="sock.reconnect()"
>
重连
</NButton>
</div>
</div>
<NAlert
type="warning"
:show-icon="true"
title="管理员终端"
>
命令在 server 宿主进程部署时即 runtime 容器内执行权限等同容器内 shell
安装 ACP 客户端示例<code>npm i -g @anthropic-ai/claude-code</code>
安装完成后将 <code>binary_path</code> 填入对应的 ACP 客户端配置
</NAlert>
<div
ref="containerEl"
class="terminal-host"
/>
</div>
</AppShell>
</template>
<script setup lang="ts">
import { computed, onBeforeUnmount, onMounted, ref } from 'vue'
import { NAlert, NButton, NTag } from 'naive-ui'
import { Terminal } from '@xterm/xterm'
import { FitAddon } from '@xterm/addon-fit'
import '@xterm/xterm/css/xterm.css'
import AppShell from '@/layouts/AppShell.vue'
import { useTerminalSocket } from '@/composables/useTerminalSocket'
const containerEl = ref<HTMLElement | null>(null)
const term = new Terminal({
fontFamily: 'Menlo, Consolas, "DejaVu Sans Mono", monospace',
fontSize: 13,
cursorBlink: true,
convertEol: false,
scrollback: 5000,
})
const fitAddon = new FitAddon()
term.loadAddon(fitAddon)
// 在 setup 同步阶段建立连接(onScopeDispose 才能正确挂到组件 scope)。
// term.write 在 open() 前会被 xterm 内部缓冲,待挂载后渲染。
const sock = useTerminalSocket({
onData: (text) => term.write(text),
initialRows: term.rows,
initialCols: term.cols,
})
term.onData((d) => sock.sendInput(d))
let resizeObserver: ResizeObserver | null = null
function fitAndReport() {
if (!containerEl.value) return
try {
fitAddon.fit()
} catch {
return
}
sock.sendResize(term.rows, term.cols)
}
onMounted(() => {
if (!containerEl.value) return
term.open(containerEl.value)
fitAndReport()
resizeObserver = new ResizeObserver(() => fitAndReport())
resizeObserver.observe(containerEl.value)
term.focus()
})
onBeforeUnmount(() => {
resizeObserver?.disconnect()
resizeObserver = null
sock.close()
term.dispose()
})
const statusLabel = computed(() => {
switch (sock.status.value) {
case 'connecting':
return '连接中'
case 'open':
return '已连接'
case 'error':
return '错误'
case 'closed':
return '已断开'
default:
return '空闲'
}
})
const statusType = computed<'success' | 'warning' | 'error' | 'default'>(() => {
switch (sock.status.value) {
case 'open':
return 'success'
case 'connecting':
return 'warning'
case 'error':
return 'error'
default:
return 'default'
}
})
</script>
<style scoped>
.terminal-host {
width: 100%;
height: calc(100vh - 220px);
min-height: 320px;
background: #000;
padding: 8px;
border-radius: 6px;
overflow: hidden;
}
</style>