Merge remote-tracking branch 'origin/master' into feature/subagent-policy-inheritance
# Conflicts: # .agents/notes/implemented/feature/2026-07-06-sandbox.i18n.yaml # docs/cordis-catalog/services.md # docs/persistence-catalog.md # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl # packages/core/session/README.i18n.yaml # packages/subagent/subagent-inprocess/README.i18n.yaml # packages/ui/user-approval/src/index.ts
This commit is contained in:
kingwl
@@ -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
|
||||
README.md: 9dfabe4042598bbff9ce4f1bddbcda54f7d19498
|
||||
README.zh.md: a2ad538cb4ed9ad2050867178eb239e5adbaeeea
|
||||
README.md: aba0f00a2960eca3db06dda267cc954f602c1b09
|
||||
README.zh.md: ea17fb00ec22cb3a94ae22dc4a37aa7e2f73a225
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
Semantic durability policy for persisted agents. It checkpoints the event-sourced session before a model adapter receives a request, before a top-level tool body may produce an external side effect, and after a step has recorded its complete assistant message and ordered tool results. The final `turn/end` checkpoint remains owned by `dsh-agent-loop`.
|
||||
Semantic durability policy for persisted agents. It checkpoints the event-sourced session before a model adapter receives a request, before a top-level tool body may produce an external side effect, and at each `agent/step` boundary so the preceding response and ordered tool results are durable before the next request.
|
||||
|
||||
## Plugin (namespace: `session-checkpoint-policy`)
|
||||
|
||||
@@ -16,13 +16,11 @@ This zero-config function plugin consumes `ctx.sessions`, `ctx.llm`, `ctx.tools`
|
||||
name: '@deepseek-ai/dsh-session-checkpoint-policy'
|
||||
```
|
||||
|
||||
Persistence and checkpoint scheduling are intentionally separate Cordis plugins. A persistence backend makes each requested `session/flush` durable; this policy chooses the request, tool-dispatch, and completed-step checkpoints. Loading a backend without this policy is valid and retains checkpoints requested by the loop, including final `turn/end`, but crash recovery may lose the rest of an in-flight turn. First-party persisted apps and runtimes mount both plugins explicitly; a specialized deployment may deliberately omit or replace the policy.
|
||||
Persistence and checkpoint scheduling are intentionally separate Cordis plugins. A persistence backend eagerly writes `session/event` appends and makes each requested `session/flush` an observation barrier; this policy chooses the request, tool-dispatch, and next-step barriers. Loading a backend without this policy is valid, but a crash may lose the latest eagerly buffered events. First-party persisted apps and runtimes mount both plugins explicitly; a specialized deployment may deliberately omit or replace the policy.
|
||||
|
||||
The policy wraps `llm/stream` lazily, so the downstream stream is not constructed until the live session's buffered request events are durable. It wraps `tools/execute` after pre-execute policy and guards; a top-level tool body runs only after its recorded call is durable. If cancellation lands while that flush is pending, the wrapper returns the canonical `ABORTED_BEFORE_DISPATCH` result without entering the tool body. Nested tool dispatches reuse the outer model-visible call's checkpoint. `agent/post-step` persists the complete response/result batch before continuation work.
|
||||
The policy wraps `llm/stream` lazily, so the downstream stream is not constructed until the live session's buffered request events are durable. It wraps `tools/execute` after pre-execute policy and guards; a top-level tool body runs only after its recorded call is durable. If cancellation lands while that flush is pending, the wrapper returns the canonical `ABORTED_BEFORE_DISPATCH` result without entering the tool body. Nested tool dispatches reuse the outer model-visible call's checkpoint. `agent/step` persists the preceding response/result batch before request derivation.
|
||||
|
||||
The loop records its assistant message and ordered tool results before dispatching `agent/post-step`, so the policy always captures that core batch. An event appended by another `agent/post-step` listener is captured at this checkpoint only when that listener is registered before the policy; Cordis registration order is the explicit composition rule for such extensions.
|
||||
|
||||
Checkpoint rejection is fail-closed at the model and tool boundaries: neither the adapter nor the top-level tool body runs. A post-step rejection fails the turn before another request starts. Concurrent tool checkpoints share the session store's serialized persistence drain and cannot duplicate sequence numbers.
|
||||
Checkpoint rejection is fail-closed at the model and tool boundaries: neither the adapter nor the top-level tool body runs. A step-boundary rejection fails the turn before another request starts. Concurrent tool checkpoints share the session store's serialized persistence drain and cannot duplicate sequence numbers.
|
||||
|
||||
## Model Experience
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
持久化 agent 的语义持久性策略。它会在模型适配器收到请求前、顶层工具正文可产生外部副作用前，以及步骤已记录完整 assistant 消息和有序工具结果后，为事件溯源会话创建检查点。最终 `turn/end` 检查点仍由 `dsh-agent-loop` 负责。
|
||||
持久化 agent 的语义持久性策略。它会在模型适配器收到请求前、顶层工具正文可产生外部副作用前，以及每个 `agent/step` 边界为事件溯源会话创建检查点，使前一响应与有序工具结果在下一个请求前持久。
|
||||
|
||||
## 插件（命名空间：`session-checkpoint-policy`）
|
||||
|
||||
@@ -16,13 +16,11 @@
|
||||
name: '@deepseek-ai/dsh-session-checkpoint-policy'
|
||||
```
|
||||
|
||||
持久化与检查点调度刻意拆分为独立 Cordis 插件。持久化后端使每个已请求 `session/flush` 持久；该策略选择请求、工具分派和已完成步骤检查点。不带此策略加载后端是有效的，仍保留 loop 请求的检查点，包括最终 `turn/end`；但崩溃恢复可能丢失正在进行轮次的其余部分。第一方持久化应用和运行时显式挂载两个插件；专用部署可以刻意省略或替换策略。
|
||||
持久化与检查点调度刻意拆分为独立 Cordis 插件。持久化后端会立即写入 `session/event` 追加，并把每个已请求 `session/flush` 变成观测屏障；该策略选择请求、工具分派和下一步骤屏障。不带此策略加载后端是有效的，但崩溃可能丢失最新的立即缓冲事件。第一方持久化应用和运行时显式挂载两个插件；专用部署可以刻意省略或替换策略。
|
||||
|
||||
策略延迟包装 `llm/stream`，因此下游流只会在实时会话缓冲请求事件持久后构造。它在预执行策略和保护后包装 `tools/execute`；只有在已记录调用持久后，顶层工具正文才会运行。如果取消在 flush 等待期间到达，包装层会返回规范 `ABORTED_BEFORE_DISPATCH` 结果，不进入工具正文。嵌套工具分派重用外层模型可见调用的检查点。`agent/post-step` 在继续工作前持久完整响应/结果批次。
|
||||
策略延迟包装 `llm/stream`，因此下游流只会在实时会话缓冲请求事件持久后构造。它在预执行策略和保护后包装 `tools/execute`；只有在已记录调用持久后，顶层工具正文才会运行。如果取消在 flush 等待期间到达，包装层会返回规范 `ABORTED_BEFORE_DISPATCH` 结果，不进入工具正文。嵌套工具分派重用外层模型可见调用的检查点。`agent/step` 在派生请求前持久前一响应/结果批次。
|
||||
|
||||
Loop 在分派 `agent/post-step` 前记录 assistant 消息和有序工具结果，因此策略总能捕获该核心批次。另一个 `agent/post-step` 监听器追加的事件只有在该监听器先于策略注册时才在此检查点捕获；Cordis 注册顺序是这类扩展的显式组合规则。
|
||||
|
||||
在模型和工具边界，检查点拒绝会快速失败：适配器和顶层工具正文都不运行。步骤后拒绝会在另一个请求开始前使轮次失败。并发工具检查点共享会话存储的串行持久化 drain，无法复制序列号。
|
||||
在模型和工具边界，检查点拒绝会快速失败：适配器和顶层工具正文都不运行。步骤边界拒绝会在另一个请求开始前使轮次失败。并发工具检查点共享会话存储的串行持久化 drain，无法复制序列号。
|
||||
|
||||
## 模型体验
|
||||
|
||||
|
||||
@@ -52,8 +52,8 @@ function abortedBeforeDispatchResult(): ToolExecutionResult {
|
||||
/**
|
||||
* Install semantic checkpoint listeners. Loop-built model calls checkpoint the
|
||||
* logged request before adapter dispatch; top-level tool calls checkpoint their
|
||||
* recorded call before the tool body; post-step checkpoints retain the complete
|
||||
* response/result batch. Nested tool dispatches reuse the durable outer call.
|
||||
* recorded call before the tool body; the next request boundary checkpoints
|
||||
* the preceding response/result batch. Nested tool dispatches reuse the durable outer call.
|
||||
*
|
||||
* Checkpoint failures are fail-closed at the model and tool side-effect
|
||||
* boundaries: the downstream adapter or tool body is not invoked.
|
||||
@@ -74,5 +74,7 @@ export function apply(ctx: Context): void {
|
||||
return next()
|
||||
})
|
||||
|
||||
ctx.on('agent/post-step', (agent): Promise<void> => ctx.sessions.flush(agent.session))
|
||||
// Before each request, persist everything committed by the preceding step;
|
||||
// the first step's call is an intentional no-op beyond any prompt intake.
|
||||
ctx.on('agent/step', (agent): Promise<void> => ctx.sessions.flush(agent.session))
|
||||
}
|
||||
|
||||
@@ -56,5 +56,5 @@ const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('semantic-checkpoint-crash'),
|
||||
agentOptions: { provider: 'crash', model: 'crash' },
|
||||
})
|
||||
handle.agent.followup([{ type: 'text', text: 'exercise the crash boundary' }])
|
||||
handle.agent.followup({ content: [{ type: 'text', text: 'exercise the crash boundary' }], source: { kind: 'user' } })
|
||||
await waitForCrash()
|
||||
|
||||
@@ -217,15 +217,13 @@ describe('session-checkpoint-policy tool and step boundaries', () => {
|
||||
expect(flushes).toBe(0)
|
||||
})
|
||||
|
||||
it('checkpoints the complete recorded step at agent/post-step', async () => {
|
||||
it('checkpoints before the next agent step', async () => {
|
||||
const ctx = await setup()
|
||||
const session = ctx.sessions.create(SessionId('post-step'))
|
||||
const agent = { session } as Agent
|
||||
const flushed: string[] = []
|
||||
ctx.on('session/flush', (current) => { flushed.push(current.id) })
|
||||
await agentEvents(ctx, agent).serial(
|
||||
'agent/post-step', 1, 1, new AbortController().signal,
|
||||
)
|
||||
await agentEvents(ctx, agent).serial('agent/step', 1, 1, new AbortController().signal)
|
||||
expect(flushed).toEqual([session.id])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -185,6 +185,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
private states = new Map<SessionId, SessionState>()
|
||||
/** Lifecycle and write-behind state keyed by the exact live Session. */
|
||||
private live = new Map<Session, LiveSessionState>()
|
||||
/** Exact disposed lifecycles whose eager tail is still draining. */
|
||||
private retirements = new Map<SessionId, Promise<void>>()
|
||||
/** Cold loads currently reserving an id across backend reads and repair writes. */
|
||||
private coldLoads = new Set<SessionId>()
|
||||
/**
|
||||
@@ -284,6 +286,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
* @returns the header plus the event log, ending on a balanced `turn/end`.
|
||||
*/
|
||||
async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
await this.retirements.get(id)
|
||||
const selected = await this.serialize(id, async () => {
|
||||
const live = this.ctx.sessions.get(id)
|
||||
if (live !== undefined) return { live }
|
||||
@@ -305,7 +308,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
* @returns stored header and events before any synthetic recovery closers.
|
||||
*/
|
||||
inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
return this.serialize(id, () => this.inspectCore(id, signal), signal)
|
||||
// Waiting for an in-flight retirement drain must honor cancellation too: a
|
||||
// slow drain would otherwise pin a cancelled inspect until it finishes,
|
||||
// past the documented boundary. serialize() already races the signal for
|
||||
// the queued read; do the same for the retirement wait.
|
||||
const retired = Promise.resolve(this.retirements.get(id))
|
||||
const waited = signal === undefined ? retired : observeQueuedAbort(retired, signal, () => false)
|
||||
return waited.then(() => this.serialize(id, () => this.inspectCore(id, signal), signal))
|
||||
}
|
||||
|
||||
private async inspectCore(
|
||||
@@ -488,7 +497,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
/** Start and observe one disposed session's final drain. */
|
||||
private retire(session: Session): void {
|
||||
if (!this.live.has(session)) return
|
||||
void this.retireCore(session).catch((error: unknown) => {
|
||||
const retirement = this.retireCore(session)
|
||||
this.retirements.set(session.id, retirement)
|
||||
const forget = (): void => {
|
||||
if (this.retirements.get(session.id) === retirement) this.retirements.delete(session.id)
|
||||
}
|
||||
void retirement.then(forget, forget)
|
||||
void retirement.catch((error: unknown) => {
|
||||
this.ctx.logger.warn(`${this.backend.name}: session "${session.id}" retirement failed: ${String(error)}`)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -50,6 +50,7 @@ interface CoordinatorInternals {
|
||||
states: Map<unknown, unknown>
|
||||
live: Map<unknown, { pending: unknown[]; flush: Promise<void> | undefined }>
|
||||
chains: Map<unknown, unknown>
|
||||
retirements: Map<unknown, Promise<void>>
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -451,6 +452,53 @@ describe('PersistenceCoordinator observation cancellation', () => {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects a cancelled inspect while an in-flight retirement drain is still pending', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const backend = new ControlledBackend()
|
||||
let coordinator!: PersistenceCoordinator<never>
|
||||
const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
coordinator = new PersistenceCoordinator(inner, backend)
|
||||
}, { inject: ['sessions'] }))
|
||||
const internals = coordinator as unknown as CoordinatorInternals
|
||||
const appendGate = Promise.withResolvers<boolean>()
|
||||
backend.beforeAppend = async () => { await appendGate.promise }
|
||||
|
||||
try {
|
||||
const id = SessionId('retiring-inspect')
|
||||
let session!: Session
|
||||
const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
session = inner.sessions.create(id)
|
||||
}, { inject: ['sessions'] }))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
// Dispose the session so retirement starts; its append is gated, so the
|
||||
// retirement promise stays pending in the coordinator.
|
||||
await sessionFiber.dispose()
|
||||
await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(true) })
|
||||
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('inspect cancelled during retirement')
|
||||
const pending = coordinator.inspect(id, controller.signal)
|
||||
let observedReason: unknown
|
||||
const observed = pending.catch((error: unknown) => { observedReason = error })
|
||||
|
||||
// Cancel before the gated retirement can settle: the inspect must reject
|
||||
// promptly instead of waiting for the drain, and must never reach the
|
||||
// backend read.
|
||||
controller.abort(reason)
|
||||
await vi.waitFor(() => { expect(observedReason).toBe(reason) })
|
||||
expect(backend.loadAttempts).toBe(0)
|
||||
|
||||
appendGate.resolve(true)
|
||||
await observed
|
||||
} finally {
|
||||
appendGate.resolve(true)
|
||||
await backendFiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator retirement', () => {
|
||||
|
||||
Reference in New Issue
Block a user