diff --git a/packages/agent-loop/package.json b/packages/agent-loop/package.json index 8557ba85fa..b93ebeed14 100644 --- a/packages/agent-loop/package.json +++ b/packages/agent-loop/package.json @@ -32,6 +32,7 @@ }, "devDependencies": { "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-session": "^0.0.1", "@deepseek-ai/dsh-system-prompt": "^0.0.1", diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts index f6694c7aae..063ed21a9d 100644 --- a/packages/agent-loop/src/loop.ts +++ b/packages/agent-loop/src/loop.ts @@ -151,7 +151,9 @@ export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: number): Promise { const { session } = agent - // Drain queued messages into the session — they trigger this turn. + // --- Pre-turn. A throw here (the invariant guard or a user-message append) + // is owed NO turn/end — turn/start has not been appended — so it propagates + // to runLoop's backstop untouched. const queued = agent.inbox.drainQueued() const first = queued[0] /* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */ @@ -161,91 +163,161 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: session.append('user/message', { content: message.content, source: message.source }) } - session.append('turn/start', { turn, trigger }) - ctx.emit('agent/turn-start', agent, turn) - let reason: TurnEndReason = { kind: 'completed' } let step = 0 + let turnStarted = false + let turnEnded = false + let stepOpen = false + let errorReported = false - while (true) { - step += 1 - - // Steering from the previous round's step-end/continuation listeners - // (or turn-start listeners on the first step) joins before the request. - drainSteering(ctx, agent, turn) - - session.append('step/start', { turn, step }) - ctx.emit('agent/step-start', agent, turn, step) - - const abort = new AbortController() - handle.setAbort(abort) - - let stepOutcome: { hadToolCalls: boolean } | { error: Error } - try { - stepOutcome = await runStep(ctx, agent, turn, step, abort.signal) - } catch (error: unknown) { - stepOutcome = { error: toError(error) } - } finally { - handle.setAbort(undefined) - } - - if ('error' in stepOutcome) { - // Steering that arrived during the failed step stays in the inbox — - // runLoop re-enqueues it as a queued message, so an abort-then-steer - // starts a fresh turn instead of being silently consumed. - session.append('step/end', { turn, step }) - ctx.emit('agent/step-end', agent, turn, step) - const { error } = stepOutcome - if (handle.isDisposed()) { - reason = { kind: 'disposed' } - } else if (abort.signal.aborted) { - /* v8 ignore next -- abort.signal.reason always set by agent.abort() which provides a default */ - reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') } - } else { - const coded = error as CodedError - session.append('error', { turn, step, ...errorData(coded) }) - ctx.emit('agent/error', agent, turn, step, error) - reason = { kind: 'error', ...errorData(coded) } - } - break - } - - // Steering that arrived during streaming/tool execution. - const steered = drainSteering(ctx, agent, turn) - + // Close the open step exactly once (idempotent via stepOpen). The + // agent/step-end emit is contained: a throwing step-end listener must not + // abort finalization and strand the turn open (turn/end balance > notifying + // one bad listener). Appended before the emit (ADR 0003 append-before-emit). + const closeStep = (): void => { + if (!stepOpen) return + stepOpen = false session.append('step/end', { turn, step }) - ctx.emit('agent/step-end', agent, turn, step) - - const defaultDecision = stepOutcome.hadToolCalls || steered - let shouldContinue: boolean try { - shouldContinue = await ctx.waterfall( - 'agent/turn-continuation', agent, turn, defaultDecision, - () => Promise.resolve(defaultDecision), - ) - } catch (error: unknown) { - // A broken continuation plugin ends the turn, not the loop. - const err = toError(error) - session.append('error', { turn, step, ...errorData(err) }) - ctx.emit('agent/error', agent, turn, step, err) - reason = { kind: 'error', ...errorData(err) } - break - } - - // Steering from step-end/continuation listeners (the /goal pattern) - // demands the model see it — it overrides a negative decision; the - // next iteration's drain records it. - if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true - - if (!shouldContinue || handle.isDisposed()) { - /* v8 ignore next -- disposal during continuation-decision window is a narrow race; error-path disposal is covered elsewhere */ - if (handle.isDisposed()) reason = { kind: 'disposed' } - break + ctx.emit('agent/step-end', agent, turn, step) + } catch { + // contained: step/end is already recorded, so balance holds; a throwing + // step-end listener is the listener's bug, not the loop's. } } - session.append('turn/end', { turn, reason }) - ctx.emit('agent/turn-end', agent, turn, reason) + // Record a step/turn failure exactly once: append the single `error` event, + // set the error reason, and emit agent/error (contained — trap: a throwing + // agent/error listener must not re-escape and strand the turn). Disposal and + // abort set `reason` directly without calling this (no `error` event for + // those — they are not failures). + const failTurn = (err: CodedError): void => { + if (errorReported) return + errorReported = true + session.append('error', { turn, step, ...errorData(err) }) + reason = { kind: 'error', ...errorData(err) } + try { + ctx.emit('agent/error', agent, turn, step, err) + } catch { + // contained: the error is already logged; a throwing agent/error + // listener must not prevent the turn from closing. + } + } + + // Close the turn exactly once (idempotent via turnEnded). `emit` is false on + // the error path (the failure was already surfaced via agent/error) and true + // on the normal/inline-error path. A throwing agent/turn-end listener on the + // normal path escapes to the outer catch, which surfaces it via failTurn — + // turn/end is already appended, so balance holds either way. + const closeTurn = (emit: boolean): void => { + if (turnEnded) return + turnEnded = true + session.append('turn/end', { turn, reason }) + if (emit) ctx.emit('agent/turn-end', agent, turn, reason) + } + + try { + // --- Turn boundary. Once turn/start is appended, a turn/end is owed no + // matter what throws below; the catch + closeTurn guarantee it. + session.append('turn/start', { turn, trigger }) + turnStarted = true + ctx.emit('agent/turn-start', agent, turn) + + while (true) { + step += 1 + + // Steering from the previous round's step-end/continuation listeners + // (or turn-start listeners on the first step) joins before the request. + drainSteering(ctx, agent, turn) + + session.append('step/start', { turn, step }) + stepOpen = true + ctx.emit('agent/step-start', agent, turn, step) + + const abort = new AbortController() + handle.setAbort(abort) + + let stepOutcome: { hadToolCalls: boolean } | { error: Error } + try { + stepOutcome = await runStep(ctx, agent, turn, step, abort.signal) + } catch (error: unknown) { + stepOutcome = { error: toError(error) } + } finally { + handle.setAbort(undefined) + } + + if ('error' in stepOutcome) { + // Steering that arrived during the failed step stays in the inbox — + // runLoop re-enqueues it as a queued message, so an abort-then-steer + // starts a fresh turn instead of being silently consumed. + closeStep() + const { error } = stepOutcome + if (handle.isDisposed()) { + reason = { kind: 'disposed' } + } else if (abort.signal.aborted) { + /* v8 ignore next -- abort.signal.reason always set by agent.abort() which provides a default */ + reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') } + } else { + failTurn(error) + } + break + } + + // Steering that arrived during streaming/tool execution. + const steered = drainSteering(ctx, agent, turn) + + closeStep() + + const defaultDecision = stepOutcome.hadToolCalls || steered + let shouldContinue: boolean + try { + shouldContinue = await ctx.waterfall( + 'agent/turn-continuation', agent, turn, defaultDecision, + () => Promise.resolve(defaultDecision), + ) + } catch (error: unknown) { + // A broken continuation plugin ends the turn, not the loop. + failTurn(toError(error)) + break + } + + // Steering from step-end/continuation listeners (the /goal pattern) + // demands the model see it — it overrides a negative decision; the + // next iteration's drain records it. + if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true + + if (!shouldContinue || handle.isDisposed()) { + /* v8 ignore next -- disposal during continuation-decision window is a narrow race; error-path disposal is covered elsewhere */ + if (handle.isDisposed()) reason = { kind: 'disposed' } + break + } + } + + // Normal / inline-error loop exit: close the turn and notify. + closeTurn(true) + } catch (error: unknown) { + // A pre-turn throw (turn/start append) is owed no turn/end — rethrow to + // the backstop. Otherwise a boundary emit (turn-start, step-start, the + // normal-path turn-end) or other unhandled throw escaped: close any open + // step, choose the reason (disposal wins only if no error was reported), + // record the error, and close the turn WITHOUT re-emitting agent/turn-end. + if (!turnStarted) throw error + closeStep() + // Choose the close reason. Disposal wins only if no error was already + // reported: a turn disposed mid-step sets reason=disposed in the step-error + // branch (without reporting an error), and if closeTurn(true)'s turn-end + // emit then throws, we land here and must PRESERVE disposed rather than + // overwrite it with the listener's throw. Otherwise a boundary-emit throw + // on a live agent is a real failure → failTurn. (errorReported is mutated + // only inside the failTurn closure, which the analyzer can't follow, hence + // the inline lint-disable.) + if (handle.isDisposed() && !errorReported) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition + reason = { kind: 'disposed' } + } else { + failTurn(toError(error)) + } + closeTurn(false) + } // Durability checkpoint: persistence plugins drain write-behind buffers. // A failing persistence plugin is reported but doesn't kill the agent. diff --git a/packages/agent-loop/tests/review-fixes.spec.ts b/packages/agent-loop/tests/review-fixes.spec.ts index ee45be1f5b..fb297307dd 100644 --- a/packages/agent-loop/tests/review-fixes.spec.ts +++ b/packages/agent-loop/tests/review-fixes.spec.ts @@ -6,6 +6,7 @@ import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools' import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent' import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop' +import * as Invariants from '@deepseek-ai/dsh-invariants' import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts' /** @@ -639,3 +640,250 @@ describe('P1-6: step/start is appended before agent/step-start is emitted', () = expect(observed[0]).toMatchObject({ turn: 1, step: 1, lastEventType: 'step/start', sawStepStart: true }) }) }) + +describe('P1-5: a started turn (and any open step) is always closed on a boundary throw', () => { + // Harness with the invariants plugin loaded as an oracle: it throws on + // append if the log goes unbalanced (turn/end while a step is open, + // turn/start while a turn is open, etc.), so a regression surfaces as an + // InvariantError on the NEXT turn's append rather than a silent imbalance. + async function balancedHarness(adapter: MockAdapter) { + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(Invariants, { freeze: false }) + ctx.llm.registerAdapter(['mock'], adapter) + return ctx + } + + /** Count turn/step boundary events for balance assertions. */ + function boundaryCounts(agent: LoopAgent) { + const e = [...agent.session.events] + return { + turnStart: e.filter(x => x.type === 'turn/start').length, + turnEnd: e.filter(x => x.type === 'turn/end').length, + stepStart: e.filter(x => x.type === 'step/start').length, + stepEnd: e.filter(x => x.type === 'step/end').length, + errors: e.filter(x => x.type === 'error').length, + lastTurnEnd: e.findLast(x => x.type === 'turn/end'), + } + } + + it('a throwing agent/turn-start listener still closes the turn with exactly one error and one turn/end, no step', async () => { + const adapter = new MockAdapter([textResponse('never reached')]) + const ctx = await balancedHarness(adapter) + const agent = ctx.agentLoop.create('a-turnstart', { model: 'mock' }) + + let threw = false + ctx.on('agent/turn-start', () => { if (!threw) { threw = true; throw new Error('boom turn-start') } }) + const errors: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const c = boundaryCounts(agent) + // turn opened and closed; no step ran; exactly one error logged + emitted. + expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 0, stepEnd: 0, errors: 1 }) + expect(errors.map(e => e.message)).toEqual(['boom turn-start']) + expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toEqual({ kind: 'error', message: 'boom turn-start' }) + // model was never called (we threw before the step's request). + expect(adapter.requests).toHaveLength(0) + }) + + it('a throwing agent/step-start listener closes the open step then the turn (step/end before turn/end)', async () => { + const adapter = new MockAdapter([textResponse('never reached')]) + const ctx = await balancedHarness(adapter) + const agent = ctx.agentLoop.create('a-stepstart', { model: 'mock' }) + + let threw = false + ctx.on('agent/step-start', () => { if (!threw) { threw = true; throw new Error('boom step-start') } }) + const errors: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const e = [...agent.session.events] + const c = boundaryCounts(agent) + expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 1, stepEnd: 1, errors: 1 }) + expect(errors.map(x => x.message)).toEqual(['boom step-start']) + // step/end must precede turn/end (the invariants oracle would reject + // turn/end-while-step-open, but assert the order explicitly too). + const stepEndIdx = e.findIndex(x => x.type === 'step/end') + const turnEndIdx = e.findIndex(x => x.type === 'turn/end') + expect(stepEndIdx).toBeGreaterThanOrEqual(0) + expect(stepEndIdx).toBeLessThan(turnEndIdx) + }) + + it('a throwing agent/error listener during a step-error path still balances the turn, loop survives', async () => { + // First turn: model stream ends with a finish-error → step error path → + // failTurn emits agent/error, whose listener throws. The turn must still + // close balanced. Second turn proves the loop survived. + const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }] + const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')]) + const ctx = await balancedHarness(adapter) + const agent = ctx.agentLoop.create('a-errorlistener', { model: 'mock' }) + + let threw = false + ctx.on('agent/error', () => { if (!threw) { threw = true; throw new Error('boom error-listener') } }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const c = boundaryCounts(agent) + // turn 1 balanced despite the throwing agent/error listener. + expect(c.turnStart).toBe(1) + expect(c.turnEnd).toBe(1) + expect(c.stepStart).toBe(c.stepEnd) + expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toMatchObject({ kind: 'error', message: 'provider 500' }) + + // loop survives: a second turn runs to completion (invariants oracle would + // throw on its turn/start if turn 1 had been left open). + send(agent, 'again') + await waitForIdle(ctx, agent) + const c2 = boundaryCounts(agent) + expect(c2.turnStart).toBe(2) + expect(c2.turnEnd).toBe(2) + expect(c2.stepStart).toBe(c2.stepEnd) + }) + + it('disposal during a running turn ends the turn with reason disposed (balanced)', async () => { + // The 'hang' adapter blocks in stream() until the signal aborts; disposing + // the agent's fiber mid-turn aborts the in-flight step. The turn must close + // balanced with reason disposed (no error event for a disposal). + const adapter = new MockAdapter(['hang']) + const ctx = await balancedHarness(adapter) + let agent!: LoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create('a-dispose', { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason)) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 30)) + await fiber.dispose() // dispose during the hanging step + await agent.done + + const e = [...agent.session.events] + const turnStarts = e.filter(x => x.type === 'turn/start').length + const turnEnds = e.filter(x => x.type === 'turn/end').length + expect(turnStarts).toBe(1) + expect(turnEnds).toBe(1) // balanced — the turn was closed despite disposal + expect(reasons).toEqual([{ kind: 'disposed' }]) + // no error event: disposal is not a failure. + expect(e.some(x => x.type === 'error')).toBe(false) + }) + + it('preserves reason disposed when the turn-end emit throws during disposal (outer-catch disposed branch)', async () => { + // Dispose mid-step → the step-error branch sets reason=disposed (no error + // reported). closeTurn(true) then emits agent/turn-end, whose listener + // throws → control reaches the outer catch with isDisposed() && !errorReported, + // which must PRESERVE disposed rather than overwrite it with the listener's + // throw. This is the only path that exercises that catch sub-branch. + const adapter = new MockAdapter(['hang']) + const ctx = await balancedHarness(adapter) + let agent!: LoopAgent + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + agent = inner.agentLoop.create('a-dispose-emit-throw', { model: 'mock' }) + }, { inject: ['agentLoop'] })) + + // The FIRST agent/turn-end emit throws (the disposal-driven turn end). + let threw = false + ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end during disposal') } }) + // Collect agent/error emissions to prove none is surfaced through that + // channel either (the listener throw must be fully contained). + const errorEmits: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errorEmits.push(error)) + + send(agent, 'go') + await new Promise(r => setTimeout(r, 30)) + await fiber.dispose() // dispose during the hanging step + await agent.done + + // The throwing turn-end listener actually fired — proving the outer-catch + // path was exercised, not skipped. + expect(threw).toBe(true) + + const e = [...agent.session.events] + // Exactly one turn/start and one turn/end (balanced); the turn/end carries + // the disposed reason, NOT an error reason from the throwing listener. + expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1) + const turnEnd = e.findLast(x => x.type === 'turn/end') + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' }) + // The throwing turn-end listener is contained: no error event is logged and + // no agent/error is emitted (disposal is not a failure; the throw is swallowed). + expect(e.some(x => x.type === 'error')).toBe(false) + expect(errorEmits).toHaveLength(0) + }) + + it('a throw at the turn/start append (before turnStarted) is rethrown to the runLoop backstop', async () => { + // A session/event listener that throws specifically on the turn/start + // event makes session.append('turn/start') throw while turnStarted is + // still false. runTurn must NOT try to close a turn it never opened — it + // rethrows, and the runLoop backstop records the error and survives. + // (Uses the plain harness — NOT the invariants oracle — because the + // throwing listener is itself a session/event subscriber.) + const adapter = new MockAdapter([textResponse('turn 2')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a-preturn', { model: 'mock' }) + + let threw = false + ctx.on('session/event', (_session, event) => { + if (!threw && event.type === 'turn/start') { threw = true; throw new Error('boom turn/start append') } + }) + const errors: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + // The backstop logged exactly one error for the failed pre-turn append. + expect(errors.map(e => e.message)).toEqual(['boom turn/start append']) + // No turn/end was appended (none is owed — the turn never opened). + expect([...agent.session.events].some(e => e.type === 'turn/end')).toBe(false) + + // loop survives: a second turn runs normally. + send(agent, 'second') + await waitForIdle(ctx, agent) + expect(adapter.requests).toHaveLength(1) + }) + + it('a step error followed by a throwing turn-end listener logs the error exactly once (no double-report)', async () => { + // The step fails (finish-error) → failTurn records ONE error and sets the + // error reason. closeTurn(true) then appends turn/end and emits + // agent/turn-end, whose listener throws → the outer catch calls failTurn + // again, but its errorReported guard makes it a no-op. Trap #1: exactly one + // error, the turn stays balanced. + const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider down' } }] + const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')]) + const ctx = await balancedHarness(adapter) + const agent = ctx.agentLoop.create('a-double', { model: 'mock' }) + + let threw = false + ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end') } }) + const errors: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const c = boundaryCounts(agent) + // exactly one error event + one agent/error emit, despite two failTurn calls. + expect(c.errors).toBe(1) + expect(errors.map(e => e.message)).toEqual(['provider down']) + expect(c.turnStart).toBe(1) + expect(c.turnEnd).toBe(1) // single turn/end, balanced + expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toMatchObject({ kind: 'error', message: 'provider down' }) + + // loop survives the compound failure. + send(agent, 'again') + await waitForIdle(ctx, agent) + expect(boundaryCounts(agent).turnEnd).toBe(2) + }) +}) diff --git a/yarn.lock b/yarn.lock index 34aa3f9a3b..e3e3a55972 100644 --- a/yarn.lock +++ b/yarn.lock @@ -558,6 +558,7 @@ __metadata: resolution: "@deepseek-ai/dsh-agent-loop@workspace:packages/agent-loop" dependencies: "@deepseek-ai/dsh-agent": "npm:^0.0.1" + "@deepseek-ai/dsh-invariants": "npm:^0.0.1" "@deepseek-ai/dsh-llm": "npm:^0.0.1" "@deepseek-ai/dsh-session": "npm:^0.0.1" "@deepseek-ai/dsh-system-prompt": "npm:^0.0.1" @@ -611,7 +612,7 @@ __metadata: languageName: unknown linkType: soft -"@deepseek-ai/dsh-invariants@workspace:packages/invariants": +"@deepseek-ai/dsh-invariants@npm:^0.0.1, @deepseek-ai/dsh-invariants@workspace:packages/invariants": version: 0.0.0-use.local resolution: "@deepseek-ai/dsh-invariants@workspace:packages/invariants" dependencies: