From 7b875b9f62379f268f87247f577e48929666e755 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Sat, 25 Jul 2026 17:06:26 +0800 Subject: [PATCH] fix(agent-loop): publish turn state only after turn/start commits A pre-commit turn/start rejection previously left the machine bricked: lastTurn had already advanced past a turn the log never recorded, so every later turn/start violated the session invariant's contiguity rule, and the admitted prompt lingered in the outbox to leak into the next turn's request. Keep admitted input on the stack (an argument to run()) until turn/start commits, then advance lastTurn, set turnOpen, and append the prompt and its additional contexts as user/message events inside the now-existing turn. A rejected turn/start therefore unwinds with zero shared state to roll back, the turn number stays reusable, and the outbox never holds input for a turn that does not exist. This also restores the documented event order: the prompt follows turn/start directly instead of waiting in the outbox behind any steering carried over by cancel({keepInbox}). --- packages/core/agent-loop/src/agent.ts | 33 ++++++++++----- .../tests/contract-regressions.spec.ts | 40 +++++++++++++++++++ .../agent-loop/tests/coverage-edges.spec.ts | 8 ++-- 3 files changed, 67 insertions(+), 14 deletions(-) diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 6285d38874..e6fc0df010 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -1,7 +1,8 @@ /** * Concrete Agent loop over two pending-input lists: queued prompts each open a - * turn, while admitted input, steering, and injected context enter through the - * outbox at step boundaries. Every request is derived from the session log. + * turn that logs its admitted input after `turn/start` commits, while steering + * and injected context enter through the outbox at step boundaries. Every + * request is derived from the session log. * * @module dsh-agent-loop/agent */ @@ -205,7 +206,9 @@ export class ReactLoopAgent implements Agent { this.done = this.loopCtx.agents.withInitiator(this, async () => { const signal = admission.signal const trigger: TurnTrigger = { kind: 'message', source: message.source } - let admitted = false + // Admitted input stays on the stack until its turn/start commits: the + // turn owns it only once the turn exists in the log. + let admitted: UserMessageData[] | undefined try { signal.throwIfAborted() const decision = await this.loopCtx.waterfall( @@ -215,11 +218,10 @@ export class ReactLoopAgent implements Agent { signal.throwIfAborted() if (decision.kind === 'allow') { - this.outbox.push({ content: decision.content ?? message.content, source: message.source }) + admitted = [{ content: decision.content ?? message.content, source: message.source }] for (const context of decision.additionalContexts ?? []) { - this.outbox.push({ content: context.content, source: context.source }) + admitted.push({ content: context.content, source: context.source }) } - admitted = true } } catch (error: unknown) { if (agentInterruptReasonOf(signal) === undefined) { @@ -228,16 +230,19 @@ export class ReactLoopAgent implements Agent { } if (this.abort === admission) this.abort = undefined - if (!admitted) { + if (admitted === undefined) { this.continueOrIdle() return } - await this.run(trigger) + await this.run(trigger, admitted) }) } - /** Run one turn and any request-error retry over input already admitted by {@link kick}. */ - private async run(trigger: TurnTrigger): Promise { + /** + * Run one turn and any request-error retry. `admitted` input enters the log + * only after `turn/start` commits; until then it has no owner state to unwind. + */ + private async run(trigger: TurnTrigger, admitted: UserMessageData[] = []): Promise { if (this.abort !== undefined) throw new Error(`agent "${this.id}" is already running`) const controller = new AbortController() this.abort = controller @@ -246,7 +251,7 @@ export class ReactLoopAgent implements Agent { emitAgentEvent(this.loopCtx, this, 'agent/status', 'running') } const signal = controller.signal - const turn = ++this.lastTurn + const turn = this.lastTurn + 1 let step = 0 let reason: TurnEndReason = { kind: 'completed' } let idle: IdleReason = { kind: 'completed' } @@ -257,7 +262,13 @@ export class ReactLoopAgent implements Agent { try { signal.throwIfAborted() this.session.append('turn/start', { turn, trigger }) + // Committed: publish the turn to the machine's own bookkeeping and let + // the admitted input enter the log it now belongs to. this.turnOpen = true + this.lastTurn = turn + for (const input of admitted) { + this.session.append('user/message', input, { surfaceOp: 'append' }) + } signal.throwIfAborted() this.drainOutbox(turn) diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index f6629571d7..e3bb0a83a2 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -736,6 +736,46 @@ describe('turn and step boundary recovery', () => { expect(stepEndIdx).toBeLessThan(turnEndIdx) }) + it('a pre-commit turn/start rejection leaves no turn state for the next prompt', async () => { + const adapter = new MockAdapter([textResponse('after recovery')]) + const ctx = await balancedHarness(adapter) + const agent = ctx.agentLoop.create(SessionId('a-turnstart-veto'), { provider: 'mock', model: 'mock' }) + let rejected = false + ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const event = args[1] as SessionEvent + if (event.type === 'turn/start' && !rejected) { + rejected = true + throw new Error('reject turn-start before commit') + } + }) + const errors: Error[] = [] + ctx.on('agent/error', (_agent, _turn, _step, error) => { + if (error instanceof Error) errors.push(error) + }) + + send(agent, 'rejected') + await waitForIdle(ctx, agent) + + // The rejected turn left nothing behind: no events, no admitted prompt. + expect(agent.session.events).toEqual([]) + expect(errors.map(error => error.message)).toEqual(['reject turn-start before commit']) + + // The next prompt reuses the never-committed turn number and carries only + // its own admitted content — invariants (mounted) accept the log. + send(agent, 'go') + await waitForIdle(ctx, agent) + + expect(boundaryCounts(agent)).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 1, stepEnd: 1 }) + const turnStart = agent.session.events.find(event => event.type === 'turn/start') + expect(turnStart?.type === 'turn/start' && turnStart.data.turn).toBe(1) + const prompts = agent.session.events.filter(event => event.type === 'user/message') + expect(prompts.map(event => event.type === 'user/message' && event.data.content)).toEqual([ + [{ type: 'text', text: 'go' }], + ]) + expect(adapter.requests).toHaveLength(1) + }) + it('a pre-commit step/start validation failure does not invent a step boundary', async () => { const adapter = new MockAdapter([textResponse('never reached')]) const ctx = await balancedHarness(adapter) diff --git a/packages/core/agent-loop/tests/coverage-edges.spec.ts b/packages/core/agent-loop/tests/coverage-edges.spec.ts index 4a0cc18d01..427005d907 100644 --- a/packages/core/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts @@ -132,10 +132,12 @@ describe('thrown-value propagation', () => { const ends = agent.session.events.filter(event => event.type === 'turn/end') const messages = agent.session.events.filter(event => event.type === 'user/message') expect(starts).toHaveLength(1) - expect(starts[0]?.type === 'turn/start' && starts[0].data.turn).toBe(2) + // The rejected turn/start committed nothing, so the survivor reuses turn 1 + // and the rejected prompt does not leak into it. + expect(starts[0]?.type === 'turn/start' && starts[0].data.turn).toBe(1) expect(ends).toHaveLength(1) - expect(messages).toHaveLength(2) - expect(messages[1]?.type === 'user/message' && messages[1].data.content).toEqual([ + expect(messages).toHaveLength(1) + expect(messages[0]?.type === 'user/message' && messages[0].data.content).toEqual([ { type: 'text', text: 'survives as the next item' }, ]) })