fix(subagent): await pending descendants during activation disposal
This commit is contained in:
1 parent
55cbf441d0
commit
4767528ed8
5 files changed
+69
-20
No files matched your search
@@ -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/README.md
|
||||
README.md: a90d160234e6169639a35481cfd9bb19eac614f4
|
||||
README.zh.md: b7c0d22fa40697736d7c982cb582078b115ae94c
|
||||
README.md: ae8b0f629a6dcb1e4ec8176b1f5dd2fa18cddf50
|
||||
README.zh.md: 822f64a3db43ab1b4bfd679f00d503c31f5d6d12
|
||||
@@ -64,7 +64,7 @@ Every exact live Agent can use `sendMessage()` with a direct continuable child;
|
||||
|
||||
### Failure and recovery
|
||||
|
||||
Unsupported provider capabilities reject creation. The caller signal cancels only unpublished work; after publication, `dispose()` cancels the exact activation and waits for its descendants and resources. `result` settles after teardown and notifications, carrying output, optional structured data, and a stop reason; capture failures reject it after teardown. Cleanup failures reject `dispose()` without replacing a captured execution result.
|
||||
Unsupported provider capabilities reject creation. The caller signal cancels only unpublished work; after publication, `dispose()` cancels the exact activation and waits for its descendants, including pending startups, and their resource cleanup. `result` settles after teardown and notifications, carrying output, optional structured data, and a stop reason; capture failures reject it after teardown. Cleanup failures reject `dispose()` without replacing a captured execution result.
|
||||
|
||||
-----
|
||||
|
||||
|
||||
@@ -64,7 +64,7 @@ kind: "package-reference"
|
||||
|
||||
### 失败与恢复
|
||||
|
||||
不受提供方支持的能力会使创建失败。调用方信号仅取消尚未发布的工作;发布后通过 `dispose()` 取消该次 activation,并等待其后代和资源释放。`result` 在清理与通知结束后完成,携带输出、可选结构化数据与停止原因;结果捕获故障在清理后使其拒绝。清理失败使 `dispose()` 拒绝,不覆盖已捕获的执行结果。
|
||||
不受提供方支持的能力会使创建失败。调用方信号仅取消尚未发布的工作;发布后通过 `dispose()` 取消该次 activation,并等待其后代(包括正在启动的后代)及其资源清理完成。`result` 在清理与通知结束后完成,携带输出、可选结构化数据与停止原因;结果捕获故障在清理后使其拒绝。清理失败使 `dispose()` 拒绝,不覆盖已捕获的执行结果。
|
||||
|
||||
-----
|
||||
|
||||
|
||||
@@ -1025,7 +1025,7 @@ export class SubagentManager {
|
||||
return 'ready'
|
||||
}
|
||||
|
||||
/** Propagate stop synchronously, then finish the child-first release. */
|
||||
/** Propagate stop synchronously and await descendant startup rollback before release. */
|
||||
private async finishDisposal(activation: Activation, stop: boolean): Promise<void> {
|
||||
this.wake(activation)
|
||||
const { childId } = activation
|
||||
@@ -1055,7 +1055,10 @@ export class SubagentManager {
|
||||
const children = [...activation.ownedChildren]
|
||||
.map(child => this.resident.get(child))
|
||||
.filter((child): child is Activation => child !== undefined)
|
||||
const pending = [...this.materializations].filter(item => item.lineage.includes(activation.handle.agent))
|
||||
const childDisposals = children.map(child => this.dispose(child))
|
||||
for (const item of pending) item.controller.abort()
|
||||
childDisposals.push(...pending.map(item => item.settled))
|
||||
const childFailures = await Promise.all(childDisposals.map(async (disposal) => {
|
||||
try {
|
||||
await disposal
|
||||
|
||||
@@ -3481,6 +3481,63 @@ describe('continuable errors', () => {
|
||||
await drained
|
||||
})
|
||||
|
||||
it.each([1, 2])('disposes a local activation only after pending startup at depth %s rolls back', async (depth) => {
|
||||
const { ctx, parent } = await setup(Array.from({ length: depth }, () => 'hang' as const))
|
||||
const started = await ctx.subagents.startActivation({ ...startSpec(parent), delivery: 'caller' })
|
||||
const child = ctx.agents.get(started.childId)!
|
||||
const entered = Promise.withResolvers<AbortSignal>()
|
||||
const releaseStartup = Promise.withResolvers<undefined>()
|
||||
const cleaning = Promise.withResolvers<undefined>()
|
||||
const releaseCleanup = Promise.withResolvers<undefined>()
|
||||
let starting: Promise<unknown> | undefined
|
||||
let disposal: Promise<void> | undefined
|
||||
try {
|
||||
let owner = child
|
||||
for (let level = 1; level < depth; level++) {
|
||||
const nested = await ctx.subagents.startActivation({ ...startSpec(owner), delivery: 'caller' })
|
||||
owner = ctx.agents.get(nested.childId)!
|
||||
}
|
||||
ctx.subagents.registerProvider({
|
||||
name: 'external', inheritsParentContext: false,
|
||||
capabilities: { agentOptions: false, outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
start: async ({ signal }) => {
|
||||
entered.resolve(signal)
|
||||
await releaseStartup.promise
|
||||
return {
|
||||
id: SessionId('pending-descendant'),
|
||||
result: Promise.resolve({ output: [], stopReason: 'completed' }),
|
||||
dispose: async () => { cleaning.resolve(undefined); await releaseCleanup.promise },
|
||||
}
|
||||
},
|
||||
})
|
||||
starting = ctx.subagents.startActivation({ ...startSpec(owner, 'external'), delivery: 'caller' })
|
||||
.catch((error: unknown) => error)
|
||||
const signal = await entered.promise
|
||||
disposal = started.dispose()
|
||||
const disposed = vi.fn()
|
||||
void disposal.then(disposed, disposed)
|
||||
expect(signal.aborted).toBe(true)
|
||||
await expect(ctx.subagents.startActivation({ ...startSpec(child), delivery: 'caller' }))
|
||||
.rejects.toMatchObject({ code: 'ACTIVATION_CLOSING' })
|
||||
releaseStartup.resolve(undefined)
|
||||
await cleaning.promise
|
||||
expect(disposed).not.toHaveBeenCalled()
|
||||
expect(ctx.agents.get(child.id)).toBe(child)
|
||||
releaseCleanup.resolve(undefined)
|
||||
await disposal
|
||||
await expect(starting).resolves.toMatchObject({ name: 'AbortError' })
|
||||
expect(ctx.agents.get(child.id)).toBeUndefined()
|
||||
expect(ctx.agents.get(owner.id)).toBeUndefined()
|
||||
expect(ctx.agents.get(parent.id)).toBe(parent)
|
||||
await expect(ctx.subagents.waitForChildren(parent)).resolves.toBe(false)
|
||||
} finally {
|
||||
releaseStartup.resolve(undefined)
|
||||
releaseCleanup.resolve(undefined)
|
||||
await Promise.allSettled([starting, disposal, started.dispose()])
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('reports a failing branch after every branch settles, without pinning the rest', async () => {
|
||||
const hold = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
@@ -3545,7 +3602,7 @@ describe('continuable errors', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('rolls the transfer back when the parent begins disposal during materialization', async () => {
|
||||
it.each(['agent/created', 'subagent/start'] as const)('rejects child admission when parent disposal begins at %s', async (event) => {
|
||||
const hold = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('parent child'), gate: hold.promise },
|
||||
@@ -3558,29 +3615,18 @@ describe('continuable errors', () => {
|
||||
expect(found).toBeDefined()
|
||||
return found!
|
||||
})
|
||||
const activations = managerState(ctx)
|
||||
const ownerAgents = activations.ownerCtx.agents
|
||||
const before = new Set(ctx.agents.list().map(agent => agent.id))
|
||||
// Open the would-be parent's disposal only once the grandchild's Agent is
|
||||
// being created: the entry hold has already passed, so the post-transfer
|
||||
// ownership registration must reject and roll the transfer back with no
|
||||
// Activation and no live Agent left behind.
|
||||
const originalCreate = ownerAgents.create.bind(ownerAgents)
|
||||
let disposal: Promise<void> | undefined
|
||||
const createSpy = vi.spyOn(ownerAgents, 'create').mockImplementation((options) => {
|
||||
disposal = outer.dispose()
|
||||
createSpy.mockRestore()
|
||||
return originalCreate(options)
|
||||
})
|
||||
const detach = ctx.on(event, () => { disposal = outer.dispose() })
|
||||
|
||||
try {
|
||||
await expect(ctx.subagents.startActivation({ ...startSpec(child), delivery: 'parent' }))
|
||||
.rejects.toMatchObject({ code: 'ACTIVATION_CLOSING' })
|
||||
.rejects.toMatchObject(event === 'agent/created' ? { name: 'AbortError' } : { code: 'ACTIVATION_CLOSING' })
|
||||
await vi.waitFor(() => {
|
||||
expect(ctx.agents.list().map(agent => agent.id).filter(id => !before.has(id))).toEqual([])
|
||||
})
|
||||
} finally {
|
||||
createSpy.mockRestore()
|
||||
detach()
|
||||
hold.resolve(undefined)
|
||||
await disposal
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user