fix(agent-loop): report turn failures at source
This commit is contained in:
@@ -41,6 +41,8 @@ type Admission =
|
||||
| { kind: 'admitted'; messages: UserMessage[] }
|
||||
| { kind: 'blocked' }
|
||||
|
||||
type StepEndReason = Extract<TurnEndReason, { kind: 'completed' | 'max-tokens' }>
|
||||
|
||||
/** Remove adapter-derived values before plugins propose the next request config. */
|
||||
function requestProposal(header: EpochHeader): LlmCallConfig {
|
||||
if (header.adapterDefaults === undefined) return header.config
|
||||
@@ -136,15 +138,19 @@ export class ReactLoopAgent implements Agent {
|
||||
} while (driver !== this.driverDone)
|
||||
}
|
||||
|
||||
/** Report one failure at its live boundary, then preserve it for driver containment. */
|
||||
private throwError(error: unknown): never {
|
||||
const turn = this.phase.kind === 'running' ? this.phase.turn : this.phase.lastTurn
|
||||
const step = this.phase.kind === 'running' ? this.phase.step : 0
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error)
|
||||
throw error
|
||||
}
|
||||
|
||||
private async kick(): Promise<void> {
|
||||
try {
|
||||
while (await this.turn()) {}
|
||||
} catch (error: unknown) {
|
||||
if (this.phase.kind !== 'idle') {
|
||||
const turn = this.phase.kind === 'collecting' ? this.phase.lastTurn : this.phase.turn
|
||||
this.setPhase({ kind: 'idle', lastTurn: turn })
|
||||
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, 0, error)
|
||||
}
|
||||
} catch (_error) {
|
||||
// Admission and turn boundaries report before rethrowing; the driver only contains the rejection.
|
||||
} finally {
|
||||
if (this.phase.kind === 'running') {
|
||||
this.setPhase({ kind: 'idle', lastTurn: this.phase.turn })
|
||||
@@ -176,7 +182,9 @@ export class ReactLoopAgent implements Agent {
|
||||
|
||||
/** Admitted input stays unowned until `turn/start` commits. */
|
||||
private async turn(): Promise<boolean> {
|
||||
if (this.phase.kind === 'idle') throw new Error(`agent "${this.id}": turn without driver reservation`)
|
||||
if (this.phase.kind === 'idle') {
|
||||
this.throwError(new Error(`agent "${this.id}": turn without driver reservation`))
|
||||
}
|
||||
const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController()
|
||||
const { signal } = abort
|
||||
const lastTurn = this.phase.kind === 'collecting' ? this.phase.lastTurn : this.phase.turn
|
||||
@@ -191,10 +199,14 @@ export class ReactLoopAgent implements Agent {
|
||||
} catch (error: unknown) {
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- cancel may abort while admission awaits
|
||||
if (signal.aborted) return this.inbox.hasPending
|
||||
throw error
|
||||
this.throwError(error)
|
||||
}
|
||||
const turn = ++phase.turn
|
||||
this.session.append('turn/start', { turn })
|
||||
try {
|
||||
this.session.append('turn/start', { turn })
|
||||
} catch (error: unknown) {
|
||||
this.throwError(error)
|
||||
}
|
||||
let turnEnds: TurnEndReason | null = null
|
||||
try {
|
||||
while (true) {
|
||||
@@ -218,7 +230,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
admission = await this.admit(false)
|
||||
if (admission.kind === 'blocked') {
|
||||
turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }
|
||||
turnEnds = { kind: 'blocked' }
|
||||
return false
|
||||
}
|
||||
signal.throwIfAborted()
|
||||
@@ -226,16 +238,27 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
// oxlint-disable-next-line typescript/no-unnecessary-condition -- cancel may abort during any awaited turn operation
|
||||
if (signal.aborted) turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }
|
||||
else turnEnds = { kind: 'error', error: errorChain(error) }
|
||||
if (signal.aborted) {
|
||||
turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause }
|
||||
} else {
|
||||
turnEnds = {
|
||||
kind: 'error',
|
||||
error: error instanceof LlmError ? error.failure : errorChain(error),
|
||||
}
|
||||
this.throwError(error)
|
||||
}
|
||||
} finally {
|
||||
// oxlint-disable-next-line typescript/no-non-null-assertion -- the turn is always ended in this block
|
||||
this.session.append('turn/end', { turn, reason: turnEnds! })
|
||||
try {
|
||||
// oxlint-disable-next-line typescript/no-non-null-assertion -- every exit assigns a turn ending
|
||||
this.session.append('turn/end', { turn, reason: turnEnds! })
|
||||
} catch (error: unknown) {
|
||||
this.throwError(error)
|
||||
}
|
||||
}
|
||||
return this.inbox.hasPending
|
||||
}
|
||||
|
||||
private async step(): Promise<TurnEndReason | null> {
|
||||
private async step(): Promise<StepEndReason | null> {
|
||||
if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": step outside running phase`)
|
||||
const { turn, step, abort: { signal } } = this.phase
|
||||
signal.throwIfAborted()
|
||||
@@ -272,7 +295,9 @@ export class ReactLoopAgent implements Agent {
|
||||
() => Promise.resolve<RequestErrorAction>(undefined),
|
||||
)
|
||||
signal.throwIfAborted()
|
||||
if (action?.kind !== 'retry') return { kind: 'error', error: finish.failure }
|
||||
if (action?.kind !== 'retry') {
|
||||
throw new LlmError(finish.failure.message, finish.failure.code, finish.failure)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { createUserMessage, freezeMessage, CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { createUserMessage, freezeMessage, CallId, LlmError, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason, type UserMessage } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
@@ -594,12 +594,20 @@ describe('a finish-error stream chunk ends the turn as error, not completed', ()
|
||||
const agent = ctx.agentLoop.create(SessionId('a-finish-error'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
const errors: unknown[] = []
|
||||
ctx.on('agent/error', (_agent, turn, step, error) => {
|
||||
expect({ turn, step }).toEqual({ turn: 1, step: 1 })
|
||||
errors.push(error)
|
||||
})
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'error', error: failure }])
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]).toBeInstanceOf(LlmError)
|
||||
expect((errors[0] as LlmError).failure).toEqual(failure)
|
||||
|
||||
const events = [...agent.session.events]
|
||||
const turnEnd = events.find(event => event.type === 'turn/end')
|
||||
@@ -809,6 +817,7 @@ describe('turn and step boundary recovery', () => {
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(errors.map(error => error.message)).toEqual([
|
||||
'reject first step-end',
|
||||
'invariant violated by "@deepseek-ai/dsh-session": turn/end 1 while step 1 is still open',
|
||||
])
|
||||
expect(boundaryCounts(agent)).toMatchObject({
|
||||
@@ -842,6 +851,7 @@ describe('turn and step boundary recovery', () => {
|
||||
kind: 'error',
|
||||
error: { message: 'provider 500', code: 'SERVER' },
|
||||
})
|
||||
expect(threw).toBe(true)
|
||||
|
||||
// loop survives: a second turn runs to completion (invariants oracle would
|
||||
// throw on its turn/start if turn 1 had been left open).
|
||||
@@ -1012,7 +1022,9 @@ describe('turn and step boundary recovery', () => {
|
||||
expect(e.some(x => x.type === 'step/end')).toBe(true)
|
||||
expect(e.some(x => x.type === 'turn/end')).toBe(true)
|
||||
expect(e.at(-1)?.type).toBe('turn/end')
|
||||
expect(errors).toEqual([])
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]).toBeInstanceOf(LlmError)
|
||||
expect((errors[0] as LlmError).failure).toEqual({ message: 'provider 500', code: 'SERVER' })
|
||||
|
||||
// loop survives.
|
||||
send(agent, 'again')
|
||||
|
||||
@@ -181,7 +181,10 @@ describe('durable error rendering', () => {
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd).toBeDefined()
|
||||
if (turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error') {
|
||||
expect(turnEnd.data.reason.error).toBe('server overloaded')
|
||||
expect(turnEnd.data.reason.error).toEqual({
|
||||
message: 'server overloaded',
|
||||
code: 'RATE_LIMIT',
|
||||
})
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { createUserMessage, CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { createUserMessage, CallId, LlmError, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
|
||||
@@ -200,7 +200,9 @@ describe('agent loop', () => {
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(0) // the request was never sent
|
||||
expect(errors).toEqual([])
|
||||
expect(errors.map(error => error.message)).toEqual([
|
||||
'prompt variable "{{cwd}}" has no value for this assembly (section "deployment:persona")',
|
||||
])
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('error')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error'
|
||||
@@ -346,7 +348,7 @@ describe('agent loop', () => {
|
||||
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('second idle steer')
|
||||
})
|
||||
|
||||
it('contains a throwing step observer and carries steering into a replacement turn', async () => {
|
||||
it('stops after a throwing step observer and retains steering until a later wakeup', async () => {
|
||||
const adapter = new MockAdapter([textResponse('recovered')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('failed-steering'), { provider: 'mock', model: 'mock' })
|
||||
@@ -361,6 +363,13 @@ describe('agent loop', () => {
|
||||
send(agent, 'prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
|
||||
expect(agent.inbox.nextStep).toHaveLength(1)
|
||||
|
||||
send(agent, 'resume')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
|
||||
expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
|
||||
@@ -658,7 +667,7 @@ describe('agent loop', () => {
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
// The first turn failed at step 1 before a model call.
|
||||
expect(errors).toEqual([])
|
||||
expect(errors.map(error => error.message)).toEqual(['boom in pre-step'])
|
||||
expect(adapter.requests.length).toBe(0)
|
||||
const firstTurnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(firstTurnEnd?.type === 'turn/end' && firstTurnEnd.data.reason).toMatchObject({ kind: 'error' })
|
||||
@@ -1085,20 +1094,24 @@ describe('agent loop', () => {
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
const errors: Error[] = []
|
||||
const errors: unknown[] = []
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => {
|
||||
if (error instanceof Error) errors.push(error)
|
||||
errors.push(error)
|
||||
})
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(errors).toEqual([])
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]).toBeInstanceOf(LlmError)
|
||||
expect((errors[0] as LlmError).failure).toEqual({
|
||||
message: 'MockAdapter: script exhausted',
|
||||
code: 'UNKNOWN',
|
||||
})
|
||||
expect(reasons[0]).toMatchObject({ kind: 'error' })
|
||||
// The durable failure lives entirely on turn/end.reason (with the failing
|
||||
// step), not a standalone error event.
|
||||
// The durable failure and live relay describe the same failed turn.
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({ kind: 'error' })
|
||||
})
|
||||
|
||||
@@ -359,7 +359,12 @@ describe('request stability across the loop', () => {
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(agent.session.events.findLast(event => event.type === 'turn/end')).toMatchObject({
|
||||
data: { reason: { kind: 'error', error: failure.message } },
|
||||
data: {
|
||||
reason: {
|
||||
kind: 'error',
|
||||
error: failure instanceof LlmError ? failure.failure : failure.message,
|
||||
},
|
||||
},
|
||||
})
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
},
|
||||
|
||||
@@ -100,6 +100,8 @@ export interface TurnEndReasonMap {
|
||||
completed: { kind: 'completed' }
|
||||
/** A cancellation request interrupted the live turn. */
|
||||
aborted: { kind: 'aborted'; reason: AgentCancelCause }
|
||||
|
||||
blocked: { kind: 'blocked' }
|
||||
/**
|
||||
* The turn failed.
|
||||
*/
|
||||
|
||||
@@ -300,6 +300,10 @@ export function apply(ctx: Context): void {
|
||||
}
|
||||
return
|
||||
case 'turn/end':
|
||||
if (event.data.reason.kind === 'max-tokens') {
|
||||
disarm(state)
|
||||
return
|
||||
}
|
||||
if (event.data.reason.kind !== 'aborted') return
|
||||
if (state.attempt?.phase === 'admitted') state.attempt.cancelled = true
|
||||
else disarm(state)
|
||||
|
||||
@@ -224,15 +224,15 @@ describe('same-session goal driving', () => {
|
||||
['rate limit', new LlmError('slow down', 'RATE_LIMIT')],
|
||||
['request error', new Error('provider broke')],
|
||||
['max tokens', maxTokensResponse('unfinished')],
|
||||
] as const)('does not attribute a %s to one goal follow-up', async (_label, response) => {
|
||||
const test = await harness(Array.from({ length: 8 }, () => response))
|
||||
] as const)('disarms automatic continuation after a %s', async (_label, response) => {
|
||||
const test = await harness([response])
|
||||
test.ctx.goals.create(test.agent, { objective: 'stop safely', maxGoalRounds: 8 })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current =>
|
||||
current?.phase === 'active' && current.activation === 'disarmed')
|
||||
|
||||
expect(goal).toMatchObject({ roundsStarted: 8, activation: 'disarmed' })
|
||||
expect(goal?.blockedReason?.code).toBe('round-limit')
|
||||
expect(test.adapter.requests).toHaveLength(8)
|
||||
expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('maps a downstream prompt veto to blocked without admitting the round', async () => {
|
||||
@@ -827,7 +827,7 @@ describe('same-session goal driving', () => {
|
||||
expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('goal-session'))
|
||||
})
|
||||
|
||||
it('ignores the failed outcome of a round made stale by human work queued at turn start', async () => {
|
||||
it('keeps terminal agent failure disarmed and defers queued human work until another wakeup', async () => {
|
||||
const test = await harness([new Error('round one broke'), textResponse('human answer')])
|
||||
let queued = false
|
||||
test.ctx.on('session/event', (session, event) => {
|
||||
@@ -841,13 +841,18 @@ describe('same-session goal driving', () => {
|
||||
})
|
||||
test.ctx.goals.create(test.agent, { objective: 'survive a stale failure', maxGoalRounds: 1 })
|
||||
|
||||
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
|
||||
await waitForGoal(test.ctx, test.agent, current =>
|
||||
current?.phase === 'active' && current.activation === 'disarmed')
|
||||
|
||||
expect(test.adapter.requests).toHaveLength(1)
|
||||
expect(test.agent.inbox.nextTurn).toHaveLength(1)
|
||||
|
||||
test.agent.steer(createUserMessage({ content: [{ type: 'text', text: 'resume after failure' }], source: { kind: 'user' } }))
|
||||
await test.agent.whenIdle()
|
||||
|
||||
// The stale round's turn-error never blocks the goal; only the durable
|
||||
// round budget does, after the interleaved human turn ran.
|
||||
expect(goal?.blockedReason?.code).toBe('round-limit')
|
||||
expect(test.adapter.requests).toHaveLength(2)
|
||||
expect(requestText(test.adapter.requests[1]!)).toContain('human interleaved')
|
||||
expect(requestText(test.adapter.requests[1]!)).toContain('resume after failure')
|
||||
})
|
||||
|
||||
it('waits for work queued by a pause observer before considering the next round', async () => {
|
||||
|
||||
@@ -93,7 +93,8 @@ export function isQuotaExceededError(detail: string): boolean {
|
||||
/**
|
||||
* Render a thrown value with its full `cause` chain and AggregateError
|
||||
* members, so transport wrappers like undici's `TypeError: fetch failed`
|
||||
* surface the underlying failure instead of masking it. Diagnostic-surface
|
||||
* surface the underlying failure instead of masking it. Plain structured
|
||||
* failures render their own data-backed `message`. Diagnostic-surface
|
||||
* rendering only (messages, notices, logs) — never parse the result; route on
|
||||
* {@link HarnessError.code}.
|
||||
* @param value - the caught value (`unknown` in catch clauses).
|
||||
@@ -109,7 +110,15 @@ export function errorChain(value: unknown): string {
|
||||
if (path.has(current)) return '<circular cause>'
|
||||
path.add(current)
|
||||
try {
|
||||
if (!(current instanceof Error)) return String(current)
|
||||
if (!(current instanceof Error)) {
|
||||
if (typeof current === 'object' && current !== null) {
|
||||
const descriptor = Object.getOwnPropertyDescriptor(current, 'message')
|
||||
if (descriptor !== undefined && 'value' in descriptor && typeof descriptor.value === 'string') {
|
||||
return descriptor.value
|
||||
}
|
||||
}
|
||||
return String(current)
|
||||
}
|
||||
const message = current.message === '' ? current.name : current.message
|
||||
const members = current instanceof AggregateError && current.errors.length > 0
|
||||
? ` [${current.errors.map(render).join('; ')}]`
|
||||
|
||||
@@ -145,6 +145,8 @@ describe('LlmService', () => {
|
||||
|
||||
it('errorChain survives non-Error values, hostile coercion, and circular causes', () => {
|
||||
expect(errorChain('plain string')).toBe('plain string')
|
||||
expect(errorChain({ message: 'structured provider failure', code: 'SERVER' }))
|
||||
.toBe('structured provider failure')
|
||||
expect(errorChain({ toString: () => { throw new Error('hostile') } })).toBe('<unrenderable value>')
|
||||
const circular = new Error('outer')
|
||||
circular.cause = circular
|
||||
|
||||
@@ -794,7 +794,7 @@ export function createTuiChat(
|
||||
liveErrors.delete(key)
|
||||
alreadyReported = true
|
||||
}
|
||||
const message = reason.error instanceof Error ? reason.error.message : String(reason.error)
|
||||
const message = errorChain(reason.error)
|
||||
if (!alreadyReported) appendNotice(message, 'error')
|
||||
break
|
||||
}
|
||||
|
||||
@@ -3755,7 +3755,7 @@ describe('pi-tui chat lifecycle and transcript', () => {
|
||||
events.session.append('turn/start', { turn: 6 })
|
||||
events.session.append('turn/end', {
|
||||
turn: 6,
|
||||
reason: { kind: 'error', error: 'structured provider failure' },
|
||||
reason: { kind: 'error', error: { message: 'structured provider failure', code: 'SERVER' } },
|
||||
})
|
||||
events.session.append('turn/start', { turn: 8 })
|
||||
events.session.append('turn/end', {
|
||||
@@ -3771,6 +3771,7 @@ describe('pi-tui chat lifecycle and transcript', () => {
|
||||
expect(events.terminal.output).toContain('durable failure')
|
||||
expect(events.terminal.output).toContain('Turn cancelled')
|
||||
expect(events.terminal.output).toContain('structured provider failure')
|
||||
expect(events.terminal.output).not.toContain('[object Object]')
|
||||
expect(events.terminal.output).toContain('output-token limit')
|
||||
expect(events.terminal.output).toContain('previous process ended')
|
||||
expect(events.terminal.output).toContain('Turn ended: plugin-policy')
|
||||
|
||||
Reference in New Issue
Block a user