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' }, ]) })