fix(host): settle pre-aborted asks at registration; make audit pairing callId-symmetric
Two races from the #572 review, still live in the ported registry: An ask whose signal aborted between the service's own check and the microtask-deferred waterfall dispatch would register its abort listener AFTER the signal fired — never invoked, entry pending forever, zombie frame on every mux replay. The answerer now settles 'cancelled' synchronously before publishing anything. The audit back-scan let a callId-less ask claim the newest unclaimed asked record even when that record carried another call's id. Pairing is now shape-symmetric: callId-bearing asks take exactly their call's record, callId-less asks take only callId-less records — neither can steal under parallel asks.
This commit is contained in:
@@ -617,6 +617,11 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
for (const pending of [...pendingApprovals.values()]) pending.resolve('cancelled')
|
for (const pending of [...pendingApprovals.values()]) pending.resolve('cancelled')
|
||||||
}, 'api-proxy: approval registry teardown')
|
}, 'api-proxy: approval registry teardown')
|
||||||
ctx.on('approval/request', (req, next) => {
|
ctx.on('approval/request', (req, next) => {
|
||||||
|
// Dispatch rides a microtask behind the service's own signal check: an
|
||||||
|
// abort landing in that window would register the abort listener AFTER
|
||||||
|
// the signal fired — never invoked, entry pending forever, zombie frame
|
||||||
|
// on every mux replay. Settle synchronously instead of publishing.
|
||||||
|
if (req.signal?.aborted === true) return Promise.resolve<ApprovalOutcome>('cancelled')
|
||||||
// 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
|
||||||
// can append several asked events before any answerer runs. THIS
|
// can append several asked events before any answerer runs. THIS
|
||||||
@@ -634,7 +639,12 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
decided.add(event.data.id)
|
decided.add(event.data.id)
|
||||||
} else if (event.type === 'approval/asked') {
|
} else if (event.type === 'approval/asked') {
|
||||||
if (decided.has(event.data.id) || claimed.has(event.data.id)) continue
|
if (decided.has(event.data.id) || claimed.has(event.data.id)) continue
|
||||||
if (req.callId !== undefined && event.data.callId !== req.callId) continue
|
// Symmetric pairing: a callId-bearing ask only takes its own call's
|
||||||
|
// record, and a callId-less ask only takes a callId-less record —
|
||||||
|
// so neither shape can steal the other's audit id under parallel
|
||||||
|
// asks. (Today every producer — the tool executor — passes callId;
|
||||||
|
// the callId-less arm guards any future non-tool asker.)
|
||||||
|
if ((req.callId ?? null) !== (event.data.callId ?? null)) continue
|
||||||
approvalId = event.data.id
|
approvalId = event.data.id
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -176,6 +176,36 @@ describe('approval pending registry', () => {
|
|||||||
abort.abort()
|
abort.abort()
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('an ask whose signal aborted before dispatch settles cancelled without publishing', async () => {
|
||||||
|
// The service checks the signal, then dispatch rides a microtask: an
|
||||||
|
// abort in that window must not register a dead listener and strand the
|
||||||
|
// entry (zombie frame on every replay). Drive the waterfall directly
|
||||||
|
// with a pre-aborted signal to hit the answerer's register-path guard.
|
||||||
|
const { ctx, api } = await harness()
|
||||||
|
const abort = new AbortController()
|
||||||
|
const mux = openMux(api, abort)
|
||||||
|
const session = ctx.sessions.create()
|
||||||
|
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||||
|
session.append('approval/asked', { id: 'pre-aborted' as ApprovalRequestId, toolName: 'bash' })
|
||||||
|
const agent = { session } as unknown as Agent
|
||||||
|
const cancelled = new AbortController()
|
||||||
|
cancelled.abort()
|
||||||
|
const outcome = await ctx.waterfall(
|
||||||
|
'approval/request',
|
||||||
|
{ agent, toolName: 'bash', signal: cancelled.signal },
|
||||||
|
() => Promise.resolve('unavailable' as const),
|
||||||
|
)
|
||||||
|
expect(outcome).toBe('cancelled')
|
||||||
|
// Nothing was published: a fresh mux open replays no approval frame.
|
||||||
|
const abort2 = new AbortController()
|
||||||
|
const mux2 = openMux(api, abort2)
|
||||||
|
await new Promise(resolve => setTimeout(resolve, 10))
|
||||||
|
expect(mux2.envelopes.some(e => e.payload.type === 'approval/requested')).toBe(false)
|
||||||
|
abort2.abort()
|
||||||
|
abort.abort()
|
||||||
|
void mux
|
||||||
|
})
|
||||||
|
|
||||||
it('gateway teardown settles pending approvals as cancelled (question-provider parity)', async () => {
|
it('gateway teardown settles pending approvals as cancelled (question-provider parity)', async () => {
|
||||||
// Mount the proxy on its own fiber so disposal exercises the teardown
|
// Mount the proxy on its own fiber so disposal exercises the teardown
|
||||||
// effect while an ask is still pending.
|
// effect while an ask is still pending.
|
||||||
|
|||||||
Reference in New Issue
Block a user