DeepSeek Harness 源码深潜（中）——工具、提示词与插件内核（附DSH源码）

tools = [{”type”: ”function”, ”function”: {”name”: ”get_weather”, ”parameters”: {...}}}] def dispatch(name, args):    if name == ”get_weather”:        return get_weather(**args)

export function defineTool(  options: DefineToolOptions,): ToolDefinition {  const parameters = parameterSchemaSpecToJsonSchema(options.parameters)  const validate = (args: unknown): string[] =>    validateJsonSchemaValue(parameters, args, '')  return {    name: options.name,    description: options.description,    parameters,    async execute(args, exec) {      const violations = validate(args)      if (violations.length > 0) throw new ToolArgsError(violations)      return userExecute(args as InferArgs, exec)    },  }}

interface BashToolArgs {  command: string  description: string      // 必填——”为什么执行”（可审计）  timeoutMs?: number  workdir?: string  run_in_background?: boolean}

function validateBashArgs(args: BashToolArgs): void {  if (args.command.trim().length === 0) {    throw new Error('invalid command: expected a non-empty string')  }  // timeoutMs 必须是正有限数  // sandbox_permissions ⇔ justification 必须配对}

function bashDescription(backgroundEnabled: boolean, escalationModes: readonly SandboxMode[]): string {  // 把当前沙箱模式、后台执行可用性写进描述}

tools/pre-execute   （瀑布事件：允许 / 拒绝 / 询问）      ↓tools/execute       （调度器分发）      ↓tools/post-execute  （结果后处理）      ↓tools/result        （结果观测）

ctx.provide(TOOL_RUNTIME_SCHEDULER, {  prepare(exec)        // 进入流水线（跑 pre-execute），返回 dispatch / post-result / final-result  dispatch(prepared)   // 真正执行工具函数  finalize(exec, result)  // 结果后处理  finish(exec, result)    // 直接收尾})

const prepared = await ctx.tools[TOOL_RUNTIME_SCHEDULER].prepare(call.exec)switch (prepared.kind) {  case 'dispatch':    promise = ctx.tools[TOOL_RUNTIME_SCHEDULER].dispatch(prepared.exec)  case 'post-result':  case 'final-result':}

defineTool({  ...,  isConcurrencySafe: true,   // 如 read_file：可并行  // 默认 false：如 bash：必须串行})

ctx.tools.register(tool, { scope: agentId })   // 只给这个 agent 注册ctx.tools.register(tool)                        // 全局注册

class ToolRuntime:    def __init__(self):        self.tools = {}        self.pre_execute_hooks = []     def register(self, tool_def: dict, scope: str = ”*”):        self.tools[(scope, tool_def[”name”])] = tool_def     def get(self, scope: str, name: str):        return self.tools.get((scope, name)) or self.tools.get((”*”, name))     def add_pre_execute_hook(self, hook):        self.pre_execute_hooks.append(hook)     async def prepare(self, scope: str, name: str, args: dict):        tool = self.get(scope, name)        if not tool:            return {”kind”: ”rejected”, ”reason”: ”tool not found”}        for hook in self.pre_execute_hooks:            decision = hook(name, args)            if decision:                return {”kind”: ”rejected”, ”reason”: decision}        return {”kind”: ”dispatch”, ”tool”: tool}     async def dispatch(self, prepared, args: dict):        return prepared[”tool”][”execute”](args) def define_tool(runtime: ToolRuntime, name: str, description: str,                parameters: dict, concurrency_safe: bool = False,                scope: str = ”*”):    def decorator(func):        def execute(args: dict):            for key, spec in parameters.get(”properties”, {}).items():                if spec.get(”required”) and key not in args:                    raise ValueError(f”缺少参数: {key}”)            return func(**args)        runtime.register({            ”name”: name, ”description”: description,            ”parameters”: parameters, ”execute”: execute,            ”concurrency_safe”: concurrency_safe,        }, scope)        return func    return decorator rt = ToolRuntime() @define_tool(rt, ”read_file”, ”读取文件（可并行）”,             {”type”: ”object”, ”properties”: {”path”: {”type”: ”string”, ”required”: True}}},             concurrency_safe=True)def read_file(path: str): return f”[内容] {path}” @define_tool(rt, ”delete_file”, ”删除文件（危险）”,             {”type”: ”object”, ”properties”: {”path”: {”type”: ”string”, ”required”: True}}})def delete_file(path: str): return f”[已删除] {path}” rt.add_pre_execute_hook(lambda name, args: f”禁止执行 {name}” if name == ”delete_file” else None) import asyncioasync def main():    print(”决策:”, await rt.prepare(”*”, ”delete_file”, {”path”: ”/etc/passwd”}))    prepared = await rt.prepare(”*”, ”read_file”, {”path”: ”a.py”})    print(”执行:”, await rt.dispatch(prepared, {”path”: ”a.py”})) asyncio.run(main())

system_prompt = f”””你是一个智能助手。当前工作目录：{cwd}可用工具：{”, ”.join(tool_names)}规则：{rules}”””

export interface PromptSection {  readonly name: string          // 唯一名——重名注册直接抛错  readonly order: number         // 排序权重  readonly text: string | ((context) => string)  readonly complete?: boolean    // ”我就是整个系统提示词”（独占模式）}

export interface PromptAssembly {  sections: AssembledSection[]     // 静态/半静态规则  contexts: AssembledContext[]     // 动态上下文  tools: ToolSchema[]              // 工具 schema  variables: Record   // 模板变量}

// 组装时：只是把文本解析出来，保留 {{var}} 原样// 渲染时：renderPrompt(assembly) 才把 {{var}} 替换成 assembly.variables 里的值

readonly complete?: boolean// 若某 section 标记 complete=true：//   组装仍跑 waterfall//   但最终只保留这一个 section 作为系统提示词// 多个 complete 同时生效 → 组装失败

from dataclasses import dataclass, fieldimport re @dataclassclass Section:    name: str    order: int    text: str | callable    complete: bool = False @dataclassclass Assembly:    sections: list[dict] = field(default_factory=list)    contexts: list[str] = field(default_factory=list)    tools: list[dict] = field(default_factory=list)    variables: dict = field(default_factory=dict) class SystemPrompt:    def __init__(self):        self._sections: list[Section] = []     def add(self, s: Section):        if any(x.name == s.name for x in self._sections):            raise ValueError(f”重复 section: {s.name}”)        if s.complete and any(x.complete for x in self._sections):            raise ValueError(”多个 complete section 冲突”)        self._sections.append(s)     def assemble(self, context: dict, tools: list[dict]) -> Assembly:        sections = []        for s in sorted(self._sections, key=lambda x: x.order):            text = s.text(context) if callable(s.text) else s.text            sections.append({”name”: s.name, ”text”: text})        completes = [x for x in sections if any(            s.name == x[”name”] and s.complete for s in self._sections)]        if completes:            sections = completes        return Assembly(sections=sections, tools=tools,                        variables=context.get(”variables”, {}))     def render(self, assembly: Assembly) -> str:        parts = [s[”text”] for s in assembly.sections]        if assembly.tools:            parts.append(”可用工具: ” + ”, ”.join(t[”name”] for t in assembly.tools))        text = ”\n\n”.join(parts)        for key, value in assembly.variables.items():            text = re.sub(r”\{\{\s*” + key + r”\s*\}\}”, str(value), text)        return text sp = SystemPrompt()sp.add(Section(”identity”, -100, ”你是自动化 agent。”))sp.add(Section(”persona”, 0,               lambda ctx: f”你是{ctx['deployment']}的助手，今天是{{{{date}}}}。”))sp.add(Section(”rules”, 150, ”调用工具前必须说明目的。”)) asm = sp.assemble({”deployment”: ”工厂质检”, ”variables”: {”date”: ”2026-08-14”}},                  [{”name”: ”read_file”}, {”name”: ”search”}])print(sp.render(asm))

加载（依赖解析、拓扑排序）卸载（副作用逆序回滚）事件（waterfall / serial / emit 三种语义）

export const name = 'tool-bash'export const inject = ['tools', 'shell', 'systemPrompt', 'shellEnv'] export function apply(ctx: Context): void {  ctx.tools.register(bashTool)}

// ① waterfall：监听器必须调用 next() 才能放行ctx.emit('tools/pre-execute', data, (decision) => { })// 权力：可以拦截、可以修改 // ② serial：按注册顺序执行，但不能改写结果ctx.emit('agent/turn-stopping', { turn, signal })// 权力：可以感知、可以追加副作用 // ③ emit：异步通知，监听器互不干扰ctx.emit('session/event', event)// 权力：只能旁观

ctx.on('tools/pre-execute', handler, { scope: agentId })ctx.provide('llm', impl, { scope: agentId })

ctx.provide('llm', impl)          // → 记录: 卸载时删除 'llm'ctx.on('tools/pre-execute', fn)   // → 记录: 卸载时移除监听器ctx.tools.register(tool)          // → 记录: 卸载时注销工具 // 卸载时：逆序执行回滚栈

import asyncio class Cordis:    def __init__(self):        self._services = {}        self._listeners = {}        self._rollbacks = []     def provide(self, name, impl, scope=”*”):        key = (name, scope)        self._services[key] = impl        self._rollbacks.append(lambda: self._services.pop(key, None))     def get(self, name, scope=”*”):        return self._services.get((name, scope)) or self._services.get((name, ”*”))     def on(self, event, handler, scope=”*”):        self._listeners.setdefault((event, scope), []).append(handler)        self._rollbacks.append(            lambda: self._listeners[(event, scope)].remove(handler))     def _collect(self, event, scope):        return (self._listeners.get((event, scope), [])                + self._listeners.get((event, ”*”), []))     async def waterfall(self, event, data, scope=”*”, default=None):        for handler in self._collect(event, scope):            result = await handler(data)            if result is not None:                return result        return default     async def serial(self, event, data, scope=”*”):        for handler in self._collect(event, scope):            await handler(data)     def emit(self, event, data, scope=”*”):        for handler in self._collect(event, scope):            asyncio.create_task(handler(data))     def load(self, plugin):        for dep in plugin.get(”inject”, []):            if self.get(dep) is None:                raise RuntimeError(f”{plugin['name']} 缺少依赖 {dep}”)        plugin[”apply”](self)        self._rollbacks.append(lambda: print(f”[卸载] {plugin['name']}”))     def unload_all(self):        for fn in reversed(self._rollbacks):            fn()

Service Definition（接口声明）   ← 接缝本身      ↑ 实现               ↑ 使用Service Provider（实现）         Consumer（消费者）

import type { } from '@deepseek-ai/dsh-fs'   // ← 只 import 接口（type-only！） export const inject = ['fs', 'tools', 'systemPrompt'] export function apply(ctx: Context): void {  const readTool = defineTool({    name: 'read_file',    async execute(args, exec) {      return ctx.fs.read(args.path, exec)   // 调用接口，不知道背后是谁    },  })  ctx.tools.register(readTool)}

from abc import ABC, abstractmethod class Shell(ABC):    @abstractmethod    def run(self, cmd: str) -> str: ... class BashLocal(Shell):    def run(self, cmd: str) -> str:        import subprocess        return subprocess.run(cmd, shell=True, capture_output=True, text=True).stdout class BashSandbox(Shell):    def run(self, cmd: str) -> str:        if ”rm” in cmd:            raise PermissionError(f”[沙箱] 拒绝危险命令: {cmd}”)        return f”[沙箱执行] {cmd} → ok” class BashRemote(Shell):    def run(self, cmd: str) -> str:        return f”[远程执行] {cmd} → ok” class ToolBash:    def __init__(self, shell: Shell):        self._shell = shell    def execute(self, cmd: str) -> str:        return self._shell.run(cmd) CONFIG = {”provider”: ”sandbox”} def make_tool() -> ToolBash:    provider = CONFIG[”provider”]    shell = {”local”: BashLocal, ”sandbox”: BashSandbox,             ”remote”: BashRemote}[provider]()    return ToolBash(shell) CONFIG[”provider”] = ”local”print(make_tool().execute(”echo hi”)) CONFIG[”provider”] = ”sandbox”print(make_tool().execute(”echo hi”))try:    make_tool().execute(”rm -rf /”)except PermissionError as e:    print(”被拦截:”, e) CONFIG[”provider”] = ”remote”print(make_tool().execute(”echo hi”))