Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(subagent): require durability participant

Browse Source
This commit is contained in:
Dudu-0223
2026-07-27 11:50:21 +08:00
committed by imccyu
parent 88f913a9ae
commit efc47b6a76
18 changed files with 150 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -746,6 +746,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
signature: 'async flush(session: Session): Promise<void>',
jsDoc: '/**\n * Dispatch the awaited `session/flush` durability checkpoint for `session`,\n * with the carrier captured at {@link enter}. THE flush entry point: the\n * store owns the carrier, so callers (the loop\'s turn-end checkpoint, idle\n * injection, teardown drains) must come through here rather than dispatch a\n * raw `ctx.parallel(\'session/flush\', …)` — one owner, one spelling, and the\n * scoped-dispatch invariant can pin it.\n * @param session - the session whose buffered events must reach durable storage.\n * @returns resolves when every flush listener has settled; after all settle,\n * rejects with the first registered listener failure if any listener failed.\n */',
},
{
signature: 'async flushRequired(session: Session): Promise<void>',
jsDoc: '/**\n * Dispatch the same awaited checkpoint as {@link flush}, but reject when its\n * scoped listener snapshot is empty. Callers use this operation when success\n * requires an installed durability participant rather than optional\n * best-effort persistence.\n * @param session - the session whose buffered events must reach durable storage.\n * @returns resolves when at least one listener participated and every\n * listener settled successfully.\n * @throws when no listener is registered or any registered listener fails.\n */',
},
{
signature: 'get(id: SessionId): Session | undefined',
jsDoc: '/**\n * Look up a live session.\n * @param id - the session id to look up.\n * @returns the session, or undefined when no live session has that id.\n */',
@@ -1410,7 +1414,7 @@ export const EVENT_API: readonly EventApiEntry[] = [
name: 'session/flush',
mode: 'parallel',
signature: '\'session/flush\'(this: Scoped<Session>, session: Session): Promise<void> | void',
jsDoc: '/**\n * Awaited parallel durability checkpoint: every listener runs and the\n * caller awaits all of them, with no waterfall veto. Dispatch through\n * {@link SessionStore.flush}. Scope-filtered dispatch\n * (`@deepseek-ai/dsh-scope`) reuses the session\'s owner scope.\n * @param session - the session whose buffered events must reach durable storage.\n * @dshScopeScan unsupported\n * @mode parallel\n */',
jsDoc: '/**\n * Awaited parallel durability checkpoint: every listener runs and the\n * caller awaits all of them, with no waterfall veto. An empty listener\n * snapshot is accepted by {@link SessionStore.flush} and rejected by\n * {@link SessionStore.flushRequired}. Scope-filtered dispatch\n * (`@deepseek-ai/dsh-scope`) reuses the session\'s owner scope.\n * @param session - the session whose buffered events must reach durable storage.\n * @dshScopeScan unsupported\n * @mode parallel\n */',
summary: 'Awaited parallel durability checkpoint: every listener runs and the caller awaits all of them, with no waterfall veto.',
},
{

4
packages/core/session/README.i18n.yaml
View File

@@ -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 packages/core/session/README.md
README.md: 9c7d41901e6fb0133fff0e210260e5310a025f75
README.zh.md: ca1292289901a09b83f9b0a794fa4edc9754b1da
README.md: 59e8694a957e9742a22662766d671dc2145c44e3
README.zh.md: 7618bc8f3a9146a4fc5afbfb19317deef7f13068

1
packages/core/session/README.md
View File

@@ -14,6 +14,7 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
- `ctx.sessions.create(id?, { seed?, meta? }?)` validates and detaches durable seed/header data, fills the version and id, defaults `createdAt` to now, publishes the session, and binds it to the calling fiber. Persisted reconstruction supplies its original `createdAt`, `seedLength`, and `delegationDepth`.
- `ctx.sessions.flush(session)` dispatches the awaited parallel durability checkpoint through the session's captured scope. Every listener starts and the call waits for all to settle before reporting failure; unpublished, detached, and stale objects reject.
- `ctx.sessions.flushRequired(session)` uses the same dispatch but also rejects an empty scoped listener snapshot. Callers use it when success requires an installed durability participant rather than optional best-effort persistence.
- `findLastMessageTurnEnd(events)` pairs message-triggered starts with their ends and returns the latest matched `turn/end`. Outcome consumers use this fold instead of the raw latest log event because between-turn records and non-message turns have no prompt outcome.
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that prefix to end outside an open turn, and create a live child session with lineage metadata.
- `ctx.sessions.get(id: SessionId): Session | undefined`

1
packages/core/session/README.zh.md
View File

@@ -14,6 +14,7 @@
- `ctx.sessions.create(id?, { seed?, meta? }?)` 校验持久种子／头部数据并生成脱离副本，补齐版本和 id，在未提供 `createdAt` 时使用当前时间，发布会话并将其绑定到调用方 fiber。持久化重建会提供原始的 `createdAt`、`seedLength` 和 `delegationDepth`。
- `ctx.sessions.flush(session)` 通过会话捕获的作用域分发受等待的并行持久性检查点。每个监听器都会启动；调用会等待全部结算后才报告失败。未发布、已脱离和陈旧的对象会被拒绝。
- `ctx.sessions.flushRequired(session)` 沿用相同的分发逻辑，但也会拒绝空的作用域监听器快照。若成功要求已安装的持久性参与方介入，而不是采用可选的尽力持久化，调用方应使用此方法。
- `findLastMessageTurnEnd(events)` 将由消息触发的开始与结束配对，并返回最近匹配的 `turn/end`。结果消费方使用该折叠逻辑，而不直接取日志中最近的事件，因为轮次间记录和非消息轮次没有提示词结果。
- `ctx.sessions.fork(source, boundary?, childSessionId?): Session`：解析实时会话对象或 id，选取截至 `boundary` 事件序号（含该事件）的种子（默认为当前最后一个事件），要求所选前缀结束时没有开放轮次，再创建带谱系元数据的实时子会话。
- `ctx.sessions.get(id: SessionId): Session | undefined`

27
packages/core/session/src/index.ts
View File

@@ -93,8 +93,9 @@ declare module 'cordis' {
'session/event'(this: Scoped<Session>, session: Session, event: SessionEvent): void
/**
* Awaited parallel durability checkpoint: every listener runs and the
* caller awaits all of them, with no waterfall veto. Dispatch through
* {@link SessionStore.flush}. Scope-filtered dispatch
* caller awaits all of them, with no waterfall veto. An empty listener
* snapshot is accepted by {@link SessionStore.flush} and rejected by
* {@link SessionStore.flushRequired}. Scope-filtered dispatch
* (`@deepseek-ai/dsh-scope`) reuses the session's owner scope.
* @param session - the session whose buffered events must reach durable storage.
* @dshScopeScan unsupported
@@ -973,9 +974,31 @@ export class SessionStore extends Service {
* rejects with the first registered listener failure if any listener failed.
*/
async flush(session: Session): Promise<void> {
await this.dispatchFlush(session, false)
}
/**
* Dispatch the same awaited checkpoint as {@link flush}, but reject when its
* scoped listener snapshot is empty. Callers use this operation when success
* requires an installed durability participant rather than optional
* best-effort persistence.
* @param session - the session whose buffered events must reach durable storage.
* @returns resolves when at least one listener participated and every
* listener settled successfully.
* @throws when no listener is registered or any registered listener fails.
*/
async flushRequired(session: Session): Promise<void> {
await this.dispatchFlush(session, true)
}
/** Dispatch one optional or required flush listener snapshot. */
private async dispatchFlush(session: Session, requireListener: boolean): Promise<void> {
const { carrier } = this.liveEntryFor(session)
const callbackArgs: unknown[] = [session]
const callbacks = collectSessionCallbacks(this.ctx, [carrier, 'session/flush', session])
if (requireListener && callbacks.length === 0) {
throw new Error(`session "${session.id}" required durability checkpoint has no registered listener`)
}
const results = await Promise.allSettled(callbacks.map((callback) => {
try {
return callback(...callbackArgs)

27
packages/core/session/tests/scoped.spec.ts
View File

@@ -80,6 +80,33 @@ describe('session dispatch carriers', () => {
})
describe('sessions.flush()', () => {
it('allows an ordinary flush with no listeners', async () => {
const ctx = await mount()
const session = ctx.sessions.create()
await expect(ctx.sessions.flush(session)).resolves.toBeUndefined()
})
it('rejects a required flush with no listeners', async () => {
const ctx = await mount()
const session = ctx.sessions.create()
await expect(ctx.sessions.flushRequired(session)).rejects.toThrow(
`session "${session.id}" required durability checkpoint has no registered listener`,
)
})
it('completes a required flush when a listener succeeds', async () => {
const ctx = await mount()
const session = ctx.sessions.create()
const flushed: Session[] = []
ctx.on('session/flush', current => void flushed.push(current))
await ctx.sessions.flushRequired(session)
expect(flushed).toEqual([session])
})
it('dispatches session/flush with the owning carrier and awaits all listeners', async () => {
const ctx = await mount()
const scope = await mintScope(ctx, 'owner')

4
packages/subagent/subagent-inprocess/README.i18n.yaml
View File

@@ -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 packages/subagent/subagent-inprocess/README.md
README.md: 1bbbfd282fe98f73b1828b22a95efd34e5ddc0ab
README.zh.md: d6dc91415beb3986ad226a8467ce2abbabce8591
README.md: afc92cf4f38830c22a2de401620e0223e7bf62d1
README.zh.md: dcf7d343901145f63758bfce0f85fa70691cb14e

2
packages/subagent/subagent-inprocess/README.md
View File

@@ -14,7 +14,7 @@ The driver follows this sequence:
2. Call `parent.ctx.agents.create` directly, passing the required request signal into the factory's creation transaction. A continuable request publishes exactly `request.continuation.sessionId` instead of an internally minted id.
3. During that transaction's unpublished setup window, install the requested persona, tool restriction, structured-output runtime, and — for a continuable request — the one-shot `agent/step` contribution that appends the `subagent/descriptor` event after the initial `turn/start` and before the first request, so the descriptor reaches persistence with that turn's flush.
4. Publish the child, retain the returned `AgentHandle`, and drive one task with `child.followup(prompt)` followed by `child.whenIdle()`.
5. For a continuable start or resume, call `child.ctx.sessions.flush(child.session)` again before returning the result. This final confirmation retries events retained after a failed turn checkpoint; if it still fails, `result` rejects with `SubagentError.code === 'DURABILITY_FAILED'`, retains the backend failure as `cause`, and names the resumability risk in its message. Activation cancellation during this await owns the unpublished result even when the completed turn was already recorded or the checkpoint subsequently fails. Foreground runs keep the loop's best-effort checkpoint behavior.
5. For a continuable start or resume, call `child.ctx.sessions.flushRequired(child.session)` again before returning the result. This final confirmation requires an installed durability listener and retries events retained after a failed turn checkpoint; if no listener participates or any listener fails, `result` rejects with `SubagentError.code === 'DURABILITY_FAILED'`, retains the checkpoint failure as `cause`, and names the resumability risk in its message. Activation cancellation during this await owns the unpublished result even when the completed turn was already recorded or the checkpoint subsequently fails. Foreground runs keep the loop's best-effort checkpoint behavior.
6. Read the child's own last assistant message and latest message-triggered turn reason, excluding any fork seed and later plugin-owned between-turn records.
The child gets the parent's working-directory/session lineage and inherits the parent provider, model, and output-token cap unless `request.agentOptions` overrides them. It gets a fresh flat registration scope: parent ownership does not import parent tool restrictions or establish an authority subset.

2
packages/subagent/subagent-inprocess/README.zh.md
View File

@@ -14,7 +14,7 @@
2. 直接调用 `parent.ctx.agents.create`，把必需的请求信号传入工厂的创建事务。可继续请求会精确发布 `request.continuation.sessionId`，而不是内部生成的 ID。
3. 在该事务未发布的设置窗口中，安装请求的 persona、工具限制和结构化输出运行时；对于可继续请求，还会安装一次性的 `agent/step` 贡献，在初始 `turn/start` 之后、首次请求之前追加 `subagent/descriptor` 事件，使描述符随该轮次的 flush 到达持久化层。
4. 发布子 agent，保留返回的 `AgentHandle`，并通过先调用 `child.followup(prompt)`、再调用 `child.whenIdle()` 来驱动一项任务。
5. 对于可继续的启动或恢复，在返回结果前再次调用 `child.ctx.sessions.flush(child.session)`。这次最终确认会重试轮次检查点失败后保留的事件；若仍然失败，`result` 会以 `SubagentError.code === 'DURABILITY_FAILED'` 拒绝，保留后端失败作为 `cause`，并在消息中指出可恢复性风险。在这次等待期间取消 activation 时，即使已记录完成的轮次，或检查点随后失败，取消仍决定尚未发布的结果。前台运行仍采用循环的尽力而为检查点行为。
5. 对于可继续启动或恢复，在返回结果前再次调用 `child.ctx.sessions.flushRequired(child.session)`。这次最终确认要求有已安装的持久性监听器参与，并会重试轮次检查点失败后保留的事件；如果没有监听器参与或任一监听器失败，`result` 会以 `SubagentError.code === 'DURABILITY_FAILED'` 拒绝，将检查点失败保留为 `cause`，并在消息中说明恢复风险。即使已记录完成的轮次，或随后检查点失败，等待期间发生的激活取消仍决定尚未发布的结果。前台运行保留循环的尽力检查点行为。
6. 读取子 agent 自身最后一条 assistant 消息，以及由消息触发的最新轮次原因；排除任何 fork 初始内容和后续由插件拥有的轮次间记录。
子 agent 会获得父 agent 的工作目录／会话谱系；除非 `request.agentOptions` 覆盖，否则还会继承父 agent 的提供方、模型和输出 token 上限。它获得全新的扁平注册作用域：父级所有权不会导入父 agent 的工具限制，也不会建立权限子集。

2
packages/subagent/subagent-inprocess/src/index.ts
View File

@@ -270,7 +270,7 @@ function driveTurn(
await child.whenIdle()
if (durability === 'required') {
try {
await child.ctx.sessions.flush(child.session)
await child.ctx.sessions.flushRequired(child.session)
} catch (error: unknown) {
if (!signal.aborted) {
throw new SubagentError(

36
packages/subagent/subagent-inprocess/tests/subagent-inprocess.spec.ts
View File

@@ -73,6 +73,40 @@ describe('startInProcessRun', () => {
expect(ctx.agents.get(run.id)).toBeUndefined()
})
it('rejects a continuable child when no durability listener is registered', async () => {
const { parent } = await setup([textResponse('driver answer')])
const run = await startInProcessRun(continuableRequest(parent), {})
const caught: unknown = await run.result.catch((error: unknown) => error)
expect(caught).toBeInstanceOf(SubagentError)
const durabilityError = caught as SubagentError
expect(durabilityError.code).toBe('DURABILITY_FAILED')
expect(durabilityError.message).toContain('required durability checkpoint has no registered listener')
await run.dispose()
})
it('rejects when the durability listener disappears before final confirmation', async () => {
const { ctx, parent } = await setup([textResponse('driver answer')])
let flushes = 0
let detach = (): void => {}
detach = ctx.on('session/flush', (session) => {
if (session.header.parentSession === undefined) return
flushes++
if (flushes === 1) detach()
})
const run = await startInProcessRun(continuableRequest(parent), {})
const caught: unknown = await run.result.catch((error: unknown) => error)
expect(caught).toBeInstanceOf(SubagentError)
const durabilityError = caught as SubagentError
expect(durabilityError.code).toBe('DURABILITY_FAILED')
expect(durabilityError.message).toContain('required durability checkpoint has no registered listener')
expect(flushes).toBe(1)
await run.dispose()
})
it('requires a final durability checkpoint for a continuable child', async () => {
const { ctx, parent } = await setup([textResponse('driver answer')])
const failure = new Error('disk full')
@@ -312,7 +346,7 @@ describe('startInProcessRun', () => {
acceptsNextStep: false,
ctx: {
sessions: {
flush: () => {
flushRequired: () => {
flushes++
return Promise.resolve()
},

27
packages/subagent/subagent/tests/continuation.spec.ts
View File

@@ -1,4 +1,4 @@
import { afterEach, describe, expect, it } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
@@ -61,10 +61,12 @@ afterEach(() => {
async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean } = {}) {
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
let disposePersistence: (() => Promise<void>) | undefined
if (options.persistence !== false) {
const root = mkdtempSync(join(tmpdir(), 'dsh-subagent-continuation-'))
roots.push(root)
await ctx.plugin(JsonlSessionPersistence, { root })
const persistenceFiber = await ctx.plugin(JsonlSessionPersistence, { root })
disposePersistence = () => persistenceFiber.dispose()
}
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(SubagentService)
@@ -74,7 +76,7 @@ async function setupWith(adapter: LlmAdapter, options: { persistence?: boolean }
await ctx.plugin(ToolTasks, {})
ctx.llm.registerAdapter(['mock'], adapter)
const parent = ctx.agentLoop.create(SessionId('parent'), { provider: 'mock', model: 'mock' })
return { ctx, parent }
return { ctx, parent, disposePersistence }
}
async function setup(script: Script, options: { persistence?: boolean } = {}) {
@@ -141,6 +143,25 @@ describe('SubagentService.startContinuable', () => {
expect(ctx.agents.get(started.childId)).toBeUndefined()
})
it('fails the Task when persistence detaches before the activation completes', async () => {
const releaseResponse = Promise.withResolvers<undefined>()
const adapter = new GatedAdapter([
{ chunks: textResponse('unconfirmed answer'), gate: releaseResponse.promise },
])
const { ctx, parent, disposePersistence } = await setupWith(adapter)
const started = ctx.subagents.startContinuable(startSpec(parent))
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
await disposePersistence!()
releaseResponse.resolve(undefined)
const snapshot = await waitTerminal(ctx, started.taskId, parent)
expect(snapshot.status).toBe('failed')
expect(snapshot.detail).toContain('durability checkpoint failed')
expect(snapshot.detail).toContain('required durability checkpoint has no registered listener')
expect(ctx.agents.get(started.childId)).toBeUndefined()
})
it('publishes the service-allocated child id and appends the turn-enclosed descriptor', async () => {
const { ctx, parent } = await setup([textResponse('answer')])
const seen: SessionEvent[] = []