diff --git a/.changeset/steer-abort-transfer.md b/.changeset/steer-abort-transfer.md new file mode 100644 index 00000000000..936f215c279 --- /dev/null +++ b/.changeset/steer-abort-transfer.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix loss of Ctrl+S steered messages when the current turn is interrupted. diff --git a/packages/agent-core-v2/src/agent/loop/loopService.ts b/packages/agent-core-v2/src/agent/loop/loopService.ts index 3c7ad890753..b40b34a8880 100644 --- a/packages/agent-core-v2/src/agent/loop/loopService.ts +++ b/packages/agent-core-v2/src/agent/loop/loopService.ts @@ -442,11 +442,11 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { return step; } - private cancelStep(job: TurnJob, step: MutableStep, request: StepRequest, reason?: unknown): boolean { + private cancelStep(job: TurnJob, step: MutableStep, request: StepRequest, reason?: unknown, abortRequest = true): boolean { if (step.state === 'completed' || step.state === 'failed' || step.state === 'cancelled') return false; const cancellation = reason ?? userCancellationReason(); step.state = 'cancelled'; - request.abort(); + if (abortRequest) request.abort(); step.controller?.abort(cancellation); step.resultControl?.resolve({ type: 'cancelled', reason: cancellation }); return true; @@ -599,8 +599,21 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { const job = this.activeTurnJob?.turn === turn ? this.activeTurnJob : undefined; if (job === undefined) return; const reason = result?.type === 'cancelled' ? result.reason : abortError('Turn ended'); + const transferred = new Map(); + for (const request of job.queue.drain()) { + if (request.state === 'pending' && !request.turnScoped) { + this.standaloneStepQueue.enqueue(request, 'tail'); + transferred.set(request.id, request); + } + } for (const step of job.steps.values()) { - if (step.state === 'queued' || step.state === 'running') step.cancel(reason); + if (step.state !== 'queued' && step.state !== 'running') continue; + const request = transferred.get(step.id); + if (request === undefined) { + step.cancel(reason); + } else { + this.cancelStep(job, step, request, reason, false); + } } this.activeTurnJob = undefined; this.maybeSettle(); diff --git a/packages/agent-core-v2/test/agent/loop/loop.test.ts b/packages/agent-core-v2/test/agent/loop/loop.test.ts index 346d6908bb4..68136247540 100644 --- a/packages/agent-core-v2/test/agent/loop/loop.test.ts +++ b/packages/agent-core-v2/test/agent/loop/loop.test.ts @@ -20,7 +20,7 @@ import { TurnStepStarted, } from '#/agent/loop/turnEvents'; import { TurnEnded } from '#/agent/loop/turnOps'; -import { RetryStepRequest } from '#/agent/prompt/promptStepRequests'; +import { RetryStepRequest, SteerStepRequest } from '#/agent/prompt/promptStepRequests'; import type { ExecutableTool } from '#/tool/toolContract'; import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry'; import { IEventBus } from '#/app/event/eventBus'; @@ -34,6 +34,7 @@ import { type TestAgentOptions, } from '../../harness'; import { recordingTelemetry, type TelemetryRecord } from '../../app/telemetry/stubs'; +import { createReminderStub } from '../../features/reminder/stubs'; type GenerateFn = NonNullable; @@ -661,6 +662,130 @@ describe('Agent loop', () => { expect(ctx.llmCalls).toHaveLength(3); }); + it('transfers an unmaterialized non-turn-scoped request into the next turn on cancel', async () => { + let started!: () => void; + const stepEntered = new Promise((resolve) => { started = resolve; }); + let release!: () => void; + const canFinish = new Promise((resolve) => { release = resolve; }); + const hook = loop.hooks.onWillBeginStep.register('test-non-turn-scoped-transfer', async (_hookCtx, next) => { + started(); + await canFinish; + await next(); + }); + + const steerCalls: unknown[] = []; + const steer = new SteerStepRequest( + { role: 'user', content: [{ type: 'text', text: 'steered' }], toolCalls: [], origin: { kind: 'user' } }, + [], + createReminderStub(), + (materialized) => { steerCalls.push(materialized); }, + () => {}, + ); + + ctx.mockNextResponse({ type: 'text', text: 'first answer' }); + const first = (await loop.enqueue(nextTurnMessage('first')).assigned).turn; + await stepEntered; + + const assignment = await loop.enqueue(steer).assigned; + expect(assignment.turn.id).toBe(first.id); + + first.cancel(userCancellationReason()); + release(); + await expect(first.result).resolves.toMatchObject({ type: 'cancelled' }); + hook.dispose(); + + expect(steer.state).toBe('pending'); + expect(steerCalls).toHaveLength(0); + await expect(assignment.step.result).resolves.toMatchObject({ type: 'cancelled' }); + expect(loop.hasPendingRequests()).toBe(true); + + ctx.mockNextResponse({ type: 'text', text: 'second answer' }); + const second = (await loop.enqueue(nextTurnMessage('second')).assigned).turn; + await expect(second.result).resolves.toMatchObject({ type: 'completed' }); + + expect(steerCalls).toHaveLength(1); + expect(steer.state).toBe('materialized'); + const texts = llmUserTexts(ctx.llmCalls.at(-1)).join('\n'); + expect(texts).toContain('second'); + expect(texts).toContain('steered'); + expect(texts.indexOf('steered')).toBeGreaterThan(texts.indexOf('second')); + }); + + it('still aborts turn-scoped queued requests when the turn is cancelled', async () => { + let started!: () => void; + const stepEntered = new Promise((resolve) => { started = resolve; }); + let release!: () => void; + const canFinish = new Promise((resolve) => { release = resolve; }); + const hook = loop.hooks.onWillBeginStep.register('test-turn-scoped-abort', async (_hookCtx, next) => { + started(); + await canFinish; + await next(); + }); + + const scoped = new MessageStepRequest( + { role: 'user', content: [{ type: 'text', text: 'scoped' }], toolCalls: [], origin: { kind: 'user' } }, + { mergeable: true, admission: 'activeTurnOnly' }, + ); + + ctx.mockNextResponse({ type: 'text', text: 'first answer' }); + const first = (await loop.enqueue(nextTurnMessage('first')).assigned).turn; + await stepEntered; + + const assignment = await loop.enqueue(scoped).assigned; + expect(assignment.turn.id).toBe(first.id); + + first.cancel(userCancellationReason()); + release(); + await expect(first.result).resolves.toMatchObject({ type: 'cancelled' }); + hook.dispose(); + + expect(scoped.state).toBe('aborted'); + await expect(assignment.step.result).resolves.toMatchObject({ type: 'cancelled' }); + expect(loop.hasPendingRequests()).toBe(false); + }); + + it('transfers an unmaterialized non-turn-scoped request when the turn fails', async () => { + let started!: () => void; + const stepEntered = new Promise((resolve) => { started = resolve; }); + let fail!: () => void; + const canFail = new Promise((resolve) => { fail = resolve; }); + const hook = loop.hooks.onWillBeginStep.register('test-non-turn-scoped-transfer-failure', async () => { + started(); + await canFail; + throw new Error('before step failed'); + }); + + const steerCalls: unknown[] = []; + const steer = new SteerStepRequest( + { role: 'user', content: [{ type: 'text', text: 'steered' }], toolCalls: [], origin: { kind: 'user' } }, + [], + createReminderStub(), + (materialized) => { steerCalls.push(materialized); }, + () => {}, + ); + + const first = (await loop.enqueue(nextTurnMessage('first')).assigned).turn; + await stepEntered; + + const assignment = await loop.enqueue(steer).assigned; + expect(assignment.turn.id).toBe(first.id); + + fail(); + await expect(first.result).resolves.toMatchObject({ type: 'failed' }); + hook.dispose(); + + expect(steer.state).toBe('pending'); + expect(steerCalls).toHaveLength(0); + + ctx.mockNextResponse({ type: 'text', text: 'second answer' }); + const second = (await loop.enqueue(nextTurnMessage('second')).assigned).turn; + await expect(second.result).resolves.toMatchObject({ type: 'completed' }); + + expect(steerCalls).toHaveLength(1); + expect(steer.state).toBe('materialized'); + expect(llmUserTexts(ctx.llmCalls.at(-1)).join('\n')).toContain('steered'); + }); + it('refuses a quiescence lease while a turn is active without cancelling it', async () => { let started!: () => void; const activeStarted = new Promise((resolve) => { @@ -1714,6 +1839,12 @@ function nextTurnMessage(text: string): MessageStepRequest { ); } +function llmUserTexts(call: TestAgentContext['llmCalls'][number] | undefined): string[] { + return (call?.history ?? []) + .filter((message) => message.role === 'user') + .flatMap((message) => message.content.filter((part) => part.type === 'text').map((part) => part.text)); +} + function createTimingRequester(): IAgentLLMRequesterService { const timing: ModelRequestTiming = { firstTokenLatencyMs: 100,