diff --git a/.changeset/durable-turn-input-cancellation.md b/.changeset/durable-turn-input-cancellation.md new file mode 100644 index 0000000000..14ef6b443f --- /dev/null +++ b/.changeset/durable-turn-input-cancellation.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Preserve queued prompts and unfinished turn context across interruptions and restarts, and stop in-flight model streams promptly when a turn is cancelled. diff --git a/.changeset/preserve-interrupted-turn-context.md b/.changeset/preserve-interrupted-turn-context.md new file mode 100644 index 0000000000..d1d91e0f52 --- /dev/null +++ b/.changeset/preserve-interrupted-turn-context.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Tell the model when a previous turn failed or was interrupted so follow-up prompts retain the unfinished request context. diff --git a/apps/kimi-code/src/tui/components/dialogs/compaction.ts b/apps/kimi-code/src/tui/components/dialogs/compaction.ts index 9ade9350c7..b00f67ad46 100644 --- a/apps/kimi-code/src/tui/components/dialogs/compaction.ts +++ b/apps/kimi-code/src/tui/components/dialogs/compaction.ts @@ -8,6 +8,8 @@ * "Compaction complete (X → Y tokens)" * - `markCanceled()` on `compaction.cancelled` → solid warning bullet + * "Compaction cancelled" + * - `markFailed()` when the compaction error event arrives without a + * terminal event → solid error bullet + "Compaction failed" * * Bullet animation mirrors `ToolCallComponent` (500ms blink) so the user * reads the same "work in progress" signal across the UI. @@ -31,6 +33,7 @@ export class CompactionComponent extends Container { private blinkTimer: ReturnType | null = null; private done = false; private canceled = false; + private failed = false; private tokensBefore: number | undefined; private tokensAfter: number | undefined; private summary: string | undefined; @@ -87,7 +90,7 @@ export class CompactionComponent extends Container { } markDone(tokensBefore?: number, tokensAfter?: number, summary?: string): void { - if (this.done || this.canceled) return; + if (this.done || this.canceled || this.failed) return; this.done = true; this.tokensBefore = tokensBefore; this.tokensAfter = tokensAfter; @@ -101,13 +104,21 @@ export class CompactionComponent extends Container { } markCanceled(): void { - if (this.done || this.canceled) return; + if (this.done || this.canceled || this.failed) return; this.canceled = true; this.stopBlink(); this.headerText.setText(this.buildHeader()); this.ui?.requestRender(); } + markFailed(): void { + if (this.done || this.canceled || this.failed) return; + this.failed = true; + this.stopBlink(); + this.headerText.setText(this.buildHeader()); + this.ui?.requestRender(); + } + setExpanded(expanded: boolean): void { if (this.expanded === expanded) return; this.expanded = expanded; @@ -164,6 +175,11 @@ export class CompactionComponent extends Container { const label = currentTheme.boldFg('warning', 'Compaction cancelled'); return `${bullet}${label}`; } + if (this.failed) { + const bullet = currentTheme.fg('error', STATUS_BULLET); + const label = currentTheme.boldFg('error', 'Compaction failed'); + return `${bullet}${label}`; + } const bullet = this.blinkOn ? currentTheme.fg('text', STATUS_BULLET) : ' '; const label = currentTheme.boldFg('primary', 'Compacting context...'); const tip = this.tip ? currentTheme.fg('textDim', ` · Tip: ${this.tip}`) : ''; diff --git a/apps/kimi-code/src/tui/controllers/session-event-handler.ts b/apps/kimi-code/src/tui/controllers/session-event-handler.ts index 82257c86cf..c17c2ef04d 100644 --- a/apps/kimi-code/src/tui/controllers/session-event-handler.ts +++ b/apps/kimi-code/src/tui/controllers/session-event-handler.ts @@ -260,7 +260,7 @@ export class SessionEventHandler { case 'goal.updated': this.handleGoalUpdated(event); break; case 'skill.activated': this.handleSkillActivated(event); break; case 'plugin_command.activated': this.handlePluginCommandActivated(event); break; - case 'error': this.handleSessionError(event); break; + case 'error': this.handleSessionError(event, sendQueued); break; case 'warning': this.handleSessionWarning(event); break; case 'compaction.started': this.handleCompactionBegin(event); break; case 'compaction.completed': this.handleCompactionEnd(event, sendQueued); break; @@ -851,10 +851,24 @@ export class SessionEventHandler { } } - private handleSessionError(event: ErrorEvent): void { + private handleSessionError( + event: ErrorEvent, + sendQueued: (item: QueuedMessage) => void, + ): void { this.host.streamingUI.flushNow(); this.host.streamingUI.resetToolUi(); this.host.streamingUI.finalizeLiveTextBuffers('idle'); + if ( + this.host.state.appState.isCompacting && + (event.code.startsWith('compaction.') || + event.code === 'auth.login_required' || + event.code === 'provider.auth_error') + ) { + // Pre-commit compaction failures have no completed/cancelled terminal: + // the error event itself is the terminal signal for the UI. + this.host.streamingUI.failCompaction(); + this.finishCompaction(sendQueued); + } if (event.code === OAUTH_LOGIN_REQUIRED_CODE) { this.host.showError(OAUTH_LOGIN_REQUIRED_STARTUP_NOTICE); return; @@ -1012,7 +1026,13 @@ export class SessionEventHandler { private finishCompaction(sendQueued: (item: QueuedMessage) => void): void { const hasActiveTurn = this.host.streamingUI.hasActiveTurn(); if (!hasActiveTurn) { - const next = this.host.shiftQueuedMessage(); + // Auto-compaction failures terminate the owning turn first and emit the + // compaction error second. finalizeTurn may therefore have already + // removed and scheduled one queued message; never drain a second item for + // the same terminal sequence. + const next = this.host.state.queuedMessageDispatchPending + ? undefined + : this.host.shiftQueuedMessage(); if (next !== undefined) { this.host.state.queuedMessageDispatchPending = true; } diff --git a/apps/kimi-code/src/tui/controllers/session-replay.ts b/apps/kimi-code/src/tui/controllers/session-replay.ts index 00ee5a64b5..245ce101c5 100644 --- a/apps/kimi-code/src/tui/controllers/session-replay.ts +++ b/apps/kimi-code/src/tui/controllers/session-replay.ts @@ -8,6 +8,7 @@ import type { Session, ToolCall, } from '@moonshot-ai/kimi-code-sdk'; +import { extractImageCompressionCaptions } from '@moonshot-ai/kimi-code-sdk'; import { ToolCallComponent } from '../components/messages/tool-call'; import { currentTheme } from '../theme'; @@ -334,8 +335,16 @@ export class SessionReplayRenderer { } this.advanceTurn(context); + const content = + message.origin?.kind === 'user' + ? message.content.map((part) => + part.type === 'text' + ? { ...part, text: extractImageCompressionCaptions(part.text).text } + : part, + ) + : message.content; this.host.appendTranscriptEntry( - replayEntry(context, 'user', contentPartsToText(message.content), 'plain'), + replayEntry(context, 'user', contentPartsToText(content), 'plain'), ); } diff --git a/apps/kimi-code/src/tui/controllers/streaming-ui.ts b/apps/kimi-code/src/tui/controllers/streaming-ui.ts index ae5eac7681..122921e059 100644 --- a/apps/kimi-code/src/tui/controllers/streaming-ui.ts +++ b/apps/kimi-code/src/tui/controllers/streaming-ui.ts @@ -752,6 +752,14 @@ export class StreamingUIController { this.host.state.ui.requestRender(); } + failCompaction(): void { + const block = this._activeCompactionBlock; + if (block === undefined) return; + block.markFailed(); + this._activeCompactionBlock = undefined; + this.host.state.ui.requestRender(); + } + // --------------------------------------------------------------------------- // Tool call grouping // --------------------------------------------------------------------------- diff --git a/apps/kimi-code/test/tui/components/dialogs/compaction.test.ts b/apps/kimi-code/test/tui/components/dialogs/compaction.test.ts index 4f415bc322..ffcd285a76 100644 --- a/apps/kimi-code/test/tui/components/dialogs/compaction.test.ts +++ b/apps/kimi-code/test/tui/components/dialogs/compaction.test.ts @@ -56,6 +56,20 @@ describe('CompactionComponent', () => { } }); + it('stops the progress animation and renders a failed terminal', () => { + const component = new CompactionComponent(); + + try { + component.markFailed(); + + const output = component.render(80).map(strip).join('\n'); + expect(output).toContain('Compaction failed'); + expect(output).not.toContain('Compacting context'); + } finally { + component.dispose(); + } + }); + it('renders a cancelled terminal state', () => { const component = new CompactionComponent(); 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 f285be77dd..0b654d465e 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 @@ -3235,6 +3235,101 @@ command = "vim" expect(transcript).not.toContain('kimi export'); }); + it('treats a compaction error as a failed terminal and leaves compacting state', async () => { + const { driver } = await makeDriver(); + const sendQueued = vi.fn(); + + driver.sessionEventHandler.handleEvent( + { + type: 'compaction.started', + agentId: 'main', + sessionId: 'ses-1', + } as Event, + sendQueued, + ); + expect(driver.state.appState.isCompacting).toBe(true); + + driver.sessionEventHandler.handleEvent( + { + type: 'error', + agentId: 'main', + sessionId: 'ses-1', + code: 'compaction.failed', + message: 'provider aborted without a cancellation signal', + retryable: false, + } as Event, + sendQueued, + ); + + expect(driver.state.appState.isCompacting).toBe(false); + expect(driver.state.appState.streamingPhase).toBe('idle'); + const transcript = stripSgr(renderTranscript(driver)); + expect(transcript).toContain('Compaction failed'); + expect(transcript).not.toContain('Compacting context'); + }); + + it('does not drain a second queued message when turn failure precedes compaction error', async () => { + vi.useFakeTimers(); + try { + const { driver } = await makeDriver(); + const sendQueued = vi.fn(); + driver.sessionEventHandler.handleEvent( + { + type: 'turn.started', + agentId: 'main', + sessionId: 'ses-1', + turnId: 1, + origin: { kind: 'user' }, + } as Event, + sendQueued, + ); + driver.sessionEventHandler.handleEvent( + { + type: 'compaction.started', + agentId: 'main', + sessionId: 'ses-1', + turnId: 1, + trigger: 'auto', + } as Event, + sendQueued, + ); + driver.state.queuedMessages = [{ text: 'first' }, { text: 'second' }]; + + driver.sessionEventHandler.handleEvent( + { + type: 'turn.ended', + agentId: 'main', + sessionId: 'ses-1', + turnId: 1, + reason: 'failed', + durationMs: 1, + error: { code: 'compaction.failed', message: 'summary failed', retryable: false }, + } as Event, + sendQueued, + ); + driver.sessionEventHandler.handleEvent( + { + type: 'error', + agentId: 'main', + sessionId: 'ses-1', + code: 'compaction.failed', + message: 'summary failed', + retryable: false, + } as Event, + sendQueued, + ); + + expect(driver.state.queuedMessages).toEqual([{ text: 'second' }]); + expect(driver.state.queuedMessageDispatchPending).toBe(true); + await vi.runAllTimersAsync(); + expect(sendQueued).toHaveBeenCalledTimes(1); + expect(sendQueued).toHaveBeenCalledWith({ text: 'first' }); + expect(driver.state.queuedMessages).toEqual([{ text: 'second' }]); + } finally { + vi.useRealTimers(); + } + }); + it('shows concise provider filter text for filtered session errors', async () => { const { driver } = await makeDriver(); const verboseMessage = diff --git a/apps/kimi-code/test/tui/message-replay.test.ts b/apps/kimi-code/test/tui/message-replay.test.ts index 5e4e11670f..95f84246f5 100644 --- a/apps/kimi-code/test/tui/message-replay.test.ts +++ b/apps/kimi-code/test/tui/message-replay.test.ts @@ -4,6 +4,7 @@ import type { AgentReplayRecord, BackgroundTaskInfo, ContentPart, + Event, GoalSnapshot, PromptOrigin, ResumedAgentState, @@ -289,6 +290,135 @@ function backgroundTask( } describe('KimiTUI resume message replay', () => { + it('does not render turn outcome reminders between visible user messages', async () => { + const driver = await replayIntoDriver([ + message('user', [{ type: 'text', text: 'Finish the requested change.' }]), + message( + 'user', + [ + { + type: 'text', + text: + '\n' + + 'The previous turn ended before producing a final response.\n\n' + + 'Error: API request failed with HTTP 500.\n\n' + + 'The preceding user request may still be unfinished. Treat the next user message as a follow-up.\n' + + '', + }, + ], + { origin: { kind: 'injection', variant: 'turn_outcome' } }, + ), + message('user', [{ type: 'text', text: 'Continue.' }]), + ]); + + expect( + driver.state.transcriptEntries + .filter((entry) => entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual(['Finish the requested change.', 'Continue.']); + const transcript = stripAnsi(driver.state.transcriptContainer.render(140).join('\n')); + expect(transcript).not.toContain('API request failed with HTTP 500'); + expect(transcript).not.toContain('preceding user request'); + }); + + it('renders ordinary user text that contains system reminder tags', async () => { + const literalText = + '\nThis text was typed by the user and is not an injection.\n'; + const driver = await replayIntoDriver([ + message('user', [{ type: 'text', text: literalText }], { + origin: { kind: 'user' }, + }), + ]); + + expect( + driver.state.transcriptEntries + .filter((entry) => entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual([literalText]); + const transcript = stripAnsi(driver.state.transcriptContainer.render(140).join('\n')); + expect(transcript).toContain('This text was typed by the user and is not an injection.'); + }); + + it('renders one synthetic tool failure without exposing its turn outcome reminder', async () => { + const abandonedOutput = + 'Tool call did not complete: the turn failed before its result was recorded. Do not assume the tool completed successfully.'; + const driver = await replayIntoDriver([ + message('user', [{ type: 'text', text: 'Run the operation.' }]), + message( + 'assistant', + [{ type: 'text', text: 'Starting the operation.' }], + { toolCalls: [toolCall('call_abandoned', 'Run', {})] }, + ), + message('tool', [{ type: 'text', text: abandonedOutput }], { + toolCallId: 'call_abandoned', + isError: true, + }), + message( + 'user', + [ + { + type: 'text', + text: + '\n' + + 'The previous turn ended before producing a final response.\n\n' + + 'Error: tool result dispatch failed\n\n' + + 'The preceding user request may still be unfinished. Treat the next user message as a follow-up.\n' + + '', + }, + ], + { origin: { kind: 'injection', variant: 'turn_outcome' } }, + ), + message('user', [{ type: 'text', text: 'Continue.' }]), + ]); + + const transcript = stripAnsi(driver.state.transcriptContainer.render(140).join('\n')); + expect(transcript.split('Tool call did not complete').length - 1).toBe(1); + expect(transcript).not.toContain('tool result dispatch failed'); + expect(transcript).not.toContain('preceding user request'); + expect( + driver.state.transcriptEntries + .filter((entry) => entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual(['Run the operation.', 'Continue.']); + }); + + it('does not render live context splice events containing turn outcome reminders', async () => { + const driver = await replayIntoDriver([]); + + // The v2 server forwards this domain event on the v1 session channel even + // though it is intentionally absent from the SDK's public Event union. + driver.sessionEventHandler.handleEvent( + { + type: 'context.spliced', + sessionId: 'ses-replay', + agentId: 'main', + start: 1, + deleteCount: 0, + messages: [ + { + role: 'user', + content: [ + { + type: 'text', + text: + '\n' + + 'The user interrupted the previous turn before it finished.\n' + + '', + }, + ], + toolCalls: [], + origin: { kind: 'injection', variant: 'turn_outcome' }, + }, + ], + } as unknown as Event, + () => {}, + ); + + expect(driver.state.transcriptEntries).toEqual([]); + const transcript = stripAnsi(driver.state.transcriptContainer.render(140).join('\n')); + expect(transcript).not.toContain('user interrupted the previous turn'); + }); + it('does not render legacy goal completion context reminders as transcript messages', async () => { const driver = await replayIntoDriver([ message( @@ -613,6 +743,30 @@ describe('KimiTUI resume message replay', () => { expect(driver.streamingUI.getToolComponent('call_agent_2')).toBeUndefined(); }); + it('hides legacy image-compression metadata when replaying a user message', async () => { + const legacyCaption = + 'Image compressed to fit model limits: original 3264x666 image/png (344 KB) -> ' + + '2000x408 image/png (282 KB). Full-detail original: /tmp/originals/shot.png.'; + const driver = await replayIntoDriver([ + message( + 'user', + [ + { + type: 'text', + text: `Inspect this${legacyCaption}ordinary user text`, + }, + ], + { origin: { kind: 'user' } }, + ), + ]); + + expect( + driver.state.transcriptEntries + .filter((entry) => entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual(['Inspect thisordinary user text']); + }); + it('groups replayed Read calls from one assistant message using live grouping', async () => { const replay: AgentReplayRecord[] = [ message('user', [{ type: 'text', text: 'read files' }]), diff --git a/apps/kimi-web/src/composables/client/useWorkspaceState.ts b/apps/kimi-web/src/composables/client/useWorkspaceState.ts index fa9891bf22..72ada9cc31 100644 --- a/apps/kimi-web/src/composables/client/useWorkspaceState.ts +++ b/apps/kimi-web/src/composables/client/useWorkspaceState.ts @@ -55,6 +55,7 @@ const MESSAGES_PAGE_SIZE = 50; // sidebar can fall back to it when a workspace's first-page size is unknown. export const SESSIONS_INITIAL_PAGE_SIZE = 5; const PROMPT_NOT_FOUND_CODE = 40402; +const PROMPT_ALREADY_COMPLETED_CODE = 40903; const WORKSPACE_NOT_FOUND_CODE = 40410; // Shared "already resolved" conflict (40902). The daemon reuses it for both // approvals and questions when a second client races the resolve, so a @@ -118,6 +119,48 @@ const pendingLocalTurnStarts = new Map>(); const afterLocalTurnsSettled = new Map void>(); let nextLocalTurnToken = 0; +/** + * A queued prompt becomes part of the active turn only after the steer request + * is admitted. Keep that short transaction visible to Stop: once the user asks + * to stop the session, a concurrent steer failure is the expected result of + * cancelling that transaction, not a second operation error to surface. + * + * There can be more than one browser request in flight for a session (for + * example, a double Ctrl+S before the first response), so track operation + * objects rather than one prompt id per session. The daemon remains the source + * of truth for queue ownership; this set records only local cancellation + * intent and is always cleared by steerPrompt's finally block. + */ +interface PendingSteerOperation { + stopRequested: boolean; +} + +const pendingSteersBySession = new Map>(); + +function registerPendingSteer(sessionId: string): PendingSteerOperation { + const operation: PendingSteerOperation = { stopRequested: false }; + const pending = pendingSteersBySession.get(sessionId) ?? new Set(); + pending.add(operation); + pendingSteersBySession.set(sessionId, pending); + return operation; +} + +function finishPendingSteer(sessionId: string, operation: PendingSteerOperation): void { + const pending = pendingSteersBySession.get(sessionId); + if (pending === undefined) return; + pending.delete(operation); + if (pending.size === 0) pendingSteersBySession.delete(sessionId); +} + +function markPendingSteersStopping(sessionId: string): boolean { + let marked = false; + for (const operation of pendingSteersBySession.get(sessionId) ?? []) { + operation.stopRequested = true; + marked = true; + } + return marked; +} + export interface LocalTurnStartState { generation: number; pending: boolean; @@ -1610,6 +1653,11 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta }; updateSessionMessages(sid, (msgs) => [...msgs, optimisticMsg]); + // Register before submitting B. prompt.submitted may reach the browser and + // replace promptIdBySession with B before the submit HTTP response returns; + // Stop must still know it is cancelling a pending A+B steer transaction in + // that window. + const pendingSteer = registerPendingSteer(sid); const localTurnToken = beginLocalTurn(sid); try { const api = getKimiWebApi(); @@ -1640,6 +1688,25 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta return updated; }); + if (pendingSteer.stopRequested) { + // Stop may have won before submit returned, when B was only queued and + // aborting A alone could let B auto-start. Pair the session-level stop + // issued by abortCurrentPrompt() with an idempotent abort(B) once its + // authoritative id is known. Never attempt the steer after Stop. + try { + await api.abortPrompt(sid, result.promptId); + } catch (error) { + const alreadyGone = + isDaemonApiError(error) && + (error.code === PROMPT_NOT_FOUND_CODE || + error.code === PROMPT_ALREADY_COMPLETED_CODE); + if (!alreadyGone) { + pushOperationFailure('abortCurrentPrompt', error, { sessionId: sid }); + } + } + return; + } + if (result.status !== 'queued') { // The turn ended while the user was typing — the prompt started a turn // of its own. Wire it up like a regular send so :abort keeps working. @@ -1650,16 +1717,29 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta try { await api.steerPrompts(sid, [result.promptId]); - } catch { - // The active turn finished between submit and steer — the daemon starts - // the parked prompt as its own turn. Nothing to roll back. + } catch (error) { + // PROMPT_NOT_FOUND is the one server response that proves the original + // active-turn target disappeared before steer admission. A concurrent + // local Stop is also expected to reject the steer request: aborting the + // reserved queued prompt linearizes as cancelling both it and the + // active prompt. Every other failure leaves the queued prompt's fate + // unknown (it may still be parked), so do not claim it auto-started — + // surface the failure while retaining its accepted transcript echo. + const activeTurnEnded = + isDaemonApiError(error) && error.code === PROMPT_NOT_FOUND_CODE; + if (!activeTurnEnded && !pendingSteer.stopRequested) { + pushOperationFailure('steer', error, { sessionId: sid }); + } } - } catch (err) { + } catch (error) { // Submit failed: drop the optimistic echo so the transcript doesn't show // a delivered-looking message the daemon never received. updateSessionMessages(sid, (msgs) => msgs.filter((m) => m.id !== tempId)); - pushOperationFailure('steer', err, { sessionId: sid }); + if (!pendingSteer.stopRequested) { + pushOperationFailure('steer', error, { sessionId: sid }); + } } finally { + finishPendingSteer(sid, pendingSteer); settleLocalTurn(sid, localTurnToken); } } @@ -1761,6 +1841,12 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta async function abortCurrentPrompt(): Promise { const sid = rawState.activeSessionId; if (!sid) return; + // Mark before the first await. The authoritative prompt id may still name + // active A or may already have been overwritten by queued B's + // prompt.submitted event; stopping either side of an admitted steer + // transaction cancels A+B, and the concurrent steer request is then + // expected to fail. + const stoppingSteer = markPendingSteersStopping(sid); const session = rawState.sessions.find((s) => s.id === sid); // 1. Authoritative id captured at submit time. @@ -1777,6 +1863,16 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta } const api = getKimiWebApi(); + // During a local steer, per-prompt abort and session abort are both needed + // to close the pre-admission window: queued B may already be cached while + // the server has not reserved it against active A yet. Start the session + // abort concurrently so a slow per-prompt request cannot delay Stop. + const steerSessionAbort = stoppingSteer + ? api.abortSession(sid).then( + () => ({ ok: true as const }), + (error: unknown) => ({ ok: false as const, error }), + ) + : undefined; // 3. If we have a real id, try the per-prompt abort first. If the daemon // reports the prompt is missing/already completed, clear the stale id and @@ -1784,29 +1880,37 @@ export function useWorkspaceState(rawState: ExtendedState, deps: UseWorkspaceSta if (promptId !== undefined) { try { const result = await api.abortPrompt(sid, promptId); - if (result.aborted) return; + if (result.aborted && !stoppingSteer) return; const nextPromptIds = { ...rawState.promptIdBySession }; delete nextPromptIds[sid]; rawState.promptIdBySession = nextPromptIds; - } catch (err) { - if (isDaemonApiError(err) && err.code === PROMPT_NOT_FOUND_CODE) { + } catch (error) { + if (isDaemonApiError(error) && error.code === PROMPT_NOT_FOUND_CODE) { // Stale id — try the session-level fallback below. const nextPromptIds = { ...rawState.promptIdBySession }; delete nextPromptIds[sid]; rawState.promptIdBySession = nextPromptIds; } else { - pushOperationFailure('abortCurrentPrompt', err, { sessionId: sid }); - return; + pushOperationFailure('abortCurrentPrompt', error, { sessionId: sid }); + if (!stoppingSteer) return; } } } + if (steerSessionAbort !== undefined) { + const outcome = await steerSessionAbort; + if (!outcome.ok) { + pushOperationFailure('abortCurrentPrompt', outcome.error, { sessionId: sid }); + } + return; + } + // 4. No real id, or the prompt id is no longer recognized: cancel whatever // is running in the session (including skill activations). try { await api.abortSession(sid); - } catch (err) { - pushOperationFailure('abortCurrentPrompt', err, { sessionId: sid }); + } catch (error) { + pushOperationFailure('abortCurrentPrompt', error, { sessionId: sid }); } } diff --git a/apps/kimi-web/test/workspace-state.test.ts b/apps/kimi-web/test/workspace-state.test.ts index bc882d0bac..4f66834524 100644 --- a/apps/kimi-web/test/workspace-state.test.ts +++ b/apps/kimi-web/test/workspace-state.test.ts @@ -23,6 +23,7 @@ const apiMock = vi.hoisted(() => ({ exportSession: vi.fn(), updateSession: vi.fn(), submitPrompt: vi.fn(), + steerPrompts: vi.fn(), respondQuestion: vi.fn(), respondApproval: vi.fn(), dismissQuestion: vi.fn(), @@ -289,6 +290,128 @@ describe('useWorkspaceState — abortCurrentPrompt', () => { }); }); +describe('useWorkspaceState — steer and Stop transaction', () => { + beforeEach(() => { + apiMock.submitPrompt.mockReset(); + apiMock.steerPrompts.mockReset(); + apiMock.abortPrompt.mockReset(); + apiMock.abortSession.mockReset(); + }); + + it('stops queued B without reporting the steer rejection caused by cancelling A+B', async () => { + let rejectSteer!: (reason: unknown) => void; + let resolvePromptAbort!: (result: { aborted: boolean }) => void; + apiMock.submitPrompt.mockResolvedValue({ + promptId: 'prompt_b', + userMessageId: 'message_b', + status: 'queued', + }); + apiMock.steerPrompts.mockReturnValue( + new Promise((_, reject) => { + rejectSteer = reject; + }), + ); + const steerCancelled = new DaemonApiError({ + code: 50000, + msg: 'steer cancelled with its active turn', + requestId: 'req_steer', + }); + apiMock.abortPrompt.mockImplementation(() => { + rejectSteer(steerCancelled); + return new Promise((resolve) => { + resolvePromptAbort = resolve; + }); + }); + apiMock.abortSession.mockResolvedValue({ aborted: true }); + const state = createState(); + state.promptIdBySession = { sess_1: 'prompt_a' }; + const deps = createDeps(); + const workspace = useWorkspaceState(state, deps); + + const steering = workspace.steerPrompt('follow up B'); + await vi.waitFor(() => { + expect(apiMock.steerPrompts).toHaveBeenCalledWith('sess_1', ['prompt_b']); + }); + + // The prompt.submitted WS event for queued B is allowed to replace A's + // cached id. With the daemon's steer reservation, abort(B) linearizes as + // cancelling both the active target A and selected queued prompt B. + state.promptIdBySession = { sess_1: 'prompt_b' }; + const stopping = workspace.abortCurrentPrompt(); + await vi.waitFor(() => { + // Session cancellation must not wait behind a slow abort(B) response. + expect(apiMock.abortSession).toHaveBeenCalledWith('sess_1'); + }); + resolvePromptAbort({ aborted: true }); + await stopping; + await steering; + + expect(apiMock.abortPrompt).toHaveBeenCalledWith('sess_1', 'prompt_b'); + expect(apiMock.abortSession).toHaveBeenCalledWith('sess_1'); + expect(deps.pushOperationFailure).not.toHaveBeenCalled(); + }); + + it('cancels A and accepted B when Stop wins before the steer submit response', async () => { + let resolveSubmit!: (result: { + promptId: string; + userMessageId: string; + status: 'queued'; + }) => void; + apiMock.submitPrompt.mockReturnValue( + new Promise((resolve) => { + resolveSubmit = resolve; + }), + ); + apiMock.abortPrompt.mockResolvedValue({ aborted: true }); + apiMock.abortSession.mockResolvedValue({ aborted: true }); + const state = createState(); + state.promptIdBySession = { sess_1: 'prompt_a' }; + const deps = createDeps(); + const workspace = useWorkspaceState(state, deps); + + const steering = workspace.steerPrompt('follow up B'); + await vi.waitFor(() => { + expect(apiMock.submitPrompt).toHaveBeenCalledOnce(); + }); + + // The WS prompt.submitted event is allowed to overtake the HTTP response. + // Stop must not treat a successful queued-only abort(B) as sufficient: + // session abort cancels A, then the steer path aborts B again once its + // accepted response arrives and never attempts prompts:steer. + state.promptIdBySession = { sess_1: 'prompt_b' }; + await workspace.abortCurrentPrompt(); + resolveSubmit({ + promptId: 'prompt_b', + userMessageId: 'message_b', + status: 'queued', + }); + await steering; + + expect(apiMock.abortPrompt).toHaveBeenCalledWith('sess_1', 'prompt_b'); + expect(apiMock.abortSession).toHaveBeenCalledWith('sess_1'); + expect(apiMock.steerPrompts).not.toHaveBeenCalled(); + expect(deps.pushOperationFailure).not.toHaveBeenCalled(); + }); + + it('reports an unexpected steer failure instead of assuming queued B started', async () => { + const error = new Error('steer transport failed'); + apiMock.submitPrompt.mockResolvedValue({ + promptId: 'prompt_b', + userMessageId: 'message_b', + status: 'queued', + }); + apiMock.steerPrompts.mockRejectedValue(error); + const deps = createDeps(); + const workspace = useWorkspaceState(createState(), deps); + + await workspace.steerPrompt('follow up B'); + + expect(deps.pushOperationFailure).toHaveBeenCalledWith('steer', error, { + sessionId: 'sess_1', + }); + }); +}); + describe('useWorkspaceState — exportSession', () => { let anchor: { href: string; diff --git a/apps/vis/server/src/lib/context-projector.ts b/apps/vis/server/src/lib/context-projector.ts index df9271b3a1..03652274b7 100644 --- a/apps/vis/server/src/lib/context-projector.ts +++ b/apps/vis/server/src/lib/context-projector.ts @@ -501,8 +501,14 @@ export function projectContext( case 'metadata': case 'forked': case 'turn.prompt': + case 'turn.deferred_prompt_started': case 'turn.steer': case 'turn.cancel': + case 'turn.input_consumed': + // A turn outcome is durable intent, not context by itself. Recovery + // materializes it later as a context.append_message record; projecting + // both would expose the reminder too early and then duplicate it. + case 'turn.outcome': case 'permission.record_approval_result': case 'full_compaction.begin': case 'full_compaction.cancel': diff --git a/apps/vis/server/test/lib/context-projector.test.ts b/apps/vis/server/test/lib/context-projector.test.ts index 0a5ded18fe..219bebdd4e 100644 --- a/apps/vis/server/test/lib/context-projector.test.ts +++ b/apps/vis/server/test/lib/context-projector.test.ts @@ -35,6 +35,41 @@ describe('context-projector', () => { expect(proj.planMode.active).toBe(false); }); + it('does not project a pending turn outcome before it is materialized into context', () => { + const reminder = 'The previous turn ended before producing a final response.'; + const entries = [ + { + lineNo: 1, + data: { + type: 'context.append_message' as const, + message: { + role: 'user' as const, + content: [{ type: 'text' as const, text: 'continue' }], + toolCalls: [], + }, + }, + raw: {}, + }, + { + lineNo: 2, + data: { + type: 'turn.outcome' as const, + outcomeId: 'outcome-1', + turnId: 0, + content: reminder, + }, + raw: {}, + }, + ]; + + for (const mode of ['model', 'full'] as const) { + const proj = projectContext(entries, mode); + expect(proj.messages.map((message) => message.message.content)).toEqual([ + [{ type: 'text', text: 'continue' }], + ]); + } + }); + it('reconstructs assistant tool-call messages and separates tool results', async () => { const entries = [ { diff --git a/apps/vis/web/src/components/wire/renderers.tsx b/apps/vis/web/src/components/wire/renderers.tsx index 6ec3eb67fd..d9dec742eb 100644 --- a/apps/vis/web/src/components/wire/renderers.tsx +++ b/apps/vis/web/src/components/wire/renderers.tsx @@ -120,6 +120,19 @@ export const WIRE_RENDERERS: RendererMap = { ), }, + 'turn.deferred_prompt_started': { + tone: 'turn', + label: 'prompt→', + headline: (r) => ({ + main: ( + + turn {r.turnId} + deferred {r.deferredPromptId} + + ), + }), + }, + 'turn.steer': { tone: 'turn', label: 'steer', @@ -165,6 +178,50 @@ export const WIRE_RENDERERS: RendererMap = { }), }, + 'turn.outcome': { + tone: 'warning', + label: 'outcome', + headline: (r) => ({ + main: ( + + turn {r.turnId} + {truncate(r.content, 80)} + + ), + }), + detail: (r) => ( +
+ + {r.outcomeId} + + + {r.turnId} + + + +
{r.content}
+
+
+
+ ), + }, + + 'turn.input_consumed': { + tone: 'meta', + label: 'consumed', + headline: (r) => ({ + main: ( + + + {r.consumedTurnInput.kind} + + turn {r.consumedTurnInput.turnId} + {r.consumedTurnInput.id} + + ), + }), + }, + 'context.append_message': { tone: 'assistant', label: 'message', diff --git a/packages/agent-core-v2/src/_base/errors/unexpectedError.ts b/packages/agent-core-v2/src/_base/errors/unexpectedError.ts index b16c766b02..a6659fd540 100644 --- a/packages/agent-core-v2/src/_base/errors/unexpectedError.ts +++ b/packages/agent-core-v2/src/_base/errors/unexpectedError.ts @@ -24,8 +24,13 @@ export function onUnexpectedError(err: unknown): void { try { currentHandler(err); } catch (handlerErr) { - // eslint-disable-next-line no-console - console.error('[unexpected] handler threw', handlerErr, 'while reporting', err); + try { + // eslint-disable-next-line no-console + console.error('[unexpected] handler threw', handlerErr, 'while reporting', err); + } catch { + // Unexpected-error reporting is the final diagnostic boundary. A broken + // custom handler and a broken console fallback must both remain isolated. + } } } diff --git a/packages/agent-core-v2/src/_base/log/logService.ts b/packages/agent-core-v2/src/_base/log/logService.ts index 0697c5b5b2..fcbf060bfb 100644 --- a/packages/agent-core-v2/src/_base/log/logService.ts +++ b/packages/agent-core-v2/src/_base/log/logService.ts @@ -89,7 +89,13 @@ export class BoundLogger extends Disposable implements ILogger { } child(ctx: LogContext): ILogger { - return new BoundLogger(this.writer, this.levelState, { ...this.bound, ...ctx }); + try { + return new BoundLogger(this.writer, this.levelState, { ...this.bound, ...ctx }); + } catch { + // A hostile binding object must not make logger creation a business + // failure. Keep the parent's bindings and return a usable child. + return new BoundLogger(this.writer, this.levelState, this.bound); + } } error(message: string, payload?: LogPayload): void { @@ -110,23 +116,28 @@ export class BoundLogger extends Disposable implements ILogger { message: string, payload?: LogPayload, ): void { - if (!levelEnabled(level, this.levelState.level)) return; - const extracted = extractPayload(payload); - if (extracted === undefined) return; - const payloadCtx = extracted.ctx; - const error = extracted.error; - const ctx = - payloadCtx !== undefined || Object.keys(this.bound).length > 0 - ? { ...payloadCtx, ...this.bound } - : undefined; - const entry: LogEntry = { - t: Date.now(), - level, - msg: message, - ...(ctx !== undefined ? { ctx } : {}), - ...(error !== undefined ? { error } : {}), - }; - this.writer.write(entry); + try { + if (!levelEnabled(level, this.levelState.level)) return; + const extracted = extractPayload(payload); + if (extracted === undefined) return; + const payloadCtx = extracted.ctx; + const error = extracted.error; + const ctx = + payloadCtx !== undefined || Object.keys(this.bound).length > 0 + ? { ...payloadCtx, ...this.bound } + : undefined; + const entry: LogEntry = { + t: Date.now(), + level, + msg: message, + ...(ctx !== undefined ? { ctx } : {}), + ...(error !== undefined ? { error } : {}), + }; + this.writer.write(entry); + } catch { + // Logging is observational. Hostile payload accessors, entry construction, + // and broken sinks must never change the caller's control flow. + } } } diff --git a/packages/agent-core-v2/src/agent/contextInjector/contextInjectorService.ts b/packages/agent-core-v2/src/agent/contextInjector/contextInjectorService.ts index 0c0934a166..7ec02fa06e 100644 --- a/packages/agent-core-v2/src/agent/contextInjector/contextInjectorService.ts +++ b/packages/agent-core-v2/src/agent/contextInjector/contextInjectorService.ts @@ -6,7 +6,7 @@ * those positions after `wire` restoration. Bound at Agent scope. */ -import { Disposable, toDisposable } from "#/_base/di/lifecycle"; +import { Disposable, toDisposable } from '#/_base/di/lifecycle'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; @@ -43,7 +43,7 @@ export class AgentContextInjectorService extends Disposable implements IAgentCon this._register( loopService.hooks.onWillBeginStep.register('context-injector', async (_ctx, next) => { await next(); - await this.inject(); + await this.inject(false); }), ); this._register( @@ -82,10 +82,10 @@ export class AgentContextInjectorService extends Disposable implements IAgentCon async injectAfterCompaction(): Promise { this.isNewTurn = true; - await this.inject(); + await this.inject(true); } - private async inject(): Promise { + private async inject(deduplicate = false): Promise { const isNewTurn = this.isNewTurn; this.isNewTurn = false; for (const entry of this.entries) { @@ -97,6 +97,7 @@ export class AgentContextInjectorService extends Disposable implements IAgentCon }); if (!this.entries.has(entry)) continue; if (content === undefined) continue; + if (deduplicate && injectedPositions.length > 0) continue; const origin = { kind: 'injection' as const, variant: entry.name }; if (typeof content === 'string') { if (content.trim().length === 0) continue; diff --git a/packages/agent-core-v2/src/agent/contextMemory/compactionHandoff.ts b/packages/agent-core-v2/src/agent/contextMemory/compactionHandoff.ts index f17242652b..815b3db699 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/compactionHandoff.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/compactionHandoff.ts @@ -12,8 +12,12 @@ export const COMPACTION_SUMMARY_PREFIX = summaryPrefixTemplate.trimEnd(); export const COMPACT_USER_MESSAGE_MAX_TOKENS = 20_000; export const COMPACT_USER_MESSAGE_HEAD_TOKENS = 2_000; export const COMPACTION_ELISION_VARIANT = 'compaction_elision'; +export const TURN_OUTCOME_VARIANT = 'turn_outcome'; + +const COMPACTION_SOURCE_INDEX = Symbol('compactionSourceIndex'); type MessageLike = ContextMessage; +type IndexedMessage = ContextMessage & { readonly [COMPACTION_SOURCE_INDEX]: number }; export interface CompactionUserSelection { readonly head: T[]; @@ -22,6 +26,11 @@ export interface CompactionUserSelection { readonly omittedTokens: number; } +export interface CompactionTurnOutcome { + readonly index: number; + readonly message: ContextMessage; +} + export interface ContextCompactionShapeInput { readonly summary: string; readonly legacySummaryMessage?: ContextMessage; @@ -31,6 +40,7 @@ export interface ContextCompactionShapeInput { readonly tokensAfter?: number; readonly keptUserMessageCount?: number; readonly keptHeadUserMessageCount?: number; + readonly keptTurnOutcomeCount?: number; readonly droppedCount?: number; readonly legacyTail?: boolean; } @@ -43,6 +53,7 @@ export interface ContextCompactionShape { readonly tokensAfter: number; readonly keptUserMessageCount: number; readonly keptHeadUserMessageCount?: number; + readonly keptTurnOutcomeCount?: number; readonly droppedCount?: number; readonly messages: readonly ContextMessage[]; } @@ -64,19 +75,19 @@ export function buildContextCompactionShape( tokensBefore: input.tokensBefore, tokensAfter: input.tokensAfter ?? estimateTokensForMessages(messages), keptUserMessageCount: 0, + keptTurnOutcomeCount: input.keptTurnOutcomeCount, droppedCount: input.droppedCount, messages, }; } - const compactableUserMessages = collectCompactableUserMessages(history); + const compactableUserMessages = collectIndexedCompactableUserMessages(history); const selection = selectCompactionUserMessages(compactableUserMessages); const elisionMessage = selection.elided ? createCompactionElisionMessage(selection.omittedTokens) : undefined; - const keptMessages = elisionMessage === undefined - ? [...selection.head, ...selection.tail] - : [...selection.head, elisionMessage, ...selection.tail]; + const turnOutcome = latestTurnOutcomeAfterLastAssistant(history); + const keptMessages = mergeCompactionHandoff(selection, elisionMessage, turnOutcome); const contextSummary = input.contextSummary ?? input.summary; const tokensAfter = input.tokensAfter ?? estimateTokens(contextSummary) + estimateTokensForMessages(keptMessages); @@ -93,6 +104,7 @@ export function buildContextCompactionShape( tokensAfter, keptUserMessageCount, keptHeadUserMessageCount, + keptTurnOutcomeCount: turnOutcome === undefined ? undefined : 1, droppedCount: input.droppedCount, messages: [...keptMessages, createCompactionSummaryMessage(contextSummary)], }; @@ -135,6 +147,30 @@ export function collectCompactableUserMessages(messages: ); } +export function latestTurnOutcomeAfterLastAssistant( + messages: readonly ContextMessage[], +): CompactionTurnOutcome | undefined { + let lastAssistantIndex = -1; + for (let index = messages.length - 1; index >= 0; index--) { + const message = messages[index]; + if (message?.role === 'assistant' && message.origin === undefined) { + lastAssistantIndex = index; + break; + } + } + for (let index = messages.length - 1; index > lastAssistantIndex; index--) { + const message = messages[index]; + if ( + message?.role === 'user' && + message.origin?.kind === 'injection' && + message.origin.variant === TURN_OUTCOME_VARIANT + ) { + return { index, message }; + } + } + return undefined; +} + export function isCompactionSummaryMessage(message: MessageLike): boolean { return message.origin?.kind === 'compaction_summary'; } @@ -260,6 +296,57 @@ function usesLegacyTailShape(input: ContextCompactionShapeInput): boolean { return input.legacyTail === true; } +function collectIndexedCompactableUserMessages( + history: readonly ContextMessage[], +): IndexedMessage[] { + const messages: IndexedMessage[] = []; + history.forEach((message, index) => { + if (!isRealUserInput(message) || isCompactionSummaryMessage(message)) return; + messages.push({ ...message, [COMPACTION_SOURCE_INDEX]: index }); + }); + return messages; +} + +function mergeCompactionHandoff( + selection: CompactionUserSelection, + elisionMessage: ContextMessage | undefined, + turnOutcome: CompactionTurnOutcome | undefined, +): ContextMessage[] { + const head = [...selection.head]; + const tail = [...selection.tail]; + if (turnOutcome !== undefined) { + const headInsertAt = head.findIndex( + (message) => message[COMPACTION_SOURCE_INDEX] > turnOutcome.index, + ); + if (headInsertAt >= 0) { + head.splice(headInsertAt, 0, withSourceIndex(turnOutcome.message, turnOutcome.index)); + } else { + const tailInsertAt = tail.findIndex( + (message) => message[COMPACTION_SOURCE_INDEX] > turnOutcome.index, + ); + tail.splice( + tailInsertAt < 0 ? tail.length : tailInsertAt, + 0, + withSourceIndex(turnOutcome.message, turnOutcome.index), + ); + } + } + const indexed = elisionMessage === undefined + ? [...head, ...tail] + : [...head, elisionMessage, ...tail]; + return indexed.map(stripSourceIndex); +} + +function withSourceIndex(message: ContextMessage, index: number): IndexedMessage { + return { ...message, [COMPACTION_SOURCE_INDEX]: index }; +} + +function stripSourceIndex(message: ContextMessage | IndexedMessage): ContextMessage { + const { [COMPACTION_SOURCE_INDEX]: _index, ...plain } = message as IndexedMessage; + void _index; + return plain; +} + function extractText(content: readonly ContentPart[]): string { let text = ''; for (const part of content) { diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts index 3334d76ab4..705ab978e5 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemory.ts @@ -1,4 +1,4 @@ -import { createDecorator } from "#/_base/di/instantiation"; +import { createDecorator } from '#/_base/di/instantiation'; import type { UndoCut } from './contextOps'; import type { LoopRecordedEvent } from './loopEventFold'; @@ -12,6 +12,7 @@ export interface ContextCompactionInput { readonly tokensAfter?: number; readonly keptUserMessageCount?: number; readonly keptHeadUserMessageCount?: number; + readonly keptTurnOutcomeCount?: number; readonly droppedCount?: number; } @@ -23,6 +24,7 @@ export interface ContextCompactionResult { tokensAfter: number; keptUserMessageCount: number; keptHeadUserMessageCount?: number; + keptTurnOutcomeCount?: number; droppedCount?: number; } @@ -33,8 +35,12 @@ export interface IAgentContextMemoryService { append(...messages: readonly ContextMessage[]): void; + appendTurnOutcome(message: ContextMessage, outcomeId: string): void; + appendLoopEvent(event: LoopRecordedEvent): void; + closeAbandonedToolExchange(output: string): number; + clear(): void; undo(count: number): UndoCut; diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts index d18d7c6942..66081498e1 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextMemoryService.ts @@ -47,7 +47,7 @@ import { isFullyUndoable, type UndoCut, } from './contextOps'; -import type { LoopRecordedEvent } from './loopEventFold'; +import { openStepUuid, pendingToolCallIds, type LoopRecordedEvent } from './loopEventFold'; import type { ContextMessage } from './types'; declare module '#/app/event/eventBus' { @@ -82,13 +82,46 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte this.publishSplice({ start, deleteCount: 0, messages: [...messages] }); } + appendTurnOutcome(message: ContextMessage, outcomeId: string): void { + const start = this.get().length; + this.wire.dispatch( + contextAppendMessage({ message, materializedTurnOutcomeId: outcomeId }), + ); + this.publishSplice({ start, deleteCount: 0, messages: [message] }); + } + appendLoopEvent(event: LoopRecordedEvent): void { this.wire.dispatch(contextAppendLoopEvent({ event })); } + closeAbandonedToolExchange(output: string): number { + const history = this.get(); + const toolCallIds = pendingToolCallIds(history); + if (toolCallIds.length === 0) return 0; + const stepUuid = openStepUuid(history); + this.wire.dispatch( + ...toolCallIds.map((toolCallId) => + contextAppendLoopEvent({ + event: { + type: 'tool.result', + parentUuid: toolCallId, + toolCallId, + result: { output, isError: true }, + }, + }), + ), + ...(stepUuid === undefined + ? [] + : [contextAppendLoopEvent({ event: { type: 'step.end', uuid: stepUuid } })]), + ); + return toolCallIds.length; + } + clear(): void { const deleteCount = this.get().length; - if (deleteCount === 0) return; + // Persist even for an empty transcript: context.clear is also the durable + // tombstone for a cancellation/failure outcome intent that may have been + // written immediately before its reminder append. this.wire.dispatch(contextClear({}), contextSizeMeasured({ length: 0, tokens: 0 })); this.publishSplice({ start: 0, deleteCount, messages: [] }); } @@ -119,6 +152,7 @@ export class AgentContextMemoryService extends Disposable implements IAgentConte tokensAfter: result.tokensAfter, keptUserMessageCount: result.keptUserMessageCount, keptHeadUserMessageCount: result.keptHeadUserMessageCount, + keptTurnOutcomeCount: result.keptTurnOutcomeCount, droppedCount: result.droppedCount, }), contextSizeMeasured({ length: result.messages.length, tokens: result.tokensAfter }), diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts index 7638a5c3d3..372a9975a4 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextOps.ts @@ -135,7 +135,10 @@ const contextMessageSchema = z.custom(); const loopRecordedEventSchema = z.custom(); export const contextAppendMessage = ContextModel.defineOp('context.append_message', { - schema: z.object({ message: contextMessageSchema }), + schema: z.object({ + message: contextMessageSchema, + materializedTurnOutcomeId: z.string().optional(), + }), apply: (state, p) => foldAppendMessage(state, p.message) as ContextMessage[], }); @@ -154,6 +157,7 @@ const contextCompactionBaseShape = { tokensAfter: z.number().optional(), keptUserMessageCount: z.number().optional(), keptHeadUserMessageCount: z.number().optional(), + keptTurnOutcomeCount: z.number().optional(), droppedCount: z.number().optional(), legacyTail: z.boolean().optional(), }; @@ -217,6 +221,7 @@ export function readContextCompactionShapeInput( tokensAfter: readOptionalNumber(fields, 'tokensAfter'), keptUserMessageCount, keptHeadUserMessageCount: readOptionalNumber(fields, 'keptHeadUserMessageCount'), + keptTurnOutcomeCount: readOptionalNumber(fields, 'keptTurnOutcomeCount'), droppedCount: readOptionalNumber(fields, 'droppedCount'), legacyTail: readOptionalBoolean(fields, 'legacyTail') ?? keptUserMessageCount === undefined, }; diff --git a/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts b/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts index 481dd5865f..368aabbe50 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/contextTranscript.ts @@ -22,8 +22,8 @@ * - `context.apply_compaction` → keep the full history, append the user-role * summary marker, recover `foldedLength` from * the recorded kept-count fields - * - `context.undo` → remove tail messages (skip injections, stop - * at compaction summaries / clear floor) + * - `context.undo` → remove the target prompt and its full tail, + * stopping at compaction summaries / clear floor * - `context.clear` → keep prior transcript entries but reset the * folded view */ @@ -34,9 +34,9 @@ import type { WireRecord } from '#/wire/record'; import { COMPACT_USER_MESSAGE_MAX_TOKENS, collectCompactableUserMessages, - isRealUserInput, selectRecentUserMessages, } from './compactionHandoff'; +import { computeUndoCut, isFullyUndoable } from './contextOps'; import type { LoopRecordedEvent } from './loopEventFold'; import type { ContextMessage } from './types'; @@ -172,18 +172,16 @@ export function createContextTranscriptReducer(): ContextTranscriptReducer { const applyUndo = (count: number): void => { if (count <= 0) return; - let removedUserCount = 0; - for (let i = transcript.length - 1; i >= clearFloor; i--) { - const message = transcript[i]!.message; - if (message.origin?.kind === 'injection') continue; - if (message.origin?.kind === 'compaction_summary') break; - transcript.splice(i, 1); - foldedLength = Math.max(0, foldedLength - 1); - if (isRealUserInput(message)) { - removedUserCount++; - if (removedUserCount >= count) break; - } - } + const visibleEntries = transcript.slice(clearFloor); + const cut = computeUndoCut( + visibleEntries.map((entry) => entry.message), + count, + ); + if (!isFullyUndoable(cut, count)) return; + const cutIndex = clearFloor + cut.cutIndex; + const removedLength = transcript.length - cutIndex; + transcript.splice(cutIndex); + foldedLength = Math.max(0, foldedLength - removedLength); resetOpenState(); }; @@ -258,9 +256,14 @@ function recoverFoldedLength( ): number { const keptUserMessageCount = readNumber(record, 'keptUserMessageCount'); const keptHeadUserMessageCount = readNumber(record, 'keptHeadUserMessageCount'); + const keptTurnOutcomeCount = readNumber(record, 'keptTurnOutcomeCount') ?? 0; const compactedCount = readNumber(record, 'compactedCount'); if (keptUserMessageCount !== undefined) { - return keptUserMessageCount + (keptHeadUserMessageCount === undefined ? 1 : 2); + return ( + keptUserMessageCount + + keptTurnOutcomeCount + + (keptHeadUserMessageCount === undefined ? 1 : 2) + ); } if (compactedCount !== undefined && compactedCount < foldedLength) { return 1 + (foldedLength - compactedCount); diff --git a/packages/agent-core-v2/src/agent/contextMemory/loopEventFold.ts b/packages/agent-core-v2/src/agent/contextMemory/loopEventFold.ts index 546efab460..1fc813690a 100644 --- a/packages/agent-core-v2/src/agent/contextMemory/loopEventFold.ts +++ b/packages/agent-core-v2/src/agent/contextMemory/loopEventFold.ts @@ -139,6 +139,14 @@ export function foldAppendMessage( return bind([...state, message], ctx); } +export function pendingToolCallIds(state: readonly ContextMessage[]): readonly string[] { + return [...ctxOf(state).pending]; +} + +export function openStepUuid(state: readonly ContextMessage[]): string | undefined { + return ctxOf(state).openStepUuid; +} + export function foldLoopEvent( state: readonly ContextMessage[], event: LoopRecordedEvent, diff --git a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts index 858ec85c36..5431f27b68 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/compactionOps.ts @@ -2,29 +2,30 @@ * `fullCompaction` domain (L4) — wire Model (`CompactionModel`) and the * `full_compaction.begin` (`fullCompactionBegin`) / `full_compaction.cancel` * (`fullCompactionCancel`) / `full_compaction.complete` - * (`fullCompactionComplete`) Ops that mirror the full-compaction lifecycle into - * a persisted, replayable phase, plus the `compaction.*` edge events + * (`fullCompactionComplete`) Ops that mirror the full-compaction lifecycle + * into a persisted, replayable phase, plus the `compaction.*` edge events * (`started` / `blocked` / `cancelled` / `completed`) declared on `DomainEventMap` * (`compaction.started` is derived from the `full_compaction.begin` Op's * `toEvent`; the rest publish directly from the service). * - * The Model is intentionally phase-only — `{ phase }` (initial `idle`). The - * richer per-compaction data is NOT resume state: `instruction` is only needed - * by the live worker (which does not survive a restart) and by telemetry, so it - * rides the `begin` payload (and is persisted on the record for audit) but is - * not stored in the Model; result numbers are consumed live by the - * `compaction.completed` signal and their durable effect (the summary message - * plus compaction metrics) already lives in `contextMemory`. The live - * `complete` payload is empty to match the v1 wire shape; legacy logs may still - * carry result numbers, and `apply` accepts and ignores them while collapsing - * to `idle`. Each `apply` returns the same reference on a no-op so the wire's - * reference-equality gate stays quiet; it carries no non-determinism. + * The Model is intentionally phase-only — `{ phase }` (initial `idle`). A + * `context.apply_compaction` cross-reducer advances a running compaction to + * `committed`, making the context replacement the durable commit point. The + * richer per-compaction data is not resume state: `instruction` is only needed + * by the live worker and telemetry, while the committed summary and accounting + * already live in `contextMemory`. A pre-commit failure uses the existing + * `full_compaction.complete` record type with `outcome: 'failed'`, preserving + * the v1 wire vocabulary while distinguishing it from a successful terminal; + * legacy complete records omit that field and remain successful. Older records + * may also carry result numbers, which `apply` ignores. Each reducer returns the + * same reference on a no-op so the wire's reference-equality gate stays quiet + * and carries no non-determinism. * * The runtime orchestration — `ActiveCompaction`, its `AbortController`, and * the in-flight worker promise — stays OUT of the Model (live-only service - * members): none of it can be resumed, and a session never restores mid-flight. - * A `running` phase stranded by a crash is reset to `idle` by the service's - * `wire.hooks.onDidRestore` hook (mirroring `goal`'s post-replay normalization). + * members): none of it can be resumed. The service's `wire.hooks.onDidRestore` + * hook fails a stranded pre-commit run and converges a committed run before + * returning the Agent to `idle`. * * The `compaction.*` events publish to `IEventBus` (`compaction.started` via the * `begin` Op's `toEvent`; the rest directly from the service); they are @@ -61,15 +62,22 @@ export interface CompactionCompletedEvent { readonly result: CompactionResult; } -export type CompactionPhase = 'idle' | 'running' | 'cancelled' | 'completed'; +export type CompactionPhase = 'idle' | 'running' | 'committed'; export interface CompactionState { readonly phase: CompactionPhase; } -export const CompactionModel = defineModel('fullCompaction', () => ({ - phase: 'idle', -})); +export const CompactionModel = defineModel( + 'fullCompaction', + () => ({ phase: 'idle' }), + { + reducers: { + 'context.apply_compaction': (state) => + state.phase === 'running' ? { phase: 'committed' } : state, + }, + }, +); declare module '#/app/event/eventBus' { interface DomainEventMap { @@ -104,6 +112,10 @@ export const fullCompactionCancel = CompactionModel.defineOp('full_compaction.ca }); export const fullCompactionComplete = CompactionModel.defineOp('full_compaction.complete', { - schema: z.object({}), + schema: z.object({ outcome: z.literal('failed').optional() }), apply: (s) => (s.phase === 'idle' ? s : { phase: 'idle' }), }); + +export function fullCompactionFail(): ReturnType { + return fullCompactionComplete({ outcome: 'failed' }); +} diff --git a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts index 03e169ac5b..f9174f3af4 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/fullCompactionService.ts @@ -1,16 +1,20 @@ -import { Disposable, type IDisposable } from "#/_base/di/lifecycle"; +import { Disposable, type IDisposable } from '#/_base/di/lifecycle'; import { InstantiationType } from '#/_base/di/extensions'; import { IInstantiationService } from '#/_base/di/instantiation'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { ILogService } from '#/_base/log/log'; -import { renderPrompt } from "#/_base/utils/render-prompt"; +import { renderPrompt } from '#/_base/utils/render-prompt'; import { estimateTokens, estimateTokensForMessage, estimateTokensForMessages, estimateTokensForTools, -} from "#/_base/utils/tokens"; -import { buildCompactionSummaryText, isRealUserInput } from '#/agent/contextMemory/compactionHandoff'; +} from '#/_base/utils/tokens'; +import { + buildCompactionSummaryText, + buildContextCompactionShape, + isRealUserInput, +} from '#/agent/contextMemory/compactionHandoff'; import { IAgentContextInjectorService } from '#/agent/contextInjector/contextInjector'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import type { ContextMessage } from '#/agent/contextMemory/types'; @@ -38,7 +42,14 @@ import { inputTotal, type TokenUsage } from '#/app/llmProtocol/usage'; import { IEventBus } from '#/app/event/eventBus'; import type { CompactionFinishedEvent } from '#/app/telemetry/events'; import { ITelemetryService } from '#/app/telemetry/telemetry'; -import { ErrorCodes, Error2, isCodedError, isError2, toKimiErrorPayload, unwrapErrorCause } from "#/errors"; +import { + ErrorCodes, + Error2, + isCodedError, + isError2, + toKimiErrorPayload, + unwrapErrorCause, +} from '#/errors'; import { IWireService } from '#/wire/wire'; import compactionInstructionTemplate from './compaction-instruction.md?raw'; import { @@ -55,6 +66,7 @@ import { fullCompactionBegin, fullCompactionCancel, fullCompactionComplete, + fullCompactionFail, } from './compactionOps'; import { type CompactionBeginData, @@ -81,6 +93,9 @@ type CompactionTelemetryProperties = Pick< interface ActiveCompaction extends FullCompactionTask { blockedByTurn: boolean; + turnId?: number; + phase: 'running' | 'cancelling' | 'committing' | 'committed' | 'finished'; + committedResult?: CompactionResult; bgRegistration?: IDisposable; } @@ -111,6 +126,10 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull private lastCompactedTokenCount: number | null = null; private consecutiveOverflowCompactions = 0; private contextInjectorService: IAgentContextInjectorService | undefined; + private systemPromptRefreshPending = false; + private systemPromptRefreshTurnId: number | undefined; + private contextReinjectionPending = false; + private postCommitMaintenance: Promise | undefined; constructor( @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, @@ -132,8 +151,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this.strategy = new RuntimeCompactionStrategy(() => this.resolveModelContextWithEffectiveMax()); this._register( this.wire.hooks.onDidRestore.register('full-compaction', async (_ctx, next) => { - this.normalizeAfterReplay(); await next(); + await this.normalizeAfterReplay(); }), ); this._register( @@ -146,8 +165,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull }), ); this._register( - this.loopService.hooks.onDidFinishStep.register('full-compaction', async (_ctx, next) => { - await this.afterStep(); + this.loopService.hooks.onDidFinishStep.register('full-compaction', async (ctx, next) => { + await this.afterStep(ctx.turnId); await next(); }), ); @@ -236,15 +255,15 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this.observedMaxContextTokensByModel.set(modelAlias, observed); } - begin(input: FullCompactionInput): boolean { - if (this._compacting) return false; + begin(input: FullCompactionInput, turnId?: number): boolean { + if (this._compacting || this.wire.getModel(CompactionModel).phase !== 'idle') return false; const data: CompactionBeginData = { source: input.source, instruction: input.instruction }; if (!this.reserveCompactionSlot(data.source)) return false; const tokenCount = this.validateCompactionStart(data.source); this.wire.dispatch(fullCompactionBegin(data)); - const active = this.createActiveCompaction(data.source, tokenCount); + const active = this.createActiveCompaction(data.source, tokenCount, turnId); this._compacting = active.task; active.task.abortController.signal.addEventListener( 'abort', @@ -282,6 +301,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull private createActiveCompaction( trigger: CompactionBeginData['source'], tokenCount: number, + turnId?: number, ): { readonly task: ActiveCompaction; readonly resolve: (result: CompactionResult) => void; @@ -301,6 +321,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull trigger, tokenCount, blockedByTurn: false, + turnId, + phase: 'running', bgRegistration: this.activity.registerBackground('compaction', abortController), }, resolve, @@ -309,7 +331,8 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull } private cancelActive(active: ActiveCompaction): boolean { - if (this._compacting !== active) return false; + if (this._compacting !== active || active.phase !== 'running') return false; + active.phase = 'cancelling'; this.wire.dispatch(fullCompactionCancel({})); this._compacting = null; active.bgRegistration?.dispose(); @@ -320,17 +343,37 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull return true; } - private markCompleted(active: ActiveCompaction): boolean { - if (this._compacting !== active) return false; - this.wire.dispatch(fullCompactionComplete({})); + private failActive(active: ActiveCompaction): boolean { + if ( + this._compacting !== active || + (active.phase !== 'running' && active.phase !== 'committing') + ) { + return false; + } + active.phase = 'finished'; + this.wire.dispatch(fullCompactionFail()); this._compacting = null; active.bgRegistration?.dispose(); return true; } - private normalizeAfterReplay(): void { - if (this.wire.getModel(CompactionModel).phase !== 'running') return; - this.wire.dispatch(fullCompactionCancel({})); + private releaseCommitted(active: ActiveCompaction): boolean { + if (this._compacting !== active || active.phase !== 'committed') return false; + active.phase = 'finished'; + this._compacting = null; + active.bgRegistration?.dispose(); + return true; + } + + private async normalizeAfterReplay(): Promise { + const phase = this.wire.getModel(CompactionModel).phase; + if (phase === 'running') { + this.wire.dispatch(fullCompactionFail()); + return; + } + if (phase !== 'committed') return; + this.schedulePostCommitMaintenance(); + await this.retryPostCommitMaintenance(false); } private resetForTurn(): void { @@ -343,7 +386,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull context: LoopErrorContext, ): Promise { this.recordOverflowRecovery(context.error); - const didStartCompaction = this.beginAutoCompaction(); + const didStartCompaction = this.beginAutoCompaction(true, context.turnId); if (!didStartCompaction && !this._compacting) return false; await this.block(context.signal, context.turnId); @@ -370,20 +413,22 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull } private async beforeStep(signal: AbortSignal, turnId?: number): Promise { - this.checkAutoCompaction(); + await this.retryPostCommitMaintenance(true); + this.checkAutoCompaction(true, turnId); if (this.strategy.shouldBlock(this.tokenCountWithPending())) { await this.block(signal, turnId); } + await this.retryPostCommitMaintenance(true); } - private async afterStep(): Promise { + private async afterStep(turnId: number): Promise { this.consecutiveOverflowCompactions = 0; if (this.strategy.checkAfterStep) { - this.checkAutoCompaction(false); + this.checkAutoCompaction(false, turnId); } } - private checkAutoCompaction(throwOnLimit = true): boolean { + private checkAutoCompaction(throwOnLimit = true, turnId?: number): boolean { if (this._compacting) return true; if ( this.lastCompactedTokenCount !== null && @@ -392,10 +437,10 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull return false; } if (!this.strategy.shouldCompact(this.tokenCountWithPending())) return false; - return this.beginAutoCompaction(throwOnLimit); + return this.beginAutoCompaction(throwOnLimit, turnId); } - private beginAutoCompaction(throwOnLimit = true): boolean { + private beginAutoCompaction(throwOnLimit = true, turnId?: number): boolean { if (this._compacting) return true; const maxCompactions = this.strategy.maxCompactionPerTurn; if (this.compactionCountInTurn >= maxCompactions) { @@ -406,13 +451,14 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull } return false; } - return this.begin({ source: 'auto' }); + return this.begin({ source: 'auto' }, turnId); } private async block(signal?: AbortSignal, turnId?: number): Promise { const active = this._compacting; if (active === null) return; active.blockedByTurn = true; + active.turnId ??= turnId; this.propagateBlockingAbort(active, signal); this.eventBus.publish({ type: 'compaction.blocked', turnId }); try { @@ -451,30 +497,24 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull try { const result = await this.compactionRound(active, data); if (this._compacting !== active) throw compactionCancelledReason(active); - try { - await this.profile.refreshSystemPrompt(); - } catch (error) { - this.log.error('failed to refresh system prompt after compaction', { error }); + return await this.finishCommittedCompaction(active, result); + } catch (error) { + if ( + active.phase === 'committing' && + this.wire.getModel(CompactionModel).phase === 'committed' + ) { + active.phase = 'committed'; } - this.lastCompactedTokenCount = result.tokensAfter; - await this.contextInjector.injectAfterCompaction(); - this.lastCompactedTokenCount = this.tokenCountWithPending(); - if (!this.markCompleted(active)) { - throw compactionCancelledReason(active); + if (active.phase === 'committed' && active.committedResult !== undefined) { + this.reportDiagnosticError('post-commit compaction work failed', error); + return await this.finishCommittedCompaction(active, active.committedResult); } - const { contextSummary: _contextSummary, ...eventResult } = result; - void _contextSummary; - this.eventBus.publish({ type: 'compaction.completed', result: eventResult }); - return result; - } catch (error) { - if (active.abortController.signal.aborted || isAbortError(error)) { + if (active.abortController.signal.aborted) { this.cancelActive(active); throw error; } const blockedByTurn = this._compacting === active && active.blockedByTurn; - if (this._compacting === active) { - this.cancelActive(active); - } + this.failActive(active); if (blockedByTurn) { throw error; } @@ -488,6 +528,20 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull } } + private async finishCommittedCompaction( + active: ActiveCompaction, + result: CompactionResult, + ): Promise { + this.lastCompactedTokenCount = result.tokensAfter; + this.schedulePostCommitMaintenance(active.turnId); + await this.retryPostCommitMaintenance(false); + this.releaseCommitted(active); + const { contextSummary: _contextSummary, ...eventResult } = result; + void _contextSummary; + this.eventBus.publish({ type: 'compaction.completed', result: eventResult }); + return result; + } + private async compactionRound( active: ActiveCompaction, data: Readonly, @@ -611,13 +665,21 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull } const summary = this.postProcessSummary(attempt.summary); - const result = this.context.applyCompaction({ + const compactionInput = { summary, contextSummary: buildCompactionSummaryText(summary), compactedCount: originalHistory.length, tokensBefore, droppedCount: droppedCount === 0 ? undefined : droppedCount, - }); + }; + const prepared = buildContextCompactionShape(this.context.get(), compactionInput); + const { messages: _messages, ...preparedResult } = prepared; + void _messages; + active.phase = 'committing'; + active.committedResult = preparedResult; + const result = this.context.applyCompaction(compactionInput); + active.phase = 'committed'; + active.committedResult = result; const properties: CompactionFinishedEvent = { source: data.source, @@ -634,7 +696,7 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull this.telemetry.track2('compaction_finished', properties); return result; } catch (error) { - if (isAbortError(error)) throw error; + if (isAbortError(error) && active.abortController.signal.aborted) throw error; this.telemetry.track2('compaction_failed', { source: data.source, tokens_before: tokensBefore, @@ -671,6 +733,81 @@ export class AgentFullCompactionService extends Disposable implements IAgentFull return this.contextSize.get().size; } + private schedulePostCommitMaintenance(turnId?: number): void { + this.systemPromptRefreshPending = true; + this.systemPromptRefreshTurnId = turnId; + this.contextReinjectionPending = true; + } + + private async retryPostCommitMaintenance(throwOnFailure: boolean): Promise { + if ( + !this.systemPromptRefreshPending && + !this.contextReinjectionPending && + this.wire.getModel(CompactionModel).phase !== 'committed' + ) { + return; + } + let maintenance = this.postCommitMaintenance; + if (maintenance === undefined) { + maintenance = this.performPostCommitMaintenance(); + this.postCommitMaintenance = maintenance; + void maintenance.finally(() => { + if (this.postCommitMaintenance === maintenance) { + this.postCommitMaintenance = undefined; + } + }); + } + const error = await maintenance; + if (throwOnFailure && error !== undefined) throw error; + } + + private async performPostCommitMaintenance(): Promise { + let firstError: Error | undefined; + if (this.systemPromptRefreshPending) { + try { + const systemPrompt = await this.profile.refreshSystemPrompt(); + if (systemPrompt !== undefined && this.systemPromptRefreshTurnId !== undefined) { + this.llmRequester.refreshTurnSystemPrompt( + this.systemPromptRefreshTurnId, + systemPrompt, + ); + } + this.systemPromptRefreshPending = false; + this.systemPromptRefreshTurnId = undefined; + } catch (error) { + firstError = error instanceof Error ? error : new Error(String(error)); + this.reportDiagnosticError('failed to refresh system prompt after compaction', error); + } + } + if (this.contextReinjectionPending) { + try { + await this.contextInjector.injectAfterCompaction(); + this.contextReinjectionPending = false; + } catch (error) { + firstError ??= error instanceof Error ? error : new Error(String(error)); + this.reportDiagnosticError('failed to inject context after compaction', error); + } + } + this.lastCompactedTokenCount = this.tokenCountWithPending(); + if ( + !this.systemPromptRefreshPending && + !this.contextReinjectionPending && + this.wire.getModel(CompactionModel).phase === 'committed' + ) { + this.wire.dispatch(fullCompactionComplete({})); + } + return firstError; + } + + private reportDiagnosticError(message: string, error: unknown): void { + try { + this.log.error(message, { error }); + } catch { + // Post-commit diagnostics must not alter compaction convergence or make + // an already-committed result appear to have failed. + } + } + private get contextInjector(): IAgentContextInjectorService { if (this.contextInjectorService === undefined) { this.contextInjectorService = this.instantiation.invokeFunction((accessor) => diff --git a/packages/agent-core-v2/src/agent/fullCompaction/types.ts b/packages/agent-core-v2/src/agent/fullCompaction/types.ts index 2d7f38f70e..1d84c915ca 100644 --- a/packages/agent-core-v2/src/agent/fullCompaction/types.ts +++ b/packages/agent-core-v2/src/agent/fullCompaction/types.ts @@ -21,6 +21,7 @@ export interface CompactionResult { * backward compatibility with older wire records. */ keptHeadUserMessageCount?: number; + keptTurnOutcomeCount?: number; /** * Oldest messages trimmed from the summarizer input when the compaction * request overflowed the model window; not covered by the produced summary. diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts index 259ecf019b..e27513347e 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequester.ts @@ -59,6 +59,14 @@ export interface LLMRequestOverrides { export interface IAgentLLMRequesterService { readonly _serviceBrand: undefined; + /** + * Replace only the system-prompt field of a turn's request snapshot after + * full compaction re-renders the active profile. Model/provider/thinking + * choices remain frozen for the turn, and ordinary config updates still + * take effect only on a later turn. + */ + refreshTurnSystemPrompt(turnId: number, systemPrompt: string): void; + request( overrides?: LLMRequestOverrides, onPart?: LLMRequestPartHandler, diff --git a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts index 7a27849d62..a00d2a1a6f 100644 --- a/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts +++ b/packages/agent-core-v2/src/agent/llmRequester/llmRequesterService.ts @@ -146,6 +146,12 @@ export class AgentLLMRequesterService implements IAgentLLMRequesterService { @IFaultInjectionService private readonly faultInjection: IFaultInjectionService, ) {} + refreshTurnSystemPrompt(turnId: number, systemPrompt: string): void { + const snapshot = this.turnConfigs.get(turnId); + if (snapshot === undefined) return; + this.turnConfigs.set(turnId, { ...snapshot, systemPrompt }); + } + async request( overrides: LLMRequestOverrides = {}, onPart: LLMRequestPartHandler = noopOnPart, diff --git a/packages/agent-core-v2/src/agent/loop/loopService.ts b/packages/agent-core-v2/src/agent/loop/loopService.ts index f15f148d4f..56a7aee48d 100644 --- a/packages/agent-core-v2/src/agent/loop/loopService.ts +++ b/packages/agent-core-v2/src/agent/loop/loopService.ts @@ -23,8 +23,10 @@ * `stepRetry` re-enqueues the failed driver after backoff, `fullCompaction` * compacts and re-enqueues it — so the loop only learns caught-or-not, while * an unclaimed or uncaught error fails the turn. Emits `turn.*` / delta - * events through `event`, persists loop events through `contextMemory`, and - * reads the step budget from `config`. Bound at Agent scope. + * events through `event`, persists loop events through `contextMemory`, appends + * abnormal-turn context through `systemReminder`, reads the step budget from + * `config`, and reports reminder-persistence failures through `log`. Bound at + * Agent scope. */ import { randomUUID } from 'node:crypto'; @@ -34,6 +36,7 @@ import { createControlledPromise } from '@antfu/utils'; import { InstantiationType } from '#/_base/di/extensions'; import { Disposable, toDisposable, type IDisposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { ILogService } from '#/_base/log/log'; import { abortError, isAbortError, isUserCancellation, userCancellationReason } from '#/_base/utils/abort'; import { toErrorMessage } from '#/_base/errors/errorMessage'; import { IAgentLLMRequesterService, type LLMRequestFinish } from '#/agent/llmRequester/llmRequester'; @@ -49,6 +52,7 @@ import { OrderedHookSlot } from '#/hooks'; import type { ActivityLease } from '#/activity/activity'; import { IAgentActivityService } from '#/activity/activity'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; +import { IAgentSystemReminderService } from '#/agent/systemReminder/systemReminder'; import { IAgentTelemetryContextService } from '#/app/telemetry/agentTelemetryContext'; import type { TurnEndedEvent as TurnEndedTelemetryEvent, @@ -81,7 +85,18 @@ import { type TurnSeed, } from './stepRequest'; import { StepRequestQueue, type StepRequestBatch } from './stepRequestQueue'; -import { cancelTurn, promptTurn, TurnModel } from './turnOps'; +import { + cancelTurn, + promptTurn, + TurnModel, + turnOutcome, + type TurnOutcomeIntent, +} from './turnOps'; +import { + renderTurnCancellationReminder, + renderTurnFailureReminder, + TURN_OUTCOME_REMINDER_VARIANT, +} from './turnOutcomeReminder'; // Loads the `DomainEventMap` augmentation for the `turn.*` / delta events this // service publishes (the augmentation lives with the event definitions; // without an import it would not enter every consumer's program). @@ -103,10 +118,13 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { private readonly pendingTurns: TurnJob[] = []; private activeTurnJob: TurnJob | undefined; private nextReservedTurnId: number | undefined; + private readonly cancellingTurnIds = new Set(); private disposing = false; constructor( @IAgentContextMemoryService private readonly context: IAgentContextMemoryService, + @IAgentSystemReminderService private readonly reminders: IAgentSystemReminderService, + @ILogService private readonly log: ILogService, @IAgentLLMRequesterService private readonly llmRequester: IAgentLLMRequesterService, @IEventBus private readonly eventBus: IEventBus, @IAgentToolExecutorService private readonly toolExecutor: IAgentToolExecutorService, @@ -117,6 +135,12 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { @IAgentTelemetryContextService private readonly telemetryContext: IAgentTelemetryContextService, ) { super(); + this._register( + this.wire.hooks.onDidRestore.register('loop-turn-outcome', async (_ctx, next) => { + await next(); + this.materializePendingTurnOutcomes(true); + }), + ); } override dispose(): void { @@ -198,7 +222,15 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { private cancelActiveTurn(turnId: number | undefined, cancellation: unknown): boolean { const turn = this.activeTurnJob?.turn; if (turn === undefined || (turnId !== undefined && turn.id !== turnId)) return false; - this.wire.dispatch(cancelTurn({ turnId })); + if (this.cancellingTurnIds.has(turn.id)) return true; + this.cancellingTurnIds.add(turn.id); + const outcome = this.createCancellationOutcome(turn.id, cancellation); + this.wire.dispatch(cancelTurn({ + turnId, + outcomeId: outcome?.outcomeId, + outcomeTurnId: outcome?.turnId, + outcomeContent: outcome?.content, + })); return this.activity.cancel(cancellation); } @@ -321,19 +353,37 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { return true; } - private pumpTurns(): void { + private pumpTurns(continueAfterStartFailure = false): void { if (this.disposing || this.activeTurnJob !== undefined) return; - const job = this.pendingTurns.shift(); - if (job === undefined) return; - this.startTurn(job); + while (!this.disposing && this.activeTurnJob === undefined) { + const job = this.pendingTurns.shift(); + if (job === undefined) return; + try { + this.startTurn(job); + return; + } catch (error) { + if (!continueAfterStartFailure || job.turn.state !== 'failed') throw error; + } + } } private startTurn(job: TurnJob): void { const lease = this.activity.begin('turn', { origin: job.seed.origin, turnId: job.turn.id }); - this.wire.dispatch(promptTurn({ input: job.seed.input, origin: lease.origin })); job.turn.state = 'running'; job.turn.signal = lease.signal; this.activeTurnJob = job; + try { + // An abnormal-turn intent is a durable ordering barrier: the reminder + // must reach context before the next prompt can be admitted. If the + // append actually committed and only a downstream observer threw, the + // TurnModel acknowledgement below lets this continue without retrying or + // duplicating it. + this.materializePendingTurnOutcomes(true); + } catch (error) { + this.failTurnStart(job, lease, error); + throw error; + } + this.wire.dispatch(promptTurn({ input: job.seed.input, origin: lease.origin })); this.eventBus.publish({ type: 'turn.started', turnId: job.turn.id, origin: lease.origin }); void this.runTurn(job.turn, lease, job.ready).then(job.result.resolve, job.result.reject); } @@ -348,6 +398,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { const turnTelemetry = this.telemetry.withContext(telemetryContext); const { mode, provider_type, protocol } = telemetryContext; let result: TurnResult | undefined; + let terminalOutcome: TurnOutcomeIntent | undefined; try { const started: TurnStartedTelemetryEvent = { turn_id: turn.id, mode, provider_type, protocol }; turnTelemetry.track2('turn_started', started); @@ -361,12 +412,17 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { result = this.resultFromTurnError(lease, error); return result; } finally { + const error = result?.type === 'failed' ? toKimiErrorPayload(result.error) : undefined; + if (result !== undefined) { + terminalOutcome = this.ensureTurnOutcomeIntent(turn.id, result, error); + } this.settleTurnReady(ready, result); this.releaseActiveTurn(turn, result); const outcome = result?.type ?? 'failed'; lease.end(outcome, result?.type === 'failed' ? { error: result.error } : undefined); if (result !== undefined) { - const error = result.type === 'failed' ? toKimiErrorPayload(result.error) : undefined; + this.closeAbandonedToolExchange(turn.id, result, error); + if (terminalOutcome !== undefined) this.materializeTurnOutcome(terminalOutcome); this.eventBus.publish({ type: 'turn.ended', turnId: turn.id, @@ -396,10 +452,151 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { protocol, }; turnTelemetry.track2('turn_ended', ended); - this.pumpTurns(); + this.pumpTurns(true); + } + } + + private failTurnStart(job: TurnJob, lease: ActivityLease, error: unknown): void { + const failure: TurnResult = { type: 'failed', error, steps: 0 }; + for (const step of job.steps.values()) step.cancel(error); + job.controller.abort(error); + job.turn.state = 'failed'; + this.cancellingTurnIds.delete(job.turn.id); + if (this.activeTurnJob === job) this.activeTurnJob = undefined; + if ( + this.pendingTurns.length === 0 && + this.nextReservedTurnId === job.turn.id + 1 + ) { + this.nextReservedTurnId = job.turn.id; + } + job.ready.reject(error instanceof Error ? error : new Error(toErrorMessage(error))); + job.result.resolve(failure); + lease.end('failed', { error }); + } + + private ensureTurnOutcomeIntent( + turnId: number, + result: TurnResult, + error: ReturnType | undefined, + ): TurnOutcomeIntent | undefined { + if (result.type === 'completed') return undefined; + const existing = this.pendingTurnOutcome(turnId); + if (existing !== undefined) return existing; + // Active cancellation reserves its stable intent in turn.cancel. If a + // destructive context operation consumed that intent before the worker + // observed the abort, it is a tombstone—not permission to mint a second + // outcome after the clear/undo boundary. + if (result.type === 'cancelled' && this.cancellingTurnIds.has(turnId)) { + return undefined; + } + try { + const content = result.type === 'failed' + ? renderTurnFailureReminder(error) + : renderTurnCancellationReminder(result.reason, isUserCancellation(result.reason)); + const outcome = { outcomeId: randomUUID(), turnId, content }; + this.wire.dispatch(turnOutcome(outcome)); + return this.pendingTurnOutcome(turnId) ?? outcome; + } catch (outcomeError) { + this.safeDiagnostic('failed to persist turn outcome intent', { + turnId, + error: outcomeError, + }); + return undefined; + } + } + + private createCancellationOutcome( + turnId: number, + cancellation: unknown, + ): TurnOutcomeIntent | undefined { + const existing = this.pendingTurnOutcome(turnId); + if (existing !== undefined) return existing; + try { + return { + outcomeId: randomUUID(), + turnId, + content: renderTurnCancellationReminder( + cancellation, + isUserCancellation(cancellation), + ), + }; + } catch (error) { + this.safeDiagnostic('failed to render turn cancellation outcome', { turnId, error }); + return undefined; + } + } + + private pendingTurnOutcome(turnId: number): TurnOutcomeIntent | undefined { + return this.wire + .getModel(TurnModel) + .pendingOutcomes.find((outcome) => outcome.turnId === turnId); + } + + private materializePendingTurnOutcomes(strict = false): void { + for (const outcome of this.wire.getModel(TurnModel).pendingOutcomes) { + this.materializeTurnOutcome(outcome, strict); } } + private materializeTurnOutcome(outcome: TurnOutcomeIntent, strict = false): void { + if (!this.hasPendingTurnOutcome(outcome.outcomeId)) return; + try { + this.reminders.appendSystemReminder( + outcome.content, + { kind: 'injection', variant: TURN_OUTCOME_REMINDER_VARIANT }, + outcome.outcomeId, + ); + } catch (error) { + if (this.pendingTurnOutcome(outcome.turnId)?.outcomeId !== outcome.outcomeId) { + this.safeDiagnostic('turn outcome reminder committed with observer failure', { + turnId: outcome.turnId, + error, + }); + return; + } + if (strict) throw error; + this.safeDiagnostic('failed to append turn outcome reminder', { + turnId: outcome.turnId, + error, + }); + } + } + + private hasPendingTurnOutcome(outcomeId: string): boolean { + return this.wire + .getModel(TurnModel) + .pendingOutcomes.some((outcome) => outcome.outcomeId === outcomeId); + } + + private closeAbandonedToolExchange( + turnId: number, + result: TurnResult, + error: ReturnType | undefined, + ): void { + try { + const closed = this.context.closeAbandonedToolExchange( + abandonedToolResultOutput(result, error), + ); + if (closed === 0) return; + this.safeDiagnostic('closed abandoned tool exchange at turn end', { + turnId, + reason: result.type, + closed, + }); + } catch (closeError) { + this.safeDiagnostic('failed to close abandoned tool exchange', { + turnId, + error: closeError, + }); + } + } + + private safeDiagnostic(message: string, payload?: unknown): void { + try { + this.log.warn(message, payload); + } catch { } + } + private resultFromTurnError(lease: ActivityLease, error: unknown): TurnResult { if (!lease.signal.aborted) return { type: 'failed', error, steps: 0 }; return { type: 'cancelled', steps: 0, reason: lease.signal.reason ?? error }; @@ -420,6 +617,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService { private releaseActiveTurn(turn: Turn, result: TurnResult | undefined): void { (turn as MutableTurn).state = result?.type ?? 'failed'; + this.cancellingTurnIds.delete(turn.id); const job = this.activeTurnJob?.turn === turn ? this.activeTurnJob : undefined; if (job === undefined) return; const reason = result?.type === 'cancelled' ? result.reason : abortError('Turn ended'); @@ -991,6 +1189,19 @@ function interruptReasonFor( return 'error'; } +function abandonedToolResultOutput( + result: TurnResult, + error: ReturnType | undefined, +): string { + const cause = + result.type === 'cancelled' + ? 'the turn was cancelled' + : result.type === 'failed' + ? `the turn failed${error !== undefined ? ` (${error.message})` : ''}` + : 'the turn ended'; + return `Tool call did not complete: ${cause} before its result was recorded. Do not assume the tool completed successfully.`; +} + type StepExecutionResult = { readonly stopReason: FinishReason; readonly hookStopTurn: boolean; diff --git a/packages/agent-core-v2/src/agent/loop/turnOps.ts b/packages/agent-core-v2/src/agent/loop/turnOps.ts index 8cad5371a3..5ac0ef7f18 100644 --- a/packages/agent-core-v2/src/agent/loop/turnOps.ts +++ b/packages/agent-core-v2/src/agent/loop/turnOps.ts @@ -1,6 +1,6 @@ /** * `loop` domain (L4) — wire Model (`TurnModel`) and the Ops that bookkeep the - * agent's turn lifecycle on the wire. + * agent's turn lifecycle and abnormal outcome handoffs on the wire. * * Declares the next turn id as a wire Model (initial `0`). The persisted * `turn.prompt` record carries exactly v1's field set (`{ input, origin }` — @@ -8,8 +8,11 @@ * advances the counter by one, so the counter is restored by counting * turn starts. Every turn is started by `loopService.enqueue` admitting a * request that creates a new Turn, which dispatches one - * `turn.prompt` per start. As a belt-and-suspenders for v1-written logs whose - * internally-driven turns (goal continuations) have no `turn.prompt` record, + * `turn.prompt` per start. `turn.cancel` carries cancellation reminder intent + * atomically, while `turn.outcome` records failure intent; a + * `context.append_message` acknowledgement consumes the intent after the + * reminder reaches model context. As a belt-and-suspenders for v1-written logs + * whose internally-driven turns (goal continuations) have no `turn.prompt` record, * `TurnModel` also registers a cross-model reducer on * `context.append_loop_event` that raises the counter past any `turnId` * observed in a replayed loop event — the v1 `observeRestoredTurnId` @@ -27,22 +30,46 @@ import type { PromptOrigin } from '#/agent/contextMemory/types'; export interface TurnModelState { readonly nextTurnId: number; + readonly pendingOutcomes: readonly TurnOutcomeIntent[]; } -export const TurnModel = defineModel('turn', () => ({ nextTurnId: 0 }), { - reducers: { - 'context.append_loop_event': (state, { event }) => { - if (event.type === 'tool.result' || event.turnId === undefined) { - return state; - } +export interface TurnOutcomeIntent { + readonly outcomeId: string; + readonly turnId: number; + readonly content: string; +} + +export const TurnModel = defineModel( + 'turn', + () => ({ nextTurnId: 0, pendingOutcomes: [] }), + { + reducers: { + 'context.append_loop_event': (state, { event }) => { + if (event.type === 'tool.result' || event.turnId === undefined) { + return state; + } - const turnId = Number.parseInt(event.turnId, 10); - return Number.isInteger(turnId) && turnId >= state.nextTurnId - ? { nextTurnId: turnId + 1 } - : state; + const turnId = Number.parseInt(event.turnId, 10); + return Number.isInteger(turnId) && turnId >= state.nextTurnId + ? { ...state, nextTurnId: turnId + 1 } + : state; + }, + 'context.append_message': (state, payload) => { + const outcomeId = payload.materializedTurnOutcomeId; + if (outcomeId === undefined) return state; + const pendingOutcomes = state.pendingOutcomes.filter( + (outcome) => outcome.outcomeId !== outcomeId, + ); + return pendingOutcomes.length === state.pendingOutcomes.length + ? state + : { ...state, pendingOutcomes }; + }, + 'context.clear': (state) => clearPendingOutcomes(state), + 'context.undo': (state, payload) => + payload.count > 0 ? clearPendingOutcomes(state) : state, }, }, -}); +); const turnInputShape = { input: z.custom(), @@ -54,12 +81,13 @@ declare module '#/wire/types' { 'turn.prompt': typeof promptTurn; 'turn.steer': typeof steerTurn; 'turn.cancel': typeof cancelTurn; + 'turn.outcome': typeof turnOutcome; } } export const promptTurn = TurnModel.defineOp('turn.prompt', { schema: z.object(turnInputShape), - apply: (s) => ({ nextTurnId: s.nextTurnId + 1 }), + apply: (s) => ({ ...s, nextTurnId: s.nextTurnId + 1 }), }); export const steerTurn = TurnModel.defineOp('turn.steer', { @@ -68,6 +96,54 @@ export const steerTurn = TurnModel.defineOp('turn.steer', { }); export const cancelTurn = TurnModel.defineOp('turn.cancel', { - schema: z.object({ turnId: z.number().optional() }), - apply: (s) => s, + schema: z.object({ + turnId: z.number().optional(), + outcomeId: z.string().optional(), + outcomeTurnId: z.number().optional(), + outcomeContent: z.string().optional(), + }), + apply: (s, payload) => addOutcomeFromCancel(s, payload), }); + +export const turnOutcome = TurnModel.defineOp('turn.outcome', { + schema: z.object({ + outcomeId: z.string(), + turnId: z.number(), + content: z.string(), + }), + apply: (s, payload) => addPendingOutcome(s, payload), +}); + +function addOutcomeFromCancel( + state: TurnModelState, + payload: { + readonly outcomeId?: string; + readonly outcomeTurnId?: number; + readonly outcomeContent?: string; + }, +): TurnModelState { + if ( + payload.outcomeId === undefined || + payload.outcomeTurnId === undefined || + payload.outcomeContent === undefined + ) { + return state; + } + return addPendingOutcome(state, { + outcomeId: payload.outcomeId, + turnId: payload.outcomeTurnId, + content: payload.outcomeContent, + }); +} + +function addPendingOutcome( + state: TurnModelState, + outcome: TurnOutcomeIntent, +): TurnModelState { + if (state.pendingOutcomes.some((pending) => pending.turnId === outcome.turnId)) return state; + return { ...state, pendingOutcomes: [...state.pendingOutcomes, outcome] }; +} + +function clearPendingOutcomes(state: TurnModelState): TurnModelState { + return state.pendingOutcomes.length === 0 ? state : { ...state, pendingOutcomes: [] }; +} diff --git a/packages/agent-core-v2/src/agent/loop/turnOutcomeReminder.ts b/packages/agent-core-v2/src/agent/loop/turnOutcomeReminder.ts new file mode 100644 index 0000000000..176aa6b158 --- /dev/null +++ b/packages/agent-core-v2/src/agent/loop/turnOutcomeReminder.ts @@ -0,0 +1,106 @@ +/** + * `loop` domain (L4) — renders model-visible context for abnormal turn endings. + * + * Converts terminal failures and cancellations into compact, XML-safe reminders. + * Provider failures expose only public categories and status codes; raw provider + * payloads, request identifiers, and internal infrastructure details stay out of + * model context. + */ + +import { escapeXml } from '#/_base/utils/xml-escape'; + +export const TURN_OUTCOME_REMINDER_VARIANT = 'turn_outcome'; + +const MAX_ERROR_SUMMARY_LENGTH = 500; + +interface TurnErrorSummary { + readonly code: string; + readonly message: string; + readonly name?: string; + readonly details?: Readonly>; +} + +export function renderTurnFailureReminder(error: TurnErrorSummary | undefined): string { + return [ + 'The previous turn ended before producing a final response.', + '', + `Error: ${renderPublicError(error)}`, + '', + 'The preceding user request may still be unfinished. Treat the next user message as a follow-up.', + ].join('\n'); +} + +export function renderTurnCancellationReminder(reason: unknown, userInitiated: boolean): string { + const lines = [ + userInitiated + ? 'The user interrupted the previous turn before it finished.' + : 'The previous turn was interrupted by the runtime before it finished.', + ]; + const renderedReason = userInitiated ? undefined : renderInterruptionReason(reason); + if (renderedReason !== undefined) lines.push('', `Reason: ${renderedReason}`); + lines.push( + '', + 'Some operations may already have taken effect. Treat the next user message as a follow-up, and check existing state before repeating operations.', + ); + return lines.join('\n'); +} + +function renderPublicError(error: TurnErrorSummary | undefined): string { + if (error === undefined) return 'The turn failed.'; + switch (error.code) { + case 'provider.api_error': { + const statusCode = numericDetail(error, 'statusCode'); + return statusCode === undefined + ? 'The model provider API request failed.' + : `API request failed with HTTP ${String(statusCode)}.`; + } + case 'provider.rate_limit': + return 'The model provider rate-limited the request.'; + case 'provider.auth_error': + return 'Authentication with the model provider failed.'; + case 'provider.connection_error': + return error.name?.toLowerCase().includes('timeout') === true + ? 'The model provider request timed out.' + : 'The request could not reach the model provider.'; + case 'provider.overloaded': + return 'The model provider was overloaded.'; + case 'provider.filtered': + return 'The model provider blocked the response due to its safety policy.'; + case 'context.overflow': + return 'The request exceeded the model context limit.'; + case 'loop.max_steps_exceeded': { + const maxSteps = numericDetail(error, 'maxSteps'); + return maxSteps === undefined + ? 'The turn exceeded its step limit.' + : `The turn exceeded its ${String(maxSteps)}-step limit.`; + } + case 'model.not_configured': + return 'No model is configured.'; + case 'model.config_invalid': + return 'The model configuration is invalid.'; + default: + return escapeXml(normalizeSummary(error.message) || 'The turn failed.'); + } +} + +function renderInterruptionReason(reason: unknown): string | undefined { + const raw = reason instanceof Error ? reason.message : typeof reason === 'string' ? reason : undefined; + if (raw === undefined) return undefined; + const normalized = normalizeSummary(raw); + if (normalized.length === 0 || normalized === 'Aborted' || normalized === 'This operation was aborted') { + return undefined; + } + return escapeXml(normalized); +} + +function numericDetail(error: TurnErrorSummary, key: string): number | undefined { + const value = error.details?.[key]; + return typeof value === 'number' && Number.isFinite(value) ? value : undefined; +} + +function normalizeSummary(value: string): string { + const normalized = value.replaceAll(/\s+/g, ' ').trim(); + return normalized.length <= MAX_ERROR_SUMMARY_LENGTH + ? normalized + : `${normalized.slice(0, MAX_ERROR_SUMMARY_LENGTH - 3)}...`; +} diff --git a/packages/agent-core-v2/src/agent/profile/profile.ts b/packages/agent-core-v2/src/agent/profile/profile.ts index e8dbf7ad48..62ea8a2bab 100644 --- a/packages/agent-core-v2/src/agent/profile/profile.ts +++ b/packages/agent-core-v2/src/agent/profile/profile.ts @@ -100,7 +100,7 @@ export interface IAgentProfileService { getModel(): string; useProfile(profile: ResolvedAgentProfile, context: SystemPromptContext): void; applyProfile(profile: ResolvedAgentProfile, options?: ApplyProfileOptions): Promise; - refreshSystemPrompt(): Promise; + refreshSystemPrompt(): Promise; getAgentsMdWarning(): string | undefined; data(): ProfileData; getEffectiveThinkingLevel(): ThinkingEffort; diff --git a/packages/agent-core-v2/src/agent/profile/profileService.ts b/packages/agent-core-v2/src/agent/profile/profileService.ts index 98fd708486..a82cdc22b2 100644 --- a/packages/agent-core-v2/src/agent/profile/profileService.ts +++ b/packages/agent-core-v2/src/agent/profile/profileService.ts @@ -239,18 +239,20 @@ export class AgentProfileService implements IAgentProfileService { this.publishAgentsMdWarning(); } - async refreshSystemPrompt(): Promise { + async refreshSystemPrompt(): Promise { const profile = this.resolveActiveProfile(); - if (profile === undefined) return; + if (profile === undefined) return undefined; const context = await this.buildSystemPromptContext(this.cwd); + const systemPrompt = profile.systemPrompt(context); this.activeProfile = profile; this.update({ profileName: profile.name, - systemPrompt: profile.systemPrompt(context), + systemPrompt, }); this.cacheAgentsMdWarning(context); this.publishAgentsMdWarning(); + return systemPrompt; } getAgentsMdWarning(): string | undefined { diff --git a/packages/agent-core-v2/src/agent/systemReminder/systemReminder.ts b/packages/agent-core-v2/src/agent/systemReminder/systemReminder.ts index 3ecf30eaed..4de08fbd9a 100644 --- a/packages/agent-core-v2/src/agent/systemReminder/systemReminder.ts +++ b/packages/agent-core-v2/src/agent/systemReminder/systemReminder.ts @@ -1,11 +1,15 @@ -import { createDecorator } from "#/_base/di/instantiation"; +import { createDecorator } from '#/_base/di/instantiation'; import type { ContextMessage, PromptOrigin } from '#/agent/contextMemory/types'; export interface IAgentSystemReminderService { readonly _serviceBrand: undefined; - appendSystemReminder(content: string, origin: PromptOrigin): ContextMessage; + appendSystemReminder( + content: string, + origin: PromptOrigin, + materializedTurnOutcomeId?: string, + ): ContextMessage; } export const IAgentSystemReminderService = createDecorator('agentSystemReminderService'); diff --git a/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts b/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts index d590c18254..156d725a84 100644 --- a/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts +++ b/packages/agent-core-v2/src/agent/systemReminder/systemReminderService.ts @@ -1,4 +1,4 @@ -import { Disposable } from "#/_base/di/lifecycle"; +import { Disposable } from '#/_base/di/lifecycle'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; @@ -15,7 +15,11 @@ export class AgentSystemReminderService extends Disposable implements IAgentSyst super(); } - appendSystemReminder(content: string, origin: PromptOrigin): ContextMessage { + appendSystemReminder( + content: string, + origin: PromptOrigin, + materializedTurnOutcomeId?: string, + ): ContextMessage { const message: ContextMessage = { role: 'user', content: [ @@ -27,7 +31,11 @@ export class AgentSystemReminderService extends Disposable implements IAgentSyst toolCalls: [], origin, }; - this.context.append(message); + if (materializedTurnOutcomeId === undefined) { + this.context.append(message); + } else { + this.context.appendTurnOutcome(message, materializedTurnOutcomeId); + } return message; } } diff --git a/packages/agent-core-v2/src/app/llmProtocol/generate.ts b/packages/agent-core-v2/src/app/llmProtocol/generate.ts index bc33baa7b7..67c24f02e0 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/generate.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/generate.ts @@ -1,3 +1,10 @@ +/** + * `llmProtocol` domain (L0) — streams and assembles one provider response. + * + * Owns abort-aware provider, iterator, and streamed-part callback boundaries + * with best-effort cleanup while preserving post-stream tool-call dispatch. + */ + import { APIEmptyResponseError } from './errors'; import { isContentPart, @@ -49,16 +56,20 @@ export async function generate( : tools; options?.onRequestStart?.(); - const stream = await provider.generate(systemPrompt, wireTools, history, options); + const stream = await waitForAbort( + provider.generate(systemPrompt, wireTools, history, options), + options?.signal, + { onLateResolve: requestCancellation }, + ); - await throwIfAborted(options?.signal, stream); + throwIfAborted(options?.signal, stream); let serverDecodeMs = 0; let clientConsumeMs = 0; let firstPartAt: number | undefined; let lastResumeAt = 0; - for await (const part of stream) { + for await (const part of iterateWithAbort(stream, options?.signal)) { const arrivedAt = Date.now(); if (firstPartAt === undefined) { firstPartAt = arrivedAt; @@ -67,11 +78,11 @@ export async function generate( } try { - await throwIfAborted(options?.signal, stream); + throwIfAborted(options?.signal); if (callbacks?.onMessagePart !== undefined) { - await callbacks.onMessagePart(deepCopyPart(part)); - await throwIfAborted(options?.signal, stream); + await waitForAbort(callbacks.onMessagePart(deepCopyPart(part)), options?.signal); + throwIfAborted(options?.signal); } if ( @@ -104,7 +115,7 @@ export async function generate( } } - await throwIfAborted(options?.signal, stream); + throwIfAborted(options?.signal, stream); if (firstPartAt !== undefined) { serverDecodeMs += Date.now() - lastResumeAt; } @@ -146,9 +157,9 @@ export async function generate( ); } - if (callbacks?.onToolCall !== undefined) { + if (callbacks?.onToolCall !== undefined && message.toolCalls.length > 0) { + throwIfAborted(options?.signal, stream); for (const toolCall of message.toolCalls) { - await throwIfAborted(options?.signal, stream); await callbacks.onToolCall(toolCall); } } @@ -162,34 +173,144 @@ export async function generate( }; } -type CancelableStream = StreamedMessage & { +type CancelableTarget = { cancel?: () => unknown; return?: () => unknown; }; function throwAbortError(): never { - throw new DOMException('The operation was aborted.', 'AbortError'); + throw createAbortError(); } -async function cancelStream(stream: StreamedMessage): Promise { - const cancelable = stream as CancelableStream; +function requestCancellation(target: unknown): void { + if (target === null || (typeof target !== 'object' && typeof target !== 'function')) { + return; + } + const cancelable = target as CancelableTarget; + invokeCancellation(cancelable, 'cancel'); + invokeCancellation(cancelable, 'return'); +} +function invokeCancellation(target: CancelableTarget, method: keyof CancelableTarget): void { try { - await cancelable.cancel?.(); + const cleanup = target[method]?.(); + void Promise.resolve(cleanup).catch(() => {}); } catch {} +} + +interface AbortWaitOptions { + readonly onAbort?: () => void; + readonly onLateResolve?: (value: T) => void; +} + +function waitForAbort( + value: T | PromiseLike, + signal?: AbortSignal, + options?: AbortWaitOptions, +): Promise { + const promise = Promise.resolve(value); + if (signal === undefined) { + return promise; + } + + return new Promise((resolve, reject) => { + let settled = false; + const removeAbortListener = (): void => { + signal.removeEventListener('abort', onAbort); + }; + const onAbort = (): void => { + if (settled) { + return; + } + settled = true; + removeAbortListener(); + try { + options?.onAbort?.(); + } catch {} + reject(createAbortError()); + }; + + if (signal.aborted) { + onAbort(); + } else { + signal.addEventListener('abort', onAbort, { once: true }); + } + + void promise.then( + (result) => { + if (settled) { + try { + options?.onLateResolve?.(result); + } catch {} + return; + } + settled = true; + removeAbortListener(); + resolve(result); + }, + (error: unknown) => { + if (settled) { + return; + } + settled = true; + removeAbortListener(); + reject(error); + }, + ); + }); +} + +async function* iterateWithAbort( + stream: StreamedMessage, + signal?: AbortSignal, +): AsyncGenerator { + const iterator = stream[Symbol.asyncIterator](); + let completed = false; + let cancellationRequested = false; + const cancel = (): void => { + if (cancellationRequested) { + return; + } + cancellationRequested = true; + invokeCancellation(stream as CancelableTarget, 'cancel'); + invokeCancellation(iterator as CancelableTarget, 'return'); + }; try { - await cancelable.return?.(); - } catch {} + while (true) { + if (signal?.aborted) { + cancel(); + throwAbortError(); + } + const next = await waitForAbort(iterator.next(), signal, { onAbort: cancel }); + if (signal?.aborted) { + cancel(); + throwAbortError(); + } + if (next.done === true) { + completed = true; + return; + } + yield next.value; + } + } finally { + if (!completed) { + cancel(); + } + } +} + +function createAbortError(): DOMException { + return new DOMException('The operation was aborted.', 'AbortError'); } -async function throwIfAborted(signal?: AbortSignal, stream?: StreamedMessage): Promise { +function throwIfAborted(signal?: AbortSignal, stream?: StreamedMessage): void { if (!signal?.aborted) { return; } if (stream !== undefined) { - await cancelStream(stream); + requestCancellation(stream); } throwAbortError(); diff --git a/packages/agent-core-v2/src/app/llmProtocol/provider.ts b/packages/agent-core-v2/src/app/llmProtocol/provider.ts index 04547944fb..1467eebd21 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/provider.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/provider.ts @@ -44,6 +44,7 @@ export type FinishReason = export interface StreamedMessage { [Symbol.asyncIterator](): AsyncIterator; + cancel?(): void; readonly id: string | null; readonly usage: TokenUsage | null; readonly finishReason: FinishReason | null; diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts index c9a6f297c3..fdfe9db210 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/anthropic.ts @@ -48,6 +48,7 @@ import { sanitizeToolCallId, type ToolCallIdPolicy, } from './tool-call-id'; +import { cancelNativeStream } from './stream-cancellation'; function normalizeAnthropicStopReason(raw: string | null | undefined): { finishReason: FinishReason | null; @@ -586,8 +587,10 @@ class AnthropicStreamedMessage implements StreamedMessage { private _finishReason: FinishReason | null = null; private _rawFinishReason: string | null = null; private readonly _iter: AsyncGenerator; + private readonly _nativeStream: unknown; constructor(response: unknown, isStream: boolean) { + this._nativeStream = isStream ? response : undefined; if (isStream) { this._iter = this._convertStreamResponse(response as AsyncIterable); } else { @@ -632,6 +635,10 @@ class AnthropicStreamedMessage implements StreamedMessage { return this._rawFinishReason; } + cancel(): void { + cancelNativeStream(this._nativeStream); + } + async *[Symbol.asyncIterator](): AsyncIterator { yield* this._iter; } diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/google-genai.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/google-genai.ts index 893bf9d93c..cd70b03284 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/google-genai.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/google-genai.ts @@ -1,3 +1,10 @@ +/** + * `llmProtocol` domain (L0) — adapts Google GenAI requests and responses. + * + * Owns protocol translation, provider error normalization, and composition of + * configured, per-request, and transport cancellation signals. Owns no scoped state. + */ + import { APIConnectionError, APITimeoutError, @@ -21,6 +28,7 @@ import { ApiError as GoogleApiError, GoogleGenAI as GenAIClient } from '@google/ import { mergeConsecutiveUserMessages } from './merge-user-messages'; import { requireProviderApiKey, resolveAuthBackedClient } from './request-auth'; +import { cancelNativeStream } from './stream-cancellation'; function normalizeGoogleGenAIFinishReason(raw: unknown): { finishReason: FinishReason | null; @@ -185,23 +193,13 @@ function createAbortError(): DOMException { return new DOMException('The operation was aborted.', 'AbortError'); } -async function abortPromise(signal: AbortSignal | undefined): Promise { - if (signal === undefined) { - return new Promise(() => { - }); - } - if (signal.aborted) { - throw createAbortError(); - } - return new Promise((_, reject) => { - signal.addEventListener( - 'abort', - () => { - reject(createAbortError()); - }, - { once: true }, - ); - }); +function mergeAbortSignals( + ...signals: Array +): AbortSignal { + const definedSignals = signals.filter( + (signal): signal is AbortSignal => signal !== null && signal !== undefined, + ); + return AbortSignal.any(definedSignals); } function messageToGoogleGenAI(message: Message): GoogleContent { @@ -428,12 +426,15 @@ export class GoogleGenAIStreamedMessage implements StreamedMessage { private _finishReason: FinishReason | null = null; private _rawFinishReason: string | null = null; private readonly _iter: AsyncGenerator; + private readonly _nativeStream: unknown; constructor( response: AsyncIterable> | Record, isStream: boolean, signal?: AbortSignal, + private readonly _transportController?: AbortController, ) { + this._nativeStream = isStream ? response : undefined; if (isStream) { this._iter = this._convertStreamResponse( response as AsyncIterable>, @@ -460,6 +461,13 @@ export class GoogleGenAIStreamedMessage implements StreamedMessage { return this._rawFinishReason; } + cancel(): void { + try { + this._transportController?.abort(); + } catch {} + cancelNativeStream(this._nativeStream); + } + async *[Symbol.asyncIterator](): AsyncIterator { yield* this._iter; } @@ -730,10 +738,24 @@ export class GoogleGenAIChatProvider implements ChatProvider { } const contents = messagesToGoogleGenAIContents(history); + const transportController = new AbortController(); + const configuredSignal = this._generationKwargs['abortSignal'] as + | AbortSignal + | null + | undefined; + const requestSignal = mergeAbortSignals( + configuredSignal, + options?.signal, + transportController.signal, + ); + if (requestSignal.aborted) { + throw createAbortError(); + } const config: Record = { ...this._generationKwargs, systemInstruction: systemPrompt, + abortSignal: requestSignal, ...(tools.length > 0 ? { tools: tools.map((t) => toolToGoogleGenAI(t)) } : {}), }; applyResponseFormat(config, options?.responseFormat); @@ -749,25 +771,21 @@ export class GoogleGenAIChatProvider implements ChatProvider { options?.onRequestSent?.(); if (this._stream) { - const stream = await Promise.race([ - models.generateContentStream(params), - abortPromise(options?.signal), - ]); + const stream = await models.generateContentStream(params); return new GoogleGenAIStreamedMessage( stream as AsyncIterable>, true, - options?.signal, + requestSignal, + transportController, ); } - const response = await Promise.race([ - models.generateContent(params), - abortPromise(options?.signal), - ]); + const response = await models.generateContent(params); return new GoogleGenAIStreamedMessage( response as Record, false, - options?.signal, + requestSignal, + transportController, ); } catch (error: unknown) { if (error instanceof DOMException && error.name === 'AbortError') { diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/kimi.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/kimi.ts index c77e6fc106..b71b4b15cc 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/kimi.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/kimi.ts @@ -40,6 +40,7 @@ import { sanitizeToolCallId, type ToolCallIdPolicy, } from './tool-call-id'; +import { cancelNativeStream } from './stream-cancellation'; export interface KimiOptions { apiKey?: string | undefined; baseUrl?: string | undefined; @@ -228,11 +229,13 @@ class KimiStreamedMessage implements StreamedMessage { private _finishReason: FinishReason | null = null; private _rawFinishReason: string | null = null; private readonly _iter: AsyncGenerator; + private readonly _nativeStream: unknown; constructor( response: OpenAI.Chat.ChatCompletion | AsyncIterable, isStream: boolean, ) { + this._nativeStream = isStream ? response : undefined; if (isStream) { this._iter = this._convertStreamResponse( response as AsyncIterable, @@ -258,6 +261,10 @@ class KimiStreamedMessage implements StreamedMessage { return this._rawFinishReason; } + cancel(): void { + cancelNativeStream(this._nativeStream); + } + async *[Symbol.asyncIterator](): AsyncIterator { yield* this._iter; } diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-legacy.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-legacy.ts index 421cc2f310..ecae138641 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-legacy.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-legacy.ts @@ -41,6 +41,7 @@ import { sanitizeToolCallId, type ToolCallIdPolicy, } from './tool-call-id'; +import { cancelNativeStream } from './stream-cancellation'; const KNOWN_REASONING_KEYS = ['reasoning_content', 'reasoning_details', 'reasoning'] as const; const DEFAULT_OUTBOUND_REASONING_KEY = KNOWN_REASONING_KEYS[0]; @@ -295,12 +296,14 @@ export class OpenAILegacyStreamedMessage implements StreamedMessage { private _finishReason: FinishReason | null = null; private _rawFinishReason: string | null = null; private readonly _iter: AsyncGenerator; + private readonly _nativeStream: unknown; constructor( response: OpenAI.Chat.ChatCompletion | AsyncIterable, isStream: boolean, reasoningKey: string | undefined, ) { + this._nativeStream = isStream ? response : undefined; if (isStream) { this._iter = this._convertStreamResponse( response as AsyncIterable, @@ -330,6 +333,10 @@ export class OpenAILegacyStreamedMessage implements StreamedMessage { return this._rawFinishReason; } + cancel(): void { + cancelNativeStream(this._nativeStream); + } + async *[Symbol.asyncIterator](): AsyncIterator { yield* this._iter; } diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts index 9953a0d148..c890a4193f 100644 --- a/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/openai-responses.ts @@ -37,6 +37,7 @@ import { sanitizeOpenAIResponsesCallId, type ToolCallIdPolicy, } from './tool-call-id'; +import { cancelNativeStream } from './stream-cancellation'; function normalizeResponsesFinishReason( status: string | null | undefined, @@ -621,8 +622,10 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { private _finishReason: FinishReason | null = null; private _rawFinishReason: string | null = null; private readonly _iter: AsyncGenerator; + private readonly _nativeStream: unknown; constructor(response: unknown, isStream: boolean) { + this._nativeStream = isStream ? response : undefined; if (isStream) { this._iter = this._convertStreamResponse(response as AsyncIterable); } else { @@ -646,6 +649,10 @@ export class OpenAIResponsesStreamedMessage implements StreamedMessage { return this._rawFinishReason; } + cancel(): void { + cancelNativeStream(this._nativeStream); + } + async *[Symbol.asyncIterator](): AsyncIterator { yield* this._iter; } diff --git a/packages/agent-core-v2/src/app/llmProtocol/providers/stream-cancellation.ts b/packages/agent-core-v2/src/app/llmProtocol/providers/stream-cancellation.ts new file mode 100644 index 0000000000..40f9547fd0 --- /dev/null +++ b/packages/agent-core-v2/src/app/llmProtocol/providers/stream-cancellation.ts @@ -0,0 +1,28 @@ +/** + * `llmProtocol` domain (L0) — best-effort native stream transport cancellation. + * + * Bridges provider SDK stream handles to the protocol-level cancellation + * contract. Owns no scoped state. + */ + +type NativeCancelableStream = { + abort?: () => unknown; + controller?: { + abort(): unknown; + }; +}; + +export function cancelNativeStream(stream: unknown): void { + if (stream === null || (typeof stream !== 'object' && typeof stream !== 'function')) { + return; + } + + const cancelable = stream as NativeCancelableStream; + try { + if (typeof cancelable.abort === 'function') { + void Promise.resolve(cancelable.abort()).catch(() => {}); + return; + } + void Promise.resolve(cancelable.controller?.abort()).catch(() => {}); + } catch {} +} diff --git a/packages/agent-core-v2/src/app/telemetry/telemetryService.ts b/packages/agent-core-v2/src/app/telemetry/telemetryService.ts index e5eb2d68d1..36b9ed3b01 100644 --- a/packages/agent-core-v2/src/app/telemetry/telemetryService.ts +++ b/packages/agent-core-v2/src/app/telemetry/telemetryService.ts @@ -55,7 +55,14 @@ export class TelemetryService implements ITelemetryService { withContext(patch: TelemetryContextPatch): ITelemetryService { const child = new TelemetryService(); - child.appenders = this.appenders.map((appender) => appender.withContext?.(patch) ?? appender); + child.appenders = this.appenders.map((appender) => { + try { + return appender.withContext?.(patch) ?? appender; + } catch (error) { + onUnexpectedError(error); + return appender; + } + }); child.context = { ...this.context, ...patch }; child.enabled = this.enabled; return child; @@ -64,7 +71,11 @@ export class TelemetryService implements ITelemetryService { setContext(patch: TelemetryContextPatch): void { this.context = { ...this.context, ...patch }; for (const appender of this.appenders) { - appender.setContext?.(patch); + try { + appender.setContext?.(patch); + } catch (error) { + onUnexpectedError(error); + } } } diff --git a/packages/agent-core-v2/test/_base/errors/errors.test.ts b/packages/agent-core-v2/test/_base/errors/errors.test.ts index 67eabc4ffa..43b3317e67 100644 --- a/packages/agent-core-v2/test/_base/errors/errors.test.ts +++ b/packages/agent-core-v2/test/_base/errors/errors.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { onUnexpectedError, @@ -43,6 +43,21 @@ describe('onUnexpectedError + setUnexpectedErrorHandler', () => { expect(() => onUnexpectedError(new Error('original'))).not.toThrow(); }); + it('does not propagate when the handler and console fallback both throw', () => { + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => { + throw new Error('console failed'); + }); + setUnexpectedErrorHandler(() => { + throw new Error('handler failed'); + }); + + try { + expect(() => onUnexpectedError(new Error('original'))).not.toThrow(); + } finally { + consoleError.mockRestore(); + } + }); + it('resetUnexpectedErrorHandler restores the module default', () => { const seen: unknown[] = []; setUnexpectedErrorHandler((err) => seen.push(err)); diff --git a/packages/agent-core-v2/test/_base/log/logService.test.ts b/packages/agent-core-v2/test/_base/log/logService.test.ts index 24865b14c6..b64e0954b3 100644 --- a/packages/agent-core-v2/test/_base/log/logService.test.ts +++ b/packages/agent-core-v2/test/_base/log/logService.test.ts @@ -91,6 +91,34 @@ describe('BoundLogger', () => { expect(sink.entries.map((e) => e.msg)).not.toContain('proxy payload'); }); + it('does not let throwing Error accessors escape into caller flow', () => { + const hostile = new Error('placeholder'); + Object.defineProperty(hostile, 'message', { + get() { + throw new Error('message getter failed'); + }, + }); + Object.defineProperty(hostile, 'stack', { + get() { + throw new Error('stack getter failed'); + }, + }); + + expect(() => logger.error('hostile error', hostile)).not.toThrow(); + expect(sink.entries.map((e) => e.msg)).not.toContain('hostile error'); + }); + + it('does not let a throwing writer escape into caller flow', () => { + const brokenWriter = { + write: () => { + throw new Error('sink failed'); + }, + }; + const brokenLogger = new BoundLogger(brokenWriter, { level: 'info' }); + + expect(() => brokenLogger.warn('diagnostic only')).not.toThrow(); + }); + it('merges object payload into ctx', () => { const debugLogger = new BoundLogger(sink, { level: 'debug' }); debugLogger.info('with ctx', { requestId: 'r1', count: 2 }); @@ -112,6 +140,24 @@ describe('BoundLogger', () => { leaf.info('evt'); expect(sink.entries[0]?.ctx).toEqual({ a: 1, b: 2 }); }); + + it('creates a usable child when binding accessors throw', () => { + const hostileContext = new Proxy( + {}, + { + ownKeys() { + throw new Error('binding ownKeys failed'); + }, + }, + ); + + let child!: ReturnType; + expect(() => { + child = logger.child(hostileContext); + }).not.toThrow(); + expect(() => child.warn('child survived')).not.toThrow(); + expect(sink.entries.at(-1)?.msg).toBe('child survived'); + }); }); describe('levelEnabled', () => { diff --git a/packages/agent-core-v2/test/agent/contextInjector/contextInjector.test.ts b/packages/agent-core-v2/test/agent/contextInjector/contextInjector.test.ts index eb748ba943..053484fe7c 100644 --- a/packages/agent-core-v2/test/agent/contextInjector/contextInjector.test.ts +++ b/packages/agent-core-v2/test/agent/contextInjector/contextInjector.test.ts @@ -311,4 +311,25 @@ describe('AgentContextInjectorService', () => { { kind: 'injection', variant: 'per_turn_test' }, ]); }); + + it('does not duplicate a surviving injection when post-compaction work retries', async () => { + let calls = 0; + injector(ix).register('retry_test', () => { + calls += 1; + return 'retry-safe reminder'; + }); + + await injector(ix).injectAfterCompaction(); + await injector(ix).injectAfterCompaction(); + + expect(calls).toBe(2); + expect( + context + .get() + .filter( + (message) => + message.origin?.kind === 'injection' && message.origin.variant === 'retry_test', + ), + ).toHaveLength(1); + }); }); diff --git a/packages/agent-core-v2/test/agent/contextMemory/context.test.ts b/packages/agent-core-v2/test/agent/contextMemory/context.test.ts index 859ead8399..a933c1ef48 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/context.test.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/context.test.ts @@ -116,6 +116,63 @@ describe('Agent context', () => { ]); }); + it('closes an abandoned tool exchange before releasing deferred reminders', () => { + context.appendLoopEvent({ type: 'step.begin', uuid: 'step_abandoned' }); + context.appendLoopEvent({ + type: 'tool.call', + stepUuid: 'step_abandoned', + toolCallId: 'call_abandoned', + name: 'Run', + args: {}, + }); + context.appendLoopEvent({ + type: 'tool.call', + stepUuid: 'step_abandoned', + toolCallId: 'call_also_abandoned', + name: 'Read', + args: {}, + }); + ctx.appendSystemReminder('Turn failed.', { + kind: 'injection', + variant: 'turn_outcome', + }); + + expect(context.get().map((message) => message.role)).toEqual(['assistant']); + expect( + context.closeAbandonedToolExchange( + 'Tool call did not complete because the turn failed.', + ), + ).toBe(2); + + expect(context.get()).toMatchObject([ + { + role: 'assistant', + toolCalls: [ + { id: 'call_abandoned', name: 'Run' }, + { id: 'call_also_abandoned', name: 'Read' }, + ], + }, + { + role: 'tool', + toolCallId: 'call_abandoned', + isError: true, + content: [{ type: 'text', text: 'Tool call did not complete because the turn failed.' }], + }, + { + role: 'tool', + toolCallId: 'call_also_abandoned', + isError: true, + content: [{ type: 'text', text: 'Tool call did not complete because the turn failed.' }], + }, + { + role: 'user', + origin: { kind: 'injection', variant: 'turn_outcome' }, + content: [{ type: 'text', text: '\nTurn failed.\n' }], + }, + ]); + expect(context.closeAbandonedToolExchange('unused')).toBe(0); + }); + it('drops empty text parts only in LLM projection', () => { const history: ContextMessage[] = [ { @@ -663,6 +720,104 @@ describe('Agent context', () => { ]); }); + it('keeps the latest unseen turn outcome intact before an oversized follow-up', () => { + const outcomeText = 'The previous turn was interrupted before it completed.'; + context.append( + userMessage('old request', { kind: 'user' }), + { + role: 'assistant', + content: [{ type: 'text', text: 'partial answer' }], + toolCalls: [], + }, + userMessage(outcomeText, { kind: 'injection', variant: 'turn_outcome' }), + userMessage(`follow-up ${'f'.repeat(100_000)}`, { kind: 'user' }), + ); + + const result = context.applyCompaction({ + summary: 'summary', + compactedCount: context.get().length, + tokensBefore: estimateTokensForMessages(context.get()), + }); + + const messages = context.get(); + const outcomes = messages.filter( + (message) => + message.origin?.kind === 'injection' && message.origin.variant === 'turn_outcome', + ); + const outcomeIndex = messages.indexOf(outcomes[0]!); + const followUpIndex = messages.findIndex( + (message) => message.origin?.kind === 'user' && textOf(message).startsWith('follow-up '), + ); + expect(outcomes).toHaveLength(1); + expect(textOf(outcomes[0]!)).toBe(outcomeText); + expect(outcomeIndex).toBeLessThan(followUpIndex); + expect(result.keptTurnOutcomeCount).toBe(1); + expect(result.keptUserMessageCount).toBe(3); + }); + + it('keeps only the latest unseen turn outcome and ignores hook assistant messages', () => { + context.append( + userMessage('request', { kind: 'user' }), + { + role: 'assistant', + content: [{ type: 'text', text: 'model answer' }], + toolCalls: [], + }, + userMessage('older pending outcome', { kind: 'injection', variant: 'turn_outcome' }), + userMessage('latest pending outcome', { kind: 'injection', variant: 'turn_outcome' }), + { + role: 'assistant', + content: [{ type: 'text', text: 'hook blocked without an LLM request' }], + toolCalls: [], + origin: { kind: 'hook_result', event: 'UserPromptSubmit', blocked: true }, + }, + userMessage('continue', { kind: 'user' }), + ); + + context.applyCompaction({ + summary: 'summary', + compactedCount: context.get().length, + tokensBefore: estimateTokensForMessages(context.get()), + }); + + expect( + context + .get() + .filter( + (message) => + message.origin?.kind === 'injection' && message.origin.variant === 'turn_outcome', + ) + .map(textOf), + ).toEqual(['latest pending outcome']); + }); + + it('drops a turn outcome that a later model assistant has already seen', () => { + context.append( + userMessage('request', { kind: 'user' }), + userMessage('seen outcome', { kind: 'injection', variant: 'turn_outcome' }), + { + role: 'assistant', + content: [{ type: 'text', text: 'model continued' }], + toolCalls: [], + }, + userMessage('next request', { kind: 'user' }), + ); + + const result = context.applyCompaction({ + summary: 'summary', + compactedCount: context.get().length, + tokensBefore: estimateTokensForMessages(context.get()), + }); + + expect( + context.get().some( + (message) => + message.origin?.kind === 'injection' && message.origin.variant === 'turn_outcome', + ), + ).toBe(false); + expect(result.keptTurnOutcomeCount).toBeUndefined(); + }); + describe('notification projection', () => { it('does not merge a cron-fire envelope into an adjacent user message', () => { const cronEnvelope = diff --git a/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts b/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts index d6e2170d7a..ef4ae6b62b 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/contextTranscript.test.ts @@ -29,6 +29,10 @@ function assistantMessage(text: string): ContextMessage { return { role: 'assistant', content: [{ type: 'text', text }], toolCalls: [] }; } +function turnOutcomeMessage(text: string): ContextMessage { + return userMessage(text, { kind: 'injection', variant: 'turn_outcome' }); +} + function appendMessage(message: ContextMessage): WireRecord { return { type: 'context.append_message', message }; } @@ -50,6 +54,7 @@ function compaction( compactedCount: number, keptUserMessageCount?: number, keptHeadUserMessageCount?: number, + keptTurnOutcomeCount?: number, ): WireRecord { return { type: 'context.apply_compaction', @@ -60,6 +65,7 @@ function compaction( tokensAfter: 100, ...(keptUserMessageCount === undefined ? {} : { keptUserMessageCount }), ...(keptHeadUserMessageCount === undefined ? {} : { keptHeadUserMessageCount }), + ...(keptTurnOutcomeCount === undefined ? {} : { keptTurnOutcomeCount }), }; } @@ -120,6 +126,17 @@ describe('reduceContextTranscript', () => { expect(result.foldedLength).toBe(4); }); + it('accounts for a protected turn outcome separately from real user messages', () => { + const result = reduceContextTranscript([ + appendMessage(userMessage('u1')), + appendMessage(turnOutcomeMessage('outcome')), + appendMessage(userMessage('u2')), + compaction('SUM', 3, 2, undefined, 1), + ]); + + expect(result.foldedLength).toBe(4); + }); + it('carries the originating wire record time per entry', () => { const result = reduceContextTranscript([ { type: 'context.append_message', message: userMessage('u1'), time: 100 }, @@ -172,7 +189,44 @@ describe('reduceContextTranscript', () => { expect(texts(result)).toEqual(['message A', 'reply A']); }); - it('undo stops at a compaction summary', () => { + it('removes a turn outcome reminder when its user prompt is undone', () => { + const result = reduceContextTranscript([ + appendMessage(userMessage('message A')), + appendMessage(turnOutcomeMessage('outcome A')), + undo(1), + ]); + + expect(texts(result)).toEqual([]); + expect(result.foldedLength).toBe(0); + }); + + it('keeps the preceding turn outcome reminder when only the latest prompt is undone', () => { + const result = reduceContextTranscript([ + appendMessage(userMessage('message A')), + appendMessage(turnOutcomeMessage('outcome A')), + appendMessage(userMessage('message B')), + appendMessage(turnOutcomeMessage('outcome B')), + undo(1), + ]); + + expect(texts(result)).toEqual(['message A', 'outcome A']); + expect(result.foldedLength).toBe(2); + }); + + it('removes the full two-turn tail when undoing two prompts', () => { + const result = reduceContextTranscript([ + appendMessage(userMessage('message A')), + appendMessage(turnOutcomeMessage('outcome A')), + appendMessage(userMessage('message B')), + appendMessage(turnOutcomeMessage('outcome B')), + undo(2), + ]); + + expect(texts(result)).toEqual([]); + expect(result.foldedLength).toBe(0); + }); + + it('leaves the transcript unchanged when the requested undo crosses a compaction boundary', () => { const result = reduceContextTranscript([ appendMessage(userMessage('old')), compaction('SUM', 1, 1), @@ -180,7 +234,9 @@ describe('reduceContextTranscript', () => { appendMessage(assistantMessage('answer')), undo(2), ]); - expect(texts(result)).toEqual(['old', 'SUM']); + + expect(texts(result)).toEqual(['old', 'SUM', 'recent', 'answer']); + expect(result.foldedLength).toBe(4); }); it('clear keeps prior transcript entries but resets the folded view', () => { @@ -206,6 +262,19 @@ describe('reduceContextTranscript', () => { expect(result.foldedLength).toBe(0); }); + it('leaves the transcript unchanged when the requested undo crosses a clear floor', () => { + const result = reduceContextTranscript([ + appendMessage(userMessage('u1')), + { type: 'context.clear' }, + appendMessage(userMessage('u2')), + appendMessage(assistantMessage('a2')), + undo(2), + ]); + + expect(texts(result)).toEqual(['u1', 'u2', 'a2']); + expect(result.foldedLength).toBe(2); + }); + it('folds tool calls and results from loop events', () => { const result = reduceContextTranscript([ appendMessage(userMessage('q')), diff --git a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts index ba6aef5624..50be0d211a 100644 --- a/packages/agent-core-v2/test/agent/contextMemory/stubs.ts +++ b/packages/agent-core-v2/test/agent/contextMemory/stubs.ts @@ -52,7 +52,13 @@ export function stubContextMemory(eventBus?: IEventBus): StubContextMemory { messages.push(...inserted); publishSplice(eventBus, { start, deleteCount: 0, messages: [...inserted] }); }, + appendTurnOutcome: (message) => { + const start = messages.length; + messages.push(message); + publishSplice(eventBus, { start, deleteCount: 0, messages: [message] }); + }, appendLoopEvent: () => {}, + closeAbandonedToolExchange: () => 0, clear: () => { const deleteCount = messages.length; if (deleteCount === 0) return; @@ -100,12 +106,18 @@ class StubContextMemoryService implements IAgentContextMemoryService { append(...messages: readonly ContextMessage[]): void { this.impl.append(...messages); } + appendTurnOutcome(message: ContextMessage, outcomeId: string): void { + this.impl.appendTurnOutcome(message, outcomeId); + } clear(): void { this.impl.clear(); } appendLoopEvent(event: LoopRecordedEvent): void { this.impl.appendLoopEvent(event); } + closeAbandonedToolExchange(output: string): number { + return this.impl.closeAbandonedToolExchange(output); + } undo(count: number): UndoCut { return this.impl.undo(count); } diff --git a/packages/agent-core-v2/test/agent/fullCompaction/compactionOps.test.ts b/packages/agent-core-v2/test/agent/fullCompaction/compactionOps.test.ts index 167d5d656e..97258a29a0 100644 --- a/packages/agent-core-v2/test/agent/fullCompaction/compactionOps.test.ts +++ b/packages/agent-core-v2/test/agent/fullCompaction/compactionOps.test.ts @@ -10,7 +10,9 @@ import { fullCompactionBegin, fullCompactionCancel, fullCompactionComplete, + fullCompactionFail, } from '#/agent/fullCompaction/compactionOps'; +import { contextApplyCompaction } from '#/agent/contextMemory/contextOps'; import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; @@ -58,7 +60,7 @@ async function readRecords(key = KEY): Promise { } describe('fullCompaction ops (wire-backed)', () => { - it('begin/complete/cancel drive the phase and persist flat records', async () => { + it('begin/complete/cancel/fail drive the phase and persist flat records', async () => { expect(wire.getModel(CompactionModel).phase).toBe('idle'); wire.dispatch(fullCompactionBegin({ source: 'manual', instruction: 'keep facts' })); @@ -72,12 +74,19 @@ describe('fullCompaction ops (wire-backed)', () => { wire.dispatch(fullCompactionCancel({})); expect(wire.getModel(CompactionModel).phase).toBe('idle'); + wire.dispatch(fullCompactionBegin({ source: 'manual' })); + expect(wire.getModel(CompactionModel).phase).toBe('running'); + wire.dispatch(fullCompactionFail()); + expect(wire.getModel(CompactionModel).phase).toBe('idle'); + const records = await readRecords(); expect(records.map((record) => record.type)).toEqual([ 'full_compaction.begin', 'full_compaction.complete', 'full_compaction.begin', 'full_compaction.cancel', + 'full_compaction.begin', + 'full_compaction.complete', ]); expect(records.every((record) => 'payload' in record === false)).toBe(true); expect(records[0]).toEqual( @@ -88,6 +97,22 @@ describe('fullCompaction ops (wire-backed)', () => { }), ); expect(records[1]).toEqual({ type: 'full_compaction.complete', time: expect.any(Number) }); + expect(records.at(-1)).toEqual({ + type: 'full_compaction.complete', + outcome: 'failed', + time: expect.any(Number), + }); + }); + + it('treats context replacement as the durable commit point', () => { + wire.dispatch(fullCompactionBegin({ source: 'manual' })); + wire.dispatch(contextApplyCompaction({ + summary: 'Committed summary', + compactedCount: 0, + tokensBefore: 0, + })); + + expect(wire.getModel(CompactionModel).phase).toBe('committed'); }); it('apply returns the same reference on a no-op (gate stays quiet)', () => { diff --git a/packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts b/packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts index b5f4d8f4fe..a59c09477c 100644 --- a/packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts +++ b/packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts @@ -36,7 +36,15 @@ import { MASTER_ENV } from '#/app/flag/flagService'; import { estimateTokensForMessages } from '#/_base/utils/tokens'; import { recordingTelemetry, type TelemetryRecord } from '../../app/telemetry/stubs'; import type { TestAgentContext, TestAgentOptions, TestAgentServiceOverride } from '../../harness'; -import { appServices, createCommandRunner, execEnvServices, hostEnvironmentServices, sessionServices, testAgent } from '../../harness'; +import { + appServices, + createCommandRunner, + execEnvServices, + hostEnvironmentServices, + logServices, + sessionServices, + testAgent, +} from '../../harness'; import { IAgentFullCompactionService, IOAuthService, @@ -50,7 +58,10 @@ import { } from '#/index'; import { IAgentLoopService } from '#/agent/loop/loop'; import { IAgentContextSizeService } from '#/agent/contextSize/contextSize'; +import { IAgentContextInjectorService } from '#/agent/contextInjector/contextInjector'; import { IAgentGoalService } from '#/agent/goal/goal'; +import { CompactionModel } from '#/agent/fullCompaction/compactionOps'; +import { IWireService } from '#/wire/wire'; import { HostFileSystem } from '#/os/backends/node-local/hostFsService'; type GenerateFn = NonNullable; @@ -354,6 +365,77 @@ describe('FullCompaction', () => { } }); + it('uses the captured compaction prompt when retrying the same turn', async () => { + const homeDir = mkdtempSync(join(tmpdir(), 'kimi-compact-turn-refresh-home-')); + const workDir = mkdtempSync(join(tmpdir(), 'kimi-compact-turn-refresh-work-')); + try { + writeFileSync(join(workDir, 'AGENTS.md'), 'old project instructions', 'utf-8'); + let callCount = 0; + const systemPrompts: string[] = []; + const modelNames: string[] = []; + const thinkingEfforts: Array = []; + const generate: GenerateFn = async ( + provider, + systemPrompt, + _tools, + _history, + callbacks, + ) => { + callCount += 1; + systemPrompts.push(systemPrompt); + modelNames.push(provider.modelName); + thinkingEfforts.push(provider.thinkingEffort); + if (callCount === 1) { + writeFileSync(join(workDir, 'AGENTS.md'), 'new project instructions', 'utf-8'); + throw new APIContextOverflowError(400, 'Context length exceeded', 'req-refresh-turn'); + } + if (callCount === 2) return textResult('Compacted summary.'); + if (callCount === 3) { + await callbacks?.onMessagePart?.({ + type: 'text', + text: 'Retried with refreshed instructions.', + }); + return textResult('Retried with refreshed instructions.'); + } + throw new Error(`Unexpected generate call ${String(callCount)}`); + }; + const ctx = testAgent( + execEnvServices({ hostFs: new HostFileSystem() }), + hostEnvironmentServices(homeDir), + { autoConfigure: false, cwd: workDir, generate }, + ); + ctx.configureRuntimeModel(CATALOGUED_PROVIDER, CATALOGUED_MODEL_CAPABILITIES); + const profile = ctx.get(IAgentProfileService); + await profile.applyProfile(EXACT_COMPACTION_REFRESH_PROFILE); + ctx.appendExchange(1, 'old user one', 'old assistant one', 20); + ctx.emitter.once('compaction.completed', () => { + profile.update({ + systemPrompt: 'ORDINARY-MID-TURN-CONFIG', + thinkingLevel: 'high', + }); + }); + + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Retry after overflow' }] }); + await ctx.untilTurnEnd(); + + expect(systemPrompts).toEqual([ + exactCompactionRefreshPrompt(workDir, 'old project instructions'), + exactCompactionRefreshPrompt(workDir, 'old project instructions'), + exactCompactionRefreshPrompt(workDir, 'new project instructions'), + ]); + expect(modelNames).toEqual(['kimi-code', 'kimi-code', 'kimi-code']); + expect(thinkingEfforts[2]).toBe(thinkingEfforts[0]); + expect(profile.data()).toMatchObject({ + systemPrompt: 'ORDINARY-MID-TURN-CONFIG', + thinkingLevel: 'on', + }); + await ctx.expectResumeMatches(); + } finally { + rmSync(homeDir, { recursive: true, force: true }); + rmSync(workDir, { recursive: true, force: true }); + } + }); + it('rejects a manual compaction while a turn is active', async () => { const ctx = testAgent(execEnvServices({ processRunner: createCommandRunner('should-not-run') })); ctx.configure({ @@ -957,7 +1039,7 @@ describe('FullCompaction', () => { await ctx.expectResumeMatches(); }); - it('cancels the compaction lifecycle when manual compaction generation fails', async () => { + it('fails the compaction lifecycle when manual compaction generation fails', async () => { const records: TelemetryRecord[] = []; const generate: GenerateFn = async () => { throw new Error('compaction exploded'); @@ -977,12 +1059,18 @@ describe('FullCompaction', () => { const events = ctx.newEvents(); expect(events).toEqual( expect.arrayContaining([ - expect.objectContaining({ type: '[wire]', event: 'full_compaction.cancel' }), - expect.objectContaining({ type: '[rpc]', event: 'compaction.cancelled' }), + expect.objectContaining({ + type: '[wire]', + event: 'full_compaction.complete', + args: expect.objectContaining({ outcome: 'failed' }), + }), expect.objectContaining({ type: '[rpc]', event: 'error' }), ]), ); - expect(eventIndex(events, 'compaction.cancelled')).toBeLessThan(eventIndex(events, 'error')); + expect(countEvents(events, 'full_compaction.cancel')).toBe(0); + expect(countEvents(events, 'compaction.cancelled')).toBe(0); + expect(countEvents(events, 'compaction.completed')).toBe(0); + expect(eventIndex(events, 'full_compaction.complete')).toBeLessThan(eventIndex(events, 'error')); expect(ctx.compactHistory()).toEqual([ { role: 'user', text: 'old user one' }, { role: 'assistant', text: 'old assistant one' }, @@ -1006,6 +1094,163 @@ describe('FullCompaction', () => { await ctx.expectResumeMatches(); }); + it('treats a provider-originated AbortError as failure rather than cancellation', async () => { + const generate: GenerateFn = async () => { + const error = new Error('provider aborted its own request'); + error.name = 'AbortError'; + throw error; + }; + const ctx = testAgent({ generate }); + ctx.configure({ + provider: CATALOGUED_PROVIDER, + modelCapabilities: CATALOGUED_MODEL_CAPABILITIES, + }); + ctx.appendExchange(1, 'old user one', 'old assistant one', 20); + ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80); + const failed = ctx.once('error'); + + await ctx.rpc.beginCompaction({}); + await failed; + + const events = ctx.newEvents(); + expect(countEvents(events, 'full_compaction.complete')).toBe(1); + expect(events).toContainEqual( + expect.objectContaining({ + type: '[wire]', + event: 'full_compaction.complete', + args: expect.objectContaining({ outcome: 'failed' }), + }), + ); + expect(countEvents(events, 'full_compaction.cancel')).toBe(0); + expect(countEvents(events, 'compaction.cancelled')).toBe(0); + expect(countEvents(events, 'compaction.completed')).toBe(0); + await ctx.expectResumeMatches(); + }); + + it('keeps a committed summary and retries transient reinjection before the next request', async () => { + const ctx = testAgent( + logServices({ + info() {}, + warn() {}, + error() { + throw new Error('error sink failed'); + }, + debug() {}, + }), + ); + ctx.configure({ + provider: CATALOGUED_PROVIDER, + modelCapabilities: CATALOGUED_MODEL_CAPABILITIES, + }); + ctx.appendExchange(1, 'old user one', 'old assistant one', 20); + ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80); + const inject = vi + .spyOn(ctx.get(IAgentContextInjectorService), 'injectAfterCompaction') + .mockRejectedValueOnce(new Error('transient reinjection failure')) + .mockResolvedValue(undefined); + const completed = ctx.once('compaction.completed'); + + ctx.mockNextResponse({ type: 'text', text: 'Committed summary.' }); + await ctx.rpc.beginCompaction({}); + await completed; + + expect(ctx.get(IWireService).getModel(CompactionModel).phase).toBe('committed'); + expect(ctx.compactHistory().map((entry) => entry.text).join('\n')).toContain( + 'Committed summary.', + ); + expect(countEvents(ctx.allEvents, 'full_compaction.complete')).toBe(0); + expect(countEvents(ctx.allEvents, 'compaction.cancelled')).toBe(0); + + ctx.mockNextResponse({ type: 'text', text: 'Continued after maintenance.' }); + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Continue.' }] }); + await ctx.untilTurnEnd(); + + expect(inject).toHaveBeenCalledTimes(2); + expect(ctx.get(IWireService).getModel(CompactionModel).phase).toBe('idle'); + expect(countEvents(ctx.allEvents, 'full_compaction.complete')).toBe(1); + expect(countEvents(ctx.allEvents, 'full_compaction.cancel')).toBe(0); + expect(countEvents(ctx.allEvents, 'compaction.completed')).toBe(1); + expect( + ctx.llmCalls + .at(-1)! + .history.map(messageText) + .some((text) => text.includes('Committed summary.')), + ).toBe(true); + await ctx.expectResumeMatches(); + }); + + it('retries a transient system-prompt refresh without re-running successful reinjection', async () => { + const ctx = testAgent(); + ctx.configure({ + provider: CATALOGUED_PROVIDER, + modelCapabilities: CATALOGUED_MODEL_CAPABILITIES, + }); + ctx.appendExchange(1, 'old user one', 'old assistant one', 20); + ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80); + const refresh = vi + .spyOn(ctx.get(IAgentProfileService), 'refreshSystemPrompt') + .mockRejectedValueOnce(new Error('transient refresh failure')) + .mockResolvedValue(undefined); + const inject = vi.spyOn( + ctx.get(IAgentContextInjectorService), + 'injectAfterCompaction', + ); + const completed = ctx.once('compaction.completed'); + + ctx.mockNextResponse({ type: 'text', text: 'Committed summary.' }); + await ctx.rpc.beginCompaction({}); + await completed; + + expect(ctx.get(IWireService).getModel(CompactionModel).phase).toBe('committed'); + + ctx.mockNextResponse({ type: 'text', text: 'Continued after refresh.' }); + await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Continue.' }] }); + await ctx.untilTurnEnd(); + + expect(refresh).toHaveBeenCalledTimes(2); + expect(inject).toHaveBeenCalledTimes(1); + expect(ctx.get(IWireService).getModel(CompactionModel).phase).toBe('idle'); + expect(countEvents(ctx.allEvents, 'full_compaction.complete')).toBe(1); + expect(countEvents(ctx.allEvents, 'full_compaction.cancel')).toBe(0); + await ctx.expectResumeMatches(); + }); + + it('does not cancel or roll back after the context replacement commits', async () => { + const ctx = testAgent(); + ctx.configure({ + provider: CATALOGUED_PROVIDER, + modelCapabilities: CATALOGUED_MODEL_CAPABILITIES, + }); + ctx.appendExchange(1, 'old user one', 'old assistant one', 20); + ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80); + const maintenanceStarted = deferred(); + const releaseMaintenance = deferred(); + vi.spyOn(ctx.get(IAgentContextInjectorService), 'injectAfterCompaction').mockImplementation( + async () => { + maintenanceStarted.resolve(); + await releaseMaintenance.promise; + }, + ); + const completed = ctx.once('compaction.completed'); + + ctx.mockNextResponse({ type: 'text', text: 'Committed summary.' }); + await ctx.rpc.beginCompaction({}); + await maintenanceStarted.promise; + await ctx.rpc.cancelCompaction({}); + releaseMaintenance.resolve(); + await completed; + + expect(ctx.get(IWireService).getModel(CompactionModel).phase).toBe('idle'); + expect(ctx.compactHistory().map((entry) => entry.text).join('\n')).toContain( + 'Committed summary.', + ); + expect(countEvents(ctx.allEvents, 'full_compaction.complete')).toBe(1); + expect(countEvents(ctx.allEvents, 'full_compaction.cancel')).toBe(0); + expect(countEvents(ctx.allEvents, 'compaction.cancelled')).toBe(0); + expect(countEvents(ctx.allEvents, 'compaction.completed')).toBe(1); + await ctx.expectResumeMatches(); + }); + it('fails a blocked turn when auto compaction generation fails', async () => { let attempts = 0; const generate: GenerateFn = async () => { @@ -2918,7 +3163,7 @@ describe('prompt deferral during full compaction', () => { }); ctx.appendExchange(1, 'old user one', 'old assistant one', 20); ctx.appendExchange(2, 'recent user two', 'recent assistant two', 80); - const cancelled = ctx.once('compaction.cancelled'); + const failed = ctx.once('full_compaction.complete'); await ctx.rpc.beginCompaction({}); await compactionRequested.promise; @@ -2928,9 +3173,18 @@ describe('prompt deferral during full compaction', () => { expect(launch).toBeUndefined(); releaseCompaction.resolve(); - await cancelled; + await failed; const events = await ctx.untilTurnEnd(); + expect(countEvents(events, 'full_compaction.complete')).toBe(1); + expect(events).toContainEqual( + expect.objectContaining({ + type: '[wire]', + event: 'full_compaction.complete', + args: expect.objectContaining({ outcome: 'failed' }), + }), + ); + expect(countEvents(events, 'compaction.cancelled')).toBe(0); expect(countEvents(events, 'compaction.completed')).toBe(0); expect(events).toContainEqual( expect.objectContaining({ diff --git a/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts b/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts index 4b5626b8bc..e1630323da 100644 --- a/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts +++ b/packages/agent-core-v2/test/agent/llmRequester/llmRequesterService.test.ts @@ -260,6 +260,34 @@ describe('AgentLLMRequesterService strict resend', () => { }); }); +describe('AgentLLMRequesterService turn prompt refresh', () => { + it('updates only an existing turn snapshot', async () => { + const calls = { value: 0 }; + const inputs: LLMRequestInput[] = []; + const { service } = createService(createModel(calls, null, [], inputs), undefined); + + await service.request({ source: { type: 'turn', turnId: 1, step: 1 } }); + service.refreshTurnSystemPrompt(1, 'refreshed after compaction'); + await service.request({ source: { type: 'turn', turnId: 1, step: 2 } }); + + expect(inputs.map((input) => input.systemPrompt)).toEqual([ + 'system', + 'refreshed after compaction', + ]); + }); + + it('does not create a turn snapshot before its first request', async () => { + const calls = { value: 0 }; + const inputs: LLMRequestInput[] = []; + const { service } = createService(createModel(calls, null, [], inputs), undefined); + + service.refreshTurnSystemPrompt(1, 'premature compaction prompt'); + await service.request({ source: { type: 'turn', turnId: 1, step: 1 } }); + + expect(inputs[0]?.systemPrompt).toBe('system'); + }); +}); + describe('AgentLLMRequesterService media-stripped resend', () => { const IMAGE_FORMAT_400 = new APIStatusError( 400, 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 889d83135a..7c8f0c95f1 100644 --- a/packages/agent-core-v2/test/agent/loop/loop.test.ts +++ b/packages/agent-core-v2/test/agent/loop/loop.test.ts @@ -1,17 +1,24 @@ -import { type ToolCall } from '#/app/llmProtocol/message'; +import { APIStatusError } from '#/app/llmProtocol/errors'; +import { type Message, type ToolCall } from '#/app/llmProtocol/message'; import { emptyUsage } from '#/app/llmProtocol/usage'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { IAgentProfileService } from '#/index'; import { IAgentLLMRequesterService, type LLMStreamTiming } from '#/agent/llmRequester/llmRequester'; import { IAgentGoalService } from '#/agent/goal/goal'; +import { IAgentContextProjectorService } from '#/agent/contextProjector/contextProjector'; +import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import { IAgentLoopService, type Turn } from '#/agent/loop/loop'; +import { TurnModel, turnOutcome } from '#/agent/loop/turnOps'; import { ContinuationStepRequest, MessageStepRequest } from '#/agent/loop/stepRequest'; +import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor'; import type { ExecutableTool } from '#/tool/toolContract'; import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry'; import { IAgentUsageService } from '#/agent/usage/usage'; import { IEventBus } from '#/app/event/eventBus'; import { userCancellationReason } from '#/_base/utils/abort'; +import { ILogService } from '#/_base/log/log'; +import { IWireService } from '#/wire/wire'; import { agentService, @@ -21,9 +28,21 @@ import { type TestAgentOptions, } from '../../harness'; import { recordingTelemetry, type TelemetryRecord } from '../../app/telemetry/stubs'; +import { stubLog } from '../../_base/log/stubs'; +import { stubToolExecutor } from './stubs'; type GenerateFn = NonNullable; +function throwingDiagnosticLog(): { + readonly log: ILogService; + readonly warn: ReturnType; +} { + const warn = vi.fn(() => { + throw new Error('diagnostic sink failed'); + }); + return { log: { ...stubLog(), warn }, warn }; +} + describe('Agent loop', () => { let ctx: TestAgentContext; let loop: IAgentLoopService; @@ -115,7 +134,9 @@ describe('Agent loop', () => { [emit] agent.activity.updated { "lifecycle": "ready", "turn": { "turnId": 0, "origin": { "kind": "user" }, "phase": "running", "step": 1, "ending": false, "pendingApprovals": [], "activeToolCalls": [], "since": "