fix(subagent): harden continuable persistence

packages/subagent/subagent-inprocess/README.i18n.yaml

@@ -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: 7587b6dfc44bef90756c9f2aba96d54872935fee
-README.zh.md: 751e745c6c7a64831debd2df58ed8c3d7861f84d
+README.md: eb5d973566f01c05b43f4f56eff746b7af93f60b
+README.zh.md: 5be640f9b6da6402ece0e1d15997d9e2970a7d1c

packages/subagent/subagent-inprocess/README.md

@@ -12,9 +12,10 @@ The driver follows this sequence:
 
 1. Validate the parent depth and optional absolute `maxDepth`, then derive child depth as parent depth plus one and persist it in the child session header.
 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/pre-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.
+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. Read the child's own last assistant message and latest message-triggered turn reason, excluding any fork seed and later plugin-owned zero-step turns.
+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. 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.
 
@@ -22,7 +23,7 @@ When the optional sandbox-policy or approval service is composed, the driver sna
 
 ## Cold resume
 
-`resumeInProcessRun(request): Promise<SubagentRun>` reconstructs a persisted continuable child under the live parent's scope: `parent.ctx.agents.resume` loads the child's own transcript through persistence (a fork child's log already contains its seed prefix, so resume never re-forks current parent history), the descriptor's persona and tool filter are reapplied in the unpublished setup window, and the descriptor's `agentProvider`/`agentModel` become the runtime options. The persisted header stays authoritative for lineage and the delegation-depth floor. The activation's result boundary is the resumed log length: only this follow-up turn's output becomes the run result. Publication, abort handoff, and disposal follow the same contract as start.
+`resumeInProcessRun(request): Promise<SubagentRun>` reconstructs a persisted continuable child under the live parent's scope: `parent.ctx.agents.resume` loads the child's own transcript through persistence (a fork child's log already contains its seed prefix, so resume never re-forks current parent history), the descriptor's persona and tool filter are reapplied in the unpublished setup window, and the descriptor's `agentProvider`/`agentModel` become the runtime options. The persisted header stays authoritative for lineage and the delegation-depth floor. The activation's result boundary is the resumed log length: only this follow-up turn's output becomes the run result. Publication, final durability confirmation, abort handoff, and disposal follow the same contract as a continuable start.
 
 ## Cancellation and ownership
 
@@ -30,7 +31,7 @@ The required request signal covers both startup and the live run. Before publica
 
 After fulfillment, the caller owns the run. Provider-plugin unload does not revoke it. `dispose()` removes the live abort listener, records cancellation, and delegates to the returned `AgentHandle.dispose()`, whose memoized quiescence transaction stops the loop, removes the agent and session, and unwinds scoped registrations. Cancellation owns every non-completed in-flight outcome and reports `aborted`; an already-completed turn remains completed.
 
-Runs expose the strict `steer` capability: the synchronous checks and the `Agent.steer()` call share one frame, so delivery joins the observed turn or throws. Delivery requires `AgentStatus.running`, an open turn in the child log (status stays `running` through a closed turn's durability flush, where the loop would strand the message), an open step (between steps the loop may sit at its continuation/turn-stop checkpoints, where steering was already folded and a terminal stop discards a later arrival; a message accepted during an open step is recorded at that step's settlement before any terminal decision), and no committed structured capture (whose terminal stop makes the loop discard late steering). The Agent-level idle fallback (queue and start a new turn) is deliberately not reachable through the run — that would start an untracked turn after the run's result was read.
+Runs expose the strict `steer` capability: the synchronous checks and the `Agent.trySteer()` call share one frame, so delivery joins the observed step or throws. Delivery requires `AgentStatus.running`, an open turn and step in the child log, no committed structured capture, and acceptance before that step's final drain begins. Admission, between-step processing such as `agent/turn-stopping`, and a closed turn's durability flush all reject delivery. The Agent-level idle fallback (queue and start a new turn) is deliberately not reachable through the run — that would start an untracked turn after the run's result was read.
 
 ## Spawn and fork inputs
 

packages/subagent/subagent-inprocess/README.zh.md

@@ -12,9 +12,10 @@
 
 1. 校验父 agent 深度和可选的绝对 `maxDepth`，然后把子 agent 深度推导为父 agent 深度加一，并将其持久化到子 agent 会话 header。
 2. 直接调用 `parent.ctx.agents.create`，把必需的请求信号传入工厂的创建事务。可继续请求会精确发布 `request.continuation.sessionId`，而不是内部生成的 ID。
-3. 在该事务未发布的设置窗口中，安装请求的 persona、工具限制和结构化输出运行时；对于可继续请求，还会安装一次性的 `agent/pre-step` 贡献，在初始 `turn/start` 之后、首次请求之前追加 `subagent/descriptor` 事件，使描述符随该轮次的 flush 到达持久化层。
+3. 在该事务未发布的设置窗口中，安装请求的 persona、工具限制和结构化输出运行时；对于可继续请求，还会安装一次性的 `agent/step` 贡献，在初始 `turn/start` 之后、首次请求之前追加 `subagent/descriptor` 事件，使描述符随该轮次的 flush 到达持久化层。
 4. 发布子 agent，保留返回的 `AgentHandle`，并通过先调用 `child.followup(prompt)`、再调用 `child.whenIdle()` 来驱动一项任务。
-5. 读取子 agent 自身最后一条 assistant 消息，以及由消息触发的最新轮次原因；排除任何 fork 初始内容和后续由插件拥有的零步骤轮次。
+5. 对于可继续的启动或恢复，在返回结果前再次调用 `child.ctx.sessions.flush(child.session)`。这次最终确认会重试轮次检查点失败后保留的事件；若仍然失败，`result` 会以 `SubagentError.code === 'DURABILITY_FAILED'` 拒绝，保留后端失败作为 `cause`，并在消息中指出可恢复性风险。前台运行仍采用循环的尽力而为检查点行为。
+6. 读取子 agent 自身最后一条 assistant 消息，以及由消息触发的最新轮次原因；排除任何 fork 初始内容和后续由插件拥有的轮次间记录。
 
 子 agent 会获得父 agent 的工作目录／会话谱系；除非 `request.agentOptions` 覆盖，否则还会继承父 agent 的提供方、模型和输出 token 上限。它获得全新的扁平注册作用域：父级所有权不会导入父 agent 的工具限制，也不会建立权限子集。
 
@@ -22,7 +23,7 @@
 
 ## 冷恢复
 
-`resumeInProcessRun(request): Promise<SubagentRun>` 会在当前父级作用域下重建持久化的可继续子 agent：`parent.ctx.agents.resume` 通过持久化层加载子 agent 自身的 transcript（文本记录；fork 子 agent 的日志已经包含初始前缀，因此恢复绝不会再次 fork 当前父级历史），在未发布的设置窗口中重新应用描述符中的 persona 和工具过滤器，并把描述符中的 `agentProvider` / `agentModel` 作为运行时选项。持久化 header 对谱系和委派深度下限保持权威性。activation 的结果边界是恢复后日志的长度：只有此次后续轮次的输出会成为运行结果。发布、中止交接和 dispose 遵循与启动相同的契约。
+`resumeInProcessRun(request): Promise<SubagentRun>` 会在当前父级作用域下重建持久化的可继续子 agent：`parent.ctx.agents.resume` 通过持久化层加载子 agent 自身的 transcript（文本记录；fork 子 agent 的日志已经包含初始前缀，因此恢复绝不会再次 fork 当前父级历史），在未发布的设置窗口中重新应用描述符中的 persona 和工具过滤器，并把描述符中的 `agentProvider` / `agentModel` 作为运行时选项。持久化 header 对谱系和委派深度下限保持权威性。activation 的结果边界是恢复后日志的长度：只有此次后续轮次的输出会成为运行结果。发布、最终持久性确认、中止交接和 dispose 遵循与可继续启动相同的契约。
 
 ## 取消与所有权
 
@@ -30,7 +31,7 @@
 
 兑现后，调用方拥有该运行。提供方插件卸载不会撤销它。`dispose()` 会移除实时中止监听器、记录取消，并委托给返回的 `AgentHandle.dispose()`；后者通过可复用的完全停稳事务停止循环、移除 agent 和会话，并展开有作用域的注册。取消决定所有尚未完成的进行中结果，并将其报告为 `aborted`；已经完成的轮次仍保持完成状态。
 
-运行公开严格的 `steer` 功能：同步的 `AgentStatus.running` 检查与 `Agent.steer()` 调用位于同一个调用栈帧中，因此消息要么加入观察到的轮次，要么抛错。运行不会触达 Agent 层在空闲时排队并启动新轮次的 fallback；否则会在运行结果读取后启动一个未被跟踪的轮次。
+运行公开严格的 `steer` 功能：同步检查与 `Agent.trySteer()` 调用位于同一个调用栈帧中，因此消息要么加入观察到的步骤，要么抛错。交付要求 `AgentStatus.running`、子 agent 日志中有开放的轮次和步骤、没有已提交的结构化捕获，并且在该步骤的最终 drain 开始前获接纳。提示词接纳、`agent/turn-stopping` 等步骤间处理，以及已关闭轮次的持久性 flush 都会拒绝交付。运行不会触达 Agent 层在空闲时排队并启动新轮次的 fallback；否则会在运行结果读取后启动一个未被跟踪的轮次。
 
 ## Spawn 与 fork 输入
 

packages/subagent/subagent-inprocess/src/index.ts

@@ -11,8 +11,8 @@ import { randomUUID } from 'node:crypto'
 import type { Context } from 'cordis'
 import type { Agent, AgentHandle, AgentOptions } from '@deepseek-ai/dsh-agent'
 import { findLastMessageTurnEnd, SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
-import { createUserMessage, type ContentBlock } from '@deepseek-ai/dsh-llm'
-import { assertSubagentMaxDepth, delegationDepthOf } from '@deepseek-ai/dsh-subagent'
+import { createUserMessage, errorChain, type ContentBlock } from '@deepseek-ai/dsh-llm'
+import { assertSubagentMaxDepth, delegationDepthOf, SubagentError } from '@deepseek-ai/dsh-subagent'
 import type {
   SubagentDescriptorData,
   SubagentResult,
@@ -67,6 +67,9 @@ export interface InProcessRunOptions {
   readonly seed?: SessionEvent[]
 }
 
+/** Whether one activation must prove its final state durable before success. */
+type Durability = 'best-effort' | 'required'
+
 /** Error used when cancellation wins before the child publication boundary. */
 function prePublicationAbort(): Error {
   return new Error('subagent request was aborted before child publication')
@@ -168,7 +171,15 @@ export async function startInProcessRun(
     signal: request.signal,
     setup,
   })
-  return driveTurn(handle, request.signal, request.prompt, childId, seedLength, structured)
+  return driveTurn(
+    handle,
+    request.signal,
+    request.prompt,
+    childId,
+    seedLength,
+    request.continuation === undefined ? 'best-effort' : 'required',
+    structured,
+  )
 }
 
 /**
@@ -203,14 +214,15 @@ export async function resumeInProcessRun(request: SubagentResumeRequest): Promis
   // The result boundary is this activation's own work: everything already in
   // the resumed transcript belongs to earlier turns.
   const resumePoint = handle.agent.session.events.length
-  return driveTurn(handle, request.signal, request.prompt, request.sessionId, resumePoint)
+  return driveTurn(handle, request.signal, request.prompt, request.sessionId, resumePoint, 'required')
 }
 
 /**
  * Drive one activation turn on a published child and wrap it as a run. The
  * caller has already created or resumed the agent; this owns the
  * signal-handoff race, the live abort listener, result collection past
- * `boundary`, strict steering, and disposal.
+ * `boundary`, the continuable-run durability confirmation, strict steering,
+ * and disposal.
  */
 function driveTurn(
   handle: AgentHandle,
@@ -218,6 +230,7 @@ function driveTurn(
   prompt: ContentBlock[],
   childId: SessionId,
   boundary: number,
+  durability: Durability,
   structured?: StructuredAttachment,
 ): SubagentRun | Promise<never> {
   const child = handle.agent
@@ -238,6 +251,17 @@ function driveTurn(
     try {
       child.followup(createUserMessage({ content: prompt, source: { kind: 'user' } }))
       await child.whenIdle()
+      if (durability === 'required') {
+        try {
+          await child.ctx.sessions.flush(child.session)
+        } catch (error: unknown) {
+          throw new SubagentError(
+            `subagent "${childId}" durability checkpoint failed; the latest child state was not confirmed persisted and may be unavailable or stale on resume: ${errorChain(error)}`,
+            'DURABILITY_FAILED',
+            { cause: error },
+          )
+        }
+      }
       return readResult(
         child,
         boundary,

packages/subagent/subagent-inprocess/tests/subagent-inprocess.spec.ts

@@ -9,7 +9,7 @@ import InvariantService from '@deepseek-ai/dsh-invariants'
 import * as SessionInvariant from '@deepseek-ai/dsh-session/invariant'
 import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant'
 import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
-import SubagentService, { SUBAGENT_DESCRIPTOR_VERSION } from '@deepseek-ai/dsh-subagent'
+import SubagentService, { SUBAGENT_DESCRIPTOR_VERSION, SubagentError } from '@deepseek-ai/dsh-subagent'
 import { maxTokensResponse, MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
 import { resumeInProcessRun, startInProcessRun } from '../src/index.ts'
 
@@ -38,6 +38,22 @@ function request(parent: Agent, signal = new AbortController().signal) {
   return { prompt: [{ type: 'text' as const, text: 'child task' }], parent, signal }
 }
 
+function continuableRequest(parent: Agent) {
+  const sessionId = SessionId('continuable-child')
+  return {
+    ...request(parent),
+    continuation: {
+      sessionId,
+      descriptor: {
+        version: SUBAGENT_DESCRIPTOR_VERSION,
+        provider: 'spawn',
+        agentProvider: 'mock',
+        agentModel: 'mock',
+      },
+    },
+  }
+}
+
 function text(blocks: readonly { type: string; text?: string }[]): string {
   return blocks.filter(block => block.type === 'text').map(block => block.text).join('')
 }
@@ -56,6 +72,59 @@ describe('startInProcessRun', () => {
     expect(ctx.agents.get(run.id)).toBeUndefined()
   })
 
+  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')
+    let flushes = 0
+    ctx.on('session/flush', (session) => {
+      if (session.header.parentSession === undefined) return
+      flushes++
+      throw failure
+    })
+
+    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.cause).toBe(failure)
+    expect(durabilityError.message).toContain(
+      'the latest child state was not confirmed persisted and may be unavailable or stale on resume: disk full',
+    )
+    expect(flushes).toBe(2)
+    await run.dispose()
+  })
+
+  it('completes a continuable child when the final checkpoint retries a transient flush failure', async () => {
+    const { ctx, parent } = await setup([textResponse('driver answer')])
+    let flushes = 0
+    ctx.on('session/flush', (session) => {
+      if (session.header.parentSession === undefined) return
+      flushes++
+      if (flushes === 1) throw new Error('temporary append failure')
+    })
+
+    const run = await startInProcessRun(continuableRequest(parent), {})
+    await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
+    expect(flushes).toBe(2)
+    await run.dispose()
+  })
+
+  it('keeps foreground runs best-effort when their turn checkpoint fails', async () => {
+    const { ctx, parent } = await setup([textResponse('driver answer')])
+    let flushes = 0
+    ctx.on('session/flush', (session) => {
+      if (session.header.parentSession === undefined) return
+      flushes++
+      throw new Error('disk full')
+    })
+
+    const run = await startInProcessRun(request(parent), {})
+    await expect(run.result).resolves.toMatchObject({ stopReason: 'completed' })
+    expect(flushes).toBe(1)
+    await run.dispose()
+  })
+
   it('reports the message-turn outcome when a later non-message turn completes during flush', async () => {
     const { ctx, parent } = await setup([maxTokensResponse('partial answer')])
     let injected = false
@@ -201,13 +270,21 @@ describe('startInProcessRun', () => {
 
   it('resumes without inventing undeclared agent model options', async () => {
     const childId = SessionId('resumed-child')
+    let flushes = 0
     const child = {
       id: childId,
       options: {},
       session: new Session(childId),
       status: 'idle',
       acceptsNextStep: false,
-      ctx: new Context(),
+      ctx: {
+        sessions: {
+          flush: () => {
+            flushes++
+            return Promise.resolve()
+          },
+        },
+      } as unknown as Context,
       send(): void {},
       reserveTurnAdmission: () => undefined,
       updateInbox: () => 'not-found',
@@ -238,6 +315,7 @@ describe('startInProcessRun', () => {
     })
     expect(resumedOptions).toEqual({})
     await expect(run.result).resolves.toMatchObject({ stopReason: 'error' })
+    expect(flushes).toBe(1)
     await run.dispose()
   })