From b1103bb3431b5de9911011768095f885ee4f6ef6 Mon Sep 17 00:00:00 2001 From: 7Sageer <7sageer@djwcb.cn> Date: Mon, 6 Jul 2026 16:34:38 +0800 Subject: [PATCH] fix(server-v2): broadcast interaction question/approval events over v1 WS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 5 + .../ws/v1/sessionEventBroadcaster.ts | 137 +++++++++++++- .../test/sessionEventBroadcaster.test.ts | 170 +++++++++++++++++- 3 files changed, 304 insertions(+), 8 deletions(-) create mode 100644 .changeset/server-v2-interaction-events.md diff --git a/.changeset/server-v2-interaction-events.md b/.changeset/server-v2-interaction-events.md new file mode 100644 index 000000000..188658489 --- /dev/null +++ b/.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. diff --git a/packages/server-v2/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/server-v2/src/transport/ws/v1/sessionEventBroadcaster.ts index bcc4251cb..3b6f9bd66 100644 --- a/packages/server-v2/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/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; readonly lifecycleDisposables: IDisposable[]; + /** Interactions already announced (or pre-existing at activation): id → kind. */ + readonly knownInteractions: Map; } 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 { 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; + 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 | undefined { diff --git a/packages/server-v2/test/sessionEventBroadcaster.test.ts b/packages/server-v2/test/sessionEventBroadcaster.test.ts index a9fa1e6f4..370174ab1 100644 --- a/packages/server-v2/test/sessionEventBroadcaster.test.ts +++ b/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, 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']); + }); });