你是否也曾好奇那些看起来神奇的 AI 编程工具背后到底是怎么运作的当我第一次深入研究 Claude Code、OpenCode、Pi 这些主流 coding agent 时脑袋里全是问号—— agent 的边界在哪里 harness 又是什么经过几个月的源码拆解我得出了一个反直觉的结论真正可工程化的部分全在 harness编排脚手架层而非模型本身。模型权重和核心 API 行为是固定的但 harness 里的状态管理、上下文注入、执行隔离、可观测性……这些才是让 coding agent 真正可靠、上下文感知、高性能的关键。这篇文章我带你从零开始构建一个 coding agent harness扒开那些黑箱里的核心组件。一、Agent vs Harness边界在哪里很多人一想到编程 agent脑子里就是 LLM 在循环里调工具。那个 ReAct 循环确实是心脏——但它也是最小的部分。真正让 agent 变得有用、安全、可靠的是周围的基础设施。来看一下agent的核心有多小# src/decode/agent/factory.py (simplified) from pydantic_ai import Agent, DeferredToolRequests from decode.agent.deps import AgentDeps agent Agent( build_model(settings.llm_provider), # gemini | openrouter | modal deps_typeAgentDeps, output_type[str, DeferredToolRequests], # final answer, or tool calls paused for approval ) register_tools(agent) # read, edit, bash, grep, ... async with agent.iter(prompt, message_historyhistory) as run: async for node in run: # model request → tool calls → repeat stream_events(node) # everything else is the harness就这么点东西。真正的复杂性都在 harness 里。核心 Harness 架构Context Memory Hygiene通过滑动窗口压缩summary tail保持模型专注注入持久化的工作区内存AGENTS.md / MEMORY.md。Safe Execution Environment通过沙箱Docker / Modal隔离命令执行严格运行时权限门控。Steering Interactivity用优先级队列解耦输入处理用户可以在循环中途 steer、interrupt 或 enqueue 任务。Orchestration Fan-Out管理 headless 运行时生命周期、并行 subagent forks、human-in-the-loopHITL检查点。Observability Feedback发出结构化 traces、实时 TUI 事件流、持续评估指标。二、高层架构三层分离理解现代 coding agent第一步是把它拆成三层三层架构Layer 1Agent Loop核心智能基于 Pydantic AI 构建的轻量级 ReAct 循环。模型评估状态选择 action 或 tool call执行它接收 observation迭代直到完成。Layer 2Harness系统脚手架包裹循环的运营支柱。处理 phase machine runner、steering queues、permission gates、execution sandboxes、动态内存、skill registries、context compaction、LSP 代码智能。Layer 3Interfaces消费与控制连接 harness 的交互面。涵盖本地交互终端TUI和分布式远程执行headless workflows。技术栈两大接口模式交互模式TUI输入与渲染基于 prompt-toolkit 构建用户输入Rich 实时终端渲染。队列路由用户输入路由到 Steering Queue标准 Enter或 Follow-up QueueAltEnter。优先级门控评估边界MODEL_REQUEST vs WOULD_STOP决定哪个队列优先执行。事件流式将实时文本 delta、thinking delta、tool calls 和交互式权限提示分派回终端。远程模式Kitaru Runtime并行编排利用 Kitaru runtimeZenML并行编排和分发 headless harness 实例。核心能力HITLhuman-in-the-loop pause/resume checkpoints、Replays精确 session 重放、Durability跨进程重启的状态存活。Swarm隔离为每个远程任务分配一个独立的、全新的无头 harness 实例。三、Agent Loop~20 行定义Agent在理解 harness 之前先看它包裹了什么。Agent loop 是一个基于 Pydantic AI 的 ReAct 循环核心就这 ~20 行# src/decode/agent/loop.py async def _run_turn(self, ctx, *, promptNone, deferred_resultsNone): async with self._agent.iter( prompt, depsself._deps, message_historyself.message_history, deferred_tool_resultsdeferred_results, ) as run: async for node in run: if Agent.is_model_request_node(node): await self._stream_model_node(ctx, node, run) elif Agent.is_call_tools_node(node): await self._stream_tool_node(ctx, node, run) # Carry the whole conversation into the next turn self.message_history run.all_messages() self._last_input_tokens _leg_input_tokens(self.message_history)Agent Loop 的两个节点类型ModelRequestNode生成与推理处理原始模型推理实时流式传输部分文本 delta 和内部 thinking/reasoning delta。CallToolsNode执行与状态更新管理 tool 交互的完整生命周期宣布请求的 tool call、执行它、发出结果 observation。关键设计DeferredToolRequestsoutput_type[str, DeferredToolRequests]str模型产生了最终答案turn 结束DeferredToolRequests模型发出了需要人工批准的 tool callsturn 暂停等待决议后恢复这个 deferred seam 就是实现 mid-turn steering 的关键也是 durable HITL 的未来保障。Agent Loop Streaming Nodes# src/decode/agent/loop.py async def _run_turn(self, ctx, *, promptNone, deferred_resultsNone): async with self._agent.iter( prompt, depsself._deps, message_historyself.message_history, deferred_tool_resultsdeferred_results, ) as run: async for node in run: if Agent.is_model_request_node(node): await self._stream_model_node(ctx, node, run) elif Agent.is_call_tools_node(node): await self._stream_tool_node(ctx, node, run) # Carry the whole conversation into the next turn self.message_history run.all_messages() self._last_input_tokens _leg_input_tokens(self.message_history)四、Runner单飞 phase machineRunnersrc/decode/harness/runner.py是驱动 turn 逐一执行的编排器。为什么是单飞single-flight一个 turn 可能碎成多个legiter → deferred pause → resume → follow-up。锁跨越整个多-leg turn# src/decode/harness/runner.py class Phase(enum.Enum): IDLE idle DISPATCHING dispatching # synchronous window before first await RUNNING running class Boundary(enum.Enum): MODEL_REQUEST model_request # drain steering before next model call WOULD_STOP would_stop # drain follow-up; empty → idle关键不变式Phase 在第一个 await 之前同步设置——竞态提交观察busy协作式 abortEsc 设置 flagturn 在下一个 boundary 停止绝不会在流中间中断完成的历史保留pending 的输入在 abort 时丢弃五、Steering Queue Priority Gate当用户发送新命令但 agent 正在执行任务时——立即注入会破坏正在进行的 tool call忽略它又让 agent 不可 steering。解决方案双队列交互模型。# src/decode/harness/queue.py dataclass(slotsTrue) class InteractionQueues: steering: asyncio.Queue[str] field(default_factoryasyncio.Queue) follow_up: asyncio.Queue[str] field(default_factoryasyncio.Queue) def drain_steering(self) - list[str]: Non-blocking: remove and return every queued steering message, oldest first. return _drain(self.steering) def drain_follow_up(self) - list[str]: Non-blocking: remove and return every queued follow-up message, oldest first. return _drain(self.follow_up)Steering Queue时机每个 model-request leg 之前 drain内容忙碌时用户输入的消息普通 Enter安全边界注入绝不在流中间或 tool call 中途插入Follow-up Queue时机只在 WOULD_STOP 边界 drain内容作为下一个 leg 继续对话的消息AltEnter安全只在模型完成当前推理后才处理六、Permission GateAllow / Ask / Deny权限层是最重要的 guardrail。它在每个 action 之前问你或者根据模式只问危险的操作。# src/decode/permissions/gate.py class PermissionGate: def __init__(self, modePermissionMode.DEFAULT, *, user_rulesNone): self._mode mode self._user_rules user_rules or RuleSet() self._agent_rules RuleSet() def check(self, request: PermissionRequest) - PermissionDecision: Return the gates verdict: deny → allow → mode → ask. decision self._decide_with_rules(request) return decision def _decide_with_rules(self, request: PermissionRequest) - PermissionDecision: # Walk every sources deny list, then every allow list, then fall through to mode. sources self._rule_sources() for rule_set in sources: denied rule_set.matching_deny(request) if denied is not None: return PermissionDecision.deny(modeself._mode, reason_deny_reason(denied)) for rule_set in sources: if rule_set.matching_allow(request) is not None: return PermissionDecision.allow(modeself._mode) return self._decide_by_mode(request.kind)权限模式决策通道当 gate 返回 ASK 时tool 抛出 ApprovalRequired。Pydantic AI 暂停运行并返回 DeferredToolRequests。循环通过 DecisionChannel 路由# src/decode/harness/decisions.py class DecisionChannel: def __init__(self): self._pending: asyncio.Future[str] | None None async def request(self) - str: future asyncio.get_running_loop().create_future() self._pending future return await future # blocks until resolve() or cancel() def resolve(self, line: str) - bool: future self._pending if future is None or future.done(): return False future.set_result(line) return TrueTUI 展示权限提示用户输入 y/n/aalways。一个输入面无并发提示无死锁。# src/decode/harness/decisions.py class DecisionChannel: def __init__(self): self._pending: asyncio.Future[str] | None None async def request(self) - str: future asyncio.get_running_loop().create_future() self._pending future return await future # blocks until resolve() or cancel() def resolve(self, line: str) - bool: future self._pending if future is None or future.done(): return False future.set_result(line) return True def cancel(self) - None: future self._pending if future is not None and not future.done(): future.cancel()七、Tool Registry扁平化、门控、受限工具在扁平列表中声明一次没有插件机制。每个 tool 都有驱动 permission gate 的 ToolKind# src/decode/tools/registry.py TOOL_SPECS: list[ToolSpec] [ # Read-only: no disk/exec side effect → auto-allowed ToolSpec(nameread, funcfiles.read, kindToolKind.READ_ONLY), ToolSpec(nameglob, funcfiles.glob, kindToolKind.READ_ONLY), ToolSpec(namegrep, funcfiles.grep, kindToolKind.READ_ONLY), # Mutating file tools: FILE_EDIT → edit mode auto-allows ToolSpec(namewrite, funcfiles.write, kindToolKind.FILE_EDIT), ToolSpec(nameedit, funcfiles.edit, kindToolKind.FILE_EDIT), # Bash: shell execution → OTHER (edit mode still asks) ToolSpec(namebash, funcbash_module.bash, kindToolKind.OTHER), # ... ]Active-Agent 限制每个 tool 用 prepare callback 注册在 active agent 不允许时从模型 schema 中隐藏def register_tools(agent): for spec in TOOL_SPECS: agent.tool(spec.func, prepare_prepare_for(spec.name), retriesspec.retries) async def _restrict_to_active_agent(tool_name): async def prepare(ctx, tool_def): if tool_name in ctx.deps.active_agent.tools: return tool_def return None # hidden from the models schema for this run return prepare一个 Agent无需重建切换 active agent 在下一个 turn 改变可见工具集。八、File Tools读、写、编辑与安全文件工具src/decode/tools/files.py是 load-bearing 的。每个路径保持在 cwd 下路径逃逸被拒绝# src/decode/tools/files.py def _resolve_in_cwd(cwd: Path, raw: str) - Path: Resolve raw under cwd; raise ModelRetry when it escapes. base cwd.resolve() target (base / raw).resolve() if not _is_within(base, target): raise ModelRetry( fPath {raw!r} resolves outside the working directory. ) return target九、SandboxFresh-Exec 接缝bash 和 file tools 通过 executor seam 执行。模式none/docker/modal在每个 session 读取一次# src/decode/tools/bash.py _EXECUTOR: CommandExecutor LocalExecutor() _executor_selected False def _get_executor() - CommandExecutor: global _EXECUTOR, _executor_selected if not _executor_selected: _executor_selected True mode settings.sandbox_mode if mode ! none: from decode.sandbox import select_executor _EXECUTOR select_executor(mode) return _EXECUTORFresh-Exec 不变式沙箱文件系统在调用间持久化但每个命令作为全新进程运行。cd 和 export 不会在调用间携带。沙盒后端接口# src/decode/sandbox/executor.py class SandboxBackend(Protocol): async def create(self, workspace: Path) - None: ... async def exec(self, *args: str, timeout_s: float) - ExecResult: ... async def read_bytes(self, rel: str) - bytes: ... async def write_bytes(self, rel: str, data: bytes) - None: ... async def stat(self, rel: str) - FileStat | None: ... async def export(self) - None: ... async def destroy(self) - None: ... class SandboxExecutor: One CommandExecutor over a SandboxBackend. Fresh-exec: cd/export do NOT persist. async def run(self, command: str, *, cwd: Path, timeout_s: float) - ExecResult: await self._ensure_created(cwd) return await self._backend.exec(bash, -lc, command, timeout_stimeout_s)十、Memory Skills上下文工程MemoryPi Memory# src/decode/memory/service.py def assemble_memory(cwd: Path) - str: Read discovered memory files and return the prompt block to inject. blocks: list[str] [] for path in discover_memory_files(cwd): content _read_text(path) if content is None: continue if path.name MEMORY.md: content _cap(content) # clip to 200 lines / 25 KB blocks.append(f# From {path}\n{content}) return \n\n.join(blocks)AGENTS.md项目指令从 cwd 到文件系统根目录遍历MEMORY.md学到的 factssession 退出时提取为一句话总结即时读取无重量级索引repo 用 grep 动态探索Skills渐进式披露# src/decode/skills/catalog.py def assemble_skills_catalog(cwd: Path) - str: skills load_skills(cwd) if not skills: return ordered sorted(skills.values(), keylambda s: s.name) lines [ f- { .join(skill.name.split())} — { .join(skill.description.split())} for skill in ordered ] return _CATALOG_CUE \n \n.join(lines)catalog 每个 turn 都广告每个 skill 的 name 一行描述。skill() tool 按需返回完整内容所以 context window 不会膨胀# src/decode/tools/skills.py async def skill(ctx, name: str) - str: Return the payload of the skill called name. Ungated — loading instructions is harmless. home ctx.deps.harness_home or ctx.deps.cwd catalog load_skills(home) found catalog.get(name) if found is None: raise ModelRetry(fNo skill named {name!r}.) return format_skill_payload(found, cwdhome)十一、Context Compaction管理 Token 预算Context window 是预算。太长的 context 会让模型困惑context decay。Decode 实现了在 ~80% window 阈值触发的两级联级。# src/decode/context/compaction.py class CompactOutcome(enum.Enum): COMPACTED compacted NOTHING_TO_COMPACT nothing_to_compact SUMMARIZER_FAILED summarizer_failed async def summarize_for_compaction(messages, *, model): Summarize older history into a fixed skeleton with one cheap LLM call. transcript _render_transcript(messages) if not transcript: return None agent Agent(model, instructions_COMPACTION_INSTRUCTIONS) result await agent.run(transcript) return result.output.strip() or None def split_tail(messages, *, keep_recent_tokens: int) - int: Index where the kept recent tail begins — snapped to Compaction Boundaries so tool-call/result pairs are never split. # Walk from end, accumulating tokens, snap back to nearest boundary ... def microcompact(messages, *, keep_recent_tokens: int): Blank old tool-output bodies in-memory — the no-LLM tier. boundary split_tail(messages, keep_recent_tokenskeep_recent_tokens) # Replace ToolReturnPart / RetryPromptPart content with placeholder ...Two-Tier Cascade# src/decode/context/compaction.py class CompactOutcome(enum.Enum): COMPACTED compacted NOTHING_TO_COMPACT nothing_to_compact SUMMARIZER_FAILED summarizer_failed async def summarize_for_compaction(messages, *, model): Summarize older history into a fixed skeleton with one cheap LLM call. transcript _render_transcript(messages) if not transcript: return None agent Agent(model, instructions_COMPACTION_INSTRUCTIONS) result await agent.run(transcript) return result.output.strip() or None def split_tail(messages, *, keep_recent_tokens: int) - int: Index where the kept recent tail begins — snapped to Compaction Boundaries so tool-call/result pairs are never split. # Walk from end, accumulating tokens, snap back to nearest boundary ... def microcompact(messages, *, keep_recent_tokens: int): Blank old tool-output bodies in-memory — the no-LLM tier. boundary split_tail(messages, keep_recent_tokenskeep_recent_tokens) # Replace ToolReturnPart / RetryPromptPart content with placeholder ...为什么是 ~80%因为模型响应和后续 tool outputs 仍需要 headroom。在 limit compaction 会立即再次触发。十二、TUI一输入面两模式TUIsrc/decode/tui/app.py基于 prompt-toolkit输入和 Rich输出构建。Append 风格像对话日志。关键绑定# src/decode/tui/app.py def _build_key_bindings(*, on_cycle_mode, on_toggle_verbose): bindings KeyBindings() bindings.add(escape, enter) def _follow_up(event): event.app.exit(result(InputIntent.FOLLOW_UP, event.app.current_buffer.text)) bindings.add(escape) def _abort(event): event.app.exit(result(InputIntent.ABORT, event.app.current_buffer.text)) bindings.add(s-tab) def _cycle_mode(event): on_cycle_mode() event.app.invalidate() # redraw footer bindings.add(c-o) def _toggle_verbose(event): on_toggle_verbose() event.app.invalidate() return bindingsEvent Sink: Line-Buffered Streaming# src/decode/tui/app.py def _make_event_sink(console: Console): state {need_prefix: False, buffer: , style: render.CONVERSATION_BG} def _stream(text: str, style: str) - None: if state[buffer] and state[style] ! style: _flush() state[style] style state[buffer] text while \n in state[buffer]: line, state[buffer] state[buffer].split(\n, 1) _emit_line(line) def on_event(event: events.Event) - None: if isinstance(event, events.AssistantTextDelta): _stream(event.text, render.CONVERSATION_BG) return if isinstance(event, events.ThinkingDelta): _stream(event.text, dim italic) return _flush() if isinstance(event, events.TurnStarted): state[need_prefix] True console.print(render.render_event(event)) return on_eventFooterContext Fill Gauge Spinner# src/decode/tui/app.py def _bottom_toolbar(deps, gate, handler, runner, decisions): window deps.context_window_tokens or settings.compaction_context_window_tokens fraction handler.last_input_tokens / window if window 0 else 0.0 warn_at 1 - settings.microcompaction_reserve_fraction danger_at 1 - settings.compaction_reserve_fraction label, color render.context_gauge(fraction, warn_atwarn_at, danger_atdanger_at) hint footer_hint(deps.active_agent.name, gate.mode.value, verbosedeps.verbose.enabled) if runner.phase is Phase.IDLE or decisions.pending: spinner else: frame render.spinner_frame(int(time.monotonic() / _FOOTER_REFRESH_S)) spinner f{frame} working… return HTML(f{spinner} {label} {hint})Footer: Context Fill Gauge Spinner# src/decode/tui/app.py def _bottom_toolbar(deps, gate, handler, runner, decisions): window deps.context_window_tokens or settings.compaction_context_window_tokens fraction handler.last_input_tokens / window if window 0 else 0.0 warn_at 1 - settings.microcompaction_reserve_fraction danger_at 1 - settings.compaction_reserve_fraction label, color render.context_gauge(fraction, warn_atwarn_at, danger_atdanger_at) hint footer_hint(deps.active_agent.name, gate.mode.value, verbosedeps.verbose.enabled) if runner.phase is Phase.IDLE or decisions.pending: spinner else: frame render.spinner_frame(int(time.monotonic() / _FOOTER_REFRESH_S)) spinner f{frame} working… return HTML(f{spinner} {label} {hint})十三、Remote Runtime持久化 Headless 流Headless harness 通过 Kitaru runtime 远程运行。Bypass Flow: Everything Inline# src/decode/runtime/flow.py flow(image_runtime_image()) def run_agent_task_hitl(task: str, modelNone, repoNone, localFalse) - str: Run task with durable HITL approvals ask_user waits. tool_scope _prepare_headless_tool_scope(repo, local) durable_agent _build_hitl_runtime_agent(model) deps _build_hitl_deps(tool_scope, model) with _durable_sleeper(): try: with observability.root_span(decode_run_hitl, inputtask): result durable_agent.run_sync(task, depsdeps) except _ToolApprovalDenied: return _capture_runtime_output(_HITL_DENIED_MESSAGE) return _capture_runtime_output(result.output)HITL Flow: Durable Approvalsflow(image_runtime_image()) def run_agent_task_hitl(task: str, modelNone, repoNone, localFalse) - str: Run task with durable HITL approvals ask_user waits. tool_scope _prepare_headless_tool_scope(repo, local) durable_agent _build_hitl_runtime_agent(model) deps _build_hitl_deps(tool_scope, model) with _durable_sleeper(): try: with observability.root_span(decode_run_hitl, inputtask): result durable_agent.run_sync(task, depsdeps) except _ToolApprovalDenied: return _capture_runtime_output(_HITL_DENIED_MESSAGE) return _capture_runtime_output(result.output)Headless Deps: BYPASS Mode# src/decode/runtime/flow.py def _build_headless_deps(cwdNone, modelNone) - AgentDeps: home Path.cwd() return AgentDeps( cwdcwd or home, harness_homehome, emit_headless_emit, # logs only, no TUI gatePermissionGate(modePermissionMode.BYPASS), resolve_permission_deny_permission_resolver, resolve_user_questiondeny_user_question_resolver, context_window_tokensresolve_context_window(model), )Replay: What-If Re-runs# src/decode/runtime/flow.py def replay_agent_task(exec_id: str, *, from_: str, model: str | None) - ReplayResult: Replay a recorded bypass run from from_ with an optional Model Override. handle run_agent_task.replay(exec_id, from_from_, modelmodel) return ReplayResult( exec_idhandle.exec_id, original_exec_idexec_id, output_load_runtime_output(handle.exec_id), )核心能力DurabilityKill -9 一个 run从上一个检查点恢复HITLrun 在问题处冻结不消耗计算力。几个小时后回答时恢复Replay完成的 run 可以用不同的 model 或 fixed prompt 重放Swarms并行启动 5–10 个实现每个有自己独立的 PR十四、可观测性与 Evals证明它有效每个 session 用 OpikComet 的开源 LLM 可观测性平台追踪。关键洞察P99 是 P50 的 60 倍这正是 harness 存在的原因。原始 ReAct 循环会愉快地烧一小时 token。Compaction cascade、steering queues、permission gate 的存在就是为了防止尾部吃掉预算。# From the agent loop and runtime with observability.root_span(chat_turn, thread_idsession_id, inputprompt) as span: # ... run the turn ... observability.record_output(span, output)# src/decode/entities/events.py Event ( TurnStarted | TurnFinished | AssistantTextDelta | ThinkingDelta | ToolCallStarted | ToolResult | PermissionRequested | AskUserRequested | TaskListUpdated | ContextCompacted | ContextMicrocompacted | AgentError )每个 event 都是 frozen、hashable、exhaustively matched by renderer。事件流是 TUI 和远程可观测性的单一真实来源。十五、结论为什么从零构建在现有开源 harness 上添加自定义逻辑并不难。但知道该加什么——基于这些 harness 的内部原理——才是真正重要的。那是基本功。是让 AI Engineer 依然有价值的神秘酱汁。从零构建一次 coding agent——从 bare ~20 行 agent loop 到远程 agent swarm——你就有了为任何 AI agent 工作流构建自定义 harness 的能力。我现在理解的东西✅ Runner 的单飞 phase machine 保证 turn 安全和可 steering✅ 双队列交互模型防止 mid-turn 破坏✅ Permission Gate 的规则 union mode × kind 矩阵提供灵活 guardrails✅ 扁平 tool registry active-agent 限制让 persona 切换即时完成✅ Sandbox seam 的 fresh-exec 模型让本地和远程模式字节级一致✅ Context compaction 把 context window 当作受管预算✅ TUI 的一输入面设计防止死锁并启用实时流✅ Headless runtime 的 durability、HITL、replay 让远程自动化成为可能Harness而不是模型才是让 coding agent 变得优秀的东西。 总结Coding agent 的核心魅力在于模型能力是固定的但 harness 的设计是无限的。理解 harness 架构你就能从被动使用者变成 AI power user获得构建定制化 agent 工作流的架构直觉在模型能力之外找到真正的竞争力来源这篇文章的所有源码都来自作者对 Claude Code、OpenCode、Pi、Hermes 等主流工具的深度拆解。理论实践这才是真正有用的学习方式。 讨论你在使用 coding agent 时有没有好奇过它的内部运作机制有没有什么具体场景让你觉得这东西还不够智能欢迎在评论区分享你的想法。