fix(server-v2): broadcast interaction question/approval events over v1 WS

The v1-compat SessionEventBroadcaster only forwarded agent wire signals
and model catalog events, so pending questions and approvals created via
the session interaction service never reached v1 WS clients (kimi-web)
in real time — they only appeared after a page refresh via the REST
snapshot. Subscribe to ISessionInteractionService per session and
synthesize durable event.question.requested/answered/dismissed and
event.approval.requested/resolved envelopes through the existing
seq/journal/fan-out pipeline.

.changeset/server-v2-interaction-events.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fix approval and question prompts not appearing in real time for web clients connected to the v2 server; they previously only showed up after a page refresh.

packages/server-v2/src/transport/ws/v1/sessionEventBroadcaster.ts

@@ -10,7 +10,9 @@
  *   1. Subscribes to every agent's `IAgentWireService.onEmission` via
  *      `IAgentLifecycleService` reach-down-via-handle (and `onDidCreate` /
  *      `onDidDispose` for late agents); `record` emissions are persisted and not
- *      broadcast (see step 3).
+ *      broadcast (see step 3). Also subscribes to the session's
+ *      `ISessionInteractionService` and synthesizes the v1 approval/question
+ *      protocol events from pending-set changes and resolutions.
  *   2. Attaches `agentId`/`sessionId` to build the wire `Event`.
  *   3. Classifies durable vs volatile — `isVolatileSignal` for the agent
  *      wire-emission path (`isVolatileEventType` remains for the global/model path).
@@ -28,9 +30,12 @@
  */
 
 import type {
+  ApprovalResponse,
   DomainEvent,
   IAgentScopeHandle,
   IDisposable,
+  Interaction,
+  InteractionKind,
   ISessionScopeHandle,
   Scope,
   WireEmission,
@@ -39,6 +44,7 @@ import {
   IAgentLifecycleService,
   IAgentWireService,
   IEventService,
+  ISessionInteractionService,
   ISessionLifecycleService,
 } from '@moonshot-ai/agent-core-v2';
 import type {
@@ -49,6 +55,8 @@ import type {
 } from '@moonshot-ai/protocol';
 import { isVolatileEventType } from '@moonshot-ai/protocol';
 
+import { toWireApproval } from '../../../routes/approvals';
+import { toWireQuestion } from '../../../routes/questions';
 import { InFlightTurnTracker } from './inFlightTurnTracker';
 import {
   type EventEnvelope,
@@ -91,6 +99,8 @@ interface SessionState {
   /** agentId → sink subscription. */
   readonly agentDisposables: Map<string, IDisposable>;
   readonly lifecycleDisposables: IDisposable[];
+  /** Interactions already announced (or pre-existing at activation): id → kind. */
+  readonly knownInteractions: Map<string, InteractionKind>;
 }
 
 export const DEFAULT_MAX_BUFFER_SIZE = 1000;
@@ -211,9 +221,11 @@ export class SessionEventBroadcaster {
       queue: Promise.resolve(),
       agentDisposables: new Map(),
       lifecycleDisposables: [],
+      knownInteractions: new Map(),
     };
     this.sessions.set(sessionId, state);
     this.attachAgents(sessionId, session, state);
+    this.attachInteractions(sessionId, session, state);
     return state;
   }
 
@@ -234,6 +246,7 @@ export class SessionEventBroadcaster {
       queue: Promise.resolve(),
       agentDisposables: new Map(),
       lifecycleDisposables: [],
+      knownInteractions: new Map(),
     };
     this.sessions.set(GLOBAL_SESSION_ID, state);
     return state;
@@ -302,6 +315,49 @@ export class SessionEventBroadcaster {
       .catch(() => {});
   }
 
+  /**
+   * Bridge the session's interaction kernel (approvals / questions) onto the
+   * v1 event stream. The kernel only emits in-process notifications
+   * (`onDidChangePending` / `onDidResolve`), so the v1 protocol events
+   * (`event.question.requested`, `event.approval.requested`, ...) are
+   * synthesized here — mirroring what v1's question/approval services
+   * published through the Core firehose.
+   */
+  private attachInteractions(
+    sessionId: string,
+    session: ISessionScopeHandle,
+    state: SessionState,
+  ): void {
+    const interactions = session.accessor.get(ISessionInteractionService);
+    // Seed silently: interactions already pending at activation are surfaced
+    // by the snapshot route (`pending_questions` / `pending_approvals`), so
+    // announcing them again would duplicate the snapshot.
+    for (const i of interactions.listPending()) {
+      state.knownInteractions.set(i.id, i.kind);
+    }
+    state.lifecycleDisposables.push(
+      interactions.onDidChangePending(() => {
+        for (const i of interactions.listPending()) {
+          if (state.knownInteractions.has(i.id)) continue;
+          state.knownInteractions.set(i.id, i.kind);
+          const event = interactionRequestedEvent(i, sessionId);
+          if (event !== undefined) this.enqueueDurable(state, event);
+        }
+      }),
+      interactions.onDidResolve(({ id, response }) => {
+        const kind = state.knownInteractions.get(id);
+        if (kind === undefined) return;
+        state.knownInteractions.delete(id);
+        const event = interactionResolvedEvent(kind, id, response, sessionId);
+        if (event !== undefined) this.enqueueDurable(state, event);
+      }),
+    );
+  }
+
+  private enqueueDurable(state: SessionState, event: Event): void {
+    state.queue = state.queue.then(() => this.dispatch(state, event, false)).catch(() => {});
+  }
+
   private async dispatch(state: SessionState, event: Event, volatile: boolean): Promise<void> {
     const { journal, tracker, tail, targets, sessionId } = state;
     const annotation = tracker.apply(sessionId, event);
@@ -388,6 +444,85 @@ function isGlobalEvent(type: string): boolean {
   );
 }
 
+// ---------------------------------------------------------------------------
+// Interaction → v1 protocol event synthesis. Event names and payload shapes
+// mirror v1's question/approval services
+// (`packages/server/src/services/{question,approval}/*Service.ts`); the wire
+// request bodies are the same projections the REST/snapshot routes use.
+// ---------------------------------------------------------------------------
+
+function interactionRequestedEvent(interaction: Interaction, sessionId: string): Event | undefined {
+  const agentId = interaction.origin.agentId ?? 'main';
+  switch (interaction.kind) {
+    case 'question':
+      return {
+        type: 'event.question.requested',
+        agentId,
+        sessionId,
+        ...toWireQuestion(interaction, sessionId),
+      } as unknown as Event;
+    case 'approval':
+      return {
+        type: 'event.approval.requested',
+        agentId,
+        sessionId,
+        ...toWireApproval(interaction, sessionId),
+      } as unknown as Event;
+    default:
+      // 'user_tool' has no v1 protocol event.
+      return undefined;
+  }
+}
+
+function interactionResolvedEvent(
+  kind: InteractionKind,
+  id: string,
+  response: unknown,
+  sessionId: string,
+): Event | undefined {
+  const resolvedAt = new Date().toISOString();
+  switch (kind) {
+    case 'question': {
+      // `null` marks a dismissal (see `ISessionQuestionService.dismiss`).
+      if (response === null) {
+        return {
+          type: 'event.question.dismissed',
+          agentId: 'main',
+          sessionId,
+          question_id: id,
+          dismissed_at: resolvedAt,
+        } as unknown as Event;
+      }
+      // `QuestionResult` is either `{ answers, method? }` or a bare answers record.
+      const answers = (response as { answers?: unknown }).answers ?? response;
+      return {
+        type: 'event.question.answered',
+        agentId: 'main',
+        sessionId,
+        question_id: id,
+        answers,
+        resolved_at: resolvedAt,
+      } as unknown as Event;
+    }
+    case 'approval': {
+      const r = response as Partial<ApprovalResponse>;
+      return {
+        type: 'event.approval.resolved',
+        agentId: 'main',
+        sessionId,
+        approval_id: id,
+        decision: r.decision,
+        scope: r.scope,
+        feedback: r.feedback,
+        selected_label: r.selectedLabel,
+        resolved_at: resolvedAt,
+      } as unknown as Event;
+    }
+    default:
+      return undefined;
+  }
+}
+
 function modelCatalogChangedPayload(
   payload: unknown,
 ): Pick<ModelCatalogChangedEvent, 'changed' | 'unchanged' | 'failed'> | undefined {

packages/server-v2/test/sessionEventBroadcaster.test.ts

@@ -6,12 +6,14 @@ import { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 
-import type { IScopeHandle, Scope } from '@moonshot-ai/agent-core-v2';
+import type { IScopeHandle, Scope, WireEmission } from '@moonshot-ai/agent-core-v2';
 import {
-  IAgentEventSinkService,
   IAgentLifecycleService,
+  IAgentWireService,
   IEventService,
+  ISessionInteractionService,
   ISessionLifecycleService,
+  SessionInteractionService,
 } from '@moonshot-ai/agent-core-v2';
 import type { AgentEvent } from '@moonshot-ai/protocol';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
@@ -27,8 +29,8 @@ import type { EventEnvelope } from '../src/transport/ws/v1/sessionEventJournal';
 // ---------------------------------------------------------------------------
 
 class FakeSink {
-  private handlers: Array<(e: AgentEvent) => void> = [];
-  on(handler: (e: AgentEvent) => void) {
+  private handlers: Array<(e: WireEmission) => void> = [];
+  onEmission(handler: (e: WireEmission) => void) {
     this.handlers.push(handler);
     return {
       dispose: () => {
@@ -38,7 +40,8 @@ class FakeSink {
     };
   }
   emit(e: AgentEvent): void {
-    for (const h of [...this.handlers]) h(e);
+    const emission = { type: 'signal', signal: e } as unknown as WireEmission;
+    for (const h of [...this.handlers]) h(emission);
   }
 }
 
@@ -64,7 +67,7 @@ class FakeAgentHandle {
   readonly accessor;
   constructor(readonly id: string) {
     this.accessor = {
-      get: (t: unknown) => (t === IAgentEventSinkService ? this.sink : undefined),
+      get: (t: unknown) => (t === IAgentWireService ? this.sink : undefined),
     };
   }
   dispose(): void {}
@@ -72,6 +75,8 @@ class FakeAgentHandle {
 
 class FakeLifecycle {
   readonly handles: FakeAgentHandle[] = [];
+  /** Real interaction kernel — served at the session accessor. */
+  readonly interactions = new SessionInteractionService();
   private createHandlers: Array<(h: IScopeHandle) => void> = [];
   private disposeHandlers: Array<(id: string) => void> = [];
   list(): readonly FakeAgentHandle[] {
@@ -111,7 +116,11 @@ function makeCore(sessions: Map<string, FakeLifecycle>, eventBus = new FakeEvent
             const lifecycle = sessions.get(sid);
             if (lifecycle === undefined) return undefined;
             const sessionAccessor = {
-              get: (t: unknown) => (t === IAgentLifecycleService ? lifecycle : undefined),
+              get: (t: unknown) => {
+                if (t === IAgentLifecycleService) return lifecycle;
+                if (t === ISessionInteractionService) return lifecycle.interactions;
+                return undefined;
+              },
             };
             return { id: sid, kind: 1, accessor: sessionAccessor, dispose: () => {} };
           },
@@ -297,4 +306,151 @@ describe('SessionEventBroadcaster', () => {
     const { target } = collectingTarget();
     expect(await bc.subscribe('nope', target)).toBe(false);
   });
+
+  it('broadcasts question requested / answered as durable v1 events', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    const { target, envelopes } = collectingTarget();
+    await bc.subscribe('s1', target);
+
+    lc.interactions.enqueue({
+      id: 'q1',
+      kind: 'question',
+      payload: {
+        toolCallId: 'call_1',
+        questions: [{ question: 'Pick one', options: [{ label: 'A' }, { label: 'B' }] }],
+      },
+    });
+    await bc.getCursor('s1');
+
+    expect(envelopes).toHaveLength(1);
+    expect(envelopes[0]).toMatchObject({
+      type: 'event.question.requested',
+      seq: 1,
+      session_id: 's1',
+      payload: {
+        type: 'event.question.requested',
+        agentId: 'main',
+        sessionId: 's1',
+        question_id: 'q1',
+        session_id: 's1',
+        tool_call_id: 'call_1',
+        questions: [{ id: 'q_0', question: 'Pick one', options: [{ id: 'opt_0_0', label: 'A' }, { id: 'opt_0_1', label: 'B' }] }],
+      },
+    });
+    expect(envelopes[0]!.volatile).toBeUndefined();
+
+    lc.interactions.respond('q1', { answers: { q_0: 'opt_0_0' }, method: 'enter' });
+    await bc.getCursor('s1');
+
+    expect(envelopes).toHaveLength(2);
+    expect(envelopes[1]).toMatchObject({
+      type: 'event.question.answered',
+      seq: 2,
+      session_id: 's1',
+      payload: {
+        question_id: 'q1',
+        answers: { q_0: 'opt_0_0' },
+      },
+    });
+    expect((envelopes[1]!.payload as { resolved_at?: string }).resolved_at).toBeTypeOf('string');
+  });
+
+  it('broadcasts question dismissed when resolved with null', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    const { target, envelopes } = collectingTarget();
+    await bc.subscribe('s1', target);
+
+    lc.interactions.enqueue({
+      id: 'q1',
+      kind: 'question',
+      payload: { questions: [{ question: 'Pick', options: [{ label: 'A' }] }] },
+    });
+    lc.interactions.respond('q1', null); // = ISessionQuestionService.dismiss
+    await bc.getCursor('s1');
+
+    expect(envelopes.map((e) => e.type)).toEqual([
+      'event.question.requested',
+      'event.question.dismissed',
+    ]);
+    expect(envelopes[1]!.payload).toMatchObject({ question_id: 'q1' });
+    expect((envelopes[1]!.payload as { dismissed_at?: string }).dismissed_at).toBeTypeOf('string');
+  });
+
+  it('broadcasts approval requested / resolved as durable v1 events', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    const { target, envelopes } = collectingTarget();
+    await bc.subscribe('s1', target);
+
+    lc.interactions.enqueue({
+      id: 'a1',
+      kind: 'approval',
+      payload: {
+        toolCallId: 'call_9',
+        toolName: 'Bash',
+        action: 'run',
+        display: { kind: 'command', command: 'ls' },
+      },
+      origin: { turnId: 3 },
+    });
+    await bc.getCursor('s1');
+
+    expect(envelopes).toHaveLength(1);
+    expect(envelopes[0]).toMatchObject({
+      type: 'event.approval.requested',
+      seq: 1,
+      session_id: 's1',
+      payload: {
+        approval_id: 'a1',
+        session_id: 's1',
+        turn_id: 3,
+        tool_call_id: 'call_9',
+        tool_name: 'Bash',
+        action: 'run',
+        tool_input_display: { kind: 'command', command: 'ls' },
+      },
+    });
+
+    lc.interactions.respond('a1', { decision: 'approved', scope: 'session' });
+    await bc.getCursor('s1');
+
+    expect(envelopes).toHaveLength(2);
+    expect(envelopes[1]).toMatchObject({
+      type: 'event.approval.resolved',
+      seq: 2,
+      session_id: 's1',
+      payload: {
+        approval_id: 'a1',
+        decision: 'approved',
+        scope: 'session',
+      },
+    });
+    expect((envelopes[1]!.payload as { resolved_at?: string }).resolved_at).toBeTypeOf('string');
+  });
+
+  it('does not re-announce interactions already pending at activation, but still broadcasts their resolution', async () => {
+    const lc = new FakeLifecycle();
+    lc.addAgent('main');
+    sessions.set('s1', lc);
+    // Pending before the session is activated — the snapshot covers it.
+    lc.interactions.enqueue({
+      id: 'q0',
+      kind: 'question',
+      payload: { questions: [{ question: 'Early', options: [{ label: 'A' }] }] },
+    });
+
+    const { target, envelopes } = collectingTarget();
+    await bc.subscribe('s1', target);
+    await bc.getCursor('s1');
+    expect(envelopes).toHaveLength(0);
+
+    lc.interactions.respond('q0', { answers: { q_0: 'opt_0_0' } });
+    await bc.getCursor('s1');
+    expect(envelopes.map((e) => e.type)).toEqual(['event.question.answered']);
+  });
 });