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

// permission_service.go 实装 ACP 工具调用权限审批框架。
//
// 代理通过 session/request_permission 发起工具调用授权请求，relay 为每个请求起
// 独立 goroutine 调用 PermissionHandler.Handle → PermissionService.Decide。Decide
// 分两条路径：
//   - 命中 agent kind 的 tool_allowlist：自动放行（status=auto），立即返回。
//   - 未命中：落库为 pending、通过 WS 推送 permission_request 事件、阻塞等待人工
//     决策（Resolve）/ 超时（默认拒绝）/ 会话终止（CancelSession）。
//
// 阻塞安全：relay.handleAgentRequest 对每个 agent 请求起独立 goroutine，因此在
// Decide 内阻塞不会卡住 reader 主循环；ctx 来自该请求的 relay.Run，会话关闭时取消。
package acp

import (
	"context"
	"encoding/json"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"

	"github.com/yan1h/agent-coding-workflow/internal/acp/handlers"
	"github.com/yan1h/agent-coding-workflow/internal/audit"
	"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
	"github.com/yan1h/agent-coding-workflow/internal/infra/notify"
)

// DefaultPermissionTimeout 是未配置时的人工决策超时；到期默认拒绝。
const DefaultPermissionTimeout = 5 * time.Minute

// permissionTopic 是 HITL 待审权限通知的 topic（inbox + 跨通道路由用）。
const permissionTopic = "acp.permission_pending"

// PermissionNotifier 是 PermissionService 投递"待人工审批"通知所需的窄接口。
// notify.Dispatcher 满足它；用窄接口便于测试断言 Dispatch 调用。
type PermissionNotifier interface {
	Dispatch(ctx context.Context, msg notify.Message) error
}

// PermissionMetrics 是 PermissionService 记录决策计数所需的窄接口。
// metrics.Metrics 满足它；nil 时不记录（部分测试 / metrics 关闭）。
type PermissionMetrics interface {
	RecordPermissionDecision(outcome string)
}

// RelayLocator 让 PermissionService 按 session id 取到 *Relay 以推送 WS 事件。
// Supervisor 满足该接口（GetRelay）。用窄接口避免 permSvc 依赖整个 Supervisor。
type RelayLocator interface {
	GetRelay(sid uuid.UUID) *Relay
}

// ProjectMaintainer 让 PermissionService 判定某用户是否对会话所属 project 有写权限
// （project maintainer：owner / global-admin / project admin|member 角色）。允许任意
// project maintainer 处理待审权限请求，而不仅是会话 owner（autonomy roadmap §11）。
// app 层用 projectAccessAdapter.ResolveByID 实现；nil 时退化为仅 owner/admin。
type ProjectMaintainer interface {
	// CanWriteProject 报告 (callerID, isAdmin) 是否对 projectID 有写权限。
	CanWriteProject(ctx context.Context, callerID uuid.UUID, isAdmin bool, projectID uuid.UUID) (bool, error)
}

// decision 是投递给阻塞中 Decide 的人工/系统决策结果。
type decision struct {
	optionID string
}

// waiter 关联一个挂起的权限请求与其所属会话，便于会话终止时定向清理。
type waiter struct {
	ch        chan decision
	sessionID uuid.UUID
}

// PermissionService 是进程级单例，注入到所有 relay 的 PermissionHandler。
type PermissionService struct {
	repo    Repository
	audit   audit.Recorder
	log     *slog.Logger
	timeout time.Duration

	mu      sync.Mutex
	waiters map[uuid.UUID]*waiter // requestID -> waiter

	locMu sync.RWMutex
	loc   RelayLocator

	// HITL：待审权限通知 + 决策计数。装配期经 SetNotifier/SetMetrics 注入，
	// 二者均可为 nil（部分测试 / metrics 关闭）。
	notify  PermissionNotifier
	metrics PermissionMetrics

	// maintainer 允许任意 project maintainer（非仅会话 owner）处理待审权限请求。
	// 装配期经 SetProjectMaintainer 注入；nil 时退化为仅 owner/global-admin。
	maintainer ProjectMaintainer
}

// SetNotifier 注入待审权限通知投递器（装配期，best-effort）。
func (s *PermissionService) SetNotifier(n PermissionNotifier) { s.notify = n }

// SetMetrics 注入权限决策计数器（装配期）。
func (s *PermissionService) SetMetrics(m PermissionMetrics) { s.metrics = m }

// SetProjectMaintainer 注入 project maintainer 判定器（装配期）。注入后，会话
// 所属 project 的任意 maintainer 均可处理该会话的待审权限请求。
func (s *PermissionService) SetProjectMaintainer(m ProjectMaintainer) { s.maintainer = m }

// NewPermissionService 构造 PermissionService。timeout<=0 时用 DefaultPermissionTimeout。
func NewPermissionService(repo Repository, rec audit.Recorder, timeout time.Duration, log *slog.Logger) *PermissionService {
	if log == nil {
		log = slog.Default()
	}
	if timeout <= 0 {
		timeout = DefaultPermissionTimeout
	}
	return &PermissionService{
		repo:    repo,
		audit:   rec,
		log:     log,
		timeout: timeout,
		waiters: map[uuid.UUID]*waiter{},
	}
}

// SetRelayLocator 注入 relay 定位器。因 Supervisor 与 PermissionService 互相依赖，
// 装配时先构造二者再回填（与 workspace CloneRunner.SetScheduler 同模式）。
func (s *PermissionService) SetRelayLocator(loc RelayLocator) {
	s.locMu.Lock()
	s.loc = loc
	s.locMu.Unlock()
}

// Decide 实现 handlers.PermissionDecider。返回选中的 option id；空串表示拒绝。
func (s *PermissionService) Decide(ctx context.Context, sess handlers.SessionContext, agentReqID string,
	toolCall json.RawMessage, options []handlers.PermOption) string {

	toolName := extractToolName(toolCall)
	optionsJSON, _ := json.Marshal(options)

	// 路径一：白名单命中 → 自动放行。
	if allowlistHit(sess.ToolAllowlist, toolName) {
		chosen := chooseAllowOption(options)
		_, err := s.repo.InsertPermissionRequest(ctx, &PermissionRequest{
			ID:             uuid.New(),
			SessionID:      sess.SessionID,
			AgentRequestID: agentReqID,
			ToolName:       toolName,
			ToolCall:       jsonOrNull(toolCall),
			Options:        optionsJSON,
			Status:         PermissionAuto,
			ChosenOptionID: strPtrOrNil(chosen),
		})
		if err != nil {
			s.log.Error("acp.permission.insert_auto", "session_id", sess.SessionID, "err", err.Error())
		}
		s.recordDecision(ctx, sess.UserID, sess.SessionID, toolName, "auto", nil)
		s.recordMetric("auto")
		return chosen
	}

	// 路径二：落库 pending 并阻塞等待决策。
	reqID := uuid.New()
	row, err := s.repo.InsertPermissionRequest(ctx, &PermissionRequest{
		ID:             reqID,
		SessionID:      sess.SessionID,
		AgentRequestID: agentReqID,
		ToolName:       toolName,
		ToolCall:       jsonOrNull(toolCall),
		Options:        optionsJSON,
		Status:         PermissionPending,
	})
	if err != nil {
		// 落库失败时无法人工审批，保守默认拒绝。
		s.log.Error("acp.permission.insert_pending", "session_id", sess.SessionID, "err", err.Error())
		return ""
	}

	w := &waiter{ch: make(chan decision, 1), sessionID: sess.SessionID}
	s.mu.Lock()
	s.waiters[reqID] = w
	s.mu.Unlock()
	defer s.removeWaiter(reqID)

	s.fanout(sess.SessionID, &EventEnvelope{
		Kind:    "permission_request",
		Payload: mustJSON(permissionRequestPayload(row)),
	})

	// HITL：在阻塞前 best-effort 推送 inbox 通知，使关闭页面的用户也能被提醒去
	// /approvals 审批（fire-and-forget，错误不影响 deny-on-timeout 安全默认）。
	s.notifyPending(ctx, sess, row, toolName)

	select {
	case d := <-w.ch:
		return d.optionID
	case <-time.After(s.timeout):
		// 超时 → CAS 标记 expired（系统决策，decided_by=NULL）。
		if _, ok, _ := s.repo.DecidePermissionRequest(context.WithoutCancel(ctx), reqID, PermissionExpired, nil, nil); ok {
			s.fanout(sess.SessionID, resolvedEnvelope(reqID, PermissionExpired, ""))
			s.recordMetric("expired")
		}
		return ""
	case <-ctx.Done():
		// relay/进程关闭：CancelSession 已处理 DB；这里仅退出，agent 收默认拒绝。
		return ""
	}
}

// Resolve 落地一次人工决策（approve/deny）。鉴权：session owner、global-admin，或
// 会话所属 project 的任意 maintainer（autonomy roadmap §11）。
func (s *PermissionService) Resolve(ctx context.Context, c Caller, reqID uuid.UUID, approve bool, optionID string) error {
	req, err := s.repo.GetPermissionRequestByID(ctx, reqID)
	if err != nil {
		return err
	}
	sessRow, err := s.repo.GetSessionByID(ctx, req.SessionID)
	if err != nil {
		return err
	}
	if !s.canManageSession(ctx, c, sessRow) {
		return errs.New(errs.CodeAcpSessionNotOwned, "无权处理该会话的权限请求")
	}

	var options []handlers.PermOption
	_ = json.Unmarshal(req.Options, &options)

	var chosen string
	var status PermissionStatus
	if approve {
		chosen = optionID
		if chosen == "" {
			chosen = chooseAllowOption(options)
		} else if !optionExists(options, chosen) {
			return errs.New(errs.CodeAcpPermissionInvalidOption, "无效的选项 id")
		}
		status = PermissionApproved
	} else {
		chosen = chooseRejectOption(options)
		status = PermissionDenied
	}

	decidedBy := c.UserID
	_, ok, err := s.repo.DecidePermissionRequest(ctx, reqID, status, strPtrOrNil(chosen), &decidedBy)
	if err != nil {
		return err
	}
	if !ok {
		// 已被超时/会话终止/重复决策抢先落地。
		return errs.New(errs.CodeAcpPermissionAlreadyDecided, "该权限请求已被处理")
	}

	s.wake(reqID, decision{optionID: chosen})
	s.fanout(req.SessionID, resolvedEnvelope(reqID, status, chosen))
	s.recordDecision(ctx, c.UserID, req.SessionID, req.ToolName, string(status), &c.UserID)
	if approve {
		s.recordMetric("approved")
	} else {
		s.recordMetric("denied")
	}
	return nil
}

// ListPendingForUser 返回该用户跨所有会话的待审权限请求（HITL 跨会话审批 inbox）。
// 包装 repo.ListPendingPermissionRequestsByUser（已 JOIN acp_sessions.user_id）。
// admin 仅看自己拥有会话的待审项（与现有 inbox 语义一致；全局视图走 dashboard）。
func (s *PermissionService) ListPendingForUser(ctx context.Context, c Caller) ([]*PermissionRequest, error) {
	return s.repo.ListPendingPermissionRequestsByUser(ctx, c.UserID)
}

// ListPendingBySession 返回某会话的待审请求。鉴权同 Resolve。
func (s *PermissionService) ListPendingBySession(ctx context.Context, c Caller, sessionID uuid.UUID) ([]*PermissionRequest, error) {
	sessRow, err := s.repo.GetSessionByID(ctx, sessionID)
	if err != nil {
		return nil, err
	}
	if !s.canManageSession(ctx, c, sessRow) {
		return nil, errs.New(errs.CodeAcpSessionNotOwned, "无权查看该会话")
	}
	return s.repo.ListPendingPermissionRequestsBySession(ctx, sessionID)
}

// canManageSession 报告 caller 是否可处理/查看该会话的权限请求：session owner、
// global-admin，或会话所属 project 的任意 maintainer（注入 ProjectMaintainer 时）。
// maintainer 查询失败按"无权"处理（fail-closed），但 owner/admin 短路不受影响。
func (s *PermissionService) canManageSession(ctx context.Context, c Caller, sess *Session) bool {
	if c.IsAdmin || sess.UserID == c.UserID {
		return true
	}
	if s.maintainer == nil {
		return false
	}
	ok, err := s.maintainer.CanWriteProject(ctx, c.UserID, c.IsAdmin, sess.ProjectID)
	if err != nil {
		return false
	}
	return ok
}

// CancelSession 在会话终止时把其全部 pending 请求标 expired，并唤醒阻塞的 Decide
// goroutine 使其立即返回（agent 收默认拒绝）。supervisor.onExit 调用。
func (s *PermissionService) CancelSession(ctx context.Context, sessionID uuid.UUID) {
	if _, err := s.repo.ExpirePendingPermissionRequestsBySession(ctx, sessionID); err != nil {
		s.log.Error("acp.permission.cancel_session", "session_id", sessionID, "err", err.Error())
	}
	s.mu.Lock()
	for reqID, w := range s.waiters {
		if w.sessionID == sessionID {
			select {
			case w.ch <- decision{optionID: ""}:
			default:
			}
			delete(s.waiters, reqID)
		}
	}
	s.mu.Unlock()
}

// ExpireStale 把超过 before 仍 pending 的请求标 expired（启动 reaper / 兜底）。
func (s *PermissionService) ExpireStale(ctx context.Context, before time.Time) (int64, error) {
	return s.repo.ExpireStalePermissionRequests(ctx, before)
}

// ===== 内部辅助 =====

func (s *PermissionService) removeWaiter(reqID uuid.UUID) {
	s.mu.Lock()
	delete(s.waiters, reqID)
	s.mu.Unlock()
}

func (s *PermissionService) wake(reqID uuid.UUID, d decision) {
	s.mu.Lock()
	w, ok := s.waiters[reqID]
	if ok {
		delete(s.waiters, reqID)
	}
	s.mu.Unlock()
	if ok {
		select {
		case w.ch <- d:
		default:
		}
	}
}

func (s *PermissionService) fanout(sessionID uuid.UUID, env *EventEnvelope) {
	s.locMu.RLock()
	loc := s.loc
	s.locMu.RUnlock()
	if loc == nil {
		return
	}
	if relay := loc.GetRelay(sessionID); relay != nil {
		relay.FanoutControl(env)
	}
}

func (s *PermissionService) recordDecision(ctx context.Context, actor, sessionID uuid.UUID, toolName, outcome string, decidedBy *uuid.UUID) {
	if s.audit == nil {
		return
	}
	a := actor
	_ = s.audit.Record(context.WithoutCancel(ctx), audit.Entry{
		UserID:     &a,
		Action:     "acp.permission." + outcome,
		TargetType: "acp_session",
		TargetID:   sessionID.String(),
		Metadata: map[string]any{
			"tool_name":  toolName,
			"decided_by": decidedBy,
		},
	})
}

// recordMetric 记录一次权限决策计数（auto/approved/denied/expired）。metrics 为 nil 时 no-op。
func (s *PermissionService) recordMetric(outcome string) {
	if s.metrics == nil {
		return
	}
	s.metrics.RecordPermissionDecision(outcome)
}

// notifyPending best-effort 向会话所有者推送"待人工审批"通知。错误仅记日志，
// 绝不影响 Decide 的阻塞 / deny-on-timeout 安全默认。
func (s *PermissionService) notifyPending(ctx context.Context, sess handlers.SessionContext, row *PermissionRequest, toolName string) {
	if s.notify == nil {
		return
	}
	if err := s.notify.Dispatch(context.WithoutCancel(ctx), notify.Message{
		UserID:   sess.UserID,
		Topic:    permissionTopic,
		Severity: notify.SeverityWarning,
		Title:    "Approval needed",
		Body:     "An agent is requesting permission to run a tool and needs your approval.",
		Link:     "/approvals",
		Metadata: map[string]any{
			"request_id": row.ID.String(),
			"session_id": sess.SessionID.String(),
			"tool_name":  toolName,
		},
	}); err != nil {
		s.log.Warn("acp.permission.notify_pending_failed", "session_id", sess.SessionID, "err", err.Error())
	}
}

func permissionRequestPayload(p *PermissionRequest) map[string]any {
	return map[string]any{
		"request_id": p.ID.String(),
		"session_id": p.SessionID.String(),
		"tool_name":  p.ToolName,
		"tool_call":  json.RawMessage(p.ToolCall),
		"options":    json.RawMessage(p.Options),
		"status":     string(p.Status),
		"created_at": p.CreatedAt,
	}
}

func resolvedEnvelope(reqID uuid.UUID, status PermissionStatus, chosen string) *EventEnvelope {
	return &EventEnvelope{
		Kind: "permission_resolved",
		Payload: mustJSON(map[string]any{
			"request_id":       reqID.String(),
			"status":           string(status),
			"chosen_option_id": chosen,
		}),
	}
}

// extractToolName 尝试从 toolCall JSON 中提取真实工具标识。
//
// 安全约束（自动放行依据）：只接受真实工具名字段（name / toolName）。绝不回退到
// ACP toolCall 的 'kind'（粗粒度枚举 read/edit/execute…）或 'title'（自由文本）——
// 二者都由不可信 agent 自行填写，agent 可把一个 execute/shell 调用标成 kind='read'
// 或冒用白名单里的 title 来绕过人工审批闸门。取不到真实名时返回空串，使 allowlistHit
// 一律不放行（保守走人工审批），同时保留空名默认拒绝的语义。
func extractToolName(toolCall json.RawMessage) string {
	if len(toolCall) == 0 {
		return ""
	}
	var m map[string]any
	if err := json.Unmarshal(toolCall, &m); err != nil {
		return ""
	}
	for _, k := range []string{"name", "toolName"} {
		if v, ok := m[k].(string); ok && v != "" {
			return v
		}
	}
	return ""
}

// allowlistHit 判断工具是否在白名单内。含 "*" 表示放行全部可识别的工具；
// toolName 为空（无法识别的 tool call）时一律不放行，保守走人工审批。
func allowlistHit(allowlist []string, toolName string) bool {
	if toolName == "" {
		return false
	}
	for _, a := range allowlist {
		if a == "*" || a == toolName {
			return true
		}
	}
	return false
}

// chooseAllowOption 选一个“批准”类选项 id：优先包含 allow/yes 的，否则取首个。
func chooseAllowOption(options []handlers.PermOption) string {
	if id := matchOption(options, []string{"allow", "yes", "approve", "accept"}); id != "" {
		return id
	}
	if len(options) > 0 {
		return options[0].ID
	}
	return ""
}

// chooseRejectOption 选一个“拒绝”类选项 id：优先包含 reject/deny/no 的，否则空串。
func chooseRejectOption(options []handlers.PermOption) string {
	return matchOption(options, []string{"reject", "deny", "no"})
}

func matchOption(options []handlers.PermOption, keywords []string) string {
	for _, opt := range options {
		lc := strings.ToLower(opt.ID)
		for _, kw := range keywords {
			if strings.Contains(lc, kw) {
				return opt.ID
			}
		}
	}
	return ""
}

func optionExists(options []handlers.PermOption, id string) bool {
	for _, opt := range options {
		if opt.ID == id {
			return true
		}
	}
	return false
}

func strPtrOrNil(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}

func jsonOrNull(b json.RawMessage) []byte {
	if len(b) == 0 {
		return []byte("null")
	}
	return b
}