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

fix(subagent): preserve published run failures

Browse Source
This commit is contained in:
Dudu-0223
2026-07-31 14:02:03 +08:00
committed by Tianyi Cui
parent 8b0a7a5d8d
commit a977ef30ee
57 changed files with 444 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/subagent/subagent-inprocess/README.i18n.yaml
+2 -2
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: 800ba24b65cedbbff31008a94b45957d5a65657e
README.zh.md: f6e0bd6fcb2340d195469ad57e2e96941c684fdd
README.md: ea6443e9a862c03ce4e43b32d5c1b1d8996f670f
README.zh.md: fb5710a363faa828d0c7252723cc552094eb3eb2
packages/subagent/subagent-inprocess/README.md
+4 -4
View File
@@ -6,13 +6,13 @@ This package is the shared run driver for the two in-process providers' one-shot
## Start contract
`startInProcessRun(request, options): Promise<SubagentRun>` fulfills only after the child is published in `ctx.agents`. A rejected start has already quiesced the agent factory's unpublished creation transaction, so the caller never receives a half-created handle.
`startInProcessRun(request, options): Promise<SubagentRun>` fulfills as soon as the child is published in `ctx.agents`. A rejected start has already quiesced the agent factory's unpublished creation transaction, while turn or infrastructure failures after publication settle through the returned run without hiding the child id.
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. Mint a fresh child session id and call `parent.ctx.agents.create` directly, passing the optional fork seed and required request signal into the factory's creation transaction. During the unpublished setup window, install the requested persona, tool restriction, structured-output runtime, and a one-shot `agent/step` contribution that appends the resolved `subagent/descriptor` event after the initial `turn/start` and before the first request.
3. Publish the child, retain the returned `AgentHandle`, and drive one task with `child.followup(prompt)` followed by `child.whenIdle()`.
3. Publish the child, retain the returned `AgentHandle`, and return its holder-owned run. The run's `result` drives one task with `child.followup(prompt)` followed by `child.whenIdle()`.
4. Read the child's own last assistant message and latest message-triggered turn reason, excluding the fork seed prefix so a seeded parent message is never mistaken for child output.
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.
@@ -21,9 +21,9 @@ When the optional sandbox-policy or approval service is composed, the driver sna
## Cancellation and ownership
The required request signal covers both startup and the live run. Before publication, `AgentCreationTransaction` observes it, rolls back, and rejects. The factory detaches that creation-only listener before returning; the driver immediately checks the signal once more before installing a minimal live-run listener, closing the handoff race. After publication, abort cancels the child.
The required request signal covers both startup and the live run. Before publication, `AgentCreationTransaction` observes it, rolls back, and rejects. The factory detaches that creation-only listener before returning; the published run immediately installs its own listener and checks the signal again, closing the handoff race. Once publication has occurred, an abort preserves the returned child id, prevents unsubmitted work, and resolves an incomplete result as `aborted`; an abort during the turn cancels the child.
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.
After fulfillment, the caller owns the run. Provider-plugin unload does not revoke it. `dispose()` removes the live abort listener, records cancellation, and awaits both `result` and the returned `AgentHandle.dispose()`; the handle's memoized quiescence transaction stops the loop, removes the agent and session, and unwinds scoped registrations. A result rejection remains on `result`; `dispose()` rejects only when handle disposal fails, after both operations settle. Cancellation owns every non-completed in-flight outcome and reports `aborted`; an already-completed turn remains completed.
## Spawn and fork inputs
packages/subagent/subagent-inprocess/README.zh.md
+4 -4
View File
@@ -6,13 +6,13 @@
## 启动契约
`startInProcessRun(request, options): Promise<SubagentRun>` 只在子 agent 发布到 `ctx.agents` 后才兑现。启动被拒绝时，agent 工厂的未发布创建事务已经完全停稳，因此调用方绝不会收到创建到一半的句柄。
`startInProcessRun(request, options): Promise<SubagentRun>` 会在子 agent 发布到 `ctx.agents` 后立即兑现。启动被拒绝时，agent 工厂的未发布创建事务已经完全停稳；发布后的轮次或基础设施故障则通过返回的 run 结算，且不会隐藏 child id。
驱动器按以下顺序运行：
1. 校验父 agent 深度和可选的绝对 `maxDepth`，然后把子 agent 深度推导为父 agent 深度加一，并将其持久化到子 agent 会话 header。
2. 生成全新的子 agent 会话 id，并直接调用 `parent.ctx.agents.create`，把可选的 fork 初始内容和必需的请求信号传入工厂的创建事务。在未发布的设置窗口中，安装请求的 persona、工具限制、结构化输出运行时，以及一次性的 `agent/step` contribution；该 contribution 会在初始 `turn/start` 之后、首次请求之前追加已解析的 `subagent/descriptor` 事件。
3. 发布子 agent，保留返回的 `AgentHandle`，并通过先调用 `child.followup(prompt)`、再调用 `child.whenIdle()` 来驱动一项任务。
3. 发布子 agent，保留返回的 `AgentHandle`，并返回由持有方拥有的 run。该 run 的 `result` 会通过先调用 `child.followup(prompt)`、再调用 `child.whenIdle()` 来驱动一项任务。
4. 读取子 agent 自身最后一条 assistant 消息，以及由消息触发的最新轮次原因；排除 fork 初始内容前缀，确保作为初始内容的父 agent 消息绝不会被误认为子 agent 输出。
子 agent 会获得父 agent 的工作目录／会话谱系；除非 `request.agentOptions` 覆盖，否则还会继承父 agent 的提供方、模型和输出 token 上限。它获得全新的扁平注册作用域：父级所有权不会导入父 agent 的工具限制，也不会建立权限子集。
@@ -20,9 +20,9 @@
当组合中挂载了可选的沙箱策略或审批服务时，驱动器会在创建子 agent 前对父级的显式会话覆盖项获取快照，并在未发布的设置阶段追加一条带来源标记的事件，使其位于所有 fork 历史之后、会话发布之前。它绝不复制部署默认值或一次性授权；子 agent 后续的切换仍然优先。参见[策略继承决策](../../../.agents/notes/implemented/feature/2026-07-25-subagent-policy-inheritance.md)。
## 取消与所有权
必需的请求信号同时覆盖启动阶段和实时运行。发布前，`AgentCreationTransaction` 会观察该信号、回滚并拒绝。工厂返回前会移除仅用于创建阶段的监听器；驱动器随即再次检查信号，然后安装最小化的实时运行监听器，从而消除交接竞态。发布后，中止会取消子 agent。
必需的请求信号同时覆盖启动阶段和实时运行。发布前，`AgentCreationTransaction` 会观察该信号、回滚并拒绝。工厂返回前会移除仅用于创建阶段的监听器；已发布的 run 会立即安装自己的监听器并再次检查信号，从而消除交接竞态。一旦完成发布，中止会保留已返回的 child id、阻止尚未提交的工作，并以 `aborted` 兑现未完成的结果；轮次期间发生中止时，则会取消子 agent。
兑现后，调用方拥有该运行。提供方插件卸载不会撤销它。`dispose()` 会移除实时中止监听器、记录取消，并委托给返回的 `AgentHandle.dispose()`；后者通过可复用的完全停稳事务停止循环、移除 agent 和会话，并展开有作用域的注册。取消决定所有尚未完成的进行中结果，并将其报告为 `aborted`；已经完成的轮次仍保持完成状态。
兑现后，调用方拥有该运行。提供方插件卸载不会撤销它。`dispose()` 会移除实时中止监听器、记录取消，并同时等待 `result` 和返回的 `AgentHandle.dispose()`；该句柄通过可复用的完全停稳事务停止循环、移除 agent 和会话，并展开有作用域的注册。`result` 的 rejection 仍归 `result` 通道；只有句柄释放失败时，`dispose()` 才会在两项操作都结算后拒绝。取消决定所有尚未完成的进行中结果，并将其报告为 `aborted`；已经完成的轮次仍保持完成状态。
## Spawn 与 fork 输入
packages/subagent/subagent-inprocess/src/index.ts
+25 -20
View File
@@ -85,12 +85,14 @@ function attachDescriptorAppend(childCtx: Context, descriptor: SubagentDescripto
/**
* Establish and drive one in-process one-shot child. Fulfillment means the agent
* is already published in the registry; rejection means the agent factory's
* creation transaction and any partially-created child have reached quiescence.
* Every start appends its resolved descriptor inside the child's initial turn.
* is already published in the registry and transfers its turn, cancellation,
* and disposal work through the returned run. Rejection means the agent
* factory's unpublished creation transaction reached quiescence without
* publishing a child. Every start appends its resolved descriptor inside the
* child's initial turn.
* @param request - the trusted typed start request, including its required signal.
* @param options - the optional fork seed.
* @returns a ready holder-owned run.
* @returns a published holder-owned run.
*/
export async function startInProcessRun(
request: ResolvedSubagentStartRequest,
@@ -139,7 +141,7 @@ export async function startInProcessRun(
signal: request.signal,
setup,
})
return driveTurn(
return drivePublishedRun(
handle,
request.signal,
request.prompt,
@@ -150,36 +152,35 @@ export async function startInProcessRun(
}
/**
* Drive one turn on a published child and wrap it as a run. The caller has
* already created the agent; this owns the signal-handoff race, the live abort
* listener, result collection past `boundary`, and disposal.
* Wrap a published child in the single run lifecycle that owns signal handoff,
* one turn, result settlement, and quiescent disposal.
*/
function driveTurn(
function drivePublishedRun(
handle: AgentHandle,
signal: AbortSignal,
prompt: ContentBlock[],
childId: SessionId,
boundary: number,
structured: StructuredAttachment | undefined,
): SubagentRun | Promise<never> {
): SubagentRun {
const child = handle.agent
// Agent creation detaches its creation-only abort listener before returning.
// Close the narrow handoff race before installing the live-run listener.
if (signal.aborted) {
return handle.dispose().then(() => { throw prePublicationAbort() })
}
const flags = { cancelled: false }
const onAbort = (): void => {
flags.cancelled = true
child.cancel({ kind: 'parent' })
}
signal.addEventListener('abort', onAbort, { once: true })
// Agent creation detaches its creation-only listener before returning. The
// post-registration check closes that handoff without treating an already
// published child as a failed start.
if (signal.aborted) onAbort()
const result: Promise<SubagentResult> = (async () => {
try {
child.followup(createUserMessage({ content: prompt, source: { kind: 'user' } }))
await child.whenIdle()
if (!flags.cancelled) {
child.followup(createUserMessage({ content: prompt, source: { kind: 'user' } }))
await child.whenIdle()
}
return readResult(
child,
boundary,
@@ -195,10 +196,14 @@ function driveTurn(
id: childId,
localAgent: child,
result,
dispose(): Promise<void> {
async dispose(): Promise<void> {
signal.removeEventListener('abort', onAbort)
flags.cancelled = true
return handle.dispose()
const settlements = await Promise.allSettled([handle.dispose(), result])
const disposal = settlements[0]
// The result channel owns run faults; disposal reports only failure to
// release the published handle after both operations settle.
if (disposal.status === 'rejected') throw disposal.reason
},
}
}
packages/subagent/subagent-inprocess/tests/subagent-inprocess.spec.ts
+40 -4
View File
@@ -96,6 +96,40 @@ describe('startInProcessRun', () => {
await run.dispose()
})
it('keeps published run and handle disposal failures on separate channels', async () => {
const { ctx, parent } = await setup([])
const runError = new Error('published run failed')
const disposalError = new Error('published handle disposal failed')
const beforeAgents = ctx.agents.list().length
const beforeSessions = ctx.sessions.list().length
const parentWithFailedDisposal = {
options: parent.options,
session: parent.session,
ctx: {
get: () => undefined,
agents: {
create: async (options: Parameters<typeof ctx.agents.create>[0]) => {
const handle = await ctx.agents.create(options)
handle.agent.followup = () => { throw runError }
return {
...handle,
dispose: async () => {
await handle.dispose()
throw disposalError
},
}
},
},
},
} as unknown as Agent
const run = await startInProcessRun(request(parentWithFailedDisposal), {})
expect(ctx.agents.get(run.id)).toBeDefined()
await expect(run.result).rejects.toBe(runError)
await expect(run.dispose()).rejects.toBe(disposalError)
expect(ctx.agents.list()).toHaveLength(beforeAgents)
expect(ctx.sessions.list()).toHaveLength(beforeSessions)
})
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
@@ -272,7 +306,7 @@ describe('startInProcessRun', () => {
expect(ctx.sessions.list()).toHaveLength(beforeSessions)
})
it('closes the abort handoff after the factory detaches its creation listener', async () => {
it('treats abort after factory publication as a cancelled run with an id', async () => {
const { ctx, parent } = await setup([])
const controller = new AbortController()
const beforeAgents = ctx.agents.list().length
@@ -288,15 +322,17 @@ describe('startInProcessRun', () => {
create: async (options: Parameters<typeof ctx.agents.create>[0]) => {
const handle = await ctx.agents.create(options)
// `create()` has detached its creation-only listener, but the
// provider continuation has not installed its live-run listener.
// published run has not installed its live listener yet.
controller.abort('handoff race')
return handle
},
},
},
} as unknown as Agent
await expect(startInProcessRun(request(parentWithAbortAtHandoff, controller.signal), {}))
.rejects.toThrow('aborted before child publication')
const run = await startInProcessRun(request(parentWithAbortAtHandoff, controller.signal), {})
expect(ctx.agents.get(run.id)).toBeDefined()
await expect(run.result).resolves.toEqual({ output: [], stopReason: 'aborted' })
await run.dispose()
expect(ctx.agents.list()).toHaveLength(beforeAgents)
expect(ctx.sessions.list()).toHaveLength(beforeSessions)
})