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

// 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 (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/exec"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"

	"github.com/yan1h/agent-coding-workflow/internal/acp/handlers"
	"github.com/yan1h/agent-coding-workflow/internal/acp/sandbox"
	"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"
	procgrp "github.com/yan1h/agent-coding-workflow/internal/infra/proc"
	"github.com/yan1h/agent-coding-workflow/internal/mcp"
	"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
	// AgentHomesDir 是 agent CLI 受管 home 的根目录（<DataDir>/agent-homes）。
	// 空串时跳过配置物化与重定向 env 注入（部分测试）。
	AgentHomesDir string
	// DefaultProjectBudgetUSD 是 per-project ACP 累计花费的全局软上限（0 = 不限）。
	// 当 session 与 agent-kind 均未设置成本上限时，用作 fallback 的预算判定基线。
	DefaultProjectBudgetUSD float64

	// ===== 沙箱（per-session 隔离）=====
	// Sandbox 是 pluggable 沙箱实现（none|uid|container），在 cmd/env 构建后、
	// proc.Group.Prepare 之前 Apply。为 nil 时等同 mode=none（无隔离）。
	Sandbox sandbox.Sandbox
	// SandboxMode 记录当前 mode，用于落库 acp_sessions.sandbox_mode（审计/取证）。
	SandboxMode sandbox.Mode
	// SandboxBaseUID 是 per-session 低权 UID 的基址：实际 uid = BaseUID + offset，
	// offset 由 session 派生（保证不同活跃 session 不复用同一 uid）。0 = 不分配。
	SandboxBaseUID int
	// DataRoot 是要被屏蔽的 /data 根（除本 session worktree + per-user home 外）。
	DataRoot string
	// EgressProxyURL 是注入子进程 env 的 HTTP(S)_PROXY（egress 白名单代理）。
	EgressProxyURL string
	// SandboxRlimits 是套用到子进程（及其进程树）的 OS 资源上限。
	SandboxRlimits sandbox.Limits
}

// 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
	mcpTokens   mcp.TokenService // spec §6.3 — onExit revokes system token
	crypto      *crypto.Encryptor
	permSvc     *PermissionService  // 工具调用权限审批；可为 nil（部分测试）
	turnBus     TurnBus             // 回合完成事件总线；可为 nil（部分测试）
	orchRedrive OrchestratorRedrive // 崩溃的编排器 session 再驱动回调；可为 nil
	prices      ModelPriceLookup    // model 价格查询；可为 nil（无 model 价格时成本=0/agent 自报）
	agentsMD    AgentsMDRenderer    // AGENTS.md 渲染回调（project memory）；可为 nil
	metrics     SessionExitRecorder // 会话退出计数；可为 nil（metrics 关闭）
	cfg         SupervisorConfig
	log         *slog.Logger
}

// SessionExitRecorder 是 Supervisor 上报会话退出指标所需的窄接口。
// metrics.Metrics 满足它；nil 时不记录。
type SessionExitRecorder interface {
	RecordSessionExit(status string)
}

// SetMetrics 注入会话退出计数器（装配期）。
func (s *Supervisor) SetMetrics(m SessionExitRecorder) { s.metrics = m }

// AgentsMDRenderer renders the AGENTS.md content for a (project, workspace) from
// project memory. Injected via SetAgentsMDRenderer to avoid an acp →
// projectmemory import cycle. May be nil (no seeding).
type AgentsMDRenderer func(ctx context.Context, projectID uuid.UUID, wsID *uuid.UUID) (string, error)

// SetAgentsMDRenderer 注入 AGENTS.md 渲染回调（装配期）。
func (s *Supervisor) SetAgentsMDRenderer(fn AgentsMDRenderer) { s.agentsMD = fn }

// agentsMDFileName is the generated file written into the worktree root.
const agentsMDFileName = "AGENTS.md"

// seedAgentsMD renders project memory into <cwd>/AGENTS.md before spawn. Skipped
// when no renderer is configured, CwdPath is empty, or ProjectID is nil. Failures
// are logged, never fatal — the agent still spawns without seeded knowledge.
func (s *Supervisor) seedAgentsMD(ctx context.Context, sess *Session) {
	if s.agentsMD == nil || sess.CwdPath == "" || sess.ProjectID == uuid.Nil {
		return
	}
	var wsID *uuid.UUID
	if sess.WorkspaceID != uuid.Nil {
		id := sess.WorkspaceID
		wsID = &id
	}
	content, err := s.agentsMD(ctx, sess.ProjectID, wsID)
	if err != nil {
		s.log.Warn("acp.agents_md.render_failed", "session_id", sess.ID, "err", err.Error())
		return
	}
	dst := filepath.Join(sess.CwdPath, agentsMDFileName)
	if err := os.WriteFile(dst, []byte(content), 0o644); err != nil {
		s.log.Warn("acp.agents_md.write_failed", "session_id", sess.ID, "path", dst, "err", err.Error())
	}
}

// OrchestratorRedrive 在一个由编排器 step 驱动的 session 崩溃退出时被调用，由编排器
// 据此把对应 step 重新入队（带 backoff），让回合在同一 worktree 上恢复。注入为 func
// 以避免 acp → orchestrator 的 import 循环（与 notify dispatcher / permission service
// 的注入方式一致）。实现必须是非阻塞、不持 supervisor 锁的纯 DB 写（jobs Enqueue）。
type OrchestratorRedrive func(ctx context.Context, stepID uuid.UUID, reason string)

// SetOrchestratorRedrive 回填编排器再驱动回调（装配期）。
func (s *Supervisor) SetOrchestratorRedrive(fn OrchestratorRedrive) { s.orchRedrive = fn }

// SetPermissionService 注入权限审批服务。与 Supervisor 互相依赖，故装配时回填
// （permSvc 需要 Supervisor 作为 RelayLocator，Supervisor 需要 permSvc 注入 handler）。
func (s *Supervisor) SetPermissionService(p *PermissionService) { s.permSvc = p }

// SetTurnBus 注入回合事件总线。Spawn 时每个 relay 构造一个 TurnTracker 复用它。
// 可为 nil——此时回合仍落库（若 repo 可用），但不发布内部事件。
func (s *Supervisor) SetTurnBus(b TurnBus) { s.turnBus = b }

// SetModelPriceLookup 注入 model 价格查询（成本核算用）。装配期回填，避免 acp →
// chat 的构造期依赖。可为 nil——此时无 model 价格的会话成本为 0 或取 agent 自报。
func (s *Supervisor) SetModelPriceLookup(p ModelPriceLookup) { s.prices = p }

// NewSupervisor 构造空 Supervisor。Start 不需要 — supervisor 是被动的（spawn 时
// 起 goroutine，无主循环）。Stop 在 ShutdownAll 中实现。
func NewSupervisor(repo Repository, rec audit.Recorder, disp *notify.Dispatcher,
	wtSvc workspace.WorktreeService, wsRepo workspace.Repository,
	mcpTokens mcp.TokenService,
	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,
		mcpTokens: mcpTokens,
		crypto:    enc,
		cfg:       cfg,
		log:       log,
	}
}

// Process 是单个 agent 子进程的运行时句柄。
type Process struct {
	SessionID   uuid.UUID
	WorkspaceID uuid.UUID
	UserID      uuid.UUID
	IsMain      bool // 主 worktree 占用 → release 走不同路径

	Cmd    *exec.Cmd
	group  procgrp.Group // 进程树句柄（整树终止），Spawn 时 Prepare+Adopt
	Stdin  io.WriteCloser
	Stdout io.ReadCloser
	Stderr io.ReadCloser
	Relay  *Relay

	StderrBuf  *stderrRing // 排障 ring buffer
	StartedAt  time.Time
	KilledByUs atomic.Bool   // true → exited，false → crashed
	done       chan struct{} // monitor 退出时 close

	// relayCancel 取消 relay 的会话级 context。该 context 由 Spawn 用
	// context.Background() 派生，独立于调用方的 spawn ctx（spawn ctx 在 Spawn
	// 返回后立即取消），故 relay 的 reader/writer/persist/permission Decide 在整个
	// 会话生命周期内都使用未取消的 ctx。onExit 调用 relayCancel 完成 relay teardown。
	relayCancel context.CancelFunc

	// sandboxCleanup 拆除沙箱临时资源（卸载 bind mount、停止容器、删临时目录）。
	// 由 monitorProcess 在 group.Close() 之后调用，仅调用一次。可为 nil（mode=none）。
	sandboxCleanup func()
}

// 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
}

// GetRelay returns the relay for an active session, or nil if not running.
// WS handler 用此判断是否进 live 模式。
func (s *Supervisor) GetRelay(sid uuid.UUID) *Relay {
	s.mu.RLock()
	defer s.mu.RUnlock()
	p, ok := s.procs[sid]
	if !ok {
		return nil
	}
	return p.Relay
}

// Spawn 启动一个子进程并完成 ACP handshake。返回的 *Process 已注册到 procs map；
// handshake 失败时调用 Kill 回滚并返回错误。调用方（SessionService）负责前置的
// worktree 占用与 acp_sessions 行 INSERT；本方法仅负责 OS 层资源与协议层握手。
func (s *Supervisor) Spawn(ctx context.Context, sess *Session, kind *AgentKind, extraEnv map[string]string) (*Process, error) {
	// AGENTS.md seeding：在 spawn 前把 project memory 渲染成 worktree 根的
	// AGENTS.md，让 agent 启动即有结构化项目知识。best-effort，失败仅告警不阻断。
	s.seedAgentsMD(ctx, sess)

	// 受管 home：物化 DB 中的配置文件并注入重定向 env（优先级最低，可被
	// AgentKind env / extraEnv 覆盖）。AgentHomesDir 未配置时跳过（部分测试）。
	envMap := map[string]string{}
	var agentHome string
	if s.cfg.AgentHomesDir != "" {
		home, err := s.materializeAgentHome(ctx, sess, kind)
		if err != nil {
			return nil, fmt.Errorf("materialize agent home: %w", err)
		}
		agentHome = home
		envMap = agentHomeEnv(home, kind.ClientType)
	}

	kindEnv, err := decryptEnv(s.crypto, kind.EncryptedEnv)
	if err != nil {
		return nil, err
	}
	for k, v := range kindEnv {
		envMap[k] = v
	}

	// System override: extraEnv overrides agent_kinds same-name keys (spec §6.1 #11)
	for k, v := range extraEnv {
		envMap[k] = v
	}

	// strip env：只保留 PATH + 解密 overrides，避免泄漏 master key、git 凭据等。
	osEnv := []string{"PATH=" + os.Getenv("PATH")}
	for k, v := range envMap {
		osEnv = append(osEnv, fmt.Sprintf("%s=%s", k, v))
	}

	cmd := exec.Command(kind.BinaryPath, kind.Args...)
	cmd.Dir = sess.CwdPath
	cmd.Env = osEnv

	// 沙箱：在 cmd/env 构建完成后、proc.Group.Prepare（Setpgid）之前 Apply。
	// Apply 只 ADD 到 cmd.SysProcAttr（Credential / namespace），绝不替换，
	// 故不破坏后续 group.Prepare 的 Setpgid 与负 pgid 整树终止。返回的 cleanup
	// 在 monitorProcess 的 group.Close() 之后调用一次。mode=none 时为 no-op。
	sandboxCleanup := func() {}
	if s.cfg.Sandbox != nil {
		spec := s.buildSandboxSpec(sess, agentHome)
		cleanup, serr := s.cfg.Sandbox.Apply(cmd, spec)
		if serr != nil {
			return nil, fmt.Errorf("sandbox apply: %w", serr)
		}
		if cleanup != nil {
			sandboxCleanup = cleanup
		}
		// 记录本 session 运行所用的 sandbox mode（审计/取证）。best-effort。
		if s.cfg.SandboxMode != "" {
			if err := s.repo.UpdateSessionSandboxMode(ctx, sess.ID, string(s.cfg.SandboxMode)); err != nil {
				s.log.Warn("acp.sandbox.record_mode", "session_id", sess.ID, "err", err.Error())
			}
		}
	}

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

	stdin, err := cmd.StdinPipe()
	if err != nil {
		return nil, fmt.Errorf("stdin pipe: %w", err)
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, fmt.Errorf("stdout pipe: %w", err)
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return nil, fmt.Errorf("stderr pipe: %w", err)
	}

	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("start: %w", err)
	}
	if err := group.Adopt(cmd); err != nil {
		_ = cmd.Process.Kill()
		_ = cmd.Wait() // 回收进程资源，避免 Unix 僵尸进程
		return nil, fmt.Errorf("adopt process group: %w", err)
	}

	var decider handlers.PermissionDecider
	if s.permSvc != nil {
		decider = s.permSvc
	}
	relay := NewRelay(
		sess.ID,
		NewDecoder(stdout, s.cfg.EventMaxPayload),
		NewEncoder(stdin),
		s.repo,
		handlers.NewFsHandler(),
		handlers.NewPermissionHandler(decider),
		RelayConfig{
			EventMaxPayload:    s.cfg.EventMaxPayload,
			EventTruncateField: s.cfg.EventTruncateField,
			WSSendBuffer:       256,
		},
		s.log,
	)

	// 回合追踪器：被动解析 session/update + session/prompt 响应。复用同一 acpRepo
	// 与共享 TurnBus，无需额外连接池。onExit 通过 relay.Tracker().Abort 闭合悬挂回合。
	sessCtx := handlers.SessionContext{
		SessionID:      sess.ID,
		WorkspaceID:    sess.WorkspaceID,
		UserID:         sess.UserID,
		CwdPath:        sess.CwdPath,
		AgentSessionID: sess.AgentSessionID,
		ToolAllowlist:  kind.ToolAllowlist,
	}
	relay.SetTracker(NewTurnTracker(s.repo, s.turnBus, sessCtx, s.log))

	// 成本核算：构造 per-session 累加器 + usage sink，seeded with 会话快照的有效
	// 预算上限与 agent-kind 的 model 价格来源。预算突破时以 detached goroutine 标记
	// terminated_reason 后 Kill（绝不阻塞 relay.reader，且 Kill 幂等）。
	caps := BudgetCaps{
		MaxCostUSD:          sess.BudgetMaxCostUSD,
		MaxTokens:           sess.BudgetMaxTokens,
		MaxWallClockSeconds: sess.BudgetMaxWallClockSeconds,
	}
	acc := newUsageAccumulator(AccumulatorParams{
		Repo:             s.repo,
		Prices:           s.prices,
		Log:              s.log,
		SessionID:        sess.ID,
		UserID:           sess.UserID,
		ProjectID:        sess.ProjectID,
		AgentKindID:      sess.AgentKindID,
		ModelID:          kind.ModelID,
		StartedAt:        time.Now(),
		Caps:             caps,
		ProjectBudgetUSD: s.cfg.DefaultProjectBudgetUSD,
	})
	sid := sess.ID
	killForBudget := func(reason BreachReason) {
		bgCtx := context.Background()
		if err := s.repo.MarkSessionTerminatedReason(bgCtx, sid, string(reason)); err != nil {
			s.log.Error("acp.budget.mark_terminated", "session_id", sid, "err", err.Error())
		}
		s.log.Warn("acp.budget.breach_kill", "session_id", sid, "reason", string(reason))
		s.Kill(bgCtx, sid, s.cfg.KillGrace)
	}
	relay.SetUsageSink(newAccumulatorSink(acc, killForBudget, s.log))

	// relay 的会话级 context：独立于调用方的 spawn ctx（生产调用方在 Spawn 返回后
	// 立即 cancel spawn ctx）。relay 的 reader/writer/persist/permission Decide 必须
	// 在整个会话生命周期内使用未取消的 ctx，否则 prompt 被丢、事件停止持久化、权限
	// 审批被强制拒绝。teardown 由 onExit 调 relayCancel + Relay.Close（关 r.done）完成。
	relayCtx, relayCancel := context.WithCancel(context.Background())

	proc := &Process{
		SessionID:      sess.ID,
		WorkspaceID:    sess.WorkspaceID,
		UserID:         sess.UserID,
		IsMain:         sess.IsMainWorktree,
		Cmd:            cmd,
		group:          group,
		Stdin:          stdin,
		Stdout:         stdout,
		Stderr:         stderr,
		Relay:          relay,
		StderrBuf:      newStderrRing(s.cfg.StderrBufferLines),
		StartedAt:      time.Now(),
		done:           make(chan struct{}),
		relayCancel:    relayCancel,
		sandboxCleanup: sandboxCleanup,
	}

	s.mu.Lock()
	s.procs[sess.ID] = proc
	s.mu.Unlock()

	go s.drainStderr(proc)
	go s.monitorProcess(proc)
	go relay.Run(relayCtx, sessCtx)

	// 握手仍走调用方的 spawn ctx，保留 SpawnTimeout 截止语义。
	if err := s.handshake(ctx, sess, kind, relay, proc, extraEnv); err != nil {
		s.log.Error("acp.handshake_failed", "session_id", sess.ID, "err", err.Error())
		s.Kill(ctx, sess.ID, s.cfg.KillGrace)
		return nil, err
	}
	return proc, nil
}

// handshake 发送 initialize → session/new，并把返回的 agent session id + pid
// 持久化到 acp_sessions 行（status → running）。SpawnTimeout 是整个握手的硬上限。
// session/new 同时下发 mcpServers：ACW 自家 MCP（extraEnv 中的 per-session
// token）+ AgentKind 第三方 MCP，按 initialize 声明的 mcpCapabilities 过滤。
func (s *Supervisor) handshake(ctx context.Context, sess *Session, kind *AgentKind,
	relay *Relay, proc *Process, extraEnv map[string]string) error {
	hsCtx, cancel := context.WithTimeout(ctx, s.cfg.SpawnTimeout)
	defer cancel()

	initResp, err := relay.Call(hsCtx, "initialize", handlers.BuildInitializeParams("dev"))
	if err != nil {
		return fmt.Errorf("initialize: %w", err)
	}
	var ir handlers.InitializeResult
	if err := json.Unmarshal(initResp.Result, &ir); err != nil {
		// 能力解析失败不阻断握手：按无 http/sse 能力处理（仅基线 stdio 可下发）。
		s.log.Warn("acp.handshake.parse_initialize_result", "session_id", sess.ID, "err", err.Error())
	}
	httpOK, sseOK := ir.McpCapabilities()

	specs, err := decryptMCPServers(s.crypto, kind.EncryptedMCPServers)
	if err != nil {
		return err
	}
	mcpServers := buildMcpServers(specs, extraEnv[EnvMCPURL], extraEnv[EnvMCPToken],
		httpOK, sseOK, s.log, sess.ID)

	resp, err := relay.Call(hsCtx, "session/new", handlers.BuildSessionNewParams(sess.CwdPath, mcpServers))
	if err != nil {
		return fmt.Errorf("session/new: %w", err)
	}
	var snr handlers.SessionNewResult
	if err := json.Unmarshal(resp.Result, &snr); err != nil {
		return fmt.Errorf("parse session/new result: %w", err)
	}
	if snr.SessionID == "" {
		return fmt.Errorf("agent did not return sessionId")
	}
	pid := int32(proc.Cmd.Process.Pid)
	if err := s.repo.UpdateSessionRunning(ctx, sess.ID, snr.SessionID, pid); err != nil {
		return fmt.Errorf("update session running: %w", err)
	}
	sess.AgentSessionID = &snr.SessionID
	sess.PID = &pid
	sess.Status = SessionRunning
	return nil
}

// BuildSpawnEnv exposes env merge logic for testing.
func (s *Supervisor) BuildSpawnEnv(kind *AgentKind, extraEnv map[string]string) ([]string, error) {
	envMap, err := decryptEnv(s.crypto, kind.EncryptedEnv)
	if err != nil {
		return nil, err
	}
	for k, v := range extraEnv {
		envMap[k] = v
	}
	out := []string{"PATH=" + os.Getenv("PATH")}
	for k, v := range envMap {
		out = append(out, fmt.Sprintf("%s=%s", k, v))
	}
	return out, nil
}

// buildSandboxSpec 从 session + 受管 home 派生本次 spawn 的沙箱 Spec。
// per-session UID = BaseUID + offset，offset 由 session ID 派生（低 16 位），
// 保证不同 session 大概率落在不同 uid（碰撞仅影响隔离粒度，不影响正确性）。
// HomeDir = 受管 home（per-user），WorktreeDir = session cwd，二者为唯一可写路径。
func (s *Supervisor) buildSandboxSpec(sess *Session, agentHome string) sandbox.Spec {
	uid := 0
	if s.cfg.SandboxBaseUID > 0 {
		// 取 session ID 的低 16 位作为 offset，限制 uid 漂移范围。
		b := sess.ID
		offset := int(b[14])<<8 | int(b[15])
		uid = s.cfg.SandboxBaseUID + offset
	}
	return sandbox.Spec{
		Mode:        s.cfg.SandboxMode,
		UID:         uid,
		GID:         uid,
		HomeDir:     agentHome,
		WorktreeDir: sess.CwdPath,
		DataRoot:    s.cfg.DataRoot,
		Rlimits:     s.cfg.SandboxRlimits,
		ProxyURL:    s.cfg.EgressProxyURL,
	}
}

// Kill 终止指定 session 的子进程。在 Unix 平台上先发 SIGTERM 等待 grace 时间，
// 仍未退出再 Process.Kill；Windows 没有可靠 SIGTERM 实现，直接 Process.Kill。
// 始终阻塞直到 monitorProcess 关闭 done channel。Kill 是幂等的——目标 session
// 不存在时立即返回。
func (s *Supervisor) Kill(ctx context.Context, sid uuid.UUID, grace time.Duration) {
	s.mu.RLock()
	proc, ok := s.procs[sid]
	s.mu.RUnlock()
	if !ok {
		return
	}

	proc.KilledByUs.Store(true)

	// 整树终止：Unix 走 SIGTERM(-pgid) + grace + SIGKILL(-pgid)；Windows 关闭
	// Job 句柄触发 KILL_ON_JOB_CLOSE。实现下沉至 internal/infra/proc。
	proc.group.TerminateGroup(proc.done, grace)

	select {
	case <-proc.done:
	case <-ctx.Done():
	}
}

// ShutdownAll 并行 Kill 所有正在运行的子进程。bounded by cfg.ShutdownGrace。
func (s *Supervisor) 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(sid uuid.UUID) {
			defer wg.Done()
			s.Kill(shutdownCtx, sid, s.cfg.KillGrace)
		}(id)
	}

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

// drainStderr 持续读取子进程 stderr，按行追加到 ring buffer。EOF 或 read error
// 时安静退出（cmd.Wait() 已在 monitorProcess 处理）。
func (s *Supervisor) drainStderr(proc *Process) {
	scanner := bufio.NewScanner(proc.Stderr)
	// 默认 token 大小 64KB，足够 ACP agent 单行 log。
	for scanner.Scan() {
		proc.StderrBuf.Append(scanner.Text())
	}
}

// monitorProcess 等待子进程退出，更新 procs map 并触发 onExit。这是子进程
// 状态唯一的合法 close 路径（done channel）。
func (s *Supervisor) monitorProcess(proc *Process) {
	defer close(proc.done)
	waitErr := proc.Cmd.Wait()

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

	s.mu.Lock()
	delete(s.procs, proc.SessionID)
	s.mu.Unlock()

	// 释放进程树句柄（Windows 关闭 Job handle；Unix no-op）。done 已 close、
	// 进程已 Wait 返回，此处 Close 不影响 onExit-must-not-relock 不变量。
	proc.group.Close()

	// 沙箱清理：在 group.Close() 之后调用一次（卸载 bind mount、停容器、删临时目录）。
	// 不持 supervisor.mu，遵守 onExit-must-not-relock 不变量。mode=none 时为 no-op。
	if proc.sandboxCleanup != nil {
		proc.sandboxCleanup()
	}

	status := SessionExited
	if !proc.KilledByUs.Load() {
		status = SessionCrashed
	}
	s.onExit(context.Background(), proc, status, exitCode, waitErr)
}

// onExit 写终止状态、释放 worktree、触发通知与 audit、最后关闭 relay。
// 不能再 acquire s.mu（monitorProcess 已 delete map 后调用本函数，但 Kill
// 也持锁等 done，会形成 monitor → Kill → onExit 的链式 deadlock）。
func (s *Supervisor) onExit(ctx context.Context, proc *Process, status SessionStatus, exitCode int32, waitErr error) {
	if s.metrics != nil {
		s.metrics.RecordSessionExit(string(status))
	}
	stderrTail := proc.StderrBuf.Tail(s.cfg.StderrTailBytes)
	var lastErr *string
	if status == SessionCrashed && stderrTail != "" {
		lastErr = &stderrTail
	}
	if err := s.repo.UpdateSessionFinished(ctx, proc.SessionID, status, &exitCode, lastErr); err != nil {
		s.log.Error("acp.on_exit.update_session", "session_id", proc.SessionID, "err", err.Error())
	}

	// 成本/预算终止：若 terminated_reason 已被预算门或 reaper 写入，发 audit + notify。
	// 读 DB（UpdateSessionFinished 之后）拿最新 terminated_reason；不覆盖既有值。
	if finished, gerr := s.repo.GetSessionByID(ctx, proc.SessionID); gerr == nil && finished != nil &&
		finished.TerminatedReason != nil && *finished.TerminatedReason != "" {
		s.recordBudgetTerminated(ctx, proc.UserID, proc.SessionID, *finished.TerminatedReason,
			finished.TotalCostUSD, finished.PromptTokens+finished.CompletionTokens+finished.ThinkingTokens)
	}

	// Release worktree。主 worktree 走 acp 表自己的 CAS；子 worktree 按 holder
	// 释放（holder = "session:<sid>"，见 workspace.Caller.Holder），与启动 reaper
	// 一致，不需要知道 worktree ID。
	if proc.IsMain {
		if _, err := s.repo.ReleaseMainWorktree(ctx, proc.WorkspaceID, proc.SessionID); err != nil {
			s.log.Error("acp.on_exit.release_main",
				"session_id", proc.SessionID, "err", err.Error())
		}
	} else if s.wtSvc != nil {
		if _, err := s.wtSvc.ReleaseByHolder(ctx, "session:"+proc.SessionID.String()); err != nil {
			s.log.Error("acp.on_exit.release_worktree",
				"session_id", proc.SessionID, "err", err.Error())
		}
	}

	if status == SessionCrashed {
		if s.notify != nil {
			_ = s.notify.Dispatch(ctx, notify.Message{
				UserID:   proc.UserID,
				Topic:    "acp.session_crashed",
				Severity: notify.SeverityError,
				Title:    "ACP session crashed",
				Body:     fmt.Sprintf("Session %s exited with code %d.", proc.SessionID, exitCode),
				Metadata: map[string]any{
					"session_id": proc.SessionID.String(),
					"exit_code":  exitCode,
				},
			})
		}
		if s.audit != nil {
			uid := proc.UserID
			_ = s.audit.Record(ctx, audit.Entry{
				UserID:     &uid,
				Action:     "acp.session.crashed",
				TargetType: "acp_session",
				TargetID:   proc.SessionID.String(),
				Metadata: map[string]any{
					"exit_code":  exitCode,
					"last_error": stderrTail,
				},
			})
		}
	} else {
		if s.audit != nil {
			uid := proc.UserID
			_ = s.audit.Record(ctx, audit.Entry{
				UserID:     &uid,
				Action:     "acp.session.terminate",
				TargetType: "acp_session",
				TargetID:   proc.SessionID.String(),
				Metadata:   map[string]any{"exit_code": exitCode},
			})
		}
	}
	_ = waitErr // 已通过 KilledByUs / ExitError 编码到 status / exitCode

	// 终止会话的全部挂起权限请求：标 expired 并唤醒阻塞的 Decide goroutine，
	// 避免 agent 端永久挂起。须在 Relay.Close 之前（Close 后 FanoutControl 静默丢弃）。
	if s.permSvc != nil {
		s.permSvc.CancelSession(ctx, proc.SessionID)
	}

	// Revoke session's system MCP tokens (spec §6.3)
	if s.mcpTokens != nil {
		if err := s.mcpTokens.RevokeBySession(ctx, proc.SessionID); err != nil {
			s.log.Error("acp.on_exit.mcp_revoke",
				"session_id", proc.SessionID, "err", err.Error())
		}
	}

	// 回合检测：闭合任何开放的 in_progress 回合并发布 session_idle，确保崩溃/被杀时
	// 编排层也能收到空闲信号。Abort 只触碰 repo+bus（不持 supervisor.mu），且必须在
	// Relay.Close 之前——Close 后 tracker 仍可用，但语义上回合应在 teardown 前闭合。
	if tracker := proc.Relay.Tracker(); tracker != nil {
		reason := string(StopCancelled)
		if status == SessionCrashed {
			reason = "crashed"
		}
		tracker.Abort(ctx, reason)
	}

	// 取消 relay 的会话级 context：停止 reader/writer，并让仍阻塞在 Decide / 已派生
	// 的 handleAgentRequest goroutine 收到 ctx.Done 后退出。在 Relay.Close 之前调，
	// 二者共同完成 teardown（relayCancel 停 ctx 派生的 goroutine，Close 关 r.done +
	// 广播 session_terminated）。relayCancel 在早期 spawn 失败路径也可能为 nil。
	if proc.relayCancel != nil {
		proc.relayCancel()
	}

	// 最后关 relay：广播 session_terminated 给 WS subscribers。
	proc.Relay.Close(string(status), &exitCode)

	// 编排器再驱动：若该 session 由编排器某 step 驱动且本次为崩溃退出，把对应 step
	// 重新入队（带 backoff），让回合在同一 worktree 上恢复。只做 DB 读 + jobs Enqueue，
	// 不持 supervisor 锁（保持 onExit-must-not-relock 不变量）。
	if status == SessionCrashed && s.orchRedrive != nil {
		if sess, gerr := s.repo.GetSessionByID(ctx, proc.SessionID); gerr == nil &&
			sess != nil && sess.OrchestratorStepID != nil {
			s.orchRedrive(ctx, *sess.OrchestratorStepID, "session_crashed")
		}
	}
}

// recordBudgetTerminated 在会话因预算/reaper 被强制终止时发 audit + owner notify。
func (s *Supervisor) recordBudgetTerminated(ctx context.Context, userID, sessionID uuid.UUID, reason string, totalCost float64, totalTokens int64) {
	if s.audit != nil {
		uid := userID
		_ = s.audit.Record(ctx, audit.Entry{
			UserID:     &uid,
			Action:     "acp.session.budget_exceeded",
			TargetType: "acp_session",
			TargetID:   sessionID.String(),
			Metadata: map[string]any{
				"reason":         reason,
				"total_cost_usd": totalCost,
				"total_tokens":   totalTokens,
			},
		})
	}
	if s.notify != nil {
		_ = s.notify.Dispatch(ctx, notify.Message{
			UserID:   userID,
			Topic:    "acp.session_budget_exceeded",
			Severity: notify.SeverityWarning,
			Title:    "ACP session terminated (budget/reaper)",
			Body:     fmt.Sprintf("Session %s was terminated: %s.", sessionID, reason),
			Metadata: map[string]any{
				"session_id":     sessionID.String(),
				"reason":         reason,
				"total_cost_usd": totalCost,
			},
		})
	}
}