fix: address codex review round 1

- listChildren() takes an optional AbortSignal and rechecks it after every
  un-signalled session-query await (the cold-resume cooperative-cancellation
  boundary); list_agents forwards exec.signal so the registry's drain of
  started tool bodies cannot block on a slow or large catalog.
- The list_agents description now presents running/complete as a stored-record
  snapshot and defers deliverability to send_message, matching the ownership-
  conflict semantics the service tests pin.
This commit is contained in:
Dudu-0223
2026-07-26 23:00:47 +08:00
committed by Tianyi Cui
parent 8bbae77ae6
commit 4bd98407a9
15 changed files with 99 additions and 22 deletions

View File

@@ -251,11 +251,15 @@ export class SubagentService extends Service {
* live-preferred corpus without loading or resuming an Agent. The lineage
* trace supplies stable candidate order and live status; each candidate is
* then inspected independently for exactly one supported descriptor in its
* own suffix.
* own suffix. Session-query reads take no signal, so cancellation is
* cooperative: the scan rechecks `signal` after every un-signalled await and
* stops between candidates instead of draining a slow or large catalog after
* the caller has gone.
* @param parentSessionId - parent whose direct children are listed.
* @param signal - caller-owned cancellation observed between query awaits.
* @returns child and diagnostic entries in lineage-trace order.
*/
async listChildren(parentSessionId: SessionId): Promise<SubagentListEntry[]> {
async listChildren(parentSessionId: SessionId, signal?: AbortSignal): Promise<SubagentListEntry[]> {
const query = this.ctx.get('sessionQuery')
if (query === undefined) {
throw new SubagentError(
@@ -266,7 +270,8 @@ export class SubagentService extends Service {
const trace = await query.traceSession(parentSessionId)
const entries: SubagentListEntry[] = []
for (const node of trace.descendants) {
const entry = await this.inspectChild(query, parentSessionId, node.session)
assertListingNotCancelled(signal)
const entry = await this.inspectChild(query, parentSessionId, node.session, signal)
if (entry !== undefined) entries.push(entry)
}
return entries
@@ -277,10 +282,12 @@ export class SubagentService extends Service {
query: SessionQueryService,
parentSessionId: SessionId,
candidate: SessionRecord,
signal?: AbortSignal,
): Promise<SubagentListEntry | undefined> {
const childId = candidate.header.id
try {
const records = await query.listEvents(childId)
assertListingNotCancelled(signal)
// Fork seeds replay ancestor events, so only this child's suffix owns its descriptor.
const seedLength = candidate.header.seedLength ?? 0
const descriptorSeqs = records
@@ -294,6 +301,7 @@ export class SubagentService extends Service {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const seq = descriptorSeqs[0]!
const window = await query.readEvent({ sessionId: childId, seq })
assertListingNotCancelled(signal)
assertSessionHeadersCompatible(window.session, candidate.header)
if (window.session.parentSession !== parentSessionId || window.target.type !== 'subagent/descriptor') {
return { kind: 'diagnostic', id: childId, reason: 'corrupt' }
@@ -452,6 +460,13 @@ export class SubagentService extends Service {
export default SubagentService
/** Stop a cooperative listing scan at its next cancellation checkpoint. */
function assertListingNotCancelled(signal: AbortSignal | undefined): void {
if (signal?.aborted) {
throw new SubagentError('subagent listing was cancelled', 'CANCELLED')
}
}
/** Map isolated session-query failures to the fixed child diagnostic taxonomy. */
function perChildDiagnosticReason(error: unknown): 'corrupt' | 'unavailable' | undefined {
if (!(error instanceof SessionQueryError)) return undefined

View File

@@ -387,6 +387,60 @@ describe('SubagentService.listChildren', () => {
])
})
it('stops the scan at the between-candidates checkpoint when the signal aborts', async () => {
const { ctx, parent } = await setup([textResponse('one'), textResponse('two')])
await startChild(ctx, parent, 'first child')
await startChild(ctx, parent, 'second child')
const controller = new AbortController()
const query = ctx.get('sessionQuery')!
const originalListEvents = query.listEvents.bind(query)
let inspected = 0
query.listEvents = (sessionId) => {
inspected += 1
// Cancel while the first candidate's read is in flight: the loop's next
// between-candidates checkpoint must stop before the second read.
controller.abort()
return originalListEvents(sessionId)
}
await expect(ctx.subagents.listChildren(parent.id, controller.signal)).rejects.toThrow(
expect.objectContaining({ code: 'CANCELLED' }) as Error,
)
expect(inspected).toBe(1)
})
it('stops after a per-child read when the signal aborts mid-inspection', async () => {
const { ctx, parent } = await setup([textResponse('done')])
await startChild(ctx, parent, 'cancelled mid-read')
const controller = new AbortController()
const query = ctx.get('sessionQuery')!
const originalReadEvent = query.readEvent.bind(query)
let exactReads = 0
query.readEvent = async (request) => {
exactReads += 1
const window = await originalReadEvent(request)
controller.abort()
return window
}
// The post-read checkpoint throws a subagent error, which is not a
// session-query failure and therefore propagates instead of becoming a
// per-child diagnostic.
await expect(ctx.subagents.listChildren(parent.id, controller.signal))
.rejects.toThrow(expect.objectContaining({ code: 'CANCELLED' }) as Error)
expect(exactReads).toBe(1)
})
it('a pre-aborted signal stops before any candidate read', async () => {
const { ctx, parent } = await setup([textResponse('done')])
await startChild(ctx, parent, 'never read')
const controller = new AbortController()
controller.abort()
const query = ctx.get('sessionQuery')!
query.listEvents = () => Promise.reject(new Error('must not be called'))
await expect(ctx.subagents.listChildren(parent.id, controller.signal)).rejects.toThrow(
expect.objectContaining({ code: 'CANCELLED' }) as Error,
)
})
it('returns an empty array for a parent with no children', async () => {
const { ctx, parent } = await setup([])
await ctx.sessions.flush(parent.session)

View File

@@ -23,10 +23,12 @@ export function apply(ctx: Context): void {
ctx.tools.register(defineTool({
name: 'list_agents',
description:
'List your background subagents: every subagent you started that can receive `send_message`, '
+ 'whether it is still working (running) or has finished its current turn (complete — a follow-up '
+ 'message starts a new turn on the same conversation). Children that could not be read are '
+ 'reported as diagnostics instead of being silently dropped.',
'List your background subagents by durable id and label. Status is a snapshot of the stored '
+ 'record: running means the subagent session is currently live in this process, complete means '
+ 'it exists only in storage and a `send_message` starts a new turn on the same conversation. '
+ 'The snapshot is not a delivery promise — `send_message` performs the authoritative check and '
+ 'may still fail. Children that could not be read are reported as diagnostics instead of being '
+ 'silently dropped.',
parameters: {},
output: {
schema: {
@@ -70,7 +72,9 @@ export function apply(ctx: Context): void {
// Non-agent callers have no session whose children could be listed.
throw new Error('list_agents requires a calling agent (exec.agent was undefined)')
}
return await ctx.subagents.listChildren(parent.id)
// The registry drains started tool bodies, so the scan must observe the
// call's signal rather than finish a slow catalog after cancellation.
return await ctx.subagents.listChildren(parent.id, exec.signal)
},
}))
}