agentic-coding-workflow/internal/run/supervisor.go

// supervisor.go 管理 run profile 子进程的内存运行态。每个 profile 同一时刻至多
// 一个运行进程。并发骨架照搬 acp/supervisor.go 已被测试验证的不变量：
//   - procs map 的写锁仅在 Spawn 插入 / monitor 删除时持有
//   - done channel 严格 close-once（monitorProcess 唯一持有）
//   - KilledByUs atomic 区分 stopped（我方终止）vs crashed
//   - onExit 不得重新 Lock s.mu（避免 monitor→Stop→onExit 链式死锁）
//
// 与 acp 的差异：无 JSON-RPC handshake / Relay 双向中继 / MCP token / worktree
// acquire；并补上经 internal/infra/proc 的整树终止（dev server 会派生子孙进程）。
package run

import (
	"bufio"
	"context"
	"io"
	"log/slog"
	"os/exec"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"

	procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc"
)

// SupervisorConfig 是从 config.RunConfig 派生的 supervisor 子集。
type SupervisorConfig struct {
	KillGrace      time.Duration
	OutBufferLines int
	OutTailBytes   int
	WSSendBuffer   int
	ShutdownGrace  time.Duration
}

// RunProc 是单个运行进程的运行时句柄。
type RunProc struct {
	ProfileID uuid.UUID
	Cmd       *exec.Cmd
	group     procgrp.Group
	relay     *runRelay

	StartedAt  time.Time
	KilledByUs atomic.Bool   // true → stopped；false → crashed
	done       chan struct{} // monitor 退出时 close
}

// RunSupervisor 是 run profile 子进程总管。被动结构：Spawn 时起 goroutine，无主循环。
type RunSupervisor struct {
	mu       sync.RWMutex
	procs    map[uuid.UUID]*RunProc
	terminal map[uuid.UUID]RunStatus // 最近一次退出的终止态（crashed/stopped），重启即清空

	repo Repository
	cfg  SupervisorConfig
	log  *slog.Logger
}

// NewSupervisor 构造空 RunSupervisor。
func NewSupervisor(repo Repository, cfg SupervisorConfig, log *slog.Logger) *RunSupervisor {
	if log == nil {
		log = slog.Default()
	}
	return &RunSupervisor{
		procs:    map[uuid.UUID]*RunProc{},
		terminal: map[uuid.UUID]RunStatus{},
		repo:     repo,
		cfg:      cfg,
		log:      log,
	}
}

// SpawnParams 由 Service 构造：命令、参数、最终环境（宿主白名单+覆盖）、工作目录。
type SpawnParams struct {
	ProfileID uuid.UUID
	Command   string
	Args      []string
	Env       []string
	Dir       string
}

// ErrAlreadyRunning / ErrNotRunning 是 supervisor 层的哨兵，Service 翻译为 errs code。
var (
	ErrAlreadyRunning = errSentinel("run: profile already running")
	ErrNotRunning     = errSentinel("run: profile not running")
)

type errSentinel string

func (e errSentinel) Error() string { return string(e) }

// Spawn 启动一个子进程并托管。调用方（Service）负责前置鉴权与 env 构造。
// 同一 profile 已在运行时返回 ErrAlreadyRunning。
func (s *RunSupervisor) Spawn(p SpawnParams) (*RunProc, error) {
	s.mu.RLock()
	_, running := s.procs[p.ProfileID]
	s.mu.RUnlock()
	if running {
		return nil, ErrAlreadyRunning
	}

	cmd := exec.Command(p.Command, p.Args...)
	cmd.Dir = p.Dir
	cmd.Env = p.Env

	// 进程树管理：Prepare 须在 Start 前（Unix Setpgid），Adopt 须在 Start 后
	// （Windows Job Object）。终止时整树清理，避免 dev server 子孙进程残留。
	group := procgrp.NewGroup()
	group.Prepare(cmd)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return nil, err
	}
	if err := cmd.Start(); err != nil {
		return nil, err
	}
	if err := group.Adopt(cmd); err != nil {
		_ = cmd.Process.Kill()
		return nil, err
	}

	proc := &RunProc{
		ProfileID: p.ProfileID,
		Cmd:       cmd,
		group:     group,
		relay:     newRunRelay(p.ProfileID, s.cfg.OutBufferLines, s.cfg.WSSendBuffer, s.log),
		StartedAt: time.Now(),
		done:      make(chan struct{}),
	}

	s.mu.Lock()
	s.procs[p.ProfileID] = proc
	delete(s.terminal, p.ProfileID) // 进入运行态，清除上一次终止态
	s.mu.Unlock()

	go s.drainPipe(proc, stdout, "info")
	go s.drainPipe(proc, stderr, "error")
	go s.monitorProcess(proc)

	return proc, nil
}

// Stop 终止指定 profile 的子进程（整树）。幂等阻塞直到 done 关闭或 ctx 取消。
// profile 未运行时返回 ErrNotRunning。
func (s *RunSupervisor) Stop(ctx context.Context, profileID uuid.UUID, grace time.Duration) error {
	s.mu.RLock()
	proc, ok := s.procs[profileID]
	s.mu.RUnlock()
	if !ok {
		return ErrNotRunning
	}
	proc.KilledByUs.Store(true)
	proc.group.TerminateGroup(proc.done, grace)
	select {
	case <-proc.done:
	case <-ctx.Done():
	}
	return nil
}

// Get 返回指定 profile 当前运行进程（用于 status/logs/subscribe）；未运行返回 false。
func (s *RunSupervisor) Get(profileID uuid.UUID) (*RunProc, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	proc, ok := s.procs[profileID]
	return proc, ok
}

// StatusOf 返回内存运行态：在 procs 中 → running；否则取最近终止态（crashed/stopped），
// 无记录（含服务端重启后）→ stopped。
func (s *RunSupervisor) StatusOf(profileID uuid.UUID) RunStatus {
	s.mu.RLock()
	defer s.mu.RUnlock()
	if _, ok := s.procs[profileID]; ok {
		return StatusRunning
	}
	if st, ok := s.terminal[profileID]; ok {
		return st
	}
	return StatusStopped
}

// Subscribe 为正在运行的 profile 注册一个日志订阅；未运行返回 (nil,false)。
func (s *RunSupervisor) Subscribe(profileID, userID uuid.UUID) (*Subscriber, *runRelay, bool) {
	s.mu.RLock()
	proc, ok := s.procs[profileID]
	s.mu.RUnlock()
	if !ok {
		return nil, nil, false
	}
	return proc.relay.Subscribe(userID), proc.relay, true
}

// LogsSince 返回正在运行进程的内存日志（ID > since）；未运行返回 nil。
func (s *RunSupervisor) LogsSince(profileID uuid.UUID, since int64, limit int) []*LogEnvelope {
	s.mu.RLock()
	proc, ok := s.procs[profileID]
	s.mu.RUnlock()
	if !ok {
		return nil
	}
	return proc.relay.since(since, limit)
}

// drainPipe 逐行读取管道并 emit 到 relay。EOF/读错误时安静退出（Wait 由 monitor 处理）。
func (s *RunSupervisor) drainPipe(proc *RunProc, pipe io.Reader, level string) {
	scanner := bufio.NewScanner(pipe)
	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
	for scanner.Scan() {
		proc.relay.emit(level, scanner.Text())
	}
}

// monitorProcess 等待进程退出，更新 procs/terminal map，释放进程树句柄并触发 onExit。
// 这是 done channel 唯一的合法关闭路径。
func (s *RunSupervisor) monitorProcess(proc *RunProc) {
	defer close(proc.done)
	waitErr := proc.Cmd.Wait()

	exitCode := int32(0)
	if exitErr, ok := waitErr.(*exec.ExitError); ok {
		exitCode = int32(exitErr.ExitCode())
	}

	status := StatusStopped
	if !proc.KilledByUs.Load() {
		status = StatusCrashed
	}

	s.mu.Lock()
	delete(s.procs, proc.ProfileID)
	s.terminal[proc.ProfileID] = status
	s.mu.Unlock()

	// 释放进程树句柄（Windows 关闭 Job handle；Unix no-op）。
	proc.group.Close()

	s.onExit(context.Background(), proc, status, exitCode)
}

// onExit 把退出信息落库并广播终止帧。不得再 Lock s.mu（monitorProcess 已 delete
// map 后调用本函数；Stop 也持读锁等 done，重入会死锁）。
func (s *RunSupervisor) onExit(ctx context.Context, proc *RunProc, status RunStatus, exitCode int32) {
	lastErr := ""
	if status == StatusCrashed {
		lastErr = proc.relay.tailText(s.cfg.OutTailBytes)
	}
	ec := exitCode
	if s.repo != nil {
		if err := s.repo.UpdateRunProfileExit(ctx, proc.ProfileID, &ec, lastErr); err != nil {
			s.log.Error("run.on_exit.update_exit", "profile_id", proc.ProfileID, "err", err.Error())
		}
	}
	proc.relay.Close(string(status), &ec)
}

// ShutdownAll 并行终止所有运行进程，受 cfg.ShutdownGrace 总时限约束。
func (s *RunSupervisor) ShutdownAll(ctx context.Context) {
	s.mu.RLock()
	ids := make([]uuid.UUID, 0, len(s.procs))
	for id := range s.procs {
		ids = append(ids, id)
	}
	s.mu.RUnlock()

	shutdownCtx, cancel := context.WithTimeout(ctx, s.cfg.ShutdownGrace)
	defer cancel()

	var wg sync.WaitGroup
	for _, id := range ids {
		wg.Add(1)
		go func(pid uuid.UUID) {
			defer wg.Done()
			_ = s.Stop(shutdownCtx, pid, s.cfg.KillGrace)
		}(id)
	}

	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-shutdownCtx.Done():
		s.log.Warn("run.shutdown_all.timeout")
	}
}