DeepSeek Harness 工具执行流水线与权限门禁

2026-09-08 00:00    #AI   #工具   #DeepSeek  

一次工具调用按固定顺序经过多个环节,每个环节负责一类策略,钩子插件可以在其中允许、拒绝或询问。

整体顺序tools/pre-execute → 单调守卫 → tools/executetools/post-executefinalizeContenttools/result

前三个 waterfall 可改写一次调用;finalizeContent 与 tools/result 在其后运行。

tools/pre-execute:可重排的策略层

承载"钩子、权限、沙箱"等可重排策略;返回类型化决策 PreToolDecision

决策含义后续行为
{ kind: 'allow' }放行继续走单调守卫与之后的环节
{ kind: 'deny'; reason: string }拒绝物化成错误结果,工具主体被跳过
{ kind: 'ask'; reason?: string }询问用户只有审批服务返回 allowed-once 才继续,否则拒绝

参数不可被改写(历史记录、审计、UI 与执行必须一致)。沙箱、权限、plan-mode 插件都用这个扩展点。

单调守卫:不可撤销的最终拒绝

waterfall 的缺点:后注册监听器可推翻前面监听器的决策。需要"最终拒绝、任何人不能撤销"的不变式时用 ctx.tools.guard()

1// ToolGuard:感知作用域的最终预分派策略
2type ToolGuard = (execution: Readonly<ToolExecution>) => string | undefined

守卫没有 allow 结果:返回字符串=拒绝,返回 undefined=维持现状。监听器顺序永远无法把一次拒绝变回允许——“只减不增,只收权限,不给权限”。

tools/execute 与 tools/post-execute

tools/execute:环绕分派,把真正调用工具主体包起来。超时、重试、指标收集在这一层做。视图是 ToolDispatchExecution——只有这个视图可以替换必需的 exec.signal 施加截止时间。替换规则:可以替换但不能移除,注册表在调用工具主体前重新融合调用方的 signal。

tools/post-execute:执行完、结果归一化之前做检查或改写,返回 PostToolDecision

决策含义
{ kind: 'accept'; content? }接受结果,可替换展示内容(保留规范值与元数据)
{ kind: 'accept'; value }接受结果,可替换规范值(会重新校验并重算内容)
{ kind: 'block'; feedback }阻止结果,把纠正反馈变成错误结果

内容替换是展示策略,不是保密策略;要隐藏程序化值,必须替换该值或阻止结果。

finalizeContent 与 tools/result

记忆法:pre-execute 决定"能不能做",execute 决定"怎么做",post-execute 决定"结果怎么呈现",result 只负责"看一眼最终结果"。

动手示例:权限门禁插件(tools/pre-execute)

 1// 文件路径:my-plugins/permission-gate/src/index.ts
 2import type { Context } from '@deepseek-ai/cordis'
 3import type { PreToolDecision, ToolExecution } from '@deepseek-ai/dsh-tools'
 4
 5const DENY_TOOLS = new Set(['fs_write', 'fs_edit'])
 6
 7async function isAllowed(exec: ToolExecution): Promise<boolean> {
 8  if (DENY_TOOLS.has(exec.name)) return false
 9  const raw = exec.arguments as { path?: string }
10  if (typeof raw.path === 'string' && raw.path.includes('.env')) return false
11  return true
12}
13
14export const name = 'permission-gate'
15
16export function apply(ctx: Context) {
17  ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
18    if (!(await isAllowed(exec))) {
19      return { kind: 'deny', reason: 'Denied by policy: this tool is not allowed in the runoob workspace.' }
20    }
21    return next()
22  })
23}

动手示例:用 tools/execute 加超时

 1// 文件路径:my-plugins/tool-guard/src/index.ts
 2import type { Context } from '@deepseek-ai/cordis'
 3
 4export const name = 'tool-guard'
 5
 6export function apply(ctx: Context) {
 7  ctx.on('tools/execute', async (exec, next) => {
 8    const originalSignal = exec.signal
 9    const deadline = AbortSignal.timeout(30_000)
10    exec.signal = AbortSignal.any([originalSignal, deadline])
11    try {
12      return await next()
13    } finally {
14      exec.signal = originalSignal
15    }
16  })
17}

动手示例:用 ctx.tools.guard() 做单调拒绝

 1// 文件路径:my-plugins/invariant-guard/src/index.ts
 2import type { Context } from '@deepseek-ai/cordis'
 3
 4export const name = 'invariant-guard'
 5
 6export function apply(ctx: Context) {
 7  const disposer = ctx.tools.guard((execution) => {
 8    if (execution.name === 'run_code') {
 9      return 'run_code is disabled in the runoob demo profile.'
10    }
11    return undefined
12  })
13
14  ctx.effect(() => disposer)
15}

自测

  1. 永远不可放行用什么?→ ctx.tools.guard()(没有 allow 结果、不可撤销)
  2. 统一超时监听什么?→ tools/execute(环绕分派、可替换 signal)
  3. 只记录最终结果用什么?→ tools/result(观测冻结权威结果)