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

fix(acp): drain continuable subagents before disposing top-level agents

Browse Source 
A continuable Activation outlives the turn that started it and owns descendant
teardown, so the bridge must drain that forest child-first before releasing the
top-level agents whose runtime the descendants depend on.

Also rewrites the authored continuable snapshot transcript for the Task-free
tool surface; the scenario's keyless replay is still under diagnosis.
This commit is contained in:
Dudu-0223
2026-07-30 14:41:21 +08:00
committed by Tianyi Cui
parent 3911b7117e
commit c8fbc111db
4 changed files with 154 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
examples/acp-agent/tests/fixtures/subagent-durability-failure.ts vendored
View File

@@ -1,41 +1,72 @@
import type { Context } from 'cordis'
import { appendFileSync } from 'node:fs'
export const name = 'subagent-durability-failure'
export const inject = ['sessionPersistence']
export const inject = ['sessionPersistence', 'subagents']
const UNKNOWN_CHILD_ID = '22222222-2222-4222-8222-222222222222'
/** Fail the child checkpoint and stabilize the authored follow-up failure ordering. */
export function apply(ctx: Context): void {
const log = (...a: unknown[]): void => {
try { appendFileSync('/tmp/probe.log', '[PROBE] ' + a.map(String).join(' ') + '\n') } catch { /* ignore */ }
}
const followupsAccepted = Promise.withResolvers<undefined>()
const persistence = ctx.sessionPersistence
const load = persistence.load.bind(persistence)
// The unavailable-child lookup is real asynchronous I/O. Fence it behind both
// authored follow-ups so runner speed cannot reorder the exact log.
persistence.load = async (id) => {
if (id === UNKNOWN_CHILD_ID) await followupsAccepted.promise
log('load', id)
if (id === UNKNOWN_CHILD_ID) { log('gating unknown-id load'); await followupsAccepted.promise; log('unknown-id load released') }
return load.call(persistence, id)
}
// Patch followup to log routing.
const subagents = ctx.subagents as unknown as { followup: (...a: unknown[]) => Promise<unknown> }
const origFollowup = subagents.followup.bind(subagents)
subagents.followup = async (...args: unknown[]) => {
log('followup childId=', args[1])
return origFollowup(...args)
}
ctx.effect(() => () => {
persistence.load = load
followupsAccepted.resolve(undefined)
}, 'subagent snapshot ordering')
// Both authored follow-ups reach the child inbox before the unknown-id lookup
// runs, so the queued FIFO order is what the transcript records.
let accepted = 0
ctx.on('agent/inbox/enqueue', (agent) => {
if (agent.session.header.parentSession === undefined) return
accepted += 1
log('child enqueue #', accepted, 'child=', agent.session.header.id)
if (accepted >= 3) followupsAccepted.resolve(undefined)
})
ctx.on('subagent/start', (info: unknown) => {
log('subagent/start id=', (info as { id?: unknown }).id)
})
ctx.on('session/event', (session, event) => {
if (session.header.parentSession === undefined) return
if (event.type === 'turn/start') log('child turn/start turn=', event.data.turn, 'child=', session.header.id)
if (event.type === 'user/message') {
const c = event.data.content?.[0]
log('child user/message text=', c && c.type === 'text' ? c.text : '?', 'child=', session.header.id)
}
})
const flushes = new WeakMap<object, number>()
const flushedTurnEnds = new WeakSet<object>()
ctx.on('session/flush', (session) => {
if (session.header.parentSession === undefined) return
const count = (flushes.get(session) ?? 0) + 1
flushes.set(session, count)
log('child flush #', count, 'child=', session.header.id)
if (session.events.at(-1)?.type !== 'turn/end') return
if (flushedTurnEnds.has(session)) throw new Error('snapshot disk full')
if (flushedTurnEnds.has(session)) {
log('THROW snapshot disk full')
throw new Error('snapshot disk full')
}
flushedTurnEnds.add(session)
})
}