fix(host): settle pending approvals as cancelled on gateway teardown
Disposability parity with the question provider: a gateway disposed while approvals are pending settles every registry entry as 'cancelled' (the service's fail-closed vocabulary), so no ctx.approval ask dangles past the proxy's lifetime and mux subscribers see the withdrawal. Spec mounts the proxy on its own fiber and drives dispose with a live ask. Addresses the ds-review-bot suggestion on PR #851.
This commit is contained in:
@@ -609,6 +609,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
// the same rpcId (the refresh-recovery baseline) — and withdraws on the
|
// the same rpcId (the refresh-recovery baseline) — and withdraws on the
|
||||||
// ask's own abort signal (turn cancel), pushing `cancelled` to subscribers.
|
// ask's own abort signal (turn cancel), pushing `cancelled` to subscribers.
|
||||||
if (ctx.get('approval') !== undefined) {
|
if (ctx.get('approval') !== undefined) {
|
||||||
|
// Teardown parity with the question provider above: a gateway disposed
|
||||||
|
// while approvals are pending settles every entry as 'cancelled' (the
|
||||||
|
// service's fail-closed vocabulary), so no ask promise dangles past the
|
||||||
|
// proxy's lifetime and subscribers see the withdrawal.
|
||||||
|
ctx.effect(() => () => {
|
||||||
|
for (const pending of [...pendingApprovals.values()]) pending.resolve('cancelled')
|
||||||
|
}, 'api-proxy: approval registry teardown')
|
||||||
ctx.on('approval/request', (req, next) => {
|
ctx.on('approval/request', (req, next) => {
|
||||||
// The audit pair `approval/asked` is already appended by the service
|
// The audit pair `approval/asked` is already appended by the service
|
||||||
// before dispatch, but dispatch rides a microtask: parallel tool calls
|
// before dispatch, but dispatch rides a microtask: parallel tool calls
|
||||||
|
|||||||
@@ -176,6 +176,31 @@ describe('approval pending registry', () => {
|
|||||||
abort.abort()
|
abort.abort()
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('gateway teardown settles pending approvals as cancelled (question-provider parity)', async () => {
|
||||||
|
// Mount the proxy on its own fiber so disposal exercises the teardown
|
||||||
|
// effect while an ask is still pending.
|
||||||
|
const ctx = new Context()
|
||||||
|
await ctx.plugin(SessionStore)
|
||||||
|
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||||
|
await ctx.plugin(UserInteractionService)
|
||||||
|
await ctx.plugin(AgentRegistry)
|
||||||
|
await ctx.plugin(ApprovalService)
|
||||||
|
let api!: ApiProxy
|
||||||
|
const fiber = ctx.plugin(Object.assign((fiberCtx: Context) => {
|
||||||
|
api = createApiProxy(fiberCtx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
|
||||||
|
}, { inject: ['sessions', 'agents', 'userInteraction', 'approval'] }))
|
||||||
|
await fiber.await()
|
||||||
|
const abort = new AbortController()
|
||||||
|
const mux = openMux(api, abort)
|
||||||
|
const asked = ctx.approval.request({ agent: agentOf(ctx), toolName: 'bash' })
|
||||||
|
const requested = requestedOf(await mux.waitFor('approval/requested'))
|
||||||
|
await fiber.dispose()
|
||||||
|
await expect(asked).resolves.toBe('cancelled')
|
||||||
|
const resolved = await mux.waitFor('approval/resolved')
|
||||||
|
expect(resolved).toMatchObject({ approvalId: requested.approvalId, outcome: 'cancelled' })
|
||||||
|
abort.abort()
|
||||||
|
})
|
||||||
|
|
||||||
it('carries callId on the frame and ignores a late abort after the answer settled', async () => {
|
it('carries callId on the frame and ignores a late abort after the answer settled', async () => {
|
||||||
const { ctx, api } = await harness()
|
const { ctx, api } = await harness()
|
||||||
const abort = new AbortController()
|
const abort = new AbortController()
|
||||||
|
|||||||
Reference in New Issue
Block a user