vrr/kimi-code
2
0
Fork 0
mirror of https://github.com/MoonshotAI/kimi-code.git synced 2026-08-08 16:26:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat(agent-core-v2): interruption reminder for user-cancelled turns (#2400)

Browse source 
* feat(agent-core-v2): interruption reminder for user-cancelled turns

When the user interrupts a turn with Esc, append a durable
<system-reminder> (origin: injection/interruption) to the agent context
via a new loop aspect watching turn.ended, so the model learns the
previous turn was deliberately cut off. The marker persists to the
wire, replays on resume, stays hidden from transcripts, skips non-user
aborts and steer, and does not stack on repeated cancels.

Two supporting fixes:

- An aborted LLM stream now persists its accumulated partial
  text/thinking as content.part loop events instead of dropping every
  produced token; gated on the turn signal so retried or
  step-cancelled attempts keep their partial output out of the record.
- The turn.cancel wire op carries an optional reason
  ('user_cancelled' | 'aborted') so cold readers can tell deliberate
  interrupts from programmatic aborts. Goal-lifecycle cancels now pass
  an explicit programmatic reason to keep that field honest.

* feat(transcript): mark user-cancelled turns with an interruption marker

Project the deliberate user interrupt onto the transcript timeline: the
live projector emits an 'interruption' marker when a turn ends with
interruptReason 'user_cancelled', and the cold fold consumes the
persisted turn.cancel reason into the same marker. Programmatic aborts
keep surfacing through their own outlets (errors, goal/task state), and
queued cancels that left no visible residue are skipped.

* fix(agent-core-v2): make user-turn cancellation idempotent and reconcile interruption reminders on restore

* fix(transcript): dedupe user-cancelled interruption markers by turn in the cold fold

* chore(agent-core-v2): regenerate state manifest after merging main

* refactor(agent-core-v2): split interruptionReminder out of the loop domain

The loop domain owns turn execution mechanics; whether an interrupted turn
should produce a model-visible reminder is a model-context policy. Move it
into its own L4 domain with its own wire model that cross-reduces the
loop's turn.cancel fact, and rename the op to interruptionReminder.recorded.

---------

Signed-off-by: Haozhe <yanghaozhe@moonshot.ai>
Co-authored-by: Haozhe <yanghaozhe@moonshot.ai>
This commit is contained in:
7Sageer 2026-07-31 18:17:14 +08:00 • committed by GitHub
parent 302b2cd680
commit 1f3f5dadaa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 859 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.changeset/interrupt-reminder.md Normal file
View file

@ -0,0 +1,5 @@
---
"@moonshot-ai/kimi-code": patch
---
Preserve the assistant's partial output when a turn is interrupted with Esc, and remind the model that the previous turn was deliberately interrupted.

106
packages/agent-core-v2/docs/wire-manifest.d.ts vendored
View file

@ -21,52 +21,53 @@
// owning model offloads inline media to blob storage), cross-reducers
// (foreign models that also reduce this record on dispatch and replay).
// Index (45 record types)
// config.update profile persisted src/agent/profile/profileOps.ts
// context_size.measured contextSize transient src/agent/contextSize/contextSizeOps.ts
// context.append_loop_event contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.append_message contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.apply_compaction contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.clear contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.undo contextMemory persisted src/agent/contextMemory/contextOps.ts
// cron.add cron transient src/session/cron/cronOps.ts
// cron.cursor cron transient src/session/cron/cronOps.ts
// cron.delete cron transient src/session/cron/cronOps.ts
// forked goal persisted src/agent/goal/goalOps.ts
// full_compaction.begin fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// full_compaction.cancel fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// full_compaction.complete fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// goal.clear goal persisted src/agent/goal/goalOps.ts
// goal.create goal persisted src/agent/goal/goalOps.ts
// goal.update goal persisted src/agent/goal/goalOps.ts
// interaction.request interaction persisted src/session/interaction/interactionOps.ts
// interaction.resolved interaction persisted src/session/interaction/interactionOps.ts
// llm.request llm.requestTrace persisted src/agent/llmRequester/llmRequestOps.ts
// llm.tools_snapshot llm.requestTrace persisted src/agent/llmRequester/llmRequestOps.ts
// mcp.tools_discovered mcp.discovery persisted src/agent/mcp/mcpDiscoveryOps.ts
// permission.record_approval_result permissionRules persisted src/agent/permissionRules/permissionRulesOps.ts
// permission.rules.add permissionRules transient src/agent/permissionRules/permissionRulesOps.ts
// permission.set_mode permissionMode persisted src/agent/permissionMode/permissionModeOps.ts
// plan_mode.cancel plan persisted src/agent/plan/planOps.ts
// plan_mode.enter plan persisted src/agent/plan/planOps.ts
// plan_mode.exit plan persisted src/agent/plan/planOps.ts
// plan.revision plan persisted src/agent/plan/planOps.ts
// profile.bind profile persisted src/agent/profile/profileOps.ts
// skill.activate skill transient src/agent/skill/skillOps.ts
// swarm_mode.enter swarm persisted src/agent/swarm/swarmOps.ts
// swarm_mode.exit swarm persisted src/agent/swarm/swarmOps.ts
// task.started task persisted src/agent/task/taskOps.ts
// task.terminated task persisted src/agent/task/taskOps.ts
// tools.register_user_tool userTool persisted src/agent/userTool/userToolOps.ts
// tools.reset_active_tools profile.activeTools persisted src/agent/profile/profileOps.ts
// tools.set_active_tools profile.activeTools persisted src/agent/profile/profileOps.ts
// tools.unregister_user_tool userTool persisted src/agent/userTool/userToolOps.ts
// tools.update_store todo persisted src/session/todo/todoOps.ts
// turn.cancel turn persisted src/agent/loop/turnOps.ts
// turn.ended turn persisted src/agent/loop/turnOps.ts
// turn.prompt turn persisted src/agent/loop/turnOps.ts
// turn.steer turn persisted src/agent/loop/turnOps.ts
// usage.record usage persisted src/agent/usage/usageOps.ts
// Index (46 record types)
// config.update profile persisted src/agent/profile/profileOps.ts
// context_size.measured contextSize transient src/agent/contextSize/contextSizeOps.ts
// context.append_loop_event contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.append_message contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.apply_compaction contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.clear contextMemory persisted src/agent/contextMemory/contextOps.ts
// context.undo contextMemory persisted src/agent/contextMemory/contextOps.ts
// cron.add cron transient src/session/cron/cronOps.ts
// cron.cursor cron transient src/session/cron/cronOps.ts
// cron.delete cron transient src/session/cron/cronOps.ts
// forked goal persisted src/agent/goal/goalOps.ts
// full_compaction.begin fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// full_compaction.cancel fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// full_compaction.complete fullCompaction persisted src/agent/fullCompaction/compactionOps.ts
// goal.clear goal persisted src/agent/goal/goalOps.ts
// goal.create goal persisted src/agent/goal/goalOps.ts
// goal.update goal persisted src/agent/goal/goalOps.ts
// interaction.request interaction persisted src/session/interaction/interactionOps.ts
// interaction.resolved interaction persisted src/session/interaction/interactionOps.ts
// interruptionReminder.recorded interruptionReminder persisted src/agent/interruptionReminder/interruptionReminderOps.ts
// llm.request llm.requestTrace persisted src/agent/llmRequester/llmRequestOps.ts
// llm.tools_snapshot llm.requestTrace persisted src/agent/llmRequester/llmRequestOps.ts
// mcp.tools_discovered mcp.discovery persisted src/agent/mcp/mcpDiscoveryOps.ts
// permission.record_approval_result permissionRules persisted src/agent/permissionRules/permissionRulesOps.ts
// permission.rules.add permissionRules transient src/agent/permissionRules/permissionRulesOps.ts
// permission.set_mode permissionMode persisted src/agent/permissionMode/permissionModeOps.ts
// plan_mode.cancel plan persisted src/agent/plan/planOps.ts
// plan_mode.enter plan persisted src/agent/plan/planOps.ts
// plan_mode.exit plan persisted src/agent/plan/planOps.ts
// plan.revision plan persisted src/agent/plan/planOps.ts
// profile.bind profile persisted src/agent/profile/profileOps.ts
// skill.activate skill transient src/agent/skill/skillOps.ts
// swarm_mode.enter swarm persisted src/agent/swarm/swarmOps.ts
// swarm_mode.exit swarm persisted src/agent/swarm/swarmOps.ts
// task.started task persisted src/agent/task/taskOps.ts
// task.terminated task persisted src/agent/task/taskOps.ts
// tools.register_user_tool userTool persisted src/agent/userTool/userToolOps.ts
// tools.reset_active_tools profile.activeTools persisted src/agent/profile/profileOps.ts
// tools.set_active_tools profile.activeTools persisted src/agent/profile/profileOps.ts
// tools.unregister_user_tool userTool persisted src/agent/userTool/userToolOps.ts
// tools.update_store todo persisted src/session/todo/todoOps.ts
// turn.cancel turn persisted src/agent/loop/turnOps.ts
// turn.ended turn persisted src/agent/loop/turnOps.ts
// turn.prompt turn persisted src/agent/loop/turnOps.ts
// turn.steer turn persisted src/agent/loop/turnOps.ts
// usage.record usage persisted src/agent/usage/usageOps.ts
/**
* model: profile · persisted
@ -306,6 +307,15 @@ interface InteractionResolvedPayload {
response: any;
}
/**
* model: interruptionReminder · persisted
* owner: src/agent/interruptionReminder/interruptionReminderOps.ts
*/
interface InterruptionReminderRecordedPayload {
_name: 'interruptionReminder.recorded';
turnId: number;
}
/**
* model: llm.requestTrace · persisted
* owner: src/agent/llmRequester/llmRequestOps.ts
@ -563,13 +573,14 @@ interface ToolsUpdateStorePayload {
}
/**
* model: turn · persisted
* model: turn · persisted · cross-reducers: interruptionReminder
* owner: src/agent/loop/turnOps.ts
*/
interface TurnCancelPayload {
_name: 'turn.cancel';
turnId?: number;
target?: 'active' | 'queued';
reason?: 'user_cancelled' | 'aborted';
}
/**
@ -688,6 +699,7 @@ interface WirePayloadMap {
"goal.update": GoalUpdatePayload;
"interaction.request": InteractionRequestPayload;
"interaction.resolved": InteractionResolvedPayload;
"interruptionReminder.recorded": InterruptionReminderRecordedPayload;
"llm.request": LlmRequestPayload;
"llm.tools_snapshot": LlmToolsSnapshotPayload;
"mcp.tools_discovered": McpToolsDiscoveredPayload;

18
packages/agent-core-v2/src/agent/goal/goalService.ts
View file

@ -622,7 +622,7 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
const state = this.requireState();
const snapshot = this.toSnapshot(state);
if (state.status === 'active' && this.liveTurnId !== undefined) {
this.loopService.cancel(this.liveTurnId);
this.loopService.cancel(this.liveTurnId, abortError('Goal cancelled'));
}
this.clearInternal(actor);
if (actor === 'user') {
@ -985,18 +985,10 @@ export class AgentGoalService extends Disposable implements IAgentGoalService {
const pending = this.pendingContinuation;
if (preserveLiveContinuation && pending?.turnId === this.liveTurnId) return;
this.pendingContinuation = undefined;
const aborted =
reason === undefined ? pending?.receipt.abort() : pending?.receipt.abort(reason);
if (
pending !== undefined &&
!aborted &&
pending.turnId !== undefined
) {
if (reason === undefined) {
this.loopService.cancel(pending.turnId);
} else {
this.loopService.cancel(pending.turnId, reason);
}
const cancellation = reason ?? abortError('Goal continuation cancelled');
const aborted = pending?.receipt.abort(cancellation);
if (pending !== undefined && !aborted && pending.turnId !== undefined) {
this.loopService.cancel(pending.turnId, cancellation);
}
}

15
packages/agent-core-v2/src/agent/interruptionReminder/interruptionReminder.ts Normal file
View file

@ -0,0 +1,15 @@
/**
* `interruptionReminder` domain (L4) — user-interruption reminder contract.
*
* Defines the Agent-scoped aspect that records a model-visible reminder after
* a user-cancelled turn. Bound at Agent scope.
*/
import { createDecorator } from '#/_base/di/instantiation';
export interface IAgentInterruptionReminderService {
readonly _serviceBrand: undefined;
}
export const IAgentInterruptionReminderService =
createDecorator<IAgentInterruptionReminderService>('agentInterruptionReminderService');

43
packages/agent-core-v2/src/agent/interruptionReminder/interruptionReminderOps.ts Normal file
View file

@ -0,0 +1,43 @@
/**
* `interruptionReminder` domain (L4) — persists and restores pending
* user-interruption reminders.
*
* Projects the `loop` domain's `turn.cancel` fact into the set of turns whose
* interruption reminder still has to reach the conversation, and owns the op
* that records a reminder's delivery. Consumed by the Agent-scope
* `interruptionReminderService`.
*/
import { z } from 'zod';
import { defineModel } from '#/wire/model';
export const InterruptionReminderModel = defineModel<readonly number[]>(
'interruptionReminder',
() => [],
{
reducers: {
'turn.cancel': (state, { turnId, target, reason }) => {
if (target !== 'active' || reason !== 'user_cancelled' || turnId === undefined) {
return state;
}
if (state.includes(turnId)) return state;
return [...state, turnId].toSorted((a, b) => a - b);
},
},
},
);
declare module '#/wire/types' {
interface PersistedOpMap {
'interruptionReminder.recorded': typeof interruptionReminderRecorded;
}
}
export const interruptionReminderRecorded = InterruptionReminderModel.defineOp(
'interruptionReminder.recorded',
{
schema: z.object({ turnId: z.number().int().nonnegative() }),
apply: (state, { turnId }) => state.filter((pendingTurnId) => pendingTurnId !== turnId),
},
);

108
packages/agent-core-v2/src/agent/interruptionReminder/interruptionReminderService.ts Normal file
View file

@ -0,0 +1,108 @@
/**
* `interruptionReminder` domain (L4) — `IAgentInterruptionReminderService` implementation.
*
* Observes turn completion through `event`, persists reminder completion through
* its own wire model, reads conversation history through `contextMemory`, and
* appends model-visible notices through `systemReminder`. Reconciles reminders
* left pending by an interrupted restore. Bound at Agent scope.
*/
import { Disposable } from '#/_base/di/lifecycle';
import { LifecycleScope, ScopeActivation, registerScopedService } from '#/_base/di/scope';
import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
import type { ContextMessage } from '#/agent/contextMemory/types';
import { isVacuousContentPart } from '#/agent/contextMemory/vacuousContent';
import { IAgentSystemReminderService } from '#/agent/systemReminder/systemReminder';
import { IEventBus } from '#/app/event/eventBus';
import { IWireService } from '#/wire/wire';
import { IAgentInterruptionReminderService } from './interruptionReminder';
import { interruptionReminderRecorded, InterruptionReminderModel } from './interruptionReminderOps';
export const INTERRUPTION_REMINDER_VARIANT = 'interruption';
const INTERRUPTION_REMINDER = [
'The previous turn was interrupted by the user before completion;',
'any partial output shown above is incomplete.',
"The user's next message continues the conversation.",
].join(' ');
export class AgentInterruptionReminderService
extends Disposable
implements IAgentInterruptionReminderService
{
declare readonly _serviceBrand: undefined;
constructor(
@IEventBus eventBus: IEventBus,
@IAgentContextMemoryService private readonly context: IAgentContextMemoryService,
@IAgentSystemReminderService private readonly reminders: IAgentSystemReminderService,
@IWireService private readonly wire: IWireService,
) {
super();
this._register(
this.wire.hooks.onDidRestore.register('interruption-reminder', async (_ctx, next) => {
this.reconcilePendingReminders();
await next();
}),
);
this._register(
eventBus.subscribe('turn.ended', (event) => {
if (event.reason !== 'cancelled' || event.interruptReason !== 'user_cancelled') return;
this.recordReminder(event.turnId, true);
}),
);
}
private reconcilePendingReminders(): void {
const pending = this.wire.getModel(InterruptionReminderModel);
for (const turnId of pending) this.recordReminder(turnId);
}
private recordReminder(turnId: number, allowUntracked = false): void {
const pending = this.wire.getModel(InterruptionReminderModel).includes(turnId);
if (!pending && !allowUntracked) return;
if (!this.appendInterruptionReminder()) return;
if (pending) this.wire.dispatch(interruptionReminderRecorded({ turnId }));
}
private appendInterruptionReminder(): boolean {
const before = this.context.get();
const origin = lastDurableMessageOrigin(before);
if (origin?.kind === 'injection' && origin.variant === INTERRUPTION_REMINDER_VARIANT) return true;
this.reminders.appendSystemReminder(INTERRUPTION_REMINDER, {
kind: 'injection',
variant: INTERRUPTION_REMINDER_VARIANT,
});
const after = this.context.get();
if (after === before) return false;
const appended = lastDurableMessageOrigin(after);
return appended?.kind === 'injection' && appended.variant === INTERRUPTION_REMINDER_VARIANT;
}
}
function lastDurableMessageOrigin(
messages: readonly ContextMessage[],
): ContextMessage['origin'] | undefined {
for (let i = messages.length - 1; i >= 0; i--) {
const message = messages[i]!;
if (
message.role === 'assistant' &&
message.partial === true &&
message.toolCalls.length === 0 &&
message.content.every(isVacuousContentPart)
) {
continue;
}
return message.origin;
}
return undefined;
}
registerScopedService(
LifecycleScope.Agent,
IAgentInterruptionReminderService,
AgentInterruptionReminderService,
ScopeActivation.OnScopeCreated,
'interruptionReminder',
);

168
packages/agent-core-v2/src/agent/loop/loopService.ts
View file

@ -44,12 +44,13 @@ import { IAgentToolExecutorService } from '#/agent/toolExecutor/toolExecutor';
import { IConfigService } from '#/app/config/config';
import { IEventBus } from '#/app/event/eventBus';
import { type FinishReason } from '#/kosong/contract/provider';
import { type StreamedMessagePart } from '#/kosong/contract/message';
import { mergeInPlace, type ContentPart, type StreamedMessagePart } from '#/kosong/contract/message';
import { type TokenUsage } from '#/kosong/contract/usage';
import { BugIndicatingError, ErrorCodes, Error2, isError2, toKimiErrorPayload } from '#/errors';
import { OrderedHookSlot } from '#/hooks';
import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
import { isVacuousContentPart } from '#/agent/contextMemory/vacuousContent';
import { IAgentStateService } from '#/agent/state/agentState';
import { IAgentTelemetryContextService } from '#/app/telemetry/agentTelemetryContext';
import type {
@ -83,7 +84,7 @@ import {
type TurnSeed,
} from './stepRequest';
import { StepRequestQueue, type StepRequestBatch } from './stepRequestQueue';
import { isDisplayablePromptOrigin, turnPromptText } from './turnEvents';
import { isDisplayablePromptOrigin, turnPromptText, type TurnInterruptReason } from './turnEvents';
import { cancelTurn, endTurn, promptTurn, TurnModel } from './turnOps';
export type LoopInterruptReason = 'aborted' | 'max_steps' | 'error';
@ -274,7 +275,10 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
private cancelActiveTurn(turnId: number | undefined, cancellation: unknown): boolean {
const job = this.activeTurnJob;
if (job === undefined || (turnId !== undefined && job.turn.id !== turnId)) return false;
this.wire.dispatch(cancelTurn({ turnId: job.turn.id, target: 'active' }));
if (job.controller.signal.aborted) return true;
this.wire.dispatch(
cancelTurn({ turnId: job.turn.id, target: 'active', reason: cancelReasonFor(cancellation) }),
);
job.controller.abort(cancellation);
return true;
}
@ -284,7 +288,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
if (index < 0) return false;
const [job] = this.pendingTurns.splice(index, 1);
if (job === undefined || job.turn.state !== 'queued') return false;
this.wire.dispatch(cancelTurn({ turnId, target: 'queued' }));
this.wire.dispatch(cancelTurn({ turnId, target: 'queued', reason: cancelReasonFor(cancellation) }));
for (const step of job.steps.values()) step.cancel(cancellation);
job.controller.abort(cancellation);
job.turn.state = 'cancelled';
@ -495,6 +499,8 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
: this.activeRequestTrace?.traceId;
if (result !== undefined) {
const error = result.type === 'failed' ? toKimiErrorPayload(result.error) : undefined;
const interruptReason =
result.type === 'completed' ? undefined : interruptReasonFor(result);
const durationMs = Date.now() - startedAt;
this.wire.dispatch(endTurn({ turnId: turn.id, reason: result.type, error, durationMs }));
this.eventBus.publish({
@ -503,14 +509,15 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
reason: result.type,
error,
durationMs,
interruptReason,
});
if (error !== undefined) this.eventBus.publish({ type: 'error', ...error });
if (result.type !== 'completed') {
if (interruptReason !== undefined) {
const interrupted: TurnInterruptedEvent = {
turn_id: turn.id,
at_step: result.steps,
mode,
interrupt_reason: interruptReasonFor(result),
interrupt_reason: interruptReason,
provider_type,
protocol,
thinking_effort: thinkingEffort,
@ -609,6 +616,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
const result = await this.executeLoopStep(
runtime.turnId,
begun.step.signal,
runtime.turnSignal,
begun.step.number,
begun.step.uuid,
options.onStarted,
@ -792,6 +800,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
private async executeLoopStep(
turnId: number,
signal: AbortSignal,
turnSignal: AbortSignal,
currentStep: number,
stepUuid: string,
onStarted: ((step: number) => void) | undefined,
@ -799,13 +808,20 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
this.activeRequestTrace = undefined;
await this.hooks.onWillBeginStep.run({ turnId, step: currentStep, signal });
const markStepStarted = this.beginStep(turnId, signal, currentStep, stepUuid, onStarted);
const streamParts = this.createStreamPartHandler(turnId, markStepStarted);
const request = this.llmRequester.start(
{ source: { type: 'turn', turnId, step: currentStep } },
this.createStreamPartHandler(turnId, markStepStarted),
streamParts.handle,
signal,
);
this.activeRequestTrace = request.trace;
const response = await request.result;
let response: AgentLLMRequestFinish;
try {
response = await request.result;
} catch (error) {
this.appendInterruptedStreamContent(turnId, currentStep, stepUuid, streamParts, turnSignal);
throw error;
}
this.lastRequestTraceId = request.trace.traceId;
this.appendResponseContent(turnId, currentStep, stepUuid, response);
const finishReason = await this.executeStepTools(
@ -868,6 +884,26 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
}
}
private appendInterruptedStreamContent(
turnId: number,
currentStep: number,
stepUuid: string,
streamParts: StreamPartCollector,
turnSignal: AbortSignal,
): void {
if (!turnSignal.aborted) return;
for (const part of streamParts.drainInterruptedContent()) {
this.context.appendLoopEvent({
type: 'content.part',
uuid: randomUUID(),
turnId: String(turnId),
step: currentStep,
stepUuid,
part,
});
}
}
private async executeStepTools(
turnId: number,
signal: AbortSignal,
@ -1022,54 +1058,69 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
private createStreamPartHandler(
turnId: number,
onResponseEvent: () => void,
): (part: StreamedMessagePart) => void {
): StreamPartCollector {
const callsByIndex = new Map<number | string | undefined, { id: string; name: string }>();
const partialContent: ContentPart[] = [];
let forceContentPartBoundary = false;
const accumulate = (part: ContentPart): void => {
const last = partialContent.at(-1);
if (!forceContentPartBoundary && last !== undefined && mergeInPlace(last, part)) return;
forceContentPartBoundary = false;
partialContent.push({ ...part });
};
return (part) => {
switch (part.type) {
case 'text':
onResponseEvent();
this.eventBus.publish({ type: 'assistant.delta', turnId, delta: part.text });
return;
case 'think':
onResponseEvent();
this.eventBus.publish({ type: 'thinking.delta', turnId, delta: part.think });
return;
case 'image_url':
case 'audio_url':
case 'video_url':
return;
case 'function': {
onResponseEvent();
callsByIndex.set(part._streamIndex, { id: part.id, name: part.name });
this.eventBus.publish({
type: 'tool.call.delta',
turnId,
toolCallId: part.id,
name: part.name,
argumentsPart: part.arguments ?? undefined,
});
return;
return {
handle: (part) => {
switch (part.type) {
case 'text':
onResponseEvent();
accumulate(part);
this.eventBus.publish({ type: 'assistant.delta', turnId, delta: part.text });
return;
case 'think':
onResponseEvent();
accumulate(part);
this.eventBus.publish({ type: 'thinking.delta', turnId, delta: part.think });
return;
case 'image_url':
case 'audio_url':
case 'video_url':
return;
case 'function': {
onResponseEvent();
forceContentPartBoundary = true;
callsByIndex.set(part._streamIndex, { id: part.id, name: part.name });
this.eventBus.publish({
type: 'tool.call.delta',
turnId,
toolCallId: part.id,
name: part.name,
argumentsPart: part.arguments ?? undefined,
});
return;
}
case 'tool_call_part': {
if (part.argumentsPart === null) return;
const toolCall = callsByIndex.get(part.index);
if (toolCall === undefined) return;
onResponseEvent();
this.eventBus.publish({
type: 'tool.call.delta',
turnId,
toolCallId: toolCall.id,
name: toolCall.name,
argumentsPart: part.argumentsPart,
});
return;
}
default: {
const _exhaustive: never = part;
return _exhaustive;
}
}
case 'tool_call_part': {
if (part.argumentsPart === null) return;
const toolCall = callsByIndex.get(part.index);
if (toolCall === undefined) return;
onResponseEvent();
this.eventBus.publish({
type: 'tool.call.delta',
turnId,
toolCallId: toolCall.id,
name: toolCall.name,
argumentsPart: part.argumentsPart,
});
return;
}
default: {
const _exhaustive: never = part;
return _exhaustive;
}
}
},
drainInterruptedContent: () =>
partialContent.splice(0).filter((part) => !isVacuousContentPart(part)),
};
}
}
@ -1128,9 +1179,18 @@ interface StepRuntime {
type BeginStepResult = { readonly step: StepRuntime } | { readonly result: LoopRunResult };
interface StreamPartCollector {
readonly handle: (part: StreamedMessagePart) => void;
drainInterruptedContent(): ContentPart[];
}
function cancelReasonFor(cancellation: unknown): 'user_cancelled' | 'aborted' {
return isUserCancellation(cancellation) ? 'user_cancelled' : 'aborted';
}
function interruptReasonFor(
result: Extract<TurnResult, { readonly type: 'cancelled' | 'failed' }>,
): TurnInterruptedEvent['interrupt_reason'] {
): TurnInterruptReason {
if (result.type === 'cancelled') {
return isUserCancellation(result.reason) ? 'user_cancelled' : 'aborted';
}

9
packages/agent-core-v2/src/agent/loop/turnEvents.ts
View file

@ -20,6 +20,14 @@ import type { TokenUsage } from '#/kosong/contract/usage';
export type TurnEndReason = 'completed' | 'cancelled' | 'failed' | 'blocked';
export type TurnInterruptReason =
| 'user_cancelled'
| 'aborted'
| 'max_steps'
| 'error'
| 'filtered'
| 'blocked';
export interface TurnStartedEvent {
readonly type: 'turn.started';
readonly turnId: number;
@ -49,6 +57,7 @@ export interface TurnEndedEvent {
readonly reason: TurnEndReason;
readonly error?: KimiErrorPayload;
readonly durationMs?: number;
readonly interruptReason?: TurnInterruptReason;
}
export interface TurnStepStartedEvent {

7
packages/agent-core-v2/src/agent/loop/turnOps.ts
View file

@ -6,7 +6,8 @@
* legacy loop-event observations. Also persists the terminal `turn.ended`
* record (reason / error / durationMs) so downstream history rebuilds can
* recover how a turn ended; the record carries no engine-restorable state, so
* its `apply` is a no-op.
* its `apply` is a no-op. Consumed by the Agent-scope `loopService`; the
* `interruptionReminder` domain projects `turn.cancel` into its own model.
*/
import { z } from 'zod';
@ -68,9 +69,11 @@ export const cancelTurn = TurnModel.defineOp('turn.cancel', {
schema: z.object({
turnId: z.number().optional(),
target: z.enum(['active', 'queued']).optional(),
reason: z.enum(['user_cancelled', 'aborted']).optional(),
}),
apply: (s, { turnId, target }) => {
if (target === undefined || turnId === undefined || turnId < s.nextTurnId) return s;
if (target === undefined || turnId === undefined) return s;
if (turnId < s.nextTurnId) return s;
return advanceTurnClock(s, s.nextTurnId, [...s.cancelledTurnIds, turnId]);
},
});

3
packages/agent-core-v2/src/index.ts
View file

@ -520,6 +520,9 @@ export * from '#/agent/loop/loop';
export * from '#/agent/loop/loopService';
export * from '#/agent/loop/loopContinuation';
export * from '#/agent/loop/loopContinuationService';
export * from '#/agent/interruptionReminder/interruptionReminder';
export * from '#/agent/interruptionReminder/interruptionReminderService';
export * from '#/agent/interruptionReminder/interruptionReminderOps';
export * from '#/agent/mcp/mcp';
export * from '#/agent/mcp/mcpService';
export * from '#/agent/mcp/mcpDiscoveryOps';

1
packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts
View file

@ -1122,6 +1122,7 @@ describe('FullCompaction', () => {
code: 'compaction.failed',
message: 'APIStatusError: Bad request',
}),
interruptReason: 'error',
},
}),
);

9
packages/agent-core-v2/test/agent/goal/goal.test.ts
View file

@ -6,6 +6,7 @@
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { isUserCancellation } from '#/_base/utils/abort';
import type { TurnEndedEvent } from '#/agent/loop/turnEvents';
import type { IDisposable } from '#/_base/di/lifecycle';
@ -1160,7 +1161,8 @@ describe('AgentGoalService core workflow hooks', () => {
await goals.cancelGoal();
expect(abort).toHaveBeenCalledOnce();
expect(cancel).toHaveBeenCalledWith(41);
expect(cancel).toHaveBeenCalledWith(41, expect.any(Error));
expect(isUserCancellation(cancel.mock.calls[0]?.[1])).toBe(false);
});
it.each(['turn', 'token', 'wall-clock'] as const)(
@ -1928,7 +1930,7 @@ describe('AgentGoalService hard wall-clock deadline', () => {
}
});
it('keeps user cancellation authoritative when it precedes the wall-clock deadline', async () => {
it('keeps the goal-cancellation abort authoritative when it precedes the wall-clock deadline', async () => {
const clock = new ManualGoalDeadlineScheduler();
const llm = blockingGenerate();
const ctx = createTestAgent(appService(IGoalDeadlineScheduler, clock), {
@ -1946,8 +1948,9 @@ describe('AgentGoalService hard wall-clock deadline', () => {
await ctx.rpc.cancelGoal({});
expect(llm.signal()).toMatchObject({
aborted: true,
reason: expect.objectContaining({ userCancelled: true }),
reason: expect.objectContaining({ message: 'Goal cancelled' }),
});
expect(isUserCancellation(llm.signal().reason)).toBe(false);
clock.advanceBy(1_000);
await ctx.untilTurnEnd();

317
packages/agent-core-v2/test/agent/loop/loop.test.ts
View file

@ -2,12 +2,15 @@ import { type ToolCall } from '#/kosong/contract/message';
import { emptyUsage } from '#/kosong/contract/usage';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import type { IDisposable } from '#/_base/di/lifecycle';
import { IAgentProfileService } from '#/index';
import { IAgentLLMRequesterService } from '#/agent/llmRequester/llmRequester';
import type { ModelRequestTiming } from '#/kosong/model/modelRequester';
import type { ContextMessage } from '#/agent/contextMemory/types';
import { IAgentGoalService } from '#/agent/goal/goal';
import { IAgentLoopService, type Turn } from '#/agent/loop/loop';
import { ContinuationStepRequest, MessageStepRequest } from '#/agent/loop/stepRequest';
import { RetryStepRequest } from '#/agent/prompt/promptStepRequests';
import type { ExecutableTool } from '#/tool/toolContract';
import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry';
import { IAgentUsageService } from '#/agent/usage/usage';
@ -130,7 +133,7 @@ describe('Agent loop', () => {
[wire] context.append_loop_event { "event": { "type": "content.part", "uuid": "<uuid-2>", "turnId": "0", "step": 1, "stepUuid": "<uuid-1>", "part": { "type": "text", "text": "blocked" } }, "time": "<time>" }
[wire] context.append_loop_event { "event": { "type": "step.end", "uuid": "<uuid-1>", "turnId": "0", "step": 1, "finishReason": "filtered", "usage": { "inputOther": 3, "output": 5, "inputCacheRead": 0, "inputCacheCreation": 0 }, "messageId": "mock-1", "providerFinishReason": "filtered", "rawFinishReason": "filtered" }, "time": "<time>" }
[wire] turn.ended { "turnId": 0, "reason": "failed", "error": { "code": "provider.filtered", "message": "Provider safety policy blocked the response.", "name": "ProviderFilteredError", "details": { "finishReason": "filtered" }, "retryable": false }, "time": "<time>" }
[emit] turn.ended { "turnId": 0, "reason": "failed", "error": { "code": "provider.filtered", "message": "Provider safety policy blocked the response.", "name": "ProviderFilteredError", "details": { "finishReason": "filtered" }, "retryable": false } }
[emit] turn.ended { "turnId": 0, "reason": "failed", "error": { "code": "provider.filtered", "message": "Provider safety policy blocked the response.", "name": "ProviderFilteredError", "details": { "finishReason": "filtered" }, "retryable": false }, "interruptReason": "filtered" }
`);
const stepCompleted = ctx.allEvents.find(
@ -953,6 +956,318 @@ describe('turn telemetry', () => {
);
});
describe('interruption reminder', () => {
let ctx: TestAgentContext;
let loop: IAgentLoopService;
beforeEach(() => {
ctx = createTestAgent();
loop = ctx.get(IAgentLoopService);
});
afterEach(async () => {
try {
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
});
function cancelOnFirstDelta(): IDisposable {
return ctx.get(IEventBus).subscribe('assistant.delta', () => {
loop.cancel();
});
}
function remindersIn(target: TestAgentContext): ContextMessage[] {
return target.contextData().history.filter(
(message) =>
message.origin?.kind === 'injection' && message.origin.variant === 'interruption',
);
}
function interruptionReminders(): ContextMessage[] {
return remindersIn(ctx);
}
function contentPartRecordsIn(target: TestAgentContext): number {
return target.allEvents.filter(
(entry) =>
entry.type === '[wire]' &&
entry.event === 'context.append_loop_event' &&
(entry.args as { event?: { type?: string } }).event?.type === 'content.part',
).length;
}
it('preserves the partial stream and appends one reminder on user cancel', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' }, { type: 'text', text: ' more' });
const subscription = cancelOnFirstDelta();
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
expect(ctx.contextData().history).toEqual([
expect.objectContaining({ role: 'user', content: [{ type: 'text', text: 'Hello' }] }),
{
role: 'assistant',
content: [{ type: 'text', text: 'partial answer' }],
toolCalls: [],
partial: true,
},
expect.objectContaining({
role: 'user',
origin: { kind: 'injection', variant: 'interruption' },
}),
]);
expect(interruptionReminders()).toHaveLength(1);
expect(interruptionReminders()[0]!.content).toEqual([
{
type: 'text',
text: '<system-reminder>\nThe previous turn was interrupted by the user before completion; any partial output shown above is incomplete. The user\'s next message continues the conversation.\n</system-reminder>',
},
]);
const cancelRecord = ctx.allEvents.find(
(entry) => entry.type === '[wire]' && entry.event === 'turn.cancel',
);
expect(cancelRecord?.args).toMatchObject({
turnId: 0,
target: 'active',
reason: 'user_cancelled',
});
const turnEnded = ctx.allEvents.find(
(entry) => entry.type === '[rpc]' && entry.event === 'turn.ended',
);
expect(turnEnded?.args).toMatchObject({
reason: 'cancelled',
interruptReason: 'user_cancelled',
});
expect(contentPartRecordsIn(ctx)).toBe(1);
});
it('writes one active cancellation when cancel repeats before the turn settles', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' }, { type: 'text', text: ' more' });
const results: boolean[] = [];
let cancelled = false;
const subscription = ctx.get(IEventBus).subscribe('assistant.delta', () => {
if (cancelled) return;
cancelled = true;
results.push(loop.cancel(), loop.cancel());
});
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
await ctx.wire.flush();
expect(results).toEqual([true, true]);
expect(
ctx.allEvents.filter(
(entry) => entry.type === '[wire]' && entry.event === 'turn.cancel',
),
).toHaveLength(1);
expect(interruptionReminders()).toHaveLength(1);
expect(
ctx.allEvents.filter(
(entry) => entry.type === '[wire]' && entry.event === 'interruptionReminder.recorded',
),
).toHaveLength(1);
});
it('preserves the partial stream but appends no reminder on programmatic abort', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' }, { type: 'text', text: ' more' });
const subscription = ctx.get(IEventBus).subscribe('assistant.delta', () => {
loop.cancel(undefined, new Error('stop'));
});
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
expect(ctx.contextData().history).toContainEqual({
role: 'assistant',
content: [{ type: 'text', text: 'partial answer' }],
toolCalls: [],
partial: true,
});
expect(interruptionReminders()).toHaveLength(0);
const cancelRecord = ctx.allEvents.find(
(entry) => entry.type === '[wire]' && entry.event === 'turn.cancel',
);
expect(cancelRecord?.args).toMatchObject({ target: 'active', reason: 'aborted' });
const turnEnded = ctx.allEvents.find(
(entry) => entry.type === '[rpc]' && entry.event === 'turn.ended',
);
expect(turnEnded?.args).toMatchObject({ reason: 'cancelled', interruptReason: 'aborted' });
});
it('does not stack a second reminder without an intervening message', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' });
const subscription = cancelOnFirstDelta();
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
expect(interruptionReminders()).toHaveLength(1);
ctx.get(IEventBus).publish({
type: 'turn.ended',
turnId: 99,
reason: 'cancelled',
interruptReason: 'user_cancelled',
});
expect(interruptionReminders()).toHaveLength(1);
});
it('appends no reminder when a queued turn is user-cancelled before starting', async () => {
let release!: () => void;
loop.hooks.onWillBeginStep.register('test-hang-queued-cancel', async (hookCtx, next) => {
await new Promise<void>((resolve) => {
release = resolve;
});
await next();
});
ctx.mockNextResponse({ type: 'text', text: 'unreached' });
const active = (await loop.enqueue(nextTurnMessage('active')).assigned).turn;
const queued = (await loop.enqueue(nextTurnMessage('queued')).assigned).turn;
expect(loop.cancel(queued.id)).toBe(true);
await expect(queued.result).resolves.toMatchObject({ type: 'cancelled', steps: 0 });
release();
loop.cancel(active.id);
await expect(active.result).resolves.toMatchObject({ type: 'cancelled' });
expect(interruptionReminders()).toHaveLength(1);
const queuedCancel = ctx.allEvents.find(
(entry) =>
entry.type === '[wire]' &&
entry.event === 'turn.cancel' &&
(entry.args as { target?: string }).target === 'queued',
);
expect(queuedCancel?.args).toMatchObject({ target: 'queued', reason: 'user_cancelled' });
});
it('sends the partial output and reminder ahead of the next user message', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' }, { type: 'text', text: ' more' });
const subscription = cancelOnFirstDelta();
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Hello' }] });
await ctx.untilTurnEnd();
subscription.dispose();
ctx.llmInputs(); // drain the interrupted turn's request
ctx.mockNextResponse({ type: 'text', text: 'second answer' });
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Next' }] });
await ctx.untilTurnEnd();
expect(ctx.lastLlmInput()).toMatchInlineSnapshot(`
messages:
<last>
assistant: text "partial answer"
user: text "<system-reminder>\\nThe previous turn was interrupted by the user before completion; any partial output shown above is incomplete. The user's next message continues the conversation.\\n</system-reminder>"
user: text "Next"
`);
});
it('removes the reminder together with the undone turn', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' });
const subscription = cancelOnFirstDelta();
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Hello' }] });
await ctx.untilTurnEnd();
subscription.dispose();
expect(interruptionReminders()).toHaveLength(1);
await ctx.undoHistory(1);
expect(ctx.contextData().history).toEqual([]);
});
it('preserves partial thinking on user cancel', async () => {
ctx.mockNextResponse({ type: 'think', think: 'pondering' }, { type: 'text', text: 'answer' });
const subscription = ctx.get(IEventBus).subscribe('thinking.delta', () => {
loop.cancel();
});
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
expect(ctx.contextData().history).toContainEqual({
role: 'assistant',
content: [{ type: 'think', think: 'pondering' }],
toolCalls: [],
partial: true,
});
expect(interruptionReminders()).toHaveLength(1);
});
it('records no partial content when the stream only produced whitespace', async () => {
ctx.mockNextResponse({ type: 'text', text: ' ' }, { type: 'text', text: 'answer' });
const subscription = cancelOnFirstDelta();
const turn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
subscription.dispose();
expect(contentPartRecordsIn(ctx)).toBe(0);
expect(ctx.contextData().history).toEqual([
expect.objectContaining({ role: 'user' }),
{ role: 'assistant', content: [], toolCalls: [], partial: true },
expect.objectContaining({ origin: { kind: 'injection', variant: 'interruption' } }),
]);
});
it('does not stack a second reminder around a vacuous retry turn', async () => {
ctx.mockNextResponse({ type: 'text', text: 'partial answer' });
const first = cancelOnFirstDelta();
const firstTurn = (await loop.enqueue(nextTurnMessage('Hello')).assigned).turn;
await expect(firstTurn.result).resolves.toMatchObject({ type: 'cancelled' });
first.dispose();
expect(interruptionReminders()).toHaveLength(1);
ctx.mockNextResponse({ type: 'text', text: 'retried answer' });
const onStepStarted = ctx.get(IEventBus).subscribe('turn.step.started', () => {
loop.cancel();
});
const retryTurn = (await loop.enqueue(new RetryStepRequest()).assigned).turn;
await expect(retryTurn.result).resolves.toMatchObject({ type: 'cancelled' });
onStepStarted.dispose();
expect(interruptionReminders()).toHaveLength(1);
ctx.mockNextResponse({ type: 'text', text: 'third answer' });
await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Next' }] });
await ctx.untilTurnEnd();
expect(interruptionReminders()).toHaveLength(1);
});
it('does not duplicate recorded content when cancelled during tool execution', async () => {
const local = createTestAgent(permissionModeServices('yolo'));
try {
const slowToolStarted = registerAbortableWorkTool(local);
const localLoop = local.get(IAgentLoopService);
local.mockNextResponse(
{ type: 'text', text: 'working' },
{ type: 'function', id: 'call-work-1', name: 'Work', arguments: '{}' },
);
local.mockNextResponse(
{ type: 'text', text: 'still working' },
{ type: 'function', id: 'call-work-2', name: 'Work', arguments: '{}' },
);
const turn = (await localLoop.enqueue(nextTurnMessage('do work')).assigned).turn;
await slowToolStarted.promise;
localLoop.cancel(turn.id);
await expect(turn.result).resolves.toMatchObject({ type: 'cancelled' });
expect(contentPartRecordsIn(local)).toBe(2);
const history = local.contextData().history;
expect(remindersIn(local)).toHaveLength(1);
expect(history.at(-1)?.origin).toEqual({ kind: 'injection', variant: 'interruption' });
expect(history.at(-2)?.role).toBe('tool');
await local.expectResumeMatches();
} finally {
await local.dispose();
}
});
});
describe('step timing split propagation', () => {
it('carries the split from the llmRequester timing event to the turn.step.completed protocol event', async () => {
const ctx = createTestAgent(agentService(IAgentLLMRequesterService, createTimingRequester()));

1
packages/agent-core-v2/test/index.test.ts
View file

@ -87,6 +87,7 @@ const V2_RECORD_TYPES: ReadonlySet<string> = new Set([
'interaction.request',
'interaction.resolved',
'plan.revision',
'interruptionReminder.recorded',
'turn.ended',
]);

63
packages/agent-core-v2/test/wire/resume.test.ts
View file

@ -65,6 +65,69 @@ describe('Agent resume', () => {
expect(persistence.records.filter((record) => record.type === 'metadata')).toHaveLength(1);
});
it('reconciles a pending user interruption after restore when the reminder is missing', async () => {
const persistence = new RecordingAgentPersistence([
resumeConfigRecord(),
{
type: 'context.append_message',
message: {
role: 'user',
content: [{ type: 'text', text: 'Hello' }],
toolCalls: [],
origin: { kind: 'user' },
},
},
{
type: 'turn.prompt',
input: [{ type: 'text', text: 'Hello' }],
origin: { kind: 'user' },
},
{
type: 'context.append_loop_event',
event: { type: 'step.begin', uuid: 'step-0', turnId: '0', step: 1 },
},
{
type: 'context.append_loop_event',
event: {
type: 'content.part',
uuid: 'part-0',
turnId: '0',
step: 1,
stepUuid: 'step-0',
part: { type: 'text', text: 'partial answer' },
},
},
{ type: 'turn.cancel', turnId: 0, target: 'active', reason: 'user_cancelled' },
] as unknown as WireRecord[]);
const ctx = testAgent({ persistence, autoConfigure: false });
try {
await ctx.restorePersisted();
expect(ctx.context.get()).toContainEqual(
expect.objectContaining({
role: 'user',
origin: { kind: 'injection', variant: 'interruption' },
}),
);
expect(persistence.appended).toContainEqual(
expect.objectContaining({
type: 'context.append_message',
message: expect.objectContaining({
origin: { kind: 'injection', variant: 'interruption' },
}),
}),
);
expect(persistence.appended).toContainEqual(
expect.objectContaining({ type: 'interruptionReminder.recorded', turnId: 0 }),
);
await ctx.expectResumeMatches();
} finally {
await ctx.dispose();
}
});
it('replays persisted records without restarting turns, compactions, plan turns, or tools', async () => {
const persistence = new RecordingAgentPersistence(resumeHistory() as unknown as WireRecord[]);
const execWithEnv = vi.fn().mockRejectedValue(new Error('Bash should not execute on resume'));

3
packages/kap-server/src/protocol/events-zod.ts
View file

@ -646,6 +646,9 @@ export const turnEndedEventSchema = z.object({
reason: turnEndReasonSchema,
error: kimiErrorPayloadSchema.optional(),
durationMs: z.number().optional(),
interruptReason: z
.enum(['user_cancelled', 'aborted', 'max_steps', 'error', 'filtered', 'blocked'])
.optional(),
});
export const turnStepStartedEventSchema = z.object({

10
packages/kap-server/src/services/transcript/coreEventMap.ts
View file

@ -323,6 +323,7 @@ export class AgentTranscriptProjector {
reason: 'completed' | 'cancelled' | 'failed' | 'blocked';
error?: { message: string };
durationMs?: number;
interruptReason?: string;
}): TranscriptOperation[] {
const ops: TranscriptOperation[] = [];
this.flushOpenFrames(ops);
@ -351,6 +352,15 @@ export class AgentTranscriptProjector {
};
ops.push({ op: 'turn.upsert', turn: this.currentTurn });
this.currentStep = undefined;
// The user-facing counterpart of the (hidden) context reminder: a
// deliberate user interrupt gets a timeline marker, mirroring the cold
// fold's `turn.cancel` handling. Programmatic aborts already surface
// through the turn's error field or goal/task state.
if (event.reason === 'cancelled' && event.interruptReason === 'user_cancelled') {
ops.push(
this.markerOp('interruption', { turnId: event.turnId, reason: event.interruptReason }),
);
}
return ops;
}

24
packages/kap-server/test/services/transcript.test.ts
View file

@ -419,6 +419,30 @@ describe('AgentTranscriptProjector', () => {
);
});
it('marks a user-cancelled turn with an interruption marker, but not programmatic aborts', () => {
const projector = new AgentTranscriptProjector('main');
const tx = new AgentTranscript('main');
const feed = (event: DomainEvent): void => void tx.apply(projector.map(event));
feed(ev({ type: 'turn.started', turnId: 0, origin: { kind: 'user' }, prompt: 'hi' }));
feed(
ev({ type: 'turn.ended', turnId: 0, reason: 'cancelled', interruptReason: 'user_cancelled' }),
);
feed(ev({ type: 'turn.started', turnId: 1, origin: { kind: 'user' }, prompt: 'again' }));
feed(ev({ type: 'turn.ended', turnId: 1, reason: 'cancelled', interruptReason: 'aborted' }));
feed(ev({ type: 'turn.started', turnId: 2, origin: { kind: 'user' }, prompt: 'legacy' }));
feed(ev({ type: 'turn.ended', turnId: 2, reason: 'cancelled' }));
const markers = tx
.getItems()
.filter((item): item is Extract<typeof item, { kind: 'marker' }> => item.kind === 'marker');
expect(markers).toHaveLength(1);
expect(markers[0]).toMatchObject({
marker: 'interruption',
payload: { turnId: 0, reason: 'user_cancelled' },
});
});
it('carries usage / finishReason / the full timing breakdown on turn.step.completed', () => {
const projector = new AgentTranscriptProjector('main');
const tx = new AgentTranscript('main');

4
packages/klient/src/contract/agent/events.ts
View file

@ -43,6 +43,10 @@ export const turnEndedEventSchema = z.object({
/** Protocol `KimiErrorPayload` — mirrored as `unknown`. */
error: z.unknown().optional(),
durationMs: z.number().optional(),
/** Why a non-completed turn stopped early; absent on completion. */
interruptReason: z
.enum(['user_cancelled', 'aborted', 'max_steps', 'error', 'filtered', 'blocked'])
.optional(),
});
export const assistantDeltaEventSchema = z.object({

13
packages/klient/test/e2e/invalid-input-matrix.test.ts
View file

@ -992,7 +992,7 @@ describe('tool exchange structure', () => {
}
}, 60_000);
it('after an abort, consecutive user messages are merged into one on the wire (projector fallback)', async () => {
it('after a user abort, an interruption reminder separates the next user message on the wire', async () => {
const ctx = await newCase(M_OPENAI, 'abort-merge');
resetMock((_req, callIndex) => (callIndex === 0 ? { kind: 'hang' } : OK_OPENAI));
@ -1006,11 +1006,14 @@ describe('tool exchange structure', () => {
expect(requests).toHaveLength(2);
const userMessages = openAiMessages(1).filter((message) => message['role'] === 'user');
// The projector folds consecutive origin=user messages into one
// (\n\n-joined) instead of sending two same-role messages in a row.
expect(userMessages).toHaveLength(1);
// A deliberate user cancel injects an interruption reminder between the
// aborted turn's prompt and the next user message, so the two prompts no
// longer merge into one wire message.
expect(userMessages).toHaveLength(3);
expect(String(userMessages[0]?.['content'])).toContain('first message');
expect(String(userMessages[0]?.['content'])).toContain('second message');
expect(String(userMessages[1]?.['content'])).toContain('<system-reminder>');
expect(String(userMessages[1]?.['content'])).toContain('interrupted by the user');
expect(String(userMessages[2]?.['content'])).toContain('second message');
expect(ctx.payloads('prompt.completed')[0]?.['reason']).toBe('completed');
}, 60_000);
});

39
packages/transcript/src/history/foldFacts.ts
View file

@ -119,6 +119,7 @@ interface TurnPromptPayload {
interface TurnCancelPayload {
readonly turnId?: unknown;
readonly target?: unknown;
readonly reason?: unknown;
}
/**
@ -269,6 +270,7 @@ export function foldWireRecordFacts(
/** Markers/taskrefs generated by the fold, appended after the base items. */
const appended: TranscriptItem[] = [];
const activeCancelTurnIds = new Set<number>();
// Marker ids continue the base's `m<N>` numbering (groupTurns uses the same
// namespace); taskref ids dedupe against refs the base already carries.
let markerSeq = 0;
@ -430,6 +432,31 @@ export function foldWireRecordFacts(
upsertTask(record);
break;
}
case 'turn.cancel': {
const payload = record as TurnCancelPayload;
if (
payload.target === 'queued' &&
typeof payload.turnId === 'number' &&
payload.turnId >= nextTurnId
) {
cancelledTurnIds.add(payload.turnId);
skipCancelledTurnIds();
break;
}
if (
payload.target !== 'active' ||
typeof payload.turnId !== 'number' ||
!Number.isInteger(payload.turnId) ||
payload.turnId < 0 ||
activeCancelTurnIds.has(payload.turnId)
) {
break;
}
activeCancelTurnIds.add(payload.turnId);
if (payload.reason !== 'user_cancelled') break;
pushMarker('interruption', record);
break;
}
case 'interaction.request': {
const payload = record as InteractionRequestPayload;
// The live path projects only approvals/questions (`user_tool`
@ -477,18 +504,6 @@ export function foldWireRecordFacts(
if (!isVisibleTurnOrigin((record as TurnPromptPayload).origin)) hiddenTurnIds.add(turnId);
break;
}
case 'turn.cancel': {
const payload = record as TurnCancelPayload;
if (
payload.target === 'queued' &&
typeof payload.turnId === 'number' &&
payload.turnId >= nextTurnId
) {
cancelledTurnIds.add(payload.turnId);
skipCancelledTurnIds();
}
break;
}
default:
break;
}

30
packages/transcript/test/layers.test.ts
View file

@ -750,6 +750,36 @@ describe('foldWireRecordFacts (cold facts)', () => {
expect(cleared.meta.goal).toBeUndefined();
});
it('marks user-cancelled turns with interruption markers, skipping unattributable cancels', () => {
const base = baseWithMarker();
const folded = foldWireRecordFacts(
[
{ type: 'turn.cancel', turnId: 0, target: 'active', reason: 'user_cancelled', time: 1000 },
{ type: 'turn.cancel', turnId: 0, target: 'active', reason: 'user_cancelled', time: 1500 },
// A queued cancel left no visible residue — no marker.
{ type: 'turn.cancel', turnId: 1, target: 'queued', reason: 'user_cancelled', time: 2000 },
// Programmatic aborts surface through their own outlets — no marker.
{ type: 'turn.cancel', turnId: 2, target: 'active', reason: 'aborted', time: 3000 },
{ type: 'turn.cancel', turnId: 4, target: 'active', reason: 'aborted', time: 3500 },
{ type: 'turn.cancel', turnId: 4, target: 'active', reason: 'user_cancelled', time: 3600 },
// Records written before the reason field existed cannot be attributed.
{ type: 'turn.cancel', turnId: 3, target: 'active', time: 4000 },
],
base,
);
expect(
folded.items.filter((item) => item.kind === 'marker' && item.marker === 'interruption'),
).toEqual([
{
kind: 'marker',
markerId: 'm2',
marker: 'interruption',
payload: { turnId: 0, target: 'active', reason: 'user_cancelled' },
at: new Date(1000).toISOString(),
},
]);
});
it('folds plan/swarm mode records into meta.modes with enter/exit markers', () => {
const base = baseWithMarker();
const folded = foldWireRecordFacts(