From bd95fd33d6a6e87501ae267054e8380d0c4d61c6 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Thu, 7 May 2026 12:12:05 +0800 Subject: [PATCH] feat(acp): supervisor skeleton + AcpConfig wiring --- config.example.yaml | 17 +++++ internal/acp/supervisor.go | 126 +++++++++++++++++++++++++++++++++++++ internal/config/config.go | 33 ++++++++++ 3 files changed, 176 insertions(+) create mode 100644 internal/acp/supervisor.go diff --git a/config.example.yaml b/config.example.yaml index e1647d2..57e2d94 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -107,3 +107,20 @@ notify: secret: "" timeout: 10s topics: [] + +# ACP module: agent subprocess pool + WebSocket fanout + event retention +acp: + enabled: true + max_active_global: 20 + max_active_per_user: 3 + spawn_timeout: 30s + kill_grace: 5s + stderr_buffer_lines: 100 + stderr_tail_bytes: 2000 + ws_ping_interval: 30s + ws_pong_timeout: 60s + ws_send_buffer: 256 + event_max_payload_bytes: 65536 + event_truncate_field_bytes: 4096 + event_retention_days: 7 + shutdown_grace: 30s diff --git a/internal/acp/supervisor.go b/internal/acp/supervisor.go new file mode 100644 index 0000000..54053c4 --- /dev/null +++ b/internal/acp/supervisor.go @@ -0,0 +1,126 @@ +// supervisor.go 管理 ACP agent 子进程池。每 session 一个 Process，spawn / kill / +// monitor 三类操作通过 mu 保护 procs map。 +// +// 生命周期 invariants（spec §6.8）： +// - procs 写锁仅在 spawn / Kill / monitorProcess delete 时持有 +// - done channel 严格 close-once（monitor 唯一持有） +// - KilledByUs atomic 防 race（Kill / monitor 并发） +// - onExit 不能再 Lock supervisor.mu（避免死锁） +package acp + +import ( + "io" + "log/slog" + "os/exec" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" + + "github.com/yan1h/agent-coding-workflow/internal/audit" + "github.com/yan1h/agent-coding-workflow/internal/infra/crypto" + "github.com/yan1h/agent-coding-workflow/internal/infra/notify" + "github.com/yan1h/agent-coding-workflow/internal/workspace" +) + +// SupervisorConfig 是从 config.AcpConfig 派生的 supervisor 子集。 +type SupervisorConfig struct { + SpawnTimeout time.Duration + KillGrace time.Duration + StderrBufferLines int + StderrTailBytes int + EventMaxPayload int + EventTruncateField int + ShutdownGrace time.Duration +} + +// Supervisor 是 ACP 子进程总管。 +type Supervisor struct { + mu sync.RWMutex + procs map[uuid.UUID]*Process + + repo Repository + audit audit.Recorder + notify *notify.Dispatcher + wtSvc workspace.WorktreeService + wsRepo workspace.Repository + crypto *crypto.Encryptor + cfg SupervisorConfig + log *slog.Logger +} + +// NewSupervisor 构造空 Supervisor。Start 不需要 — supervisor 是被动的（spawn 时 +// 起 goroutine，无主循环）。Stop 在 ShutdownAll 中实现。 +func NewSupervisor(repo Repository, rec audit.Recorder, disp *notify.Dispatcher, + wtSvc workspace.WorktreeService, wsRepo workspace.Repository, + enc *crypto.Encryptor, cfg SupervisorConfig, log *slog.Logger) *Supervisor { + if log == nil { + log = slog.Default() + } + return &Supervisor{ + procs: map[uuid.UUID]*Process{}, + repo: repo, + audit: rec, + notify: disp, + wtSvc: wtSvc, + wsRepo: wsRepo, + crypto: enc, + cfg: cfg, + log: log, + } +} + +// Process 是单个 agent 子进程的运行时句柄。 +type Process struct { + SessionID uuid.UUID + WorkspaceID uuid.UUID + UserID uuid.UUID + IsMain bool // 主 worktree 占用 → release 走不同路径 + WorktreeID *uuid.UUID // 子 worktree 时填，用于 release + + Cmd *exec.Cmd + Stdin io.WriteCloser + Stdout io.ReadCloser + Stderr io.ReadCloser + + StderrBuf *stderrRing // 排障 ring buffer + StartedAt time.Time + KilledByUs atomic.Bool // true → exited，false → crashed + done chan struct{} // monitor 退出时 close +} + +// stderrRing 是固定行数 ring buffer，monitor 退出时取 tail 写 last_error。 +type stderrRing struct { + mu sync.Mutex + lines []string + cap int +} + +func newStderrRing(cap int) *stderrRing { return &stderrRing{cap: cap} } + +func (r *stderrRing) Append(line string) { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.lines) >= r.cap { + r.lines = r.lines[1:] + } + r.lines = append(r.lines, line) +} + +func (r *stderrRing) Tail(maxBytes int) string { + r.mu.Lock() + defer r.mu.Unlock() + out := "" + for i := len(r.lines) - 1; i >= 0 && len(out) < maxBytes; i-- { + out = r.lines[i] + "\n" + out + } + if len(out) > maxBytes { + out = out[len(out)-maxBytes:] + } + return out +} + +// Relay field + GetRelay method added in Task D7 when Relay type is wired. + +// Spawn / Kill / ShutdownAll / monitorProcess / handshake are implemented in subsequent tasks (E2). diff --git a/internal/config/config.go b/internal/config/config.go index ecfbc60..7c59fa6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -41,6 +41,7 @@ type Config struct { Storage Storage `mapstructure:"storage"` Jobs JobsConfig `mapstructure:"jobs"` Notify NotifyConfig `mapstructure:"notify"` + Acp AcpConfig `mapstructure:"acp"` } // HTTPConfig 描述 HTTP 服务监听参数。 @@ -165,6 +166,24 @@ type WebhookCfg struct { Topics []string `mapstructure:"topics"` } +// AcpConfig 控制 ACP 模块的并发限流、子进程超时、WS 心跳与事件保留。 +type AcpConfig struct { + Enabled bool `mapstructure:"enabled"` + MaxActiveGlobal int `mapstructure:"max_active_global"` + MaxActivePerUser int `mapstructure:"max_active_per_user"` + SpawnTimeout time.Duration `mapstructure:"spawn_timeout"` + KillGrace time.Duration `mapstructure:"kill_grace"` + StderrBufferLines int `mapstructure:"stderr_buffer_lines"` + StderrTailBytes int `mapstructure:"stderr_tail_bytes"` + WSPingInterval time.Duration `mapstructure:"ws_ping_interval"` + WSPongTimeout time.Duration `mapstructure:"ws_pong_timeout"` + WSSendBuffer int `mapstructure:"ws_send_buffer"` + EventMaxPayload int `mapstructure:"event_max_payload_bytes"` + EventTruncateField int `mapstructure:"event_truncate_field_bytes"` + EventRetentionDays int `mapstructure:"event_retention_days"` + ShutdownGrace time.Duration `mapstructure:"shutdown_grace"` +} + // Load 从（按优先级递增）默认值 → 配置文件（如有）→ 环境变量 加载配置。 // configFile 可为空字符串，仅用环境变量。 func Load(configFile string) (*Config, error) { @@ -210,6 +229,20 @@ func Load(configFile string) (*Config, error) { v.SetDefault("jobs.job_purge.completed_retention", "168h") v.SetDefault("notify.webhook.enabled", false) v.SetDefault("notify.webhook.timeout", "10s") + v.SetDefault("acp.enabled", true) + v.SetDefault("acp.max_active_global", 20) + v.SetDefault("acp.max_active_per_user", 3) + v.SetDefault("acp.spawn_timeout", "30s") + v.SetDefault("acp.kill_grace", "5s") + v.SetDefault("acp.stderr_buffer_lines", 100) + v.SetDefault("acp.stderr_tail_bytes", 2000) + v.SetDefault("acp.ws_ping_interval", "30s") + v.SetDefault("acp.ws_pong_timeout", "60s") + v.SetDefault("acp.ws_send_buffer", 256) + v.SetDefault("acp.event_max_payload_bytes", 65536) + v.SetDefault("acp.event_truncate_field_bytes", 4096) + v.SetDefault("acp.event_retention_days", 7) + v.SetDefault("acp.shutdown_grace", "30s") if configFile != "" { v.SetConfigFile(configFile)