mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(agent-loop): await final session flush
This commit is contained in:
Tianyi Cui
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
2026-07-21-semantic-session-checkpoints.md: 0034cde40e5b07bda1573ca39fb7d51816006140
|
||||
2026-07-21-semantic-session-checkpoints.zh.md: 3351221d7eeaf1353b4adb0fa4c4dc324ec33da5
|
||||
# pnpm run verify-translation-pairing --write .agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md
|
||||
2026-07-21-semantic-session-checkpoints.md: 927a4c5d6d2aad5dea460ea29f97c1686e9d5398
|
||||
2026-07-21-semantic-session-checkpoints.zh.md: 6454b496aa8c03c172d6a4bc969e43e8dbca2430
|
||||
|
||||
@@ -10,11 +10,11 @@ Persistence buffered every synchronous `session/event` until the loop's final tu
|
||||
|
||||
## Decision
|
||||
|
||||
`dsh-session-checkpoint-policy` owns semantic durability barriers as a zero-config plugin beside a persistence backend. It wraps `llm/stream` lazily and flushes the live session after `request/header` is logged but before the adapter stream is constructed. It wraps top-level `tools/execute` after ordered pre-execute policy and flushes the recorded `tool/call` before the tool body; nested dispatches reuse the outer model-visible call. It flushes at `agent/post-step` after the assistant message and ordered results are recorded. The loop's existing final `turn/end` checkpoint remains the closing boundary.
|
||||
`dsh-session-checkpoint-policy` owns semantic durability barriers as a zero-config plugin beside a persistence backend. At `agent/step`, it flushes pending prompt input or the preceding response/result batch before the next request is derived. It wraps `llm/stream` lazily and flushes the live session after `request/header` is logged but before the adapter stream is constructed. It wraps top-level `tools/execute` after ordered pre-execute policy and flushes the recorded `tool/call` before the tool body; nested dispatches reuse the outer model-visible call. The loop's final `turn/end` checkpoint remains the closing boundary and settles before another queued turn or idle observation.
|
||||
|
||||
Persistence and checkpoint scheduling remain separate Cordis plugins. A backend makes requested `session/flush` boundaries durable but does not choose them; loading it without this policy is valid and retains the loop's coarser checkpoints. First-party persisted apps and runtimes explicitly mount both, while a specialized deployment may intentionally omit or replace the policy. Registration order governs whether events appended by other `agent/post-step` listeners join this checkpoint; the loop-owned assistant message and ordered results always precede the event.
|
||||
Persistence and checkpoint scheduling remain separate Cordis plugins. A backend makes requested `session/flush` boundaries durable but does not choose them; loading it without this policy is valid and retains the loop's coarser checkpoints. First-party persisted apps and runtimes explicitly mount both, while a specialized deployment may intentionally omit or replace the policy. Registration order governs whether events appended by other `agent/step` listeners precede this checkpoint; prompt input and the preceding loop-owned assistant message and ordered results are already in the log.
|
||||
|
||||
Checkpoint failure and cancellation are fail-closed at effect boundaries. A rejected request checkpoint prevents adapter dispatch; a rejected tool checkpoint becomes an error result without invoking the tool body. If cancellation lands while the tool checkpoint is pending, the policy rechecks the signal and returns the canonical `ABORTED_BEFORE_DISPATCH` result. A rejected post-step checkpoint stops continuation before another model request. Persistence serialization continues to belong to the coordinator, so concurrent tool checkpoints cannot duplicate event sequences.
|
||||
Checkpoint failure and cancellation are fail-closed at effect boundaries. A rejected request checkpoint prevents adapter dispatch; a rejected tool checkpoint becomes an error result without invoking the tool body. If cancellation lands while the tool checkpoint is pending, the policy rechecks the signal and returns the canonical `ABORTED_BEFORE_DISPATCH` result. A rejected between-step checkpoint closes the turn before another model request. A rejected final turn checkpoint is reported live and does not prevent later queued work. Persistence serialization continues to belong to the coordinator, so concurrent tool checkpoints cannot duplicate event sequences.
|
||||
|
||||
The ACP app owns its bridge, checkpoint policy, and persistence backend in one ordered Cordis effect. Cordis unloads sibling plugin effects concurrently, so independent mounts would let persistence detach while bridge teardown was still closing an interrupted turn. The composite lifecycle unloads the bridge first, waits for its agents to quiesce and flush the real `step/end` and `turn/end`, then removes checkpoint scheduling and persistence.
|
||||
|
||||
@@ -26,4 +26,4 @@ Flushing every event or streaming chunk minimizes loss but turns local append an
|
||||
|
||||
## Consequences
|
||||
|
||||
Hard-crash recovery retains the complete model request, durable tool intent, and complete settled step at the nearest semantic boundary while allowing partial streaming chunks since the previous boundary to remain lossy. Default CLI, TUI, ACP, Python SDK runtime, headless persistence tests, and JSON-RPC compositions mount the policy with their persistence backend. Unit tests cover ordering, cancellation during a checkpoint, fail-closed behavior, nested dispatch, disposal, and Loader shape; a real child process killed with `SIGKILL` proves request and tool-intent recovery through JSONL, and the shared persistence contract proves both recovery classifications across backends. The crash harness waits for the expected marker contents rather than path existence, so open-before-write visibility cannot trigger the kill early. Keyless ACP snapshots prove both that retry-risk guidance reaches resumed history and the next model turn and that graceful cancellation persists the loop's real closing boundaries.
|
||||
Hard-crash recovery retains the complete model request, durable tool intent, and complete settled step at the nearest semantic boundary while allowing partial streaming chunks since the previous boundary to remain lossy. Default CLI, TUI, ACP, Python SDK runtime, headless persistence tests, and JSON-RPC compositions mount the policy with their persistence backend. Unit tests cover ordering, cancellation during a checkpoint, fail-closed behavior, nested dispatch, disposal, Loader shape, and final-checkpoint ordering and failure containment; a real child process killed with `SIGKILL` proves request and tool-intent recovery through JSONL, and the shared persistence contract proves both recovery classifications across backends. The crash harness waits for the expected marker contents rather than path existence, so open-before-write visibility cannot trigger the kill early. Keyless ACP and SDK snapshots prove that retry-risk guidance reaches resumed history and the next model turn, graceful cancellation persists the loop's real closing boundaries, and SDK shutdown observes the complete persisted turn.
|
||||
|
||||
@@ -10,11 +10,11 @@ Status: implemented
|
||||
|
||||
## 决策
|
||||
|
||||
`dsh-session-checkpoint-policy` 以零配置插件的形式与持久化后端共同加载，并负责语义持久性屏障。该插件惰性包装 `llm/stream`，在记录 `request/header` 之后、构造适配器流之前，刷新活动会话。该插件还在有序的执行前策略之后包装顶层 `tools/execute`，在进入工具主体前刷新已记录的 `tool/call`；嵌套分发则复用外层模型可见调用。它还会在 `agent/post-step` 时刷新会话，此时模型消息与按序结果都已记录。现有的最终 `turn/end` 检查点仍是轮次的收尾边界。
|
||||
`dsh-session-checkpoint-policy` 以零配置插件的形式与持久化后端共同加载，并负责语义持久性屏障。在 `agent/step` 时，该插件会在推导下一个请求前刷新待持久化的提示词输入或前一批响应/结果。该插件惰性包装 `llm/stream`，在记录 `request/header` 之后、构造适配器流之前，刷新活动会话。该插件还在有序的执行前策略之后包装顶层 `tools/execute`，在进入工具主体前刷新已记录的 `tool/call`；嵌套分发则复用外层模型可见调用。循环的最终 `turn/end` 检查点仍是轮次的收尾边界，并会在处理另一个已排队轮次或观察到空闲状态之前完成。
|
||||
|
||||
持久化与检查点调度仍是相互独立的 Cordis 插件。后端使请求的 `session/flush` 边界持久化，但不选择边界；只加载后端而不加载本策略仍是有效组合，并保留循环提供的较粗检查点。第一方持久化应用与运行时会显式加载两者，专用部署则可以有意省略或替换本策略。注册顺序决定其他 `agent/post-step` 监听器追加的事件是否会纳入本检查点；循环自身记录的助手消息与有序结果始终先于该事件。
|
||||
持久化与检查点调度仍是相互独立的 Cordis 插件。后端使请求的 `session/flush` 边界持久化，但不选择边界；只加载后端而不加载本策略仍是有效组合，并保留循环提供的较粗检查点。第一方持久化应用与运行时会显式加载两者，专用部署则可以有意省略或替换本策略。注册顺序决定其他 `agent/step` 监听器追加的事件是否先于本检查点；提示词输入以及前一批由循环自身记录的助手消息与有序结果都已在日志中。
|
||||
|
||||
检查点失败与取消在副作用边界上采取失败关闭策略。请求检查点被拒绝时，系统不会分发给适配器；工具检查点被拒绝时，系统会返回错误结果，不调用工具主体。如果在工具检查点等待期间收到取消，策略会重新检查信号，并返回标准的 `ABORTED_BEFORE_DISPATCH` 结果。步骤后检查点被拒绝时，系统会在发起下一个模型请求前停止继续执行。持久化写入的串行化仍由协调器负责，因此并发的工具检查点不会产生重复的事件序号。
|
||||
检查点失败与取消在副作用边界上采取失败关闭策略。请求检查点被拒绝时，系统不会分发给适配器；工具检查点被拒绝时，系统会返回错误结果，不调用工具主体。如果在工具检查点等待期间收到取消，策略会重新检查信号，并返回标准的 `ABORTED_BEFORE_DISPATCH` 结果。步骤间检查点被拒绝时，系统会在发起下一个模型请求前结束该轮次。轮次的最终检查点被拒绝时，系统会实时报告该失败，但不会阻止后续排队工作。持久化写入的串行化仍由协调器负责，因此并发的工具检查点不会产生重复的事件序号。
|
||||
|
||||
ACP（Agent Client Protocol）应用在一个有序 Cordis effect 中统一持有其桥接层、检查点策略与持久化后端。Cordis 会并发卸载同级插件的 effect；如果分别加载，桥接层仍在为被中断的轮次收尾时，持久化后端就可能已经卸载。组合生命周期会先卸载桥接层，等待其各 agent 达到静止，并刷新真实的 `step/end` 与 `turn/end`，再移除检查点调度与持久化。
|
||||
|
||||
@@ -26,4 +26,4 @@ ACP（Agent Client Protocol）应用在一个有序 Cordis effect 中统一持
|
||||
|
||||
## 后果
|
||||
|
||||
发生硬崩溃时，崩溃恢复会在最近的语义边界保留完整的模型请求、持久化的工具意图与完整且已结束的步骤，但允许上一个边界之后的部分流式分片仍可能丢失。默认的 CLI（命令行界面）、TUI、ACP、Python SDK 运行时、headless 持久化测试与 JSON-RPC 组合都会在持久化后端旁加载该策略。单元测试覆盖顺序、检查点期间的取消、失败关闭行为、嵌套分发、dispose（资源释放）与 Loader 形状；一个被 `SIGKILL` 终止的真实子进程通过 JSONL 证明系统可以恢复请求与工具意图，共享持久化契约则证明各后端都支持这两种恢复分类。崩溃 harness 会等待预期的标记内容，而不是仅等待路径存在，因此文件在写入前因打开而可见时，不会导致该 harness 提前终止子进程。无密钥 ACP 快照既证明重试风险指引会进入恢复后的历史记录与下一个模型轮次，也证明取消流程正常收尾时，系统会持久化由循环实际生成的闭合边界。
|
||||
发生硬崩溃时，崩溃恢复会在最近的语义边界保留完整的模型请求、持久化的工具意图与完整且已结束的步骤，但允许上一个边界之后的部分流式分片仍可能丢失。默认的 CLI（命令行界面）、TUI、ACP、Python SDK 运行时、headless 持久化测试与 JSON-RPC 组合都会在持久化后端旁加载该策略。单元测试覆盖顺序、检查点期间的取消、失败关闭行为、嵌套分发、dispose（资源释放）与 Loader 形状，以及最终检查点的顺序与故障隔离；一个被 `SIGKILL` 终止的真实子进程通过 JSONL 证明系统可以恢复请求与工具意图，共享持久化契约则证明各后端都支持这两种恢复分类。崩溃 harness 会等待预期的标记内容，而不是仅等待路径存在，因此文件在写入前因打开而可见时，不会导致该 harness 提前终止子进程。无密钥 ACP 与 SDK 快照证明重试风险指引会进入恢复后的历史记录与下一个模型轮次，取消流程正常收尾时系统会持久化由循环实际生成的闭合边界，且 SDK 关闭流程会观察到已完整持久化的轮次。
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write docs/architecture.md
|
||||
architecture.md: 3b0f72400ab1e6a9157aed2966b4a17c36e0c3ac
|
||||
architecture.zh.md: fa21c82a686d88a9bbec02180feae1dd0dbf4e47
|
||||
architecture.md: f00a8fd8f7fbda05579fa63cc4f09b29500af289
|
||||
architecture.zh.md: b5d9507c02f752b23ab41a77a6fb5c266c577b66
|
||||
|
||||
@@ -143,7 +143,7 @@ The session log is authoritative. `deriveMessages()` projects model history; raw
|
||||
|
||||
**Model-visible ⟺ logged**: messages at `step/start` plus the folded `request/header` reconstruct every request; package-owned `dsh-agent-loop/invariant` can assert this through `ctx.invariants` ([reconstructability](../.agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.md)).
|
||||
|
||||
Durability is a plugin concern. Backends eagerly drain synchronous `session/event` notifications. `session/flush` barriers precede adapter dispatch, top-level tool dispatch, and the next request's `agent/step`. `SessionPersistence` stores `SessionEvent` directly and metadata in `SessionHeader`; JSONL defaults to checksummed Zstandard, while SQLite shares the contract ([decision](../.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md)).
|
||||
Durability is a plugin concern. Backends eagerly drain synchronous `session/event` notifications. `session/flush` barriers precede each request and top-level tool dispatch, then follow `turn/end` before another queued turn or idle observation. `SessionPersistence` stores `SessionEvent` directly and metadata in `SessionHeader`; JSONL defaults to checksummed Zstandard, while SQLite shares the contract ([decision](../.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md)).
|
||||
|
||||
`ctx.sessions.appendOutOfBand()` adds plugin-owned log-only events to an open turn or creates a balanced, flushed zero-step turn. `session/title` folds latest-wins with source seqs and provenance; its immediate fallback and sole optional async provider never delay response. Forks inherit titles ([decision](../.agents/notes/implemented/feature/2026-07-21-log-backed-session-titles.md)).
|
||||
|
||||
|
||||
@@ -143,7 +143,7 @@ idle inject:
|
||||
|
||||
**模型可见 ⟺ 已记录**：`step/start` 时的消息与折叠后的 `request/header` 可以重建每个请求；该包的 `dsh-agent-loop/invariant` 可通过 `ctx.invariants` 断言这一点（[可重建性](../.agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.md)）。
|
||||
|
||||
持久性由插件负责。后端会尽快排空同步的 `session/event` 通知。`session/flush` 屏障位于适配器分发前、顶层工具分发前，以及下一次请求的 `agent/step`。`SessionPersistence` 直接存储 `SessionEvent`，并将元数据存入 `SessionHeader`；JSONL 默认采用带校验和的 Zstandard，SQLite 遵循同一契约（[决策](../.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md)）。
|
||||
持久性由插件负责。后端会尽快排空同步的 `session/event` 通知。`session/flush` 屏障位于每次请求与顶层工具分发之前，并在 `turn/end` 之后、处理另一个已排队轮次或观察到空闲状态之前执行。`SessionPersistence` 直接存储 `SessionEvent`，并将元数据存入 `SessionHeader`；JSONL 默认采用带校验和的 Zstandard，SQLite 遵循同一契约（[决策](../.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md)）。
|
||||
|
||||
`ctx.sessions.appendOutOfBand()` 会把插件所属的纯日志事件加入开放轮次，或创建一个平衡且已刷写的零步骤轮次。`session/title` 按后写覆盖方式折叠，并携带源 seq 和来源信息；其即时回退标题和唯一可选异步提供方都不会延迟响应。fork 会继承标题（[决策](../.agents/notes/implemented/feature/2026-07-21-log-backed-session-titles.md)）。
|
||||
|
||||
|
||||
@@ -190,7 +190,7 @@ export class ReactLoopAgent implements Agent {
|
||||
// but the waiter must not gamble quiescence on that: a future escape
|
||||
// still counts as settled activity.
|
||||
/* v8 ignore next 3 -- the catch arm backstops rejection paths that are all currently contained */
|
||||
while (this.wakeScheduled || this.abort !== undefined || this.queued.some(item => item.wakeup)) {
|
||||
while (this.busy || this.wakeScheduled || this.abort !== undefined || this.queued.some(item => item.wakeup)) {
|
||||
await this.done.catch(() => undefined)
|
||||
}
|
||||
}
|
||||
@@ -422,6 +422,15 @@ export class ReactLoopAgent implements Agent {
|
||||
signal.removeEventListener('abort', cancelRetry)
|
||||
}
|
||||
|
||||
if (opened) {
|
||||
try {
|
||||
await this.loopCtx.sessions.flush(this.session)
|
||||
} catch (error: unknown) {
|
||||
this.loopCtx.logger.warn(`agent "${this.id}": session/flush failed at turn ${turn}: ${errorChain(error)}`)
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error)
|
||||
}
|
||||
}
|
||||
|
||||
if (retry) {
|
||||
await this.run({ kind: 'retry' })
|
||||
} else {
|
||||
|
||||
@@ -88,6 +88,79 @@ describe('Agent', () => {
|
||||
expect(statuses).toEqual(['running', 'idle'])
|
||||
})
|
||||
|
||||
it('awaits the turn-end checkpoint before claiming the next queued turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
const firstFlush = Promise.withResolvers<undefined>()
|
||||
const flushedTurns: number[] = []
|
||||
ctx.on('session/flush', async (session) => {
|
||||
const turnEnd = session.events.findLast(event => event.type === 'turn/end')
|
||||
flushedTurns.push(turnEnd?.data.turn ?? 0)
|
||||
if (turnEnd?.data.turn === 1) await firstFlush.promise
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
send(agent, 'second')
|
||||
|
||||
await vi.waitFor(() => { expect(flushedTurns).toEqual([1]) })
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
firstFlush.resolve(undefined)
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(flushedTurns).toEqual([1, 2])
|
||||
})
|
||||
|
||||
it('keeps whenIdle pending through the final turn checkpoint', async () => {
|
||||
const ctx = await harness(new MockAdapter([textResponse('done')]))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
const flush = Promise.withResolvers<undefined>()
|
||||
let flushStarted = false
|
||||
ctx.on('session/flush', () => {
|
||||
flushStarted = true
|
||||
return flush.promise
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await vi.waitFor(() => { expect(flushStarted).toBe(true) })
|
||||
let idleSettled = false
|
||||
const idle = agent.whenIdle().then(() => { idleSettled = true })
|
||||
await Promise.resolve()
|
||||
expect(idleSettled).toBe(false)
|
||||
|
||||
flush.resolve(undefined)
|
||||
await idle
|
||||
expect(agent.status).toBe('idle')
|
||||
})
|
||||
|
||||
it('reports a rejected turn-end checkpoint and continues queued work', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
const warning = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
const failure = new Error('disk unavailable')
|
||||
const errors: { turn: number; step: number; error: unknown }[] = []
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', () => {
|
||||
flushes += 1
|
||||
if (flushes === 1) throw failure
|
||||
})
|
||||
ctx.on('agent/error', (subject, turn, step, error) => {
|
||||
if (subject === agent) errors.push({ turn, step, error })
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
send(agent, 'second')
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(flushes).toBe(2)
|
||||
expect(errors).toEqual([{ turn: 1, step: 1, error: failure }])
|
||||
expect(warning).toHaveBeenCalledWith(expect.stringContaining('session/flush failed at turn 1: disk unavailable'))
|
||||
warning.mockRestore()
|
||||
})
|
||||
|
||||
it('whenIdle() resolves immediately without active work', async () => {
|
||||
const ctx = await harness(new MockAdapter([textResponse('ok')]))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
@@ -79,8 +79,9 @@ describe('startInProcessRun', () => {
|
||||
const result = await run.result
|
||||
const child = ctx.agents.get(run.id)!
|
||||
|
||||
expect(injected).toBe(true)
|
||||
expect(child.session.events.findLast(event => event.type === 'turn/end'))
|
||||
.toMatchObject({ data: { reason: { kind: 'max-tokens' } } })
|
||||
.toMatchObject({ data: { reason: { kind: 'completed' } } })
|
||||
expect(result.stopReason).toBe('max-tokens')
|
||||
await run.dispose()
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user