从零搭建 Agent Harness 系列（十三）通用工具审批与 Human-in-the-loop

发表于 分类于 Waline：
从 go-tiny-claw 当前源码出发，设计终端可用、渠道可替换的工具审批协议，并完成先审批后并发执行。

Coding Agent 和普通聊天机器人最大的区别之一，是它可以改变真实环境。读取文件通常是低风险操作；执行任意 bash、覆盖文件、修改代码则可能带来不可逆后果。

本文源码版本对应提交：8a13d9a876a86a08c2ee43171cc2a108412f18ec（add approval）。

因此，工具调用不能从模型响应直接跳到 registry.Execute。中间需要一个可审计、可替换、可取消的 Human-in-the-loop 层。本篇实现当前仓库里的 Approval 协议，并分析它距离生产级还差哪些能力。

一、为什么不把审批写进 Tool 或 Middleware

工具本身应该关注参数解析和业务执行。Registry 的 Middleware 适合做确定性的拦截，例如参数校验、路径限制、命令黑名单；但“等待用户输入”属于交互流程。

当前工具注册接口是：

1
2
3
4
5
6
7
type BaseTool interface {
	Name() string
	Definition() schema.ToolDefinition
	Execute(ctx context.Context, args json.RawMessage) (string, error)
}

type MiddlewareFunc func(ctx context.Context, call schema.ToolCall) (allowed bool, rejectReason string)

如果把审批放在 Middleware 中，Registry 执行工具时才询问用户，会出现三个问题：

  1. Engine 无法在执行前统一展示所有待审批操作。
  2. 并发执行时多个 Middleware 可能同时读取 stdin。
  3. 渠道交互逻辑进入工具注册层，未来难以接 Web 或其他渠道。

所以审批应该是 Engine 和 Registry 之间的独立阶段：

1
2
3
4
5
6
7
模型返回完整 ToolCall
        ↓
Approval Gate
        ↓
获准的 ToolCall
        ↓
Registry.Execute

二、先定义渠道无关的协议

internal/approval/interface.go 当前定义了三类决策：

1
2
3
4
5
6
7
type Decision string

const (
	AllowOnce    Decision = "allow_once"
	AllowSession Decision = "allow_session"
	Deny         Decision = "deny"
)

它们分别表示：

1
2
3
AllowOnce：只允许当前 ToolCall
AllowSession：当前会话中继续允许该工具
Deny：拒绝当前 ToolCall

风险级别也是协议的一部分：

1
2
3
4
5
6
7
type RiskLevel string

const (
	RiskSafe      RiskLevel = "safe"
	RiskMutating  RiskLevel = "mutating"
	RiskDangerous RiskLevel = "dangerous"
)

一次审批请求携带完整上下文：

1
2
3
4
5
6
7
8
9
10
11
12
13
type Request struct {
	ID        string
	SessionID string
	ToolCall  schema.ToolCall
	Risk      RiskLevel
	Reason    string
	CreatedAt time.Time
	ExpiresAt time.Time
}

type Handler interface {
	Approve(ctx context.Context, request Request) (Decision, error)
}

Handler 不知道 Agent 如何运行，也不需要知道 Registry。它只负责把请求交给具体渠道，并返回标准 Decision。

三、RiskLevel 应该由工具声明

当前 internal/tools/interface.go 通过独立接口表达工具风险：

1
2
3
type RiskedTool interface {
	RiskLevel() approval.RiskLevel
}

Registry 查询工具风险时，如果工具不存在或没有实现 RiskedTool，默认按危险处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
func (r *registryImpl) GetRiskLevel(name string) approval.RiskLevel {
	tool, exists := r.tools[name]
	if !exists {
		return approval.RiskDangerous
	}

	riskedTool, ok := tool.(RiskedTool)
	if !ok {
		return approval.RiskDangerous
	}

	return riskedTool.RiskLevel()
}

这是一种 fail closed 策略：未知工具不能因为缺少风险声明而自动放行。

当前内置工具的实际风险声明是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func (t *ReadFileTool) RiskLevel() approval.RiskLevel {
	return approval.RiskSafe
}

func (t *WriteFileTool) RiskLevel() approval.RiskLevel {
	return approval.RiskMutating
}

func (t *EditFileTool) RiskLevel() approval.RiskLevel {
	return approval.RiskMutating
}

func (t *BashTool) RiskLevel() approval.RiskLevel {
	return approval.RiskDangerous
}

风险是工具能力的粗粒度分类，不是最终授权结果。即使是 RiskSafe，仍然可以被 Middleware 的路径检查拒绝。

四、Policy 决定默认行为

Policy 不直接询问用户，它只返回策略结果：

1
2
3
4
5
6
7
8
9
10
11
type PolicyDecision string

const (
	PolicyAutoAllow PolicyDecision = "auto_allow"
	PolicyAsk       PolicyDecision = "ask"
	PolicyDeny      PolicyDecision = "deny"
)

type Policy interface {
	Evaluate(Request) PolicyDecision
}

当前默认策略是：

1
2
3
4
5
6
7
8
9
10
11
12
type DefaultPolicy struct{}

func (DefaultPolicy) Evaluate(req Request) PolicyDecision {
	switch req.Risk {
	case RiskSafe:
		return PolicyAutoAllow
	case RiskMutating, RiskDangerous:
		return PolicyAsk
	default:
		return PolicyDeny
	}
}

这个设计把“默认规则”和“用户本次选择”分开了。未来可以增加：

1
2
3
4
只允许 workspace 内的写入
只允许 go test、go fmt 等命令自动执行
生产环境全部 Dangerous 自动拒绝
开发环境允许某些命令进入会话授权

而不用修改 Handler 或 Engine。

五、GrantStore 为什么需要单独抽象

AllowSession 不是一次性结果，它需要被保存。当前接口是：

1
2
3
4
5
type GrantStore interface {
	Has(ctx context.Context, request Request) (bool, error)
	Save(ctx context.Context, grant Grant) error
	Revoke(ctx context.Context, grant Grant) error
}

GrantStore 是能力接口，MemoryGrantStore 是一个具体实现。二者不能混为一谈：

1
2
GrantStore：Engine/Gate 依赖的抽象
MemoryGrantStore：开发阶段的内存实现

当前内存实现用会话和工具名作为索引：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
type Grant struct {
	SessionID string
	ToolName  string
	WorkDir   string
	ExpiresAt time.Time
}

type grantKey struct {
	sessionID string
	toolName  string
}

type MemoryGrantStore struct {
	mu     sync.RWMutex
	grants map[grantKey]Grant
}

func NewMemoryGrantStore() *MemoryGrantStore {
	return &MemoryGrantStore{
		grants: make(map[grantKey]Grant),
	}
}

sync.RWMutex 的零值可以直接使用，因此不需要显式初始化 mu；需要初始化的是 map，否则写入时会 panic。

六、实现过期授权和并发安全

当前 Has 会检查 Context、读取授权，并删除已经过期的 Grant：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func (s *MemoryGrantStore) Has(ctx context.Context, request Request) (bool, error) {
	if err := ctx.Err(); err != nil {
		return false, err
	}

	key := grantKey{
		sessionID: request.SessionID,
		toolName:  request.ToolCall.Name,
	}

	s.mu.Lock()
	defer s.mu.Unlock()

	grant, exists := s.grants[key]
	if !exists {
		return false, nil
	}

	if !grant.ExpiresAt.IsZero() &&
		!time.Now().Before(grant.ExpiresAt) {
		delete(s.grants, key)
		return false, nil
	}

	return true, nil
}

这里使用写锁而不是读锁，是因为读取过程中可能清理过期记录。Save 会校验必要字段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func (s *MemoryGrantStore) Save(ctx context.Context, grant Grant) error {
	if err := ctx.Err(); err != nil {
		return err
	}

	if grant.SessionID == "" {
		return fmt.Errorf("grant session ID 不能为空")
	}
	if grant.ToolName == "" {
		return fmt.Errorf("grant tool name 不能为空")
	}

	key := grantKey{
		sessionID: grant.SessionID,
		toolName:  grant.ToolName,
	}

	s.mu.Lock()
	s.grants[key] = grant
	s.mu.Unlock()
	return nil
}

七、Gate 统一编排审批流程

internal/approval/gate.go 把 Policy、GrantStore 和 Handler 串起来：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
type Gate struct {
	policy  Policy
	handler Handler
	grants  GrantStore
}

func (g *Gate) Check(ctx context.Context, request Request) (Decision, error) {
	ploicyDecision := g.policy.Evaluate(request)

	switch ploicyDecision {
	case PolicyAutoAllow:
		return AllowOnce, nil
	case PolicyDeny:
		return Deny, nil
	}

	allowed, err := g.grants.Has(ctx, request)
	if err != nil {
		return "", err
	}
	if allowed {
		return AllowOnce, nil
	}

	decision, err := g.handler.Approve(ctx, request)
	if err != nil {
		return "", err
	}

	if decision == AllowSession {
		err = g.grants.Save(ctx, Grant{
			SessionID: request.SessionID,
			ToolName:  request.ToolCall.Name,
		})
	}

	return decision, err
}

上面保留了当前源码中的 ploicyDecision 拼写；它不影响逻辑，但正式重构时应改为 policyDecision，避免误导后续维护者。

流程顺序是：

1
2
3
4
5
6
PolicyAutoAllow → AllowOnce
PolicyDeny      → Deny
PolicyAsk       → 查询已有 Grant
已有 Grant      → AllowOnce
没有 Grant      → 调用 Handler
AllowSession    → 保存 Grant

Gate 不执行工具，也不负责输出日志。这样可以对它做纯单元测试。

八、Engine 必须先审批，再并发执行

当前 internal/engine/loop.go 使用 IndexedToolCall 保存模型原始顺序：

1
2
3
4
type IndexedToolCall struct {
	index int
	call  schema.ToolCall
}

审批阶段是串行的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
observationMsgs := make([]schema.Message, len(actionResp.ToolCalls))
approvedCalls := make([]IndexedToolCall, 0, len(actionResp.ToolCalls))
toolResults := make([]schema.ToolResult, len(actionResp.ToolCalls))
var wg sync.WaitGroup

for i, toolCall := range actionResp.ToolCalls {
	req := approval.Request{
		ID:        approval.NewRequestID(),
		SessionID: session.ID,
		ToolCall:  toolCall,
		Risk:      e.registry.GetRiskLevel(toolCall.Name),
	}

	decision, err := e.approvalGate.Check(ctx, req)
	if err != nil {
		return fmt.Errorf("审批请求失败: %w", err)
	}

	if decision == approval.Deny {
		observationMsgs[i] = schema.Message{
			Role:       schema.RoleUser,
			Content:    fmt.Sprintf("工具调用被拒绝: %s", toolCall.Name),
			ToolCallID: toolCall.ID,
		}
		continue
	}

	approvedCalls = append(approvedCalls, IndexedToolCall{
		index: i,
		call:  toolCall,
	})
}

只有被批准的调用才进入并发阶段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
for _, item := range approvedCalls {
	wg.Add(1)
	go func(item IndexedToolCall) {
		defer wg.Done()

		if reporter != nil {
			reporter.OnToolCall(ctx, item.call.Name, string(item.call.Arguments))
		}

		result := e.registry.Execute(ctx, item.call)
		if reporter != nil {
			displayOutput := result.Output
			if len(displayOutput) > 200 {
				displayOutput = displayOutput[:200] + "... (已截断)"
			}
			reporter.OnToolResult(ctx, item.call.Name, displayOutput, result.IsError)
		}

		observationMsgs[item.index] = schema.Message{
			Role:       schema.RoleUser,
			Content:    result.Output,
			ToolCallID: item.call.ID,
		}
		toolResults[item.index] = result
	}(item)
}

wg.Wait()
session.Append(observationMsgs...)

为什么不能审批和执行一起并发？因为终端 Handler 会从同一个 stdin 读取，多个 Goroutine 会互相抢输入，用户无法知道当前回答对应哪个请求。

为什么要保留下标？因为工具完成顺序是不确定的，但模型下一轮需要看到与 ToolCall 顺序对应的 Observation。

九、Terminal Handler 只是一个适配器

internal/approval/terminal_approval.go 当前实现了最小可用的终端交互：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func (h *TerminalApprovalHandler) Approve(ctx context.Context, request Request) (Decision, error) {
	fmt.Fprintf(h.out, "\n需要确认执行工具: %s\n", request.ToolCall.Name)
	fmt.Fprintf(h.out, "参数: %s\n", request.ToolCall.Arguments)
	fmt.Fprint(h.out, "[y]允许本次 [a]本会话允许 [n]拒绝: ")

	inputCh := make(chan string, 1)

	go func() {
		line, _ := h.reader.ReadString('\n')
		inputCh <- strings.TrimSpace(line)
	}()

	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case input := <-inputCh:
		switch strings.ToLower(input) {
		case "y":
			return AllowOnce, nil
		case "a":
			return AllowSession, nil
		default:
			return Deny, nil
		}
	}
}

它只实现 Handler，因此 Gate 不关心这是 Terminal、HTTP 还是其他渠道。

十、main.go 中完成依赖组装

当前 cmd/claw/main.go 使用同一个 Reader 组装审批器和 REPL：

1
2
3
4
5
6
7
8
9
10
11
reader := bufio.NewReader(os.Stdin)

handler := approval.NewTerminalApprovalHandler(reader, os.Stdout)
grantStore := approval.NewMemoryGrantStore()
gate := approval.NewGate(approval.DefaultPolicy{}, handler, grantStore)

eng := engine.NewAgentEngine(llmProvider, registry, gate, false, false)
	reporter := engine.NewTerminalReporter()
	sess := ctxpkg.GlobalSessionMgr.GetOrCreate("terminal_default", workDir)

repl := cli.NewREPL(reader, os.Stdout, eng, sess, reporter)

共享 Reader 很重要。若 REPL 和 Handler 各自对 os.Stdin 创建 bufio.Reader，缓冲区可能分别预读数据，造成输入丢失或错位。

十一、当前实现的生产级缺口

1. Grant 范围过宽

当前 key 只有 SessionID + ToolName。用户允许一次 bash 后，同一会话的其他 bash 命令也会命中授权。生产级授权至少应考虑：

1
2
3
4
5
工作区 WorkDir
工具名 ToolName
命令前缀或参数摘要
文件路径范围
创建时间和过期时间

2. 保存 Grant 时没有带上过期信息

协议定义了 WorkDir 和 ExpiresAt，但当前 Gate 保存时只填了 SessionID 和 ToolName：

1
2
3
4
err = g.grants.Save(ctx, Grant{
	SessionID: request.SessionID,
	ToolName:  request.ToolCall.Name,
})

这意味着目前没有真正的会话授权过期策略，需要补充策略配置。

3. 审批输入 Goroutine 可能泄漏

ReadString 不能直接被 Context 取消。中断发生时，Handler 可以返回，但后台读取 Goroutine 仍可能阻塞。后续应设计统一的输入循环或可关闭的输入源。

4. 所有执行路径都要经过 Gate

主 Agent 的工具调用已经经过 approvalGate.Check。但 Subagent 使用传入的只读 Registry 直接执行工具：

1
result := readOnlyRegistry.Execute(ctx, call)

这是有意设计的受限路径，但生产级系统必须明确记录哪些 Registry 是只读的，避免以后新增工具时错误地把写能力挂进去。

5. Gate 需要防御空依赖

当前 Engine 直接调用：

1
decision, err := e.approvalGate.Check(ctx, req)

如果某个调用方传入 nil Gate，会发生 panic。生产级构造器应该要求 Gate 非空，或者在 Engine 初始化阶段直接返回配置错误。

十二、审批测试应该验证什么

建议先给 Gate 写表驱动测试：

1
2
3
4
5
6
7
8
RiskSafe 自动返回 AllowOnce，不调用 Handler
未知风险返回 Deny
已有 Grant 返回 AllowOnce，不重复询问
Handler 返回 AllowOnce，不写 Grant
Handler 返回 AllowSession，写入 Grant
Handler 返回 Deny，不执行工具
Grant 过期后重新询问
Context 取消能从 Check 返回

再给 Engine 写集成测试：

1
2
3
4
被拒绝的 ToolCall 不会调用 Registry.Execute
批准的多个 ToolCall 可以并发执行
Observation 按原始 ToolCall 顺序写入
工具执行错误仍然会生成 Observation

总结

审批功能的核心不是一个 [y/n] 提示，而是一条完整的授权链：

1
2
3
4
5
6
7
8
9
10
11
12
13
RiskedTool
    ↓
Policy
    ↓
GrantStore
    ↓
Handler
    ↓
Gate Decision
    ↓
Approved ToolCall
    ↓
Registry.Execute

到这里，go-tiny-claw 已经具备了一个可扩展的 Terminal Human-in-the-loop 骨架：安全工具可以自动执行，危险工具在执行前请求用户确认，会话授权可以复用，执行仍然保持并发。

下一步可以继续把工具边界从进程内 Registry 扩展到 MCP：动态发现外部工具、声明工具权限，并把审批策略应用到远程工具调用。