diff --git a/.changeset/steer-mid-turn-reminder.md b/.changeset/steer-mid-turn-reminder.md new file mode 100644 index 00000000000..dc4c52045e8 --- /dev/null +++ b/.changeset/steer-mid-turn-reminder.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix messages steered into a running turn being overlooked while the agent continued its original task. diff --git a/packages/agent-core-v2/AGENTS.md b/packages/agent-core-v2/AGENTS.md index 821d54def10..c4fcb9f1ee5 100644 --- a/packages/agent-core-v2/AGENTS.md +++ b/packages/agent-core-v2/AGENTS.md @@ -79,7 +79,7 @@ Name-collision precedence is a deliberate, documented divergence from v1: v1 ran ## Model-facing reminders -Two delivery paths only — never introduce a third (no deferred-delivery queues, no mid-step splice channels), both owned by the Agent-scope eager DI service `IAgentReminderService` — obtained through constructor injection within the same Agent scope, or through the scope handle's `accessor.get(IAgentReminderService)` across scope boundaries: reminders that restate current state (goal state, plan mode, date change, …) call `register(variant, provider)` and reconcile at every step head before the request is built, re-emitting after compaction or undo; reminders that report a one-off event (goal cancelled, AGENTS.md discovered, `/init` finished, …) call `notify(content, { variant, ownerPromptId? })` at a safe event point (a step/restore hook, an idle moment, or the loop-event fold's deferred append). The service owns `` wrapping and stamps `{ kind: 'injection', variant }`; `kind: 'injection'` is a lifecycle classification (hidden from the UI, not an undo anchor, dropped by compaction), not a provenance claim, and prompt-owned attachments carry `ownerPromptId` so undo treats them as part of their host prompt. +Two delivery paths only — never introduce a third (no deferred-delivery queues, no mid-step splice channels), both owned by the Agent-scope eager DI service `IAgentReminderService` — obtained through constructor injection within the same Agent scope, or through the scope handle's `accessor.get(IAgentReminderService)` across scope boundaries: reminders that restate current state (goal state, plan mode, date change, …) call `register(variant, provider)` and reconcile at every step head before the request is built, re-emitting after compaction or undo; reminders that report a one-off event (goal cancelled, AGENTS.md discovered, `/init` finished, …) call `notify(content, { variant, ownerPromptId? })` at a safe event point (a step/restore hook, an idle moment, or the loop-event fold's deferred append). The service owns `` wrapping and stamps `{ kind: 'injection', variant }`; `kind: 'injection'` is a lifecycle classification (hidden from the UI, not an undo anchor, dropped by compaction), not a provenance claim, and prompt-owned attachments carry `ownerPromptId` so undo treats them as part of their host prompt. A one-off event whose reminder must survive same-step compaction (the steer reminder) still goes through `register`, not `notify`: arm a flag at the event, emit while armed, clear when the step finishes — a `notify` append would be dropped by the splice with nothing left to re-emit it, while the armed provider restates it on the same-step reconciliation pass. ## Docs diff --git a/packages/agent-core-v2/src/agent/prompt/promptService.ts b/packages/agent-core-v2/src/agent/prompt/promptService.ts index 81dcac6947e..16890a01959 100644 --- a/packages/agent-core-v2/src/agent/prompt/promptService.ts +++ b/packages/agent-core-v2/src/agent/prompt/promptService.ts @@ -2,6 +2,7 @@ import { z } from 'zod'; import { IInstantiationService } from '#/_base/di/instantiation'; +import { Disposable } from '#/_base/di/lifecycle'; import { LifecycleScope } from '#/app/scopes'; import { ScopeActivation, registerScopedService } from '#/_base/di/scope'; import { defineState } from '#/state/state'; @@ -219,13 +220,19 @@ function mergeSteerMessages(records: readonly Record[]): ContextMessage { export const promptLaunchingKey = defineState('prompt.launching', () => false); -export class AgentPromptService implements IAgentPromptService { +export const STEER_REMINDER = [ + 'The user sent a new message while you were working; it appears above, delivered into the running turn.', + 'Address it as you continue this turn; where it changes the current task or approach, the new message takes precedence.', +].join(' '); + +export class AgentPromptService extends Disposable implements IAgentPromptService { declare readonly _serviceBrand: undefined; private active: (Record & { turn: Turn }) | undefined; private readonly pending: Record[] = []; private readonly steered = new Map(); private readonly reservedPromptIds = new Set(); private steering = 0; + private steerReminderArmed = false; private fullCompactionService: IAgentFullCompactionService | undefined; readonly hooks = { onBeforeSubmitPrompt: new OrderedHookSlot() }; @@ -244,13 +251,19 @@ export class AgentPromptService implements IAgentPromptService { @ISessionContext private readonly sessionContext: ISessionContext, @IAgentScopeContext private readonly scopeContext: IAgentScopeContext, ) { + super(); this.states.contributeState(promptLaunchingKey); this.states.contributeState(promptAdmissionKey); this.states.contributeState(promptResolutionKey); - toolExecutor.hooks.onDidExecuteTool.register('prompt-service-delivery', async (ctx, next) => { + this._register(this.reminder.register('steer', () => (this.steerReminderArmed ? STEER_REMINDER : undefined))); + this._register(this.loop.hooks.onDidFinishStep.register('steer-reminder', async (_ctx, next) => { + this.steerReminderArmed = false; + await next(); + })); + this._register(toolExecutor.hooks.onDidExecuteTool.register('prompt-service-delivery', async (ctx, next) => { await this.deliverToolResult(ctx); await next(); - }); + })); } private get launching(): boolean { @@ -413,6 +426,7 @@ export class AgentPromptService implements IAgentPromptService { this.pending.splice(index, 1); } const request = new SteerStepRequest(rerouted, captions, this.reminder, (materialized) => { + this.steerReminderArmed = true; void this.dispatcher.dispatch( new TurnSteer({ agentId: this.scopeContext.agentId, @@ -513,6 +527,7 @@ export class AgentPromptService implements IAgentPromptService { private settle(item: Record, result: TurnResult): void { if (this.active?.id !== item.id) return; this.active = undefined; + this.steerReminderArmed = false; const state = result.type === 'cancelled' ? 'cancelled' : result.type === 'failed' ? 'failed' : 'completed'; item.state = state; item.completionDeferred.resolve({ promptId: item.id, result, state }); for (const child of this.steered.get(item.id) ?? []) { child.state = state; child.completionDeferred.resolve({ promptId: child.id, result, state }); } diff --git a/packages/agent-core-v2/src/features/reminder/reminderService.ts b/packages/agent-core-v2/src/features/reminder/reminderService.ts index f19319aeba7..d2f79b6c55f 100644 --- a/packages/agent-core-v2/src/features/reminder/reminderService.ts +++ b/packages/agent-core-v2/src/features/reminder/reminderService.ts @@ -33,7 +33,7 @@ interface ReminderEntry { readonly variant: string; } -const REMINDER_VARIANT_PRIORITY = new Map([['date_change', -1]]); +const REMINDER_VARIANT_PRIORITY = new Map([['date_change', -1], ['steer', 1]]); interface ReminderActorContext { readonly entries: Set; diff --git a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts index a0850852669..c9261b6352c 100644 --- a/packages/agent-core-v2/test/agent/prompt/promptService.test.ts +++ b/packages/agent-core-v2/test/agent/prompt/promptService.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it, onTestFinished, vi } from 'vitest'; import { Readable } from 'node:stream'; -import { DisposableStore } from '#/_base/di/lifecycle'; +import { DisposableStore, toDisposable } from '#/_base/di/lifecycle'; import { createServices } from '#/_base/di/test'; import { Event } from '#/_base/event'; import { IAgentBlobService } from '#/agent/blob/agentBlobService'; @@ -13,11 +13,12 @@ import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompacti import { IAgentLoopService } from '#/agent/loop/loop'; import { TurnSteer } from '#/agent/loop/turnOps'; import { IAgentPromptService } from '#/agent/prompt/prompt'; -import { AgentPromptService, PromptAborted, PromptCompleted, PromptQueued, PromptStarted, PromptSteered, PromptSubmitted } from '#/agent/prompt/promptService'; +import { AgentPromptService, PromptAborted, PromptCompleted, PromptQueued, PromptStarted, PromptSteered, PromptSubmitted, STEER_REMINDER } from '#/agent/prompt/promptService'; import { IAgentScopeContext, makeAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { wrapSystemReminder } from '#/features/reminder/systemReminder'; import { IAgentReminderService } from '#/features/reminder/reminderService'; -import { createReminderStub } from '../../features/reminder/stubs'; +import type { ContextInjectionContext, ContextInjectionProvider } from '#/features/reminder/types'; +import { createReminderHarness, createReminderStub } from '../../features/reminder/stubs'; import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; import { IAgentToolPolicyService } from '#/agent/toolPolicy/toolPolicy'; import { IEventBus, ISessionEventBus } from '#/app/event/eventBus'; @@ -35,7 +36,7 @@ import { IFileService } from '#/app/file/fileService'; import { ISessionMediaStore } from '#/agent/media/sessionMediaStore'; import { stubContextMemory } from '../contextMemory/stubs'; -import { stubLoopWithHooks, stubToolExecutor, stubWire, type StubLoopOptions } from '../loop/stubs'; +import { runWillBeginStepHooks, stubLoopWithHooks, stubToolExecutor, stubWire, type StubLoopOptions } from '../loop/stubs'; import { registerStateServices } from '../../state/stubs'; import { SteerStepRequest } from '#/agent/prompt/promptStepRequests'; @@ -59,21 +60,28 @@ const noopBlob: IAgentBlobService = { isBlobRef: () => false, }; -function harness(loopOptions: StubLoopOptions = { pendingTurnResult: true }) { +function harness(loopOptions: StubLoopOptions & { integrationReminder?: boolean } = { pendingTurnResult: true }) { const disposables = new DisposableStore(); onTestFinished(() => disposables.dispose()); const context = stubContextMemory(); - const reminder = createReminderStub({ - notify: (content, notification) => { - context.append({ - role: 'user', - content: [{ type: 'text', text: wrapSystemReminder(content) }], - toolCalls: [], - origin: { kind: 'injection', ...notification }, - }); - }, - }); const loop = stubLoopWithHooks(loopOptions); + const reminderProviders = new Map(); + const reminder = loopOptions.integrationReminder === true + ? createReminderHarness(loop, context) + : createReminderStub({ + register: (variant, provider) => { + reminderProviders.set(variant, provider as ContextInjectionProvider); + return toDisposable(() => { reminderProviders.delete(variant); }); + }, + notify: (content, notification) => { + context.append({ + role: 'user', + content: [{ type: 'text', text: wrapSystemReminder(content) }], + toolCalls: [], + origin: { kind: 'injection', ...notification }, + }); + }, + }); const fullCompaction = { _serviceBrand: undefined, compacting: null, @@ -124,7 +132,23 @@ function harness(loopOptions: StubLoopOptions = { pendingTurnResult: true }) { (ix.get(IEventBus) as ISessionEventBus).activateAgent( ix.get(IAgentScopeContext).agentContext, ); - return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, eventBus: ix.get(IEventBus), intake }; + return { prompt: ix.get(IAgentPromptService), loop, context, fullCompaction, eventBus: ix.get(IEventBus), intake, reminderProviders }; +} + +function injectionContext(): ContextInjectionContext { + return { injectedPositions: [], lastInjectedAt: null, lastInjection: undefined, lastDisclosure: undefined, isNewTurn: false }; +} + +async function runDidFinishStepHooks(loop: IAgentLoopService): Promise { + await loop.hooks.onDidFinishStep.run({ + turnId: 0, + step: 0, + firstStepOfTurn: false, + signal: new AbortController().signal, + usage: { inputOther: 0, output: 0, inputCacheRead: 0, inputCacheCreation: 0 }, + finishReason: 'completed', + stopTurn: false, + }); } describe('AgentPromptService', () => { @@ -230,6 +254,109 @@ describe('AgentPromptService', () => { expect(events[0]).not.toHaveProperty('promptIds'); }); + it('emits the steer reminder from materialize until the step finishes', async () => { + const { prompt, context, loop, reminderProviders } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ message: message('new direction') }); + const provider = reminderProviders.get('steer')!; + await prompt.steer([queued.id]); + expect(await provider(injectionContext())).toBeUndefined(); + loop.drainNextBatch(context); + expect(await provider(injectionContext())).toBe(STEER_REMINDER); + expect(await provider(injectionContext())).toBe(STEER_REMINDER); + await runDidFinishStepHooks(loop); + expect(await provider(injectionContext())).toBeUndefined(); + }); + + it('keeps one steer reminder armed when separate steers merge into the same step', async () => { + const { prompt, context, loop, reminderProviders } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const one = await prompt.enqueue({ message: message('one') }); + const two = await prompt.enqueue({ message: message('two') }); + await prompt.steer([one.id]); + await prompt.steer([two.id]); + loop.drainNextBatch(context); + const provider = reminderProviders.get('steer')!; + expect(await provider(injectionContext())).toBe(STEER_REMINDER); + await runDidFinishStepHooks(loop); + expect(await provider(injectionContext())).toBeUndefined(); + }); + + it('delivers the steer reminder through the step hook and re-emits it after compaction drops it', async () => { + const { prompt, context, loop } = harness({ pendingTurnResult: true, integrationReminder: true }); + const steerInjections = () => + context.get().filter((m) => m.origin?.kind === 'injection' && m.origin.variant === 'steer'); + const steerTextIndex = () => + context.get().findIndex((m) => m.content.some((p) => p.type === 'text' && p.text === 'new direction')); + + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ message: message('new direction') }); + await prompt.steer([queued.id]); + loop.drainNextBatch(context); + + await runWillBeginStepHooks(loop); + expect(steerInjections()).toHaveLength(1); + const injected = context.get().at(-1)!; + expect(injected.origin).toEqual({ kind: 'injection', variant: 'steer' }); + expect(injected.content).toEqual([{ type: 'text', text: wrapSystemReminder(STEER_REMINDER) }]); + expect(steerTextIndex()).toBeGreaterThanOrEqual(0); + expect(steerTextIndex()).toBeLessThan(context.get().length - 1); + + context.applyCompaction({ summary: 'summary', contextSummary: 'summary', compactedCount: context.get().length, tokensBefore: 0 }); + expect(steerInjections()).toHaveLength(0); + expect(steerTextIndex()).toBeGreaterThanOrEqual(0); + + await runWillBeginStepHooks(loop); + expect(steerInjections()).toHaveLength(1); + expect(context.get().at(-1)!.content).toEqual([{ type: 'text', text: wrapSystemReminder(STEER_REMINDER) }]); + expect(steerTextIndex()).toBeLessThan(context.get().length - 1); + + await runDidFinishStepHooks(loop); + await runWillBeginStepHooks(loop); + expect(steerInjections()).toHaveLength(1); + }); + + it('re-emits the steer reminder when a step re-runs its injection pass before finishing', async () => { + const { prompt, context, loop } = harness({ pendingTurnResult: true, integrationReminder: true }); + const steerInjections = () => + context.get().filter((m) => m.origin?.kind === 'injection' && m.origin.variant === 'steer'); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ message: message('retry target') }); + await prompt.steer([queued.id]); + loop.drainNextBatch(context); + await runWillBeginStepHooks(loop); + await runWillBeginStepHooks(loop); + expect(steerInjections()).toHaveLength(2); + await runDidFinishStepHooks(loop); + await runWillBeginStepHooks(loop); + expect(steerInjections()).toHaveLength(2); + }); + + it('does not arm the steer reminder for tool-injected steers', async () => { + const { prompt, context, loop, reminderProviders } = harness(); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + await prompt.inject({ role: 'user', content: [{ type: 'text', text: 'tool delivery' }], toolCalls: [] }); + loop.drainNextBatch(context); + expect(await reminderProviders.get('steer')!(injectionContext())).toBeUndefined(); + }); + + it('drops an armed steer reminder when the turn settles before the next step', async () => { + const { prompt, context, loop, reminderProviders } = harness({ manualTurnResult: true }); + const active = await prompt.enqueue({ message: message('active') }); + await active.launched; + const queued = await prompt.enqueue({ message: message('late') }); + await prompt.steer([queued.id]); + loop.drainNextBatch(context); + loop.settleActive({ type: 'cancelled', steps: 1, reason: new Error('stop') }); + await active.completion; + expect(await reminderProviders.get('steer')!(injectionContext())).toBeUndefined(); + }); + it('aborts pending prompts and settles completion', async () => { const { prompt, eventBus } = harness(); const aborted: PromptAborted[] = [];