diff --git a/.changeset/goal-steer-queued-messages.md b/.changeset/goal-steer-queued-messages.md new file mode 100644 index 0000000000..ddb5b5d849 --- /dev/null +++ b/.changeset/goal-steer-queued-messages.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix messages sent while a goal is running being rejected with a "Cannot launch a new turn while another turn is active" error; they are now steered into the active goal turn instead of being dropped. diff --git a/apps/kimi-code/src/tui/kimi-tui.ts b/apps/kimi-code/src/tui/kimi-tui.ts index 14951c6ed8..f0764c82b9 100644 --- a/apps/kimi-code/src/tui/kimi-tui.ts +++ b/apps/kimi-code/src/tui/kimi-tui.ts @@ -1309,6 +1309,22 @@ export class KimiTUI { this.beginSessionRequest(); const sdkInput = options?.parts ?? input; + // While a goal is being pursued the engine holds its active turn across the + // whole continuation loop, so a fresh prompt races the goal driver at every + // continuation boundary and is rejected with `turn.agent_busy`, dropping + // the message. Steer instead: the engine buffers it into the running goal + // turn, or launches a turn of its own if the loop just ended. + if (this.state.appState.goal?.status === 'active') { + void session.steer(sdkInput).catch((error: unknown) => { + const message = formatErrorMessage(error); + // Same reset as the prompt path: beginSessionRequest already moved the + // TUI to the waiting phase, and no turn events may follow a failed + // steer (e.g. the session is gone), which would leave the UI stuck + // queueing input behind a request that never completes. + this.failSessionRequest(`Failed to steer: ${message}`); + }); + return; + } void session.prompt(sdkInput).catch((error: unknown) => { const message = formatErrorMessage(error); this.failSessionRequest(`Failed to send: ${message}`); diff --git a/apps/kimi-code/test/tui/kimi-tui-message-flow.test.ts b/apps/kimi-code/test/tui/kimi-tui-message-flow.test.ts index 806bb43095..7918d998f5 100644 --- a/apps/kimi-code/test/tui/kimi-tui-message-flow.test.ts +++ b/apps/kimi-code/test/tui/kimi-tui-message-flow.test.ts @@ -8,7 +8,12 @@ import { resetCapabilitiesCache, setCapabilities, } from '@moonshot-ai/pi-tui'; -import type { ApprovalRequest, ApprovalResponse, Event } from '@moonshot-ai/kimi-code-sdk'; +import type { + ApprovalRequest, + ApprovalResponse, + Event, + GoalSnapshot, +} from '@moonshot-ai/kimi-code-sdk'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { ApprovalPanelComponent } from '#/tui/components/dialogs/approval-panel'; @@ -311,6 +316,29 @@ async function makeDriver( return { driver, session, harness }; } +function makeActiveGoalSnapshot(): GoalSnapshot { + return { + goalId: 'g1', + objective: 'Ship the feature', + status: 'active', + turnsUsed: 3, + tokensUsed: 100, + wallClockMs: 1000, + budget: { + tokenBudget: null, + turnBudget: null, + wallClockBudgetMs: null, + remainingTokens: null, + remainingTurns: null, + remainingWallClockMs: null, + tokenBudgetReached: false, + turnBudgetReached: false, + wallClockBudgetReached: false, + overBudget: false, + }, + }; +} + function renderTranscript(driver: MessageDriver): string { return driver.state.transcriptContainer.render(120).join('\n'); } @@ -1645,6 +1673,91 @@ command = "vim" expect(harness.track).toHaveBeenCalledWith('input_queue', undefined); }); + it('steers fresh input while a goal is active even when the streaming phase is idle', async () => { + const { driver, session } = await makeDriver(); + driver.state.appState.goal = makeActiveGoalSnapshot(); + + driver.handleUserInput('hello mid-goal'); + + expect(session.steer).toHaveBeenCalledWith('hello mid-goal'); + expect(session.prompt).not.toHaveBeenCalled(); + expect(driver.state.transcriptEntries).toEqual([ + expect.objectContaining({ + kind: 'user', + content: 'hello mid-goal', + }), + ]); + }); + + it('resets the streaming phase when steering mid-goal input fails', async () => { + const session = makeSession({ + steer: vi.fn(async () => { + throw new Error('session closed'); + }), + }); + const { driver } = await makeDriver(session); + driver.state.appState.goal = makeActiveGoalSnapshot(); + + driver.handleUserInput('hello mid-goal'); + + expect(driver.state.appState.streamingPhase).toBe('waiting'); + await vi.waitFor(() => { + expect(driver.state.appState.streamingPhase).toBe('idle'); + }); + expect(stripSgr(renderTranscript(driver))).toContain('Failed to steer: session closed'); + }); + + it('steers a queued message at a turn boundary while a goal is active', async () => { + const { driver, session } = await makeDriver(); + driver.state.appState.goal = makeActiveGoalSnapshot(); + driver.state.appState.streamingPhase = 'waiting'; + driver.handleUserInput('mid-goal note'); + expect(driver.state.queuedMessages).toEqual([{ text: 'mid-goal note', agentId: 'main' }]); + + driver.sessionEventHandler.handleEvent( + { + type: 'turn.ended', + agentId: 'main', + sessionId: 'ses-1', + turnId: 0, + reason: 'completed', + } as Event, + (item) => { + driver.sendQueuedMessage(session, item); + }, + ); + + await vi.waitFor(() => { + expect(session.steer).toHaveBeenCalledWith('mid-goal note'); + }); + expect(session.prompt).not.toHaveBeenCalled(); + expect(driver.state.queuedMessages).toEqual([]); + }); + + it('prompts the queued message as a new turn when no goal is active', async () => { + const { driver, session } = await makeDriver(); + driver.state.appState.streamingPhase = 'waiting'; + driver.handleUserInput('after the turn'); + + driver.sessionEventHandler.handleEvent( + { + type: 'turn.ended', + agentId: 'main', + sessionId: 'ses-1', + turnId: 0, + reason: 'completed', + } as Event, + (item) => { + driver.sendQueuedMessage(session, item); + }, + ); + + await vi.waitFor(() => { + expect(session.prompt).toHaveBeenCalledWith('after the turn'); + }); + expect(session.steer).not.toHaveBeenCalled(); + }); + it('cancels active streaming from Escape and Ctrl-C editor shortcuts', async () => { const { driver, session } = await makeDriver(); diff --git a/packages/agent-core/src/session/rpc.ts b/packages/agent-core/src/session/rpc.ts index a2e6f0c1e9..d12b44b6ec 100644 --- a/packages/agent-core/src/session/rpc.ts +++ b/packages/agent-core/src/session/rpc.ts @@ -128,6 +128,12 @@ export class SessionAPIImpl implements PromisableMethods { } async steer({ agentId, ...payload }: AgentScopedPayload) { + if (agentId === 'main') { + // A steer is user input like a prompt — and can even launch the + // session's first turn (e.g. goal mode) — so keep title/lastPrompt in + // sync the same way. + await this.updatePromptMetadata(promptMetadataTextFromPayload(payload)); + } return (await this.getAgent(agentId)).steer(payload); } diff --git a/packages/agent-core/test/session/prompt-metadata.test.ts b/packages/agent-core/test/session/prompt-metadata.test.ts index 248303d913..4af1ad01a6 100644 --- a/packages/agent-core/test/session/prompt-metadata.test.ts +++ b/packages/agent-core/test/session/prompt-metadata.test.ts @@ -7,12 +7,26 @@ * - an inline image-compression caption (harness metadata placed next to * the image by prompt ingestion) never leaks into titles/lastPrompt, * whether it is a standalone text part or merged into the user's text + * - SessionAPIImpl.steer updates title/lastPrompt exactly like prompt — + * a steer can launch the session's first turn (e.g. goal mode) */ -import { describe, expect, it } from 'vitest'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'pathe'; +import type { ProviderConfig } from '@moonshot-ai/kosong'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import type { ResolvedAgentProfile } from '../../src/profile'; +import type { SDKSessionRPC } from '../../src/rpc'; +import { Session } from '../../src/session'; import { promptMetadataTextFromPayload } from '../../src/session/prompt-metadata'; +import { ProviderManager } from '../../src/session/provider-manager'; +import { SessionAPIImpl } from '../../src/session/rpc'; import { buildImageCompressionCaption } from '../../src/tools/support/image-compress'; +import { createScriptedGenerate } from '../agent/harness/scripted-generate'; +import { testKaos } from '../fixtures/test-kaos'; const CAPTION = buildImageCompressionCaption({ original: { width: 3264, height: 666, byteLength: 344 * 1024, mimeType: 'image/png' }, @@ -53,3 +67,104 @@ describe('promptMetadataTextFromPayload', () => { expect(text).not.toContain('Image compressed'); }); }); + +describe('SessionAPIImpl prompt metadata', () => { + it('derives title and lastPrompt from a steer the same way as a prompt', async () => { + const sessionDir = await makeTempDir(); + const events: Array> = []; + const scripted = createScriptedGenerate(); + const session = track( + new Session({ + id: 'prompt-metadata-steer', + kaos: testKaos.withCwd(sessionDir), + homedir: sessionDir, + rpc: createSessionRpc(events), + skills: { explicitDirs: [join(sessionDir, 'missing-skills')] }, + providerManager: testProviderManager(), + }), + ); + const { agent } = await session.createAgent( + { type: 'main', generate: scripted.generate }, + { profile: testProfile() }, + ); + agent.config.update({ modelAlias: MOCK_PROVIDER.model, thinkingEffort: 'off' }); + agent.permission.setMode('yolo'); + + const api = new SessionAPIImpl(session); + await api.steer({ agentId: 'main', input: [{ type: 'text', text: 'steered goal objective' }] }); + if (agent.turn.hasActiveTurn) { + await agent.turn.waitForCurrentTurn(); + } + + expect(session.metadata.title).toBe('steered goal objective'); + expect(session.metadata.lastPrompt).toBe('steered goal objective'); + }); +}); + +const MOCK_PROVIDER = { + type: 'kimi', + apiKey: 'test-key', + model: 'mock-model', +} as const satisfies ProviderConfig; + +const tempDirs: string[] = []; +const openSessions: Session[] = []; + +function track(session: Session): Session { + openSessions.push(session); + return session; +} + +afterEach(async () => { + // Close sessions first so their async metadata/wire writes settle before the + // temp dirs are removed (otherwise rm races with a write -> ENOTEMPTY). + await Promise.allSettled(openSessions.splice(0).map((s) => s.close())); + for (const dir of tempDirs.splice(0)) { + await rm(dir, { recursive: true, force: true }); + } +}); + +async function makeTempDir(): Promise { + const dir = await mkdtemp(join(tmpdir(), 'kimi-prompt-metadata-')); + tempDirs.push(dir); + return dir; +} + +function testProviderManager(): ProviderManager { + return new ProviderManager({ + config: { + providers: { + test: { type: MOCK_PROVIDER.type, apiKey: MOCK_PROVIDER.apiKey }, + }, + models: { + [MOCK_PROVIDER.model]: { + provider: 'test', + model: MOCK_PROVIDER.model, + maxContextSize: 1_000_000, + }, + }, + }, + }); +} + +function testProfile(): ResolvedAgentProfile { + return { + name: 'test', + systemPrompt: () => '', + tools: [], + }; +} + +function createSessionRpc(events: Array>): SDKSessionRPC { + return { + emitEvent: vi.fn(async (event) => { + events.push(event); + }), + requestApproval: vi.fn(async () => ({ decision: 'cancelled' })), + requestQuestion: vi.fn(async () => null), + toolCall: vi.fn(async () => ({ + output: 'custom tools are not supported in this test', + isError: true, + })), + } as unknown as SDKSessionRPC; +}