feat(kap-server): add the WS v2 realtime protocol stack

.changeset/fix-background-task-notification.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Show background task completion notices in the session message stream.

.changeset/fix-empty-thinking-parts.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix empty thinking entries recorded for steps without reasoning content.

.changeset/fix-history-paging-past-compaction.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix earlier-message history paging returning empty pages before compaction points.

.changeset/fix-plan-approval-display.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix plan approvals and plan blocks missing the plan content in the v2 message stream.

.changeset/fix-steer-duplicate-user-message.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix steered prompts appearing twice in the session message stream.

.changeset/goal-status-transitions-only.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Emit goal status stream entries only when the goal status actually changes.

.changeset/skill-activation-user-message.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Show skill activations as user messages in the session message stream.

packages/agent-core-v2/docs/wire-manifest.d.ts

@@ -505,6 +505,7 @@ interface PlanRevisionPayload {
   key: string;
   sha256: string;
   bytes: number;
+  summary?: string;
 }
 
 /**

packages/agent-core-v2/src/agent/loop/loopService.ts

@@ -932,6 +932,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
     response: AgentLLMRequestFinish,
   ): void {
     for (const part of response.message.content) {
+      if (part.type === 'think' && part.think === '' && (part as { encrypted?: string }).encrypted === undefined) continue;
       this.context.appendLoopEvent({
         type: 'content.part',
         uuid: randomUUID(),
@@ -950,6 +951,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
     streamParts: StreamPartCollector,
   ): void {
     for (const part of streamParts.drainInterruptedContent()) {
+      if (part.type === 'think' && part.think === '' && (part as { encrypted?: string }).encrypted === undefined) continue;
       this.context.appendLoopEvent({
         type: 'content.part',
         uuid: randomUUID(),
@@ -1144,13 +1146,19 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
               new AssistantDelta({ agentId: this.scopeContext.agentId, turnId, delta: part.text }),
             );
             return;
-          case 'think':
+          case 'think': {
+            const hasPayload =
+              part.think !== '' || (part as { encrypted?: string }).encrypted !== undefined;
+            if (!hasPayload) return;
             onResponseEvent();
             accumulate(part);
-            void this.dispatcher.dispatch(
-              new ThinkingDelta({ agentId: this.scopeContext.agentId, turnId, delta: part.think }),
-            );
+            if (part.think !== '') {
+              void this.dispatcher.dispatch(
+                new ThinkingDelta({ agentId: this.scopeContext.agentId, turnId, delta: part.think }),
+              );
+            }
             return;
+          }
           case 'image_url':
           case 'audio_url':
           case 'video_url':

packages/agent-core-v2/src/agent/prompt/prompt.ts

@@ -65,7 +65,7 @@ export interface PromptLaunchResult {
 
 export interface PromptReservation extends IDisposable {
   readonly id: string;
-  submit(message: ContextMessage): Promise<PromptHandle>;
+  submit(message: ContextMessage, opts?: { steer?: boolean }): Promise<PromptHandle>;
 }
 
 export const promptAdmission = Symbol('promptAdmission');

packages/agent-core-v2/src/agent/prompt/promptService.ts

@@ -9,7 +9,7 @@ import { extractImageCompressionCaptions } from '#/agent/media/image-compress';
 import { userCancellationReason } from '#/_base/utils/abort';
 import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory';
 import { newMessageId } from '#/agent/contextMemory/messageId';
-import { USER_PROMPT_ORIGIN, type ContextMessage } from '#/agent/contextMemory/types';
+import { USER_PROMPT_ORIGIN, type ContextMessage, type PromptOrigin } from '#/agent/contextMemory/types';
 import { IAgentFullCompactionService } from '#/agent/fullCompaction/fullCompaction';
 import { IAgentLoopService, type Turn, type TurnResult } from '#/agent/loop/loop';
 import { TurnSteer } from '#/agent/loop/turnOps';
@@ -157,6 +157,8 @@ export interface PromptSubmittedPayload {
   readonly status: 'running' | 'queued';
   readonly content: ContentPart[];
   readonly createdAt: string;
+  readonly steer?: boolean;
+  readonly origin?: PromptOrigin;
 }
 
 export class PromptSubmitted extends AgentEvent2<PromptSubmittedPayload> {
@@ -179,6 +181,7 @@ export interface PromptStarted extends PromptStartedPayload {}
 interface Deferred<T> { readonly promise: Promise<T>; resolve(value: T): void; reject(reason: unknown): void }
 interface Record extends PromptSnapshot {
   state: PromptState;
+  steerIntended: boolean;
   readonly launchedDeferred: Deferred<Turn | undefined>;
   readonly completionDeferred: Deferred<PromptCompletion>;
   handle: PromptHandle;
@@ -277,7 +280,7 @@ export class AgentPromptService implements IAgentPromptService {
     let submitted = false;
     return {
       id,
-      submit: async (message) => {
+      submit: async (message, opts) => {
         if (submitted) throw new Error2(ErrorCodes.REQUEST_INVALID, 'prompt reservation already submitted');
         submitted = true;
         this.reservedPromptIds.delete(id);
@@ -288,23 +291,23 @@ export class AgentPromptService implements IAgentPromptService {
             content: stripBundledSkillBlocks(message),
           }),
         );
-        return this.enqueue({ id, message });
+        return this.enqueue({ id, message }, opts);
       },
       dispose: () => {
         this.reservedPromptIds.delete(id);
       },
     };
   }
 
-  async enqueue(input: PromptInput): Promise<PromptHandle> {
+  async enqueue(input: PromptInput, opts?: { steer?: boolean }): Promise<PromptHandle> {
     const id = input.id ?? input.message.id ?? newMessageId();
     const message = { ...input.message, id };
     const launchedDeferred = deferred<Turn | undefined>();
     const completionDeferred = deferred<PromptCompletion>();
     const record = {} as Record;
     Object.assign(record, {
       id, userMessageId: id, createdAt: new Date().toISOString(), state: 'pending', message,
-      launchedDeferred, completionDeferred,
+      launchedDeferred, completionDeferred, steerIntended: opts?.steer === true,
     });
     record.handle = {
       get id() { return record.id; }, get userMessageId() { return record.userMessageId; },
@@ -360,7 +363,7 @@ export class AgentPromptService implements IAgentPromptService {
       role: 'user',
       content: [...payload.input],
       toolCalls: [],
-    } });
+    } }, { steer: true });
     if (queued.state !== 'pending') {
       const turn = await queued.launched;
       return turn === undefined ? undefined : { turn_id: turn.id };
@@ -570,7 +573,7 @@ export class AgentPromptService implements IAgentPromptService {
   }
   private publishSubmitted(record: Record, status: 'running' | 'queued'): void {
     if ((record.message.origin ?? USER_PROMPT_ORIGIN).kind !== 'user') return;
-    void this.dispatcher.dispatch(new PromptSubmitted({ agentId: this.scopeContext.agentId, promptId: record.id, userMessageId: record.userMessageId, status, content: stripBundledSkillBlocks(record.message), createdAt: record.createdAt }));
+    void this.dispatcher.dispatch(new PromptSubmitted({ agentId: this.scopeContext.agentId, promptId: record.id, userMessageId: record.userMessageId, status, content: stripBundledSkillBlocks(record.message), createdAt: record.createdAt, steer: record.steerIntended ? true : undefined, origin: record.message.origin }));
   }
   private publishStarted(record: Record): void {
     if ((record.message.origin ?? USER_PROMPT_ORIGIN).kind !== 'user') return;

packages/agent-core-v2/src/agent/task/task.ts

@@ -64,6 +64,8 @@ export interface AgentTaskNotificationContext {
   readonly severity: 'info' | 'warning';
   readonly sourceKind: string;
   readonly sourceId: string;
+  readonly sourceAgentId?: string;
+  readonly raw?: string;
 }
 
 export interface AgentTaskWaitDelivery {

packages/agent-core-v2/src/agent/task/taskService.ts

@@ -1301,6 +1301,8 @@ export class AgentTaskService extends Disposable implements IAgentTaskService {
         severity: notification.severity,
         sourceKind: notification.source_kind,
         sourceId: notification.source_id,
+        sourceAgentId: typeof notification.agent_id === 'string' ? notification.agent_id : undefined,
+        raw: renderNotificationXml(notification),
       }),
     );
   }

packages/agent-core-v2/src/features/plan/planOps.ts

@@ -62,6 +62,7 @@ export interface PlanRevisionRecordedEvent {
   readonly key: string;
   readonly sha256: string;
   readonly bytes: number;
+  readonly summary?: string;
 }
 
 const planRevisionSchema = z.object({
@@ -71,6 +72,7 @@ const planRevisionSchema = z.object({
   key: z.string(),
   sha256: z.string(),
   bytes: z.number(),
+  summary: z.string().optional(),
 });
 
 export class PlanRevision extends AgentEvent2<PlanRevisionRecordedEvent> {

packages/agent-core-v2/src/features/plan/planService.ts

@@ -208,6 +208,11 @@ export class AgentPlanService extends Service implements IAgentPlanService {
     const scope = this.agentCtx.scope();
     const key = `plan/${id}/v${version}.md`;
     await this.blobs.put(scope, key, bytes);
+    const heading = content
+      .split('\n')
+      .map((line) => line.trim())
+      .find((line) => line.startsWith('#'));
+    const summary = heading === undefined ? undefined : heading.replace(/^#+\s*/, '').trim() || undefined;
     await this.dispatcher.dispatch(
       new PlanRevision({
         agentId: this.agentCtx.agentId,
@@ -216,6 +221,7 @@ export class AgentPlanService extends Service implements IAgentPlanService {
         key,
         sha256: createHash('sha256').update(bytes).digest('hex'),
         bytes: bytes.byteLength,
+        summary,
       }),
     );
   }

packages/agent-core-v2/src/features/todo/todoOps.ts

@@ -16,6 +16,7 @@ const toolsUpdateStoreSchema = z.object({
 export class ToolsUpdateStore extends AgentEvent2<z.infer<typeof toolsUpdateStoreSchema>> {
   static override readonly type = 'tools.update_store';
   static override readonly durable = true;
+  static override readonly observable = true;
   static override readonly schema = toolsUpdateStoreSchema;
 }
 export interface ToolsUpdateStore {

packages/agent-core-v2/test/features/plan/plan.test.ts

@@ -301,6 +301,7 @@
           key: revisionKey('rev-plan', 1),
           sha256: createHash('sha256').update(content, 'utf8').digest('hex'),
           bytes: Buffer.byteLength(content),
+          summary: 'Plan',
           time: expect.any(Number),
         },
       ]);
@@ -707,7 +708,7 @@
       ctx.mockNextResponse({ type: 'text', text: 'The safe command printed plan-safe.' });
       await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Inspect without mutating files' }] });
 
       expect(await ctx.untilTurnEnd()).toMatchInlineSnapshot(`
         [wire] permission.set_mode         { "agentId": "main", "mode": "yolo", "time": "<time>" }
         [wire] plan_mode.enter             { "agentId": "main", "id": "test-plan", "time": "<time>" }
         [emit] agent.status.updated        { "time": "<time>", "agentId": "main", "planMode": true }
@@ -790,7 +791,7 @@
       ctx.mockNextResponse({ type: 'text', text: 'The command completed.' });
       await ctx.rpc.prompt({ input: [{ type: 'text', text: 'Remove forbidden.txt' }] });
 
       expect(await ctx.untilTurnEnd()).toMatchInlineSnapshot(`
         [wire] permission.set_mode         { "agentId": "main", "mode": "yolo", "time": "<time>" }
         [wire] plan_mode.enter             { "agentId": "main", "id": "test-plan", "time": "<time>" }
         [emit] agent.status.updated        { "time": "<time>", "agentId": "main", "planMode": true }