本文基于 deepseek-harness 仓库当前检出的源码（0.1.0-rc.5）。文中的文件路径以这份源码为准；项目仍在快速演化，内部 API 可能变化。

DeepSeek Harness（下面简称 dsh）是一套 Agent 运行时。它做的事情并不神秘：接收用户输入，请求模型，执行模型要求的工具，把工具结果交回模型，直到得到最终回答。真正值得读源码的地方，是它怎样把这条流程变成可以恢复、回放和替换的程序。

本文只跟踪一个小任务：用户要求运行 echo hello。先把这个任务跑通，再解释它经过的 agent-loop、Session 日志、工具调度器、LLM 适配器和 Cordis 插件系统。这样读到后面的类型和事件时，都能知道它们在解决哪个具体问题。

1
2
3
4
5
6
7
8
9
用户输入
  -> inbox
  -> ReactLoopAgent 的 turn / step
  -> Session 投影成模型历史
  -> DeepSeek 适配器发起请求
  -> 模型返回 bash 工具调用
  -> 工具执行并写入 tool/result
  -> 新 step 再次请求模型
  -> 最终文本

This article is based on the checked-out source of deepseek-harness (0.1.0-rc.5). File paths refer to this checkout; the project is still evolving quickly, so internal APIs may change.

DeepSeek Harness, abbreviated as dsh below, is an Agent runtime. Its job is straightforward: accept user input, call the model, run the tools requested by the model, feed the results back, and stop when a final answer is available. The interesting part in the source is how this ordinary loop becomes recoverable, replayable, and replaceable.

This article follows one small task from beginning to end: the user asks the Agent to run echo hello. We first make that task concrete, then explain the agent-loop, Session log, tool scheduler, LLM adapter, and Cordis plugin system it passes through. That way, every type and event later in the article has a concrete problem to solve.

1
2
3
4
5
6
7
8
9
user input
  -> inbox
  -> turn / step in ReactLoopAgent
  -> Session projects the model history
  -> DeepSeek adapter sends a request
  -> the model returns a bash tool call
  -> the tool runs and writes tool/result
  -> a new step calls the model again
  -> final text

先把一次请求跑通

先不要把 Agent 想成一个复杂的聊天产品。对 dsh 来说，最小的 Agent 只是一个会重复推进任务的循环：拿到输入，调用模型，处理模型的决定，再决定是否继续。

用户说“执行 echo hello”时，模型不会直接执行命令。模型只能返回一个工具调用，真正执行命令的是 dsh 中注册的 bash 工具。一次完整的处理过程大致如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
第 1 个 step
  user/message
  -> 请求模型
  -> assistant/message：调用 bash
  -> tool/call
  -> bash 执行 echo hello
  -> tool/result：stdout = hello

第 2 个 step
  -> 带着 tool/result 再请求模型
  -> assistant/message：告诉用户输出是 hello

turn 结束

这里有一个必须先记住的边界：一个 step 包含一次模型请求，以及这次请求触发的工具执行；工具结果之后的下一次模型请求属于新的 step。 如果 provider 返回可重试错误，step() 可以在同一个 step 内重新请求；这和工具结果导致的新 step 不是一回事。

turn、step 和事件

turn 是一批输入驱动的完整处理过程，通常从 turn/start 开始，以 turn/end 结束。一个 turn 里可能有多个 step。step 是模型做出一次决定并等待相关工具结果的边界。

日志中的事件不等于发给模型的一条消息。可以先用下面这张表区分它们：

事件 记录的事实 会成为模型历史吗
turn/start、turn/end 一轮任务的开始和结束 否
step/start、step/end 一次模型步骤的边界 否
user/message 用户输入或插件注入的上下文 是
assistant/chunk 模型流式输出的一小段增量 否，稍后组装
assistant/message 组装完成的模型消息 是
tool/call 准备执行哪个工具、参数是什么 否
tool/result 工具返回的内容或错误 是

如果第一次读源码，建议按这个顺序跳转：先看 packages/core/agent-loop/src/agent.ts，再看 packages/core/session/src，然后看 packages/core/tools/src 和 packages/llm/llm-deepseek/src。这条顺序正好对应一次请求的运行顺序。

First, run one request through the system

Do not start by imagining an Agent as a complete chat product. In dsh, the smallest Agent is a loop that keeps a task moving: take input, call the model, handle the model’s decision, and decide whether to continue.

When the user says “run echo hello”, the model does not execute the command itself. It returns a tool call, while the registered bash tool in dsh performs the command. One complete processing path looks like this:

1
2
3
4
5
6
7
8
9
10
11
12
13
step 1
  user/message
  -> call the model
  -> assistant/message: call bash
  -> tool/call
  -> bash runs echo hello
  -> tool/result: stdout = hello

step 2
  -> call the model again with tool/result
  -> assistant/message: tell the user the output is hello

turn ends

Keep this boundary in mind: one step contains one model request and the tool executions triggered by that request; the next model request after a tool result belongs to a new step. If a provider returns a retryable error, step() may call it again inside the same step. That is different from opening a new step after a tool result.

turn, step, and events

A turn is one complete processing run driven by a batch of input. It usually starts with turn/start and ends with turn/end. A turn may contain several steps. A step is the boundary around one model decision and the tool results it is waiting for.

An event is not automatically one message sent to the model. Use this table to separate the two:

Event Fact it records Becomes model history?
turn/start, turn/end The start and outcome of a turn No
step/start, step/end The boundary of a model step No
user/message User input or plugin-injected context Yes
assistant/chunk One streaming delta from the model No, assembled later
assistant/message The assembled model message Yes
tool/call Which tool and arguments are about to run No
tool/result The tool output or error Yes

If you are reading the source for the first time, follow this order: start with packages/core/agent-loop/src/agent.ts, then packages/core/session/src, then packages/core/tools/src and packages/llm/llm-deepseek/src. It matches the runtime order of one request.

Agent 主循环：输入如何变成模型请求

ReactLoopAgent 位于 packages/core/agent-loop/src/agent.ts。这里的 React 指 reactive driver，不是 React UI：输入到达 inbox 后，它唤醒 driver，处理一轮任务，没有待处理输入时回到 idle。

可以先把它看成两层循环。下面是为了阅读源码而压缩的伪代码，省略了取消、错误和日志细节：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async kick() {
  while (await this.turn()) {}
}

async turn() {
  while (true) {
    const decision = await this.preStep(target, position)
    if (decision.kind === 'reject') return false

    session.append('step/start', position)
    await this.step(decision.assembly)
    session.append('step/end', position)

    if (this.inbox.nextStep.length === 0) break
  }

  return this.inbox.hasPending
}

外层 kick() 负责“还有没有下一轮任务”，内层 turn() 负责“当前 turn 是否还要进入下一个 step”。因此，工具结果回来后，内层循环再次进入 preStep()，然后追加新的 step/start，而不是在旧 step 里偷偷发第二次模型请求。

preStep() 先领取输入，再组装 prompt

preStep() 做的第一件事是从 inbox 领取本次要处理的消息，然后才组装 system prompt：

1
2
3
4
5
6
7
8
9
10
const claimed = this.inbox.claim(target, position.turn)
const assembly = await this.loopCtx.systemPrompt.assemble(...)
const decision = await this.dispatch.waterfall(
  'agent/pre-step',
  { messages: claimed, ...position, signal },
  () => Promise.resolve({
    kind: 'enter',
    messages: context === undefined ? claimed : [...claimed, context],
  }),
)

这个顺序是为了确定输入批次。假设 prompt 组装需要 200 毫秒，期间用户又发来一条消息：如果先组装、后领取，这条新消息可能被错误地并入当前 step；先 claim 就能把它留给下一个边界。

inbox 里主要有两个目标队列：

目标 什么时候消费 含义
next-step 当前 turn 继续时 steering 或 inject 的输入
next-turn 当前 turn 结束、下一个 turn 开始时 followup 或新的用户任务

send() 是统一入口，followup() 写入下一轮并唤醒 driver，steer() 写入当前 turn 的下一步并唤醒，inject() 写入下一步但不负责唤醒。输入被移动时会先追加 agent/inbox/spliced，再更新内存中的队列，因此重启时也可以从事件重建 inbox。

step() 的职责

step() 先从 Session 的当前 surface 得到模型历史，再通过 llm/stream 请求模型。流中的每个增量会写成 assistant/chunk；流结束后，BlockAssembler 把它们组装成一个 assistant/message。

如果消息里没有 tool-call，step 结束。如果有 tool-call，step() 把调用交给工具调度器。调度器完成后写入 tool/result，turn() 才会关闭当前 step，并让下一次循环带着这些结果再请求模型。

The agent loop: turning input into a model request

ReactLoopAgent lives in packages/core/agent-loop/src/agent.ts. “React” means a reactive driver, not React UI: input reaches the inbox, wakes the driver, one turn runs, and the driver returns to idle when no work is pending.

Read it as two nested loops. The following is pseudocode for reading the source; cancellation, errors, and log details are omitted:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
async kick() {
  while (await this.turn()) {}
}

async turn() {
  while (true) {
    const decision = await this.preStep(target, position)
    if (decision.kind === 'reject') return false

    session.append('step/start', position)
    await this.step(decision.assembly)
    session.append('step/end', position)

    if (this.inbox.nextStep.length === 0) break
  }

  return this.inbox.hasPending
}

The outer kick() answers “is there another turn to run?”, while the inner turn() answers “does this turn need another step?”. After a tool result, the inner loop enters preStep() again and appends a new step/start; it does not secretly make a second model request inside the old step.

preStep() claims input before assembling the prompt

The first thing preStep() does is claim the messages for this step from the inbox. Only then does it assemble the system prompt:

1
2
3
4
5
6
7
8
9
10
const claimed = this.inbox.claim(target, position.turn)
const assembly = await this.loopCtx.systemPrompt.assemble(...)
const decision = await this.dispatch.waterfall(
  'agent/pre-step',
  { messages: claimed, ...position, signal },
  () => Promise.resolve({
    kind: 'enter',
    messages: context === undefined ? claimed : [...claimed, context],
  }),
)

The ordering makes the input batch deterministic. Suppose prompt assembly takes 200 milliseconds and the user sends another message during that time. If the loop assembled first and claimed afterward, the new message could accidentally join the current step. Claiming first leaves it for the next boundary.

The inbox mainly has two targets:

Target Consumed when Meaning
next-step The current turn continues Steering or injected input
next-turn The current turn closes and the next starts A follow-up or new user task

send() is the unified entry. followup() queues the next turn and wakes the driver, steer() queues the next step and wakes it, and inject() queues the next step without waking it. A move first appends agent/inbox/spliced, then updates the in-memory queue, so a restart can rebuild the inbox from events too.

What step() does

step() obtains the current model history from the Session surface and calls the model through llm/stream. Each streaming delta is written as an assistant/chunk; when the stream ends, BlockAssembler assembles them into an assistant/message.

If the message has no tool call, the step ends. If it does, step() passes the calls to the tool scheduler. The scheduler writes tool/result, then turn() closes the current step and lets the next loop iteration call the model with those results.

Session 日志：模型历史是重新计算的

现在可以回答一个更基础的问题：step() 从哪里得到下一次请求的 messages[]？dsh 的答案不是“内存里一直维护一份数组”，而是事件溯源（event sourcing）：Session 把已经发生的事实按顺序追加到日志，需要历史时再从日志计算出来。

为什么不直接保存 messages[]

直接保存数组看起来更简单，但运行中的程序和磁盘上的日志就可能变成两个状态来源：

  1. 内存数组告诉模型和 UI 当前历史是什么；
  2. 磁盘日志告诉重启后的进程历史是什么。

如果工具结果已经写入日志，但进程在更新数组前退出，两份状态就会不一致。事件溯源把写入顺序固定为“先记录事实，再从事实折叠视图”：

1
2
3
4
SessionEvent[]
  -> 当前 surface 中的事件 seq
  -> Message[]
  -> DeepSeek wire messages

这四层不要混在一起：日志包含边界、流式增量和工具执行；surface 决定哪些消息对模型可见；Message[] 是 dsh 内部的统一消息；wire messages 才是 DeepSeek HTTP 请求体里的格式。

一段会话日志

仓库的测试 fixture 会把完整过程保存成 JSONL。下面是省略参数的结构化简写：

1
2
3
4
5
6
7
8
9
10
11
{"type":"turn/start","seq":1}
{"type":"step/start","seq":2,"data":{"turn":1,"step":1}}
{"type":"user/message","seq":3}
{"type":"assistant/chunk","seq":4}
{"type":"assistant/message","seq":8}
{"type":"tool/call","seq":9,"data":{"name":"bash","callId":"call-1"}}
{"type":"tool/result","seq":10,"data":{"callId":"call-1"}}
{"type":"step/end","seq":11}
{"type":"step/start","seq":12,"data":{"turn":1,"step":2}}
{"type":"assistant/message","seq":13}
{"type":"turn/end","seq":14,"data":{"reason":{"kind":"completed"}}}

assistant/chunk 和 assistant/message 都与模型有关，但用途不同：chunk 用于流式显示和回放，完整的 message 才会进入下一次模型历史。tool/call 记录“准备做什么”，tool/result 记录“实际得到什么”，两者通过 callId 配对。

一个事件怎样变成消息

packages/core/session/src/surface.ts 中的 deriveEventMessage() 只负责判断单个事件是否产生一条内部消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
export function deriveEventMessage(event: SessionEvent): Message | null {
  switch (event.type) {
    case 'user/message':
      return event.data
    case 'assistant/message':
      if (event.data.message.content.length === 0) return null
      return event.data.message
    case 'tool/result':
      return event.data.message
    default:
      return null
  }
}

这段代码说明了几个容易忽略的事实：assistant/chunk 不直接进入历史；只有有内容的 assistant/message 才进入历史；tool/result 会进入历史，但它在 dsh 内部的 role 仍是 user，发送给 DeepSeek 时再转换成 tool，后文会说明原因。

事件名和事件数据由 SessionEventMap 统一描述。插件可以通过 TypeScript declaration merging 增加自己的事件，不需要修改 Session 核心：

1
2
3
4
5
6
7
8
export interface SessionEventMap {
  'turn/start': { turn: number }
  'turn/end': { turn: number; reason: TurnEndReason }
  'user/message': UserMessage
  'assistant/message': { message: AssistantMessage }
  'tool/call': { callId: string; name: string; arguments: string }
  'tool/result': { message: ToolResultMessage }
}

外层事件信封还包含 type、seq、time 和 data。读取旧日志时遇到未知事件，只有显式带 ignorable: true 才能跳过；默认拒绝重建会话。这样做的目的不是追求严格，而是避免新版本误读旧事件后恢复出一份缺少事实的会话。

请求分派前的 invariant

“历史可以从日志计算”还不够，代码还要检查真正发出的请求有没有绕过日志。agent-loop/src/invariant.ts 在 llm/stream 上注册了一个优先监听器，对带有 agent-loop 标记的请求比较：

1
2
3
4
5
6
7
8
9
10
ctx.on('llm/stream', (options, next) => {
  if (!isAgentLoopRequest(options)) return next()

  const expected = session.deriveMessages()
  if (JSON.stringify(options.messages) !== JSON.stringify(expected)) {
    fail('model request diverges from the durable session projection')
  }

  return next()
}, { global: true, prepend: true })

如果某个插件直接修改 options.messages，却没有追加对应事件，请求会在分派时失败，而不是等重启后才发现模型少了一段历史。这让“日志是事实来源”变成了一个可执行的检查。

The Session log: model history is recomputed

We can now answer a more basic question: where does step() get the messages[] for the next request? dsh does not keep one mutable array forever. It uses event sourcing: a Session appends facts in order, then computes history from those facts when needed.

Why not store messages[] directly?

An array is easier at first, but the running process and the durable log can then become two sources of state:

  1. the in-memory array tells the model and UI what the current history is;
  2. the durable log tells a restarted process what the history is.

If a tool result reaches the log but the process exits before updating the array, the two states diverge. Event sourcing fixes the write order: “record the fact first, then fold the view from the fact”:

1
2
3
4
SessionEvent[]
  -> event seqs in the current surface
  -> Message[]
  -> DeepSeek wire messages

Do not collapse these layers. The log contains boundaries, streaming deltas, and tool execution; the surface decides which messages are visible to the model; Message[] is dsh’s provider-neutral message type; wire messages are the format in the DeepSeek HTTP request.

A session log

The repository’s test fixture stores this process as JSONL. The following is a structural excerpt with arguments omitted:

1
2
3
4
5
6
7
8
9
10
11
{"type":"turn/start","seq":1}
{"type":"step/start","seq":2,"data":{"turn":1,"step":1}}
{"type":"user/message","seq":3}
{"type":"assistant/chunk","seq":4}
{"type":"assistant/message","seq":8}
{"type":"tool/call","seq":9,"data":{"name":"bash","callId":"call-1"}}
{"type":"tool/result","seq":10,"data":{"callId":"call-1"}}
{"type":"step/end","seq":11}
{"type":"step/start","seq":12,"data":{"turn":1,"step":2}}
{"type":"assistant/message","seq":13}
{"type":"turn/end","seq":14,"data":{"reason":{"kind":"completed"}}}

assistant/chunk and assistant/message are both model-related, but they serve different purposes: chunks support streaming display and replay, while the complete message enters the next model history. tool/call records what is about to happen, and tool/result records what actually happened; callId pairs them.

How one event becomes one message

deriveEventMessage() in packages/core/session/src/surface.ts only decides whether one event produces one internal message:

1
2
3
4
5
6
7
8
9
10
11
12
13
export function deriveEventMessage(event: SessionEvent): Message | null {
  switch (event.type) {
    case 'user/message':
      return event.data
    case 'assistant/message':
      if (event.data.message.content.length === 0) return null
      return event.data.message
    case 'tool/result':
      return event.data.message
    default:
      return null
  }
}

This small function exposes several important facts: assistant/chunk never enters history directly; only a non-empty assistant/message does; and tool/result enters history even though its internal role is still user. It becomes tool only when serialized for DeepSeek, as explained later.

Event names and payloads are described by one SessionEventMap. Plugins can add their own events through TypeScript declaration merging without editing Session core:

1
2
3
4
5
6
7
8
export interface SessionEventMap {
  'turn/start': { turn: number }
  'turn/end': { turn: number; reason: TurnEndReason }
  'user/message': UserMessage
  'assistant/message': { message: AssistantMessage }
  'tool/call': { callId: string; name: string; arguments: string }
  'tool/result': { message: ToolResultMessage }
}

The outer event envelope also contains type, seq, time, and data. When reading an old log, an unknown event may be skipped only when it explicitly carries ignorable: true; otherwise reconstruction is rejected. The goal is to avoid silently rebuilding a session with missing facts.

The invariant before dispatch

“History can be computed from the log” is not enough. The runtime also checks that the request being sent did not bypass the log. agent-loop/src/invariant.ts registers a priority listener on llm/stream and compares requests marked as agent-loop requests:

1
2
3
4
5
6
7
8
9
10
ctx.on('llm/stream', (options, next) => {
  if (!isAgentLoopRequest(options)) return next()

  const expected = session.deriveMessages()
  if (JSON.stringify(options.messages) !== JSON.stringify(expected)) {
    fail('model request diverges from the durable session projection')
  }

  return next()
}, { global: true, prepend: true })

If a plugin edits options.messages without appending a matching event, the request fails at dispatch time instead of producing a confusing missing-history bug after restart. “The log is the source of facts” is therefore an executable check, not just an architectural sentence.

Surface：压缩上下文，但不删除日志

事件溯源会带来一个现实问题：会话足够长时，所有消息无法继续塞进模型的上下文窗口。如果直接删除旧事件，审计、回放和崩溃恢复都会失去依据。

dsh 把“完整日志”和“当前给模型看的历史”分开。log 保存所有事实，surface 只保存当前可见事件的 seq 列表。它不是第二份消息数组，而是从日志事件折叠出的一个视图。

假设当前 surface 是：

1
2
log:     [3, 8, 10, 13, 14]
surface: [3, 8, 10, 13, 14]

压缩前四个节点后，插件追加一个 seq 为 20 的摘要事件，并声明它替换 [3, 13]：

1
2
log:     [3, 8, 10, 13, 14, 20]
surface: [20, 14]

旧事件仍在 log 中，只有模型历史不再直接看到它们。人类读取完整 transcript 时可以看 log，模型请求时读取 surface。

surfaceOp 和 sourceEventSeqs

可以进入 surface 的消息事件会在事件信封上带两个字段：

  • surfaceOp 表示怎样进入 surface。'append' 是追加，{ op: 'replace', start, end } 是替换当前 surface 的一个闭区间；
  • sourceEventSeqs 表示这条消息由哪些事件产生。流式 assistant message 会引用它的 chunk，替换事件则必须覆盖被遮蔽的节点。

上下文压缩的核心代码可以缩写成：

1
2
3
4
5
6
7
8
session.append('user/message', {
  content: [{ type: 'text', text: summary }],
  source: { kind: 'plugin' },
  role: 'user',
}, {
  surfaceOp: { op: 'replace', start, end },
  sourceEventSeqs: shadowedSeqs,
})

这里没有修改旧事件，只是追加了一个带操作说明的新事件。SurfaceManager 会检查 start 和 end 是否在当前 surface 中，也会检查 sourceEventSeqs 是否覆盖所有被替换节点。回放时重新执行这些 append/replace 操作，就能得到相同的 surface。

因此，compaction 并不是“把历史数组截短”，而是“向日志追加一次可回放的视图变更”。这也是它能和事件溯源同时成立的原因。

Surface: compact the context without deleting the log

Event sourcing creates a practical problem: a sufficiently long session cannot fit every message into the model’s context window. Deleting old events would remove the basis for audit, replay, and crash recovery.

dsh separates the complete log from the history currently visible to the model. log keeps every fact, while surface keeps only an ordered list of visible event seqs. It is not a second message array; it is a view folded from the event log.

Suppose the current surface is:

1
2
log:     [3, 8, 10, 13, 14]
surface: [3, 8, 10, 13, 14]

After compacting the first four nodes, the plugin appends a summary event at seq 20 and says that it replaces [3, 13]:

1
2
log:     [3, 8, 10, 13, 14, 20]
surface: [20, 14]

The old events remain in log; they simply stop being directly visible in model history. A human transcript can read the full log, while a model request reads the surface.

surfaceOp and sourceEventSeqs

Message events that can enter the surface carry two fields on their event envelope:

  • surfaceOp says how the message enters the surface. 'append' adds it, while { op: 'replace', start, end } replaces a closed range in the current surface;
  • sourceEventSeqs says which events produced the message. A streaming assistant message cites its chunks, and a replacement must cover every shadowed node.

The central part of compaction can be shortened to this:

1
2
3
4
5
6
7
8
session.append('user/message', {
  content: [{ type: 'text', text: summary }],
  source: { kind: 'plugin' },
  role: 'user',
}, {
  surfaceOp: { op: 'replace', start, end },
  sourceEventSeqs: shadowedSeqs,
})

No old event is edited. A new event records the view operation. SurfaceManager checks that start and end belong to the current surface and that sourceEventSeqs covers every replaced node. Replay runs the same append/replace operations and produces the same surface.

Compaction is therefore not “truncate the history array”. It is “append a replayable view change to the log”. That is why it can coexist with event sourcing.

工具调度：执行可以并发，写回必须保序

模型一次响应可能给出多个工具调用，例如按顺序调用 A、B、C。调度器要同时满足两件事：互不影响的调用可以重叠执行；下一次模型请求仍然按 A、B、C 的顺序看到结果。

这里要区分两个顺序：模型序是 assistant message 中 tool-call block 的顺序，完成序是工具 promise 实际结束的顺序。并发只改变完成序，不能改变日志里的模型序。

先判断工具是否允许并发

工具是否可并发由 isConcurrencySafe(args) 明确声明。调度器的分类器是 fail-closed 的：只有明确返回严格 true 才允许并发。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
executionMode(exec): ToolExecutionMode {
  const tool = this.resolveExecution(
    exec.name,
    exec.agent,
    exec.parent !== undefined,
  )
  if (!tool?.isConcurrencySafe) return { kind: 'exclusive' }

  try {
    const safe = tool.isConcurrencySafe(exec.arguments)
    return safe === true
      ? { kind: 'parallel' }
      : { kind: 'exclusive' }
  } catch {
    return { kind: 'exclusive' }
  }
}

未注册、隐藏、没有声明、参数不合法或分类器抛异常时，调用都会按 exclusive 处理。这个保守规则很重要：读取两个不同文件通常可以并行，但写文件、改配置或调用有外部副作用的接口，不能因为“看起来应该没问题”就并发。

屏障和滚动池

连续的 parallel 调用组成一个有界滚动池，最多有 maxParallelToolCalls 个调用在执行。exclusive 调用是屏障，会单独成为一个 group，不与前后的调用重叠。

如果 A、B 是 parallel，C 是 exclusive，执行形状是：

1
2
group 1: [A, B]  -> 可以重叠执行
group 2: [C]     -> 等 group 1 完成后独占执行

每次 group 开始和滚动池补充调用前，调度器都会重新分类还没启动的调用。因此注册表在等待期间发生变化时，不会把已经变成 exclusive 的调用继续塞进 parallel 池。

结果按模型序提交

假设 B 在 100 毫秒后完成，A 需要 2 秒。B 的结果虽然已经准备好，但不能先写入日志，因为模型认为 A 在前面。commitReady() 只推进连续的 slot：

1
2
3
4
5
6
while (committed < group.length) {
  const slot = slots[committed]
  if (slot === undefined) break
  appendToolResult(slot)
  committed++
}

A 完成后，A 和 B 才会一起按顺序提交。这样下一步的 tool/result 消息、日志和 UI 看到的都是模型顺序，而工具 body 仍然可以利用并发节省时间。

取消和未知结果

取消时必须区分“body 已启动”和“body 尚未启动”：

1
2
export const TOOL_ABORTED = 'ABORTED'
export const TOOL_ABORTED_BEFORE_DISPATCH = 'ABORTED_BEFORE_DISPATCH'

已经启动的 promise 要先结束，之后才能把结果标成 ABORTED；没有启动的调用则记录 tool/call，再合成一个 tool/result，说明它在 dispatch 前被取消。内部 scheduler 自身崩溃时，已经写入的 tool/call 不会被假装配上成功结果，而是交给恢复器记录未知结果。

工具调度完成后，结果会成为下一次 step 的输入。这就是“可以并发执行”和“模型仍看到确定顺序”同时成立的地方。

Tool scheduling: execution may overlap, commits must be ordered

A model response may contain several tool calls, for example A, B, and C in that order. The scheduler must satisfy two requirements: independent calls may overlap, but the next model request must still see results as A, B, C.

Separate two orders: model order is the order of tool-call blocks in the assistant message, while completion order is the order in which tool promises settle. Concurrency changes completion order, never the order written to the log.

First decide whether a tool may run concurrently

Concurrency is explicitly declared by isConcurrencySafe(args). The classifier is fail-closed: only a strict true enables parallel execution.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
executionMode(exec): ToolExecutionMode {
  const tool = this.resolveExecution(
    exec.name,
    exec.agent,
    exec.parent !== undefined,
  )
  if (!tool?.isConcurrencySafe) return { kind: 'exclusive' }

  try {
    const safe = tool.isConcurrencySafe(exec.arguments)
    return safe === true
      ? { kind: 'parallel' }
      : { kind: 'exclusive' }
  } catch {
    return { kind: 'exclusive' }
  }
}

An unregistered, hidden, undeclared, invalid, or throwing classifier produces exclusive. The conservative rule matters: reading two different files may be safe to overlap, while writing a file, changing configuration, or calling an external side-effecting API should not be concurrent just because it looks harmless.

Barriers and the rolling pool

Consecutive parallel calls form a bounded rolling pool with at most maxParallelToolCalls calls in flight. An exclusive call is a barrier and forms its own group; it never overlaps calls before or after it.

If A and B are parallel while C is exclusive, the shape is:

1
2
group 1: [A, B]  -> may overlap
group 2: [C]     -> exclusive after group 1 completes

Before each group starts and before the pool is refilled, the scheduler reclassifies calls that have not started. A registry change therefore cannot quietly put a newly exclusive call into the parallel pool.

Results commit in model order

Suppose B finishes in 100 milliseconds while A takes two seconds. B is ready, but it cannot be written first because the model placed A before it. commitReady() advances only through contiguous slots:

1
2
3
4
5
6
while (committed < group.length) {
  const slot = slots[committed]
  if (slot === undefined) break
  appendToolResult(slot)
  committed++
}

Once A finishes, A and B can commit in order. The next step, the log, and the UI all see model order, while the tool bodies still benefit from concurrency.

Cancellation and unknown outcomes

Cancellation distinguishes a body that has started from one that has not:

1
2
export const TOOL_ABORTED = 'ABORTED'
export const TOOL_ABORTED_BEFORE_DISPATCH = 'ABORTED_BEFORE_DISPATCH'

A started promise must settle before its outcome is marked ABORTED. An unstarted call records tool/call and then receives a synthesized tool/result saying it was cancelled before dispatch. If the scheduler itself fails, an already-recorded tool/call is not paired with a fabricated success; recovery later records the outcome as unknown.

After scheduling completes, the results become input for the next step. This is how execution can overlap while the model still observes a deterministic order.

LLM 层：把 DeepSeek SSE 变成统一的流

agent-loop 不应该知道 DeepSeek SSE 的字段名，也不应该为每个 provider 写一套消息组装代码。LLM 层把 provider 的响应分成三层：

1
2
3
4
DeepSeek SSE
  -> DeepSeekAdapter 翻译成 StreamChunk
  -> BlockAssembler 组装成 ContentBlock
  -> agent-loop 生成 assistant/message

StreamChunk 解决什么问题

模型输出是流式的，一次响应可能同时交错 reasoning、文本和多个工具调用。dsh 用 index 标记 block，用统一的 chunk 类型描述不同 provider 的增量：

chunk 含义
block-start 一个 reasoning、text 或 tool-call block 开始
text-delta 文本增量
reasoning-delta reasoning 增量
tool-call-delta 工具名或参数的增量
block-end 一个完整内容块
usage token 用量
finish 本次流的结束原因

同一个 index 的增量会被 BlockAssembler 放入同一个 block。即使某个 provider 省略了 block-start，assembler 也会在第一次看到 index 时创建临时 block；已经结束的 block 后面再收到 delta，则会被忽略。若因 max-tokens 得到半截 tool-call，dsh 不会执行它。

为什么内部工具结果是 user

dsh 内部的 Message.role 只有三种：system、user、assistant。所以工具结果在内部表示为一个 user-role message，其中放着 tool-result block：

1
2
3
4
5
interface ToolResultMessage extends Message {
  readonly role: 'user'
  readonly content: [ToolResultBlock]
  readonly source: ToolMessageSource
}

这不是 DeepSeek API 的角色定义，而是 harness 内部的 provider-neutral 表示。序列化到 DeepSeek wire 时，serializeMessages() 才根据 source 把它展开为 role: 'tool'：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const toolResults = message.content.filter(
  block => block.type === 'tool-result',
)
const text = flattenText(message.content)

if (text.length > 0 || toolResults.length === 0) {
  wire.push({ role: 'user', content: text })
}

for (const result of toolResults) {
  wire.push({
    role: 'tool',
    tool_call_id: result.toolCallId,
    content: flattenText(result.content) || '(no output)',
  })
}

这样 Session 日志不需要知道 provider 的消息角色差异；更换 provider 时，只需要替换最外层 serializer 和 adapter。

DeepSeek 适配器的边界

packages/llm/llm-deepseek/src 负责处理 DeepSeek 特有的事实：

  • 工具参数在流中保持 raw JSON 字符串，只有收齐增量后才作为完整参数使用；
  • 要等到 [DONE] 才最终发出 block-end、usage 和 finish，因为这些信息可能分散在终止 chunk 和 trailing chunk 中；
  • thinking 模式的第一个空字符串不能开启空的 reasoning block；
  • EOF 没有 [DONE] 时抛出 STREAM_CLOSED，截断的响应不能当作成功；
  • provider 返回 stop 但没有任何内容时，报告 EMPTY_RESPONSE，交给上层重试策略处理；
  • prompt_tokens 包含 cache hit，mapUsage() 会把缓存 token 从普通 input token 中拆出来。

一次 adapter 调用只代表一次 provider attempt。adapter 不在内部偷偷重试；是否重试由 agent/request-error 决定，重试时 step() 在同一个 turn 和 step 内重建请求。

取消信号也在 adapter 这一层合并：

1
2
3
4
const consumer = new AbortController()
const upstream = options.signal === undefined
  ? consumer.signal
  : AbortSignal.any([options.signal, consumer.signal])

调用方可以取消 turn，adapter 自己也可以因为 idle watchdog 或内部清理而取消请求。finally 中终止 consumer，让所有退出路径使用同一套清理逻辑。

The LLM layer: turn DeepSeek SSE into one stream shape

agent-loop should not know DeepSeek SSE field names, and it should not contain a separate message assembler for every provider. The LLM layer separates provider responses into three layers:

1
2
3
4
DeepSeek SSE
  -> DeepSeekAdapter translates it into StreamChunk
  -> BlockAssembler builds ContentBlock values
  -> agent-loop creates assistant/message

What StreamChunk solves

Model output is streamed. Reasoning, text, and multiple tool calls may be interleaved in one response. dsh uses index to identify a block and a provider-neutral set of chunk shapes:

Chunk Meaning
block-start A reasoning, text, or tool-call block starts
text-delta A text increment
reasoning-delta A reasoning increment
tool-call-delta A tool name or argument increment
block-end A complete content block
usage Token usage
finish The end reason for the stream

BlockAssembler puts deltas with the same index into the same block. Even if a provider omits block-start, the assembler creates a temporary block when an index first appears. Deltas received after a block has ended are ignored. If max-tokens leaves a partial tool call, dsh does not execute it.

Why an internal tool result is user

The internal Message.role vocabulary has only system, user, and assistant. A tool result is therefore represented as a user-role message containing a tool-result block:

1
2
3
4
5
interface ToolResultMessage extends Message {
  readonly role: 'user'
  readonly content: [ToolResultBlock]
  readonly source: ToolMessageSource
}

This is not DeepSeek’s wire vocabulary; it is the harness’s provider-neutral representation. When serializing for DeepSeek, serializeMessages() uses source to expand it into role: 'tool':

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const toolResults = message.content.filter(
  block => block.type === 'tool-result',
)
const text = flattenText(message.content)

if (text.length > 0 || toolResults.length === 0) {
  wire.push({ role: 'user', content: text })
}

for (const result of toolResults) {
  wire.push({
    role: 'tool',
    tool_call_id: result.toolCallId,
    content: flattenText(result.content) || '(no output)',
  })
}

The Session log therefore does not need to know provider-specific role differences. Swapping providers only changes the outer serializer and adapter.

Boundaries owned by the DeepSeek adapter

packages/llm/llm-deepseek/src owns these DeepSeek-specific facts:

  • Tool arguments remain raw JSON strings during the stream and are used as complete arguments only after all deltas arrive;
  • finalization waits for [DONE], because block-end, usage, and finish may be split between terminal and trailing chunks;
  • the first empty thinking string must not open an empty reasoning block;
  • EOF without [DONE] raises STREAM_CLOSED, because a truncated response cannot be trusted;
  • a stop with no content becomes EMPTY_RESPONSE, which the upper retry policy may handle;
  • prompt_tokens includes cache hits, so mapUsage() separates cached tokens from ordinary input tokens.

One adapter call represents one provider attempt. The adapter does not retry secretly; agent/request-error decides whether to retry, and step() rebuilds the request inside the same turn and step.

The adapter also combines cancellation signals:

1
2
3
4
const consumer = new AbortController()
const upstream = options.signal === undefined
  ? consumer.signal
  : AbortSignal.any([options.signal, consumer.signal])

The caller can cancel a turn, while the adapter can cancel itself because of an idle watchdog or internal cleanup. Aborting consumer in finally gives every exit path the same cleanup behavior.

Cordis 和插件：这些组件怎样组合起来

到这里已经看完一次请求的主路径，再回头看“everything is a plugin”就不会只剩一句口号了。插件化解决的不是“把文件拆成更多文件”，而是让使用能力的代码不必依赖某个具体实现。

最小的依赖关系可以写成：

1
2
3
4
5
6
llm 插件
  -> 创建 ctx.llm 服务
llm-deepseek 插件
  -> 向 ctx.llm 注册 deepseek-official adapter
agent-loop 插件
  -> 调用 ctx.llm.stream()

agent-loop 不需要 import DeepSeekAdapter。如果换一个 provider，只要注册另一个 adapter，主循环和 Session 事件格式都可以不变。

Context、inject 和事件

Cordis 是 dsh vendored 的插件框架。可以先把 Context 理解成服务目录：ctx.llm、ctx.tools、ctx.sessions 是其它插件查找能力的入口。

插件通过 inject 声明自己需要的服务。框架会等依赖服务出现后再执行插件，而不是要求开发者手工排列一个越来越长的启动顺序。

插件之间除了直接调用服务，也可以监听类型化事件。常见派发模式的区别是：

模式 监听器关系 是否等待结果
emit 按注册顺序广播 否
waterfall 监听器通过 next() 组成一条链 有
parallel 多个监听器同时工作 等全部完成
serial 按顺序询问策略 有

waterfall 最容易出错。只观察或改写 payload 的监听器必须调用 next()：

1
2
3
4
ctx.on('agent/pre-step', async (payload, next) => {
  if (payload.messages.length === 0) return { kind: 'reject' }
  return next()
})

直接 return 表示监听器要短路并接管决定；调用 next() 才会让后面的监听器和默认实现继续。agent/pre-step 和 agent/request-error 都利用了这个规则：前者可以拒绝或改写输入，后者可以返回 { kind: 'retry' }。

scope 让不同 Agent 看到不同能力

同一个进程可能同时运行主 Agent 和 subagent。主 Agent 可以使用 bash，某个 subagent 可能只能读文件。如果工具和 prompt 都是全局注册，它们会互相污染。

每个活着的 Agent 都有自己的 scope。通过 agent.ctx 注册的工具、prompt 和监听器只对该 Agent 及允许继承的子层可见，并随 Agent 销毁而撤回。查找注册时子 scope 可以看到父 scope 的能力，同名注册由更近的一层覆盖。

scope 的 parent 关系只能按受控方式建立：

1
2
3
4
5
6
7
8
9
const scopeParents = new WeakMap<ScopeKey, ScopeKey>()

export function bindScopeParent(key: ScopeKey, parent: ScopeKey) {
  if (scopeParents.has(key)) {
    throw new Error('scope key is already bound')
  }
  linkScopeParent(key, parent)
  return { rebind(next: ScopeKey) { linkScopeParent(key, next) } }
}

这段约束避免了 scope 被任意重新连接。开发期的 scope-invariant 插件还会检查事件 carrier 和 payload subject 是否指向同一个 scope，减少“监听器静默收不到事件”的问题。

注册必须可以撤销

插件不仅要安装工具和监听器，还要负责卸载。ctx.effect() 把注册和 disposer 绑在一起，并在插件卸载时清理。多个清理动作如果有顺序要求，应该放进同一个 disposer：

1
2
3
4
5
ctx.effect(async () => {
  await stopDriver()
  await removeRegistrations()
  await closeStore()
})

多个独立的 async disposer 之间可能并发执行，不能依赖它们的注册顺序。对 agent-loop 来说，通常要先停 driver，再拆监听器和工具注册，最后关闭 Session store。

Profile、bundle 和 patch

插件如何被加载由启动配置决定：

1
2
3
4
5
6
profile
  -> bundle 列表展开成插件 entry
  -> profile cordis.patch.yml
  -> home-level patch
  -> 命令行 --patch
  -> 最终插件树

后应用的同 id entry 会替换先前的 entry，新的 id 会插入列表。dsh --profile web --dump-config 可以先查看最终配置，再只替换想改的那一行，不必复制整个 profile。

Cordis and plugins: how the components are composed

Now that the main request path is clear, “everything is a plugin” is easier to interpret. Pluginization is not merely splitting one program into more files. It lets code that consumes a capability avoid depending on one concrete implementation.

The smallest dependency chain looks like this:

1
2
3
4
5
6
llm plugin
  -> creates the ctx.llm service
llm-deepseek plugin
  -> registers the deepseek-official adapter in ctx.llm
agent-loop plugin
  -> calls ctx.llm.stream()

agent-loop does not need to import DeepSeekAdapter. To change providers, register another adapter while leaving the main loop and Session event format intact.

Context, inject, and events

Cordis is the vendored plugin framework used by dsh. Think of Context as a service directory: ctx.llm, ctx.tools, and ctx.sessions are the entry points other plugins consume.

Plugins declare required services through inject. The framework waits for those services before applying the plugin, instead of making developers maintain a long manual boot order.

Plugins can also listen to typed events instead of calling services directly. The common dispatch modes differ as follows:

Mode Listener relationship Waits for results?
emit Broadcast in registration order No
waterfall Listeners form a chain through next() Yes
parallel Independent listeners run together Waits for all
serial Strategies are asked in order Yes

waterfall is the easiest mode to misuse. A listener that only observes or rewrites a payload must call next():

1
2
3
4
ctx.on('agent/pre-step', async (payload, next) => {
  if (payload.messages.length === 0) return { kind: 'reject' }
  return next()
})

Returning directly means the listener short-circuits and owns the decision. Calling next() lets later listeners and the default implementation continue. agent/pre-step and agent/request-error use this rule: the first can reject or rewrite input, while the second can return { kind: 'retry' }.

Scope gives each Agent a different view

One process may run a main Agent and several subagents. The main Agent may have bash access while a subagent is read-only. If tools and prompts were global, those configurations would leak between agents.

Each live Agent owns a scope. Tools, prompts, and listeners registered through agent.ctx are visible only to that Agent and permitted child layers, and are removed with the Agent. A child can look up parent registrations, while a nearer registration shadows a name farther away.

The parent relationship is deliberately controlled:

1
2
3
4
5
6
7
8
9
const scopeParents = new WeakMap<ScopeKey, ScopeKey>()

export function bindScopeParent(key: ScopeKey, parent: ScopeKey) {
  if (scopeParents.has(key)) {
    throw new Error('scope key is already bound')
  }
  linkScopeParent(key, parent)
  return { rebind(next: ScopeKey) { linkScopeParent(key, next) } }
}

The constraint prevents arbitrary scope relinking. During development, the scope-invariant plugin also checks that the event carrier and payload subject refer to the same scope, reducing silent missed-listener bugs.

Registrations must be reversible

A plugin must remove the tools and listeners it installs. ctx.effect() ties registration to a disposer and cleans it up during unload. If cleanup has ordering requirements, keep it in one disposer:

1
2
3
4
5
ctx.effect(async () => {
  await stopDriver()
  await removeRegistrations()
  await closeStore()
})

Independent async disposers may run concurrently, so their registration order is not a sequencing mechanism. For agent-loop, the usual order is to stop the driver, remove listeners and tool registrations, and close the Session store last.

Profiles, bundles, and patches

Startup composition is controlled by configuration:

1
2
3
4
5
6
profile
  -> bundle list expands into plugin entries
  -> profile cordis.patch.yml
  -> home-level patch
  -> command-line --patch
  -> final plugin tree

A later entry with the same id replaces the earlier one; a new id is inserted. dsh --profile web --dump-config lets you inspect the final configuration and replace one row without copying the entire profile.

持久化与崩溃恢复：日志为什么能够继续读

事件日志只有真正落到磁盘，重启恢复才有意义。dsh 把持久化定义成一个能力接缝，上层只依赖 append、load、flush 和 revision，不直接依赖数据库 API。

当前有两个后端：

SQLite JSONL
每个 Session sessions 一行加 events 多行 一个 session.jsonl 文件
读取方式 可以按 seq 查询 主要是顺序读取
版本检查 application_id 和 user_version 头行的事件格式版本
撕裂检测 scanRows scanLog 或压缩帧扫描

两者的存储形式不同，但都要把同一组事件恢复成同一份 Session。

append 不等于已经写入磁盘

典型的写入路径是：

1
2
3
4
session.append(event)
  -> session/event 进入 write-behind 缓冲
  -> 等待批处理窗口，或收到 session/flush
  -> 后端写入磁盘

默认的批处理窗口是有限的；模型请求和工具执行开始前，checkpoint policy 会触发 flush。这样之前的 user/message 和 tool/call 不会只存在于内存中而请求已经发出。

JSONL 的物化使用临时文件和 link(tmp, final) + unlink(tmp)，避免并发写入时用 rename 静默覆盖已有文件。SQLite 会检查 PRAGMA application_id = 0x44534850，并在 schema 版本不匹配时拒绝打开，而不是在预发布阶段原地迁移。

进程中途退出怎么办

假设进程停在下面的位置：

1
2
3
4
tool/call 已写入
工具 body 可能已经启动
tool/result 尚未写入
step/end 和 turn/end 也不存在

冷加载时，packages/core/session/src/repair.ts 的恢复器不会删除整个 turn，也不会把未知结果当成成功。它会保留已有事件，并补上缺失的结构：

  1. 对未配对的 tool/call 合成结果；如果调用已经开始，错误码是 TOOL_OUTCOME_UNKNOWN，因为外部副作用是否发生无法确认；
  2. 如果 step 仍然打开，追加 step/end；
  3. 追加 turn/end { reason: { kind: 'interrupted' } }。

completed、aborted、blocked、error 和 max-tokens 可以由主循环写出，唯独 interrupted 只由冷加载恢复器合成。因此在日志中看到 turn/end: interrupted，就能知道前一个进程没有正常闭合这轮任务。

恢复事件使用接续的 seq 和最后一条真实事件的时间，保证重放确定。需要注意：工具的真实外部副作用不能由日志自动撤销，也不能在未知结果时无条件重试。是否安全，要看工具是否只读或幂等。

读到 chunk-rows 时不要把它当成新事件

流式输出会产生大量很小的 assistant/chunk。为了减少 JSONL 的重复 envelope，存储层可以把连续的同类 delta 打包成 text-chunks、reasoning-chunks 或 tool-call-chunks。

它只压缩满足条件的连续 run：同一种 delta、同一个 block index、seq 连续、turn 和 step 相同；工具调用还要保持相同的 id。任何未知字段或未来的 chunk 形状都原样存储，宁愿不压缩也不丢数据。读取时这些 packed rows 会还原成原来的 chunk 事件，所以它不改变 Agent 的运行逻辑。

事件格式版本和磁盘 schema 版本也分别管理两件事：SESSION_FORMAT_VERSION 管事件信封和事件词汇，SCHEMA_VERSION 管物理存储布局。新增事件不一定要改 SQLite 表结构，改表结构也不等于新增事件。

Persistence and crash recovery: how the log remains readable

Event logging matters for restart recovery only after facts reach disk. dsh exposes persistence as a capability seam. Upper layers depend on append, load, flush, and revision rather than on a database API.

There are two backends:

SQLite JSONL
Per Session One sessions row plus many events rows One session.jsonl file
Reading Can query by seq Mainly sequential
Version checks application_id and user_version Event-format version in the header
Torn-tail scan scanRows scanLog or compressed-frame scanning

The physical representations differ, but both backends must rebuild the same Session from the same events.

append does not mean “on disk”

The normal write path is:

1
2
3
4
session.append(event)
  -> session/event enters a write-behind buffer
  -> the batch window expires, or session/flush is requested
  -> the backend writes to disk

The batch window is finite. Before a model request or tool execution starts, checkpoint policy requests a flush, so the preceding user/message and tool/call are not only in memory while the request has already left the process.

JSONL materialization uses a temporary file and link(tmp, final) + unlink(tmp) to avoid the silent overwrite that rename could cause during concurrent writes. SQLite checks PRAGMA application_id = 0x44534850 and rejects a schema mismatch instead of doing an in-place migration during preview.

What if the process exits halfway through?

Suppose the process stops here:

1
2
3
4
tool/call is durable
the tool body may have started
tool/result is not durable yet
step/end and turn/end do not exist

During a cold load, the repairer in packages/core/session/src/repair.ts does not delete the whole turn or pretend that the unknown outcome was successful. It keeps existing events and adds only missing structure:

  1. It synthesizes a result for each unpaired tool/call. If the call started, the code is TOOL_OUTCOME_UNKNOWN, because the outside side effect cannot be determined;
  2. It appends step/end when the step is still open;
  3. It appends turn/end { reason: { kind: 'interrupted' } }.

The loop may write completed, aborted, blocked, error, and max-tokens; only the cold-load repairer synthesizes interrupted. Seeing turn/end: interrupted in a log therefore means the previous process did not close that turn normally.

Repair events continue the seq and reuse the time of the last real event, keeping replay deterministic. A tool’s external side effect cannot be undone automatically by the log, and an unknown outcome must not be retried unconditionally. Safety depends on whether the tool is read-only or idempotent.

chunk-rows is storage, not a new runtime step

Streaming output creates many small assistant/chunk events. To reduce repeated JSON envelopes, the storage layer may pack consecutive deltas into text-chunks, reasoning-chunks, or tool-call-chunks rows.

Packing is allowed only for a matching run: the same delta kind, block index, contiguous seqs, and the same turn and step; tool-call runs must also keep the same id. Unknown fields and future chunk shapes are stored verbatim. Losing compression is preferable to losing data. On read, packed rows expand back into the original chunk events, so this optimization does not change Agent behavior.

The event-format version and disk-schema version also manage different things: SESSION_FORMAT_VERSION describes the event envelope and vocabulary, while SCHEMA_VERSION describes the physical storage layout. Adding an event does not necessarily require a SQLite table change, and changing the table layout is not the same as adding an event name.

复盘 echo hello：按这条路径读源码

把全文压缩成一次请求，顺序是：

  1. 用户输入进入 durable inbox，输入事件先被记录；
  2. ReactLoopAgent.turn() 建立 turn 边界，preStep() 领取输入并组装 prompt；
  3. Session 从 surface 投影出 Message[]，agent-loop 通过 llm/stream 请求模型；
  4. DeepSeek adapter 把 SSE 翻译成 StreamChunk，BlockAssembler 组装出 bash tool-call；
  5. 工具调度器记录 tool/call，执行 bash，并按模型顺序写入 tool/result；
  6. 当前 step 结束，下一 step 从日志投影中看到工具结果，模型生成最终文本；
  7. 如果进程中途退出，冷加载 repair 补出确定的工具结果和结束边界，但不猜测未知的外部副作用。

读到这里，可以用四个问题检查自己是否真的理解了这套代码：

  1. 模型请求能不能从持久化事件重新计算，而不是依赖一份可能漂移的数组？
  2. 流式增量、完整消息、工具调用和工具结果的边界在哪里？
  3. 并发、取消和崩溃时，日志能不能说明哪些调用已经开始、哪些结果未知？
  4. 换掉 provider 或销毁 Agent 后，工具、监听器和资源能不能撤回？

对应到源码，可以继续按下面的顺序阅读：

1
2
3
4
5
6
agent-loop/src/agent.ts
  -> session/src/surface.ts
  -> tools/src/tool-calls.ts
  -> llm/llm/src/assembler.ts
  -> llm/llm-deepseek/src/adapter.ts
  -> session/src/repair.ts

dsh 的设计重点不是某一个 API，而是把这些边界落实成了事件、投影函数、调度器和恢复器。先沿着一个具体任务理解它们，再回头看 Cordis、surface replace 和 chunk-rows，源码会比从“everything is a plugin”开始更容易读。

Review echo hello: a source-reading path

The complete request can be reduced to this order:

  1. User input enters the durable inbox, and the input event is recorded first;
  2. ReactLoopAgent.turn() opens the turn boundary, while preStep() claims input and assembles the prompt;
  3. Session projects Message[] from the surface, and agent-loop calls the model through llm/stream;
  4. The DeepSeek adapter translates SSE into StreamChunk, and BlockAssembler builds the bash tool call;
  5. The scheduler records tool/call, runs bash, and writes tool/result in model order;
  6. The current step ends, the next step projects the tool result from the log, and the model produces final text;
  7. If the process exits halfway through, cold-load repair adds deterministic results and boundaries without guessing the unknown outside side effect.

Use these four questions to check whether the architecture is clear:

  1. Can the model request be recomputed from durable events instead of a mutable array that may drift?
  2. Where are streaming deltas, complete messages, tool calls, and tool results separated?
  3. During concurrency, cancellation, and a crash, can the log say which calls started and which outcomes are unknown?
  4. After swapping a provider or destroying an Agent, can tools, listeners, and resources be removed?

The corresponding source-reading order is:

1
2
3
4
5
6
agent-loop/src/agent.ts
  -> session/src/surface.ts
  -> tools/src/tool-calls.ts
  -> llm/llm/src/assembler.ts
  -> llm/llm-deepseek/src/adapter.ts
  -> session/src/repair.ts

The important part of dsh is not one API. It turns these boundaries into events, projection functions, a scheduler, and a repairer. Follow one concrete task first, then return to Cordis, surface replacement, and chunk-rows; the source is easier to understand than when starting from the slogan “everything is a plugin”.