test: restore 100% per-file coverage for agent-loop and the acp bridge
agent-loop: behavior tests for retry-while-busy, cancelled recovery windows, no-facts stream failures, idle-listener preemption, rejected driver promises under whenIdle, finish-chunk failures after step close, presentationMeta persistence, pre-aborted and torn-down create/resume signals, and configured-start failures over existing artifacts or after teardown. The remaining guards that no public path can reach carry justified v8 ignore annotations naming the invariant that starves them. acp bridge: cover the retry-adoption path (a retry turn resolves the prompt the failed turn deferred), the no-retry quiescence rejection, and the admission-blocked cancelled settlement; the synchronous send-throw catch is annotated as a future-proofing guard since the machine's send() contains listener failures.
This commit is contained in:
@@ -275,14 +275,16 @@ export function apply(ctx: Context, config: AcpConfig): void {
|
|||||||
record.inflight = inflight
|
record.inflight = inflight
|
||||||
try {
|
try {
|
||||||
record.agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
|
record.agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
|
||||||
|
// The machine's send() contains listener failures and accepts
|
||||||
|
// any typed input; this guards a future synchronous throw so the
|
||||||
|
// slot cannot wedge.
|
||||||
|
/* v8 ignore start -- future-proofing guard, see above */
|
||||||
} catch (error: unknown) {
|
} catch (error: unknown) {
|
||||||
record.inflight = undefined
|
record.inflight = undefined
|
||||||
// followup() throws only Errors (invalid input); the String arm
|
|
||||||
// is a defensive fallback for a non-Error throw.
|
|
||||||
/* v8 ignore next */
|
|
||||||
const detail = error instanceof Error ? error.message : String(error)
|
const detail = error instanceof Error ? error.message : String(error)
|
||||||
throw internalError(`prompt was not queued: ${detail}`)
|
throw internalError(`prompt was not queued: ${detail}`)
|
||||||
}
|
}
|
||||||
|
/* v8 ignore stop */
|
||||||
// Admission is pre-turn and retries outlive their failed turn, so a
|
// Admission is pre-turn and retries outlive their failed turn, so a
|
||||||
// turnless slot settles only at quiescence: a held failure rejects
|
// turnless slot settles only at quiescence: a held failure rejects
|
||||||
// (no retry adopted the prompt); no turn at all means admission
|
// (no retry adopted the prompt); no turn at all means admission
|
||||||
|
|||||||
@@ -166,4 +166,40 @@ describe('ACP prompt lifecycle', () => {
|
|||||||
.resolves.toEqual({ stopReason: 'end_turn' })
|
.resolves.toEqual({ stopReason: 'end_turn' })
|
||||||
await vi.waitFor(() => { expect(messageText(harness!)).toBe('next') })
|
await vi.waitFor(() => { expect(messageText(harness!)).toBe('next') })
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('a retry turn adopts the prompt instead of rejecting at the failed turn end', async () => {
|
||||||
|
harness = await makeBridgeHarness({ script: [errorResponse('transient boom'), textResponse('recovered')] })
|
||||||
|
// A recovery policy: schedule one retry for the failed request.
|
||||||
|
let retried = false
|
||||||
|
harness.ctx.on('agent/request-error', async (subject) => {
|
||||||
|
if (!retried) {
|
||||||
|
retried = true
|
||||||
|
subject.retry()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
const sessionId = await newSession(harness)
|
||||||
|
const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
|
||||||
|
expect(result.stopReason).toBe('end_turn')
|
||||||
|
await vi.waitFor(() => { expect(messageText(harness!)).toBe('recovered') })
|
||||||
|
})
|
||||||
|
|
||||||
|
it('a failed turn with no retry still rejects, at quiescence', async () => {
|
||||||
|
harness = await makeBridgeHarness({ script: [errorResponse('terminal boom')] })
|
||||||
|
let offered = 0
|
||||||
|
harness.ctx.on('agent/request-error', async () => { offered += 1 })
|
||||||
|
const sessionId = await newSession(harness)
|
||||||
|
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
|
||||||
|
.rejects.toThrow(/turn failed: terminal boom/)
|
||||||
|
expect(offered).toBe(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('an admission-blocked prompt settles cancelled instead of hanging', async () => {
|
||||||
|
harness = await makeBridgeHarness({ script: [] })
|
||||||
|
harness.ctx.on('agent/prompt-submit', async () => ({ kind: 'block' as const, reason: 'policy said no' }))
|
||||||
|
const sessionId = await newSession(harness)
|
||||||
|
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
|
||||||
|
.resolves.toEqual({ stopReason: 'cancelled' })
|
||||||
|
// The blocked prompt opened no turn and streamed nothing.
|
||||||
|
expect(messageText(harness)).toBe('')
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -295,6 +295,9 @@ export class ReactLoopAgent implements Agent {
|
|||||||
if (outcome.continueTurn || this.outbox.some(item => 'id' in item)) continue
|
if (outcome.continueTurn || this.outbox.some(item => 'id' in item)) continue
|
||||||
break
|
break
|
||||||
case 'request-failed': {
|
case 'request-failed': {
|
||||||
|
// step() reports request failures only after step/start commits
|
||||||
|
// and before its own step/end, so the step is always open here.
|
||||||
|
/* v8 ignore next -- unreachable false arm, see above */
|
||||||
if (this.stepOpen) {
|
if (this.stepOpen) {
|
||||||
this.stepOpen = false
|
this.stepOpen = false
|
||||||
this.session.append('step/end', { turn, step })
|
this.session.append('step/end', { turn, step })
|
||||||
@@ -315,6 +318,9 @@ export class ReactLoopAgent implements Agent {
|
|||||||
`agent "${this.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`,
|
`agent "${this.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`,
|
||||||
)
|
)
|
||||||
} finally {
|
} finally {
|
||||||
|
// Nothing else writes the window while the waterfall runs:
|
||||||
|
// cancel() only flips `requested` and a second run cannot start.
|
||||||
|
/* v8 ignore next -- unreachable false arm, see above */
|
||||||
if (this.retryWindow === retryWindow) this.retryWindow = undefined
|
if (this.retryWindow === retryWindow) this.retryWindow = undefined
|
||||||
}
|
}
|
||||||
retry = recoveryCompleted
|
retry = recoveryCompleted
|
||||||
@@ -342,6 +348,10 @@ export class ReactLoopAgent implements Agent {
|
|||||||
({ reason, idle } = this.settle(turn, step, caught, signal))
|
({ reason, idle } = this.settle(turn, step, caught, signal))
|
||||||
} finally {
|
} finally {
|
||||||
try {
|
try {
|
||||||
|
// Every step-close before this point clears the flag on both success
|
||||||
|
// and failure paths (step(), the request-failed branch, the catch),
|
||||||
|
// so the finally never finds a step still open.
|
||||||
|
/* v8 ignore next 4 -- unreachable last-resort step close, see above */
|
||||||
if (this.stepOpen) {
|
if (this.stepOpen) {
|
||||||
this.stepOpen = false
|
this.stepOpen = false
|
||||||
this.session.append('step/end', { turn, step })
|
this.session.append('step/end', { turn, step })
|
||||||
@@ -357,6 +367,9 @@ export class ReactLoopAgent implements Agent {
|
|||||||
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error)
|
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error)
|
||||||
}
|
}
|
||||||
this.retryWindow = undefined
|
this.retryWindow = undefined
|
||||||
|
// cancel() aborts but never clears the slot, and no second run can
|
||||||
|
// install a controller while this one is still unwinding.
|
||||||
|
/* v8 ignore next -- unreachable false arm, see above */
|
||||||
if (this.abort === controller) this.abort = undefined
|
if (this.abort === controller) this.abort = undefined
|
||||||
signal.removeEventListener('abort', cancelRetry)
|
signal.removeEventListener('abort', cancelRetry)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -328,6 +328,10 @@ export class AgentLoop extends Service implements AgentFactory {
|
|||||||
*/
|
*/
|
||||||
private prepare(ownerCtx: Context, id: SessionId, options: AgentOptions, session: Session, callerSignal?: AbortSignal): PreparedAgent {
|
private prepare(ownerCtx: Context, id: SessionId, options: AgentOptions, session: Session, callerSignal?: AbortSignal): PreparedAgent {
|
||||||
ownerCtx.fiber.assertActive()
|
ownerCtx.fiber.assertActive()
|
||||||
|
// Every caller reaches prepare() synchronously from a service method
|
||||||
|
// whose Cordis dispatch already requires the live factory fiber, or
|
||||||
|
// re-checks ownership itself after its awaits (resume's load barrier).
|
||||||
|
/* v8 ignore next -- unreachable backstop, see above */
|
||||||
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
|
||||||
if (callerSignal?.aborted) {
|
if (callerSignal?.aborted) {
|
||||||
throw callerSignal.reason instanceof Error
|
throw callerSignal.reason instanceof Error
|
||||||
@@ -393,15 +397,21 @@ export class AgentLoop extends Service implements AgentFactory {
|
|||||||
abort.abort(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
|
abort.abort(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
|
||||||
return dispose(true)
|
return dispose(true)
|
||||||
}, `agentLoop.lifecycle(${id})`)
|
}, `agentLoop.lifecycle(${id})`)
|
||||||
|
/* v8 ignore start -- ctx.effect throws only on an inactive fiber, which assertActive() above already rejected */
|
||||||
} catch (error: unknown) {
|
} catch (error: unknown) {
|
||||||
untrack()
|
untrack()
|
||||||
callerSignal?.removeEventListener('abort', onCallerAbort)
|
callerSignal?.removeEventListener('abort', onCallerAbort)
|
||||||
this.ownership.signal.removeEventListener('abort', onFactoryTeardown)
|
this.ownership.signal.removeEventListener('abort', onFactoryTeardown)
|
||||||
throw error
|
throw error
|
||||||
}
|
}
|
||||||
|
/* v8 ignore stop */
|
||||||
|
|
||||||
const assertLive = (): void => {
|
const assertLive = (): void => {
|
||||||
if (!abort.signal.aborted) return
|
if (!abort.signal.aborted) return
|
||||||
|
// Every fused abort source carries an Error reason: onCallerAbort and
|
||||||
|
// raceAbort wrap non-Error caller reasons, and the factory/lifecycle
|
||||||
|
// owners abort with constructed Errors.
|
||||||
|
/* v8 ignore next -- unreachable String() arm, see above */
|
||||||
throw abort.signal.reason instanceof Error ? abort.signal.reason : new Error(String(abort.signal.reason))
|
throw abort.signal.reason instanceof Error ? abort.signal.reason : new Error(String(abort.signal.reason))
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -498,3 +498,48 @@ describe('unrenderable failure settlement', () => {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe('driver bookkeeping edges', () => {
|
||||||
|
it('a whenIdle waiter survives a rejected driver promise', async () => {
|
||||||
|
const adapter = new MockAdapter([textResponse('ok')])
|
||||||
|
const ctx = await harness(adapter)
|
||||||
|
const agent = ctx.agentLoop.create(SessionId('waiter-chain'), { provider: 'mock', model: 'mock' })
|
||||||
|
// A persistent step/end veto escapes even the catch block's own close
|
||||||
|
// attempt, so the driver promise REJECTS; the waiter's catch arm must
|
||||||
|
// treat that rejection as quiescence instead of propagating it.
|
||||||
|
ctx.on('internal/dispatch', (_mode, name, args) => {
|
||||||
|
if (name !== 'session/event') return
|
||||||
|
const event = args[1] as SessionEvent
|
||||||
|
if (event.type === 'step/end') throw new Error('step close permanently rejected')
|
||||||
|
})
|
||||||
|
|
||||||
|
send(agent, 'one')
|
||||||
|
// Entered while the run owns the abort slot, the waiter awaits the
|
||||||
|
// driver promise; its rejection must count as quiescence and resolve.
|
||||||
|
await expect(agent.whenIdle()).resolves.toBeUndefined()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('a request failure that concludes recovery after step/end closed keeps the boundary balanced', async () => {
|
||||||
|
const { LlmError } = await import('@deepseek-ai/dsh-llm')
|
||||||
|
// The failure finish-chunk path returns request-failed AFTER step() has
|
||||||
|
// already appended step/end, so the request-failed branch's own
|
||||||
|
// step-close guard must see stepOpen === false and skip the append.
|
||||||
|
const adapter = new MockAdapter([
|
||||||
|
[
|
||||||
|
{ type: 'usage' as const, usage: { inputTokens: 1, outputTokens: 0 } },
|
||||||
|
{ type: 'finish' as const, reason: { kind: 'error' as const, failure: { message: 'empty', code: 'EMPTY_RESPONSE' } } },
|
||||||
|
] satisfies StreamChunk[],
|
||||||
|
])
|
||||||
|
const ctx = await harness(adapter)
|
||||||
|
const agent = ctx.agentLoop.create(SessionId('finish-after-close'), { provider: 'mock', model: 'mock' })
|
||||||
|
void LlmError
|
||||||
|
|
||||||
|
send(agent, 'go')
|
||||||
|
await agent.whenIdle()
|
||||||
|
|
||||||
|
const types = agent.session.events.map(e => e.type)
|
||||||
|
expect(types.filter(t => t === 'step/end')).toHaveLength(1)
|
||||||
|
const end = agent.session.events.findLast(e => e.type === 'turn/end')
|
||||||
|
expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error')
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|||||||
@@ -649,3 +649,110 @@ describe('creation and resume cancellation edges', () => {
|
|||||||
await disposal
|
await disposal
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe('configured-start failure edges', () => {
|
||||||
|
it('a non-Error mid-load abort reason is wrapped for the resume caller', async () => {
|
||||||
|
const sessionId = SessionId('resume-string-mid-abort')
|
||||||
|
const root = await persistSession(sessionId)
|
||||||
|
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
|
||||||
|
const gate = Promise.withResolvers<never>()
|
||||||
|
gate.promise.catch(() => undefined)
|
||||||
|
const loadStarted = Promise.withResolvers<undefined>()
|
||||||
|
ctx.sessionPersistence.load = () => {
|
||||||
|
loadStarted.resolve(undefined)
|
||||||
|
return gate.promise
|
||||||
|
}
|
||||||
|
const controller = new AbortController()
|
||||||
|
|
||||||
|
const resuming = ctx.agents.resume({
|
||||||
|
resumeSessionId: sessionId,
|
||||||
|
agentOptions: { provider: 'mock', model: 'mock' },
|
||||||
|
signal: controller.signal,
|
||||||
|
})
|
||||||
|
await loadStarted.promise
|
||||||
|
controller.abort('operator string reason')
|
||||||
|
|
||||||
|
await expect(promptly(resuming)).rejects.toThrow(/creation aborted/)
|
||||||
|
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||||
|
await ctx.fiber.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('a failing exact-id restore over an existing artifact stays loud', async () => {
|
||||||
|
const sessionId = SessionId('config-existing-corrupt')
|
||||||
|
const root = await persistSession(sessionId)
|
||||||
|
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
|
||||||
|
// The artifact exists (list reports it) but its load fails: this is
|
||||||
|
// corruption, not first creation — the failure must be reported, and no
|
||||||
|
// fresh same-id session may shadow the broken one.
|
||||||
|
ctx.sessionPersistence.load = () => Promise.reject(new Error('artifact corrupt'))
|
||||||
|
|
||||||
|
const configured = new Context()
|
||||||
|
await configured.plugin(LlmService)
|
||||||
|
await configured.plugin(SessionStore)
|
||||||
|
await configured.plugin(SystemPrompt)
|
||||||
|
await configured.plugin(ToolRegistry)
|
||||||
|
await configured.plugin(AgentRegistry)
|
||||||
|
await configured.plugin(SessionPersistenceJsonl, { root })
|
||||||
|
configured.llm.registerAdapter(['mock'], new MockAdapter([]))
|
||||||
|
configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id)
|
||||||
|
const configFailures: unknown[] = []
|
||||||
|
configured.on('agent-loop/config-start-failed', (_id, error) => { configFailures.push(error) })
|
||||||
|
const configWarnings: string[] = []
|
||||||
|
const configWarn = configured.logger.warn.bind(configured.logger)
|
||||||
|
configured.logger.warn = ((...args: unknown[]) => {
|
||||||
|
if (typeof args[0] === 'string') configWarnings.push(args[0])
|
||||||
|
return (configWarn as (...a: unknown[]) => unknown)(...args)
|
||||||
|
}) as typeof configured.logger.warn
|
||||||
|
const loop = await configured.plugin(AgentLoop, {
|
||||||
|
agents: [{ id: 'main', sessionId, provider: 'mock', model: 'mock' }],
|
||||||
|
})
|
||||||
|
await expect.poll(() => configFailures.length).toBe(1)
|
||||||
|
expect(configFailures[0]).toBeInstanceOf(Error)
|
||||||
|
expect((configFailures[0] as Error).message).toBe('artifact corrupt')
|
||||||
|
expect(configWarnings.some(w => w.includes('config-driven restore'))).toBe(true)
|
||||||
|
expect(configured.agents.get(sessionId)).toBeUndefined()
|
||||||
|
|
||||||
|
await loop.dispose()
|
||||||
|
await configured.fiber.dispose()
|
||||||
|
await ctx.fiber.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('suppresses a configured-resume failure that lands after teardown', async () => {
|
||||||
|
const sessionId = SessionId('config-late-resume-failure')
|
||||||
|
const root = await persistSession(sessionId)
|
||||||
|
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
|
||||||
|
const gate = Promise.withResolvers<never>()
|
||||||
|
gate.promise.catch(() => undefined)
|
||||||
|
const loadStarted = Promise.withResolvers<undefined>()
|
||||||
|
ctx.sessionPersistence.load = () => {
|
||||||
|
loadStarted.resolve(undefined)
|
||||||
|
return gate.promise
|
||||||
|
}
|
||||||
|
const failures: unknown[] = []
|
||||||
|
ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
|
||||||
|
|
||||||
|
const configured = new Context()
|
||||||
|
await configured.plugin(LlmService)
|
||||||
|
await configured.plugin(SessionStore)
|
||||||
|
await configured.plugin(SystemPrompt)
|
||||||
|
await configured.plugin(ToolRegistry)
|
||||||
|
await configured.plugin(AgentRegistry)
|
||||||
|
await configured.plugin(SessionPersistenceJsonl, { root })
|
||||||
|
configured.llm.registerAdapter(['mock'], new MockAdapter([]))
|
||||||
|
configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id)
|
||||||
|
configured.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
|
||||||
|
const loop = await configured.plugin(AgentLoop, {
|
||||||
|
agents: [{ id: 'main', resumeSessionId: sessionId, provider: 'mock', model: 'mock' }],
|
||||||
|
})
|
||||||
|
await loadStarted.promise
|
||||||
|
const disposal = loop.dispose()
|
||||||
|
gate.reject(new Error('late backend failure'))
|
||||||
|
await disposal
|
||||||
|
await new Promise(r => setTimeout(r, 20))
|
||||||
|
|
||||||
|
// Ownership deactivated before the failure landed: the report is dropped.
|
||||||
|
expect(failures).toEqual([])
|
||||||
|
await configured.fiber.dispose()
|
||||||
|
await ctx.fiber.dispose()
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user