diff --git a/apps/kimi-web/src/api/daemon/agentEventProjector.ts b/apps/kimi-web/src/api/daemon/agentEventProjector.ts index cf2ec7030..58af26e4b 100644 --- a/apps/kimi-web/src/api/daemon/agentEventProjector.ts +++ b/apps/kimi-web/src/api/daemon/agentEventProjector.ts @@ -16,8 +16,10 @@ // const appEvents = projector.project(rawType, payload, sessionId); // // call reset() when re-subscribing / resyncing a session -import type { AppEvent, AppInFlightTurn, AppMessage, AppSessionUsage } from '../types'; +import type { AppEvent, AppInFlightTurn, AppMessage, AppMessageContent, AppSessionUsage } from '../types'; import { i18n } from '../../i18n'; +import { toAppMessageContent } from './mappers'; +import type { WireMessageContent } from './wire'; // --------------------------------------------------------------------------- // Internal helpers @@ -131,6 +133,31 @@ function startAssistantMessage(state: SessionState, sessionId: string, promptId: return msg; } +function startUserMessage( + state: SessionState, + sessionId: string, + promptId: string, + userMessageId: string, + content: AppMessageContent[], + createdAt: string, +): AppMessage { + const msg: AppMessage = { + id: userMessageId, + sessionId, + role: 'user', + content, + createdAt, + promptId, + }; + state.messages.push(msg); + return msg; +} + +function toAppPromptContent(raw: unknown): AppMessageContent[] { + if (!Array.isArray(raw)) return []; + return raw.map((part) => toAppMessageContent(part as WireMessageContent)); +} + /** * Append a streamed text/thinking delta in stream order: continue the LAST * content part when it has the same type, otherwise open a NEW part at the @@ -365,6 +392,26 @@ export function createAgentProjector(): AgentProjector { break; } + // ----------------------------------------------------------------------- + case 'prompt.submitted': { + const promptId: string | undefined = p?.promptId; + const userMessageId: string | undefined = p?.userMessageId; + if (!promptId || !userMessageId) break; + const content = toAppPromptContent(p?.content); + if (content.length === 0) break; + s.currentPromptId = promptId; + const msg = startUserMessage( + s, + sessionId, + promptId, + userMessageId, + content, + typeof p?.createdAt === 'string' ? p.createdAt : new Date().toISOString(), + ); + out.push({ type: 'messageCreated', message: cloneMessage(msg) }); + break; + } + // ----------------------------------------------------------------------- case 'turn.started': { // Bind turnId → promptId. Generate a synthetic one if none was pre-bound. @@ -855,6 +902,7 @@ const KNOWN_AGENT_CORE_TYPES = new Set([ 'tool.progress', 'tool.result', 'agent.status.updated', + 'prompt.submitted', 'prompt.completed', 'session.meta.updated', 'compaction.started', diff --git a/apps/kimi-web/src/api/daemon/mappers.ts b/apps/kimi-web/src/api/daemon/mappers.ts index 23ea46ca4..8b077193f 100644 --- a/apps/kimi-web/src/api/daemon/mappers.ts +++ b/apps/kimi-web/src/api/daemon/mappers.ts @@ -130,7 +130,7 @@ function toAppImageSource(src: WireImageSource): ImageSource { return { kind: 'url', url: src.url }; } -function toAppMessageContent(wire: WireMessageContent): AppMessageContent { +export function toAppMessageContent(wire: WireMessageContent): AppMessageContent { switch (wire.type) { case 'text': return { type: 'text', text: wire.text }; diff --git a/apps/kimi-web/test/thinking-multi-segment.test.ts b/apps/kimi-web/test/thinking-multi-segment.test.ts index 16af3b3cb..2c7cb2146 100644 --- a/apps/kimi-web/test/thinking-multi-segment.test.ts +++ b/apps/kimi-web/test/thinking-multi-segment.test.ts @@ -123,3 +123,36 @@ describe('multi-segment thinking', () => { }); }); }); + +describe('prompt.submitted projection', () => { + it('creates the user message for a prompt sent by another client', () => { + const state = play([ + [ + 'prompt.submitted', + { + promptId: 'prompt_1', + userMessageId: 'msg_user_1', + status: 'running', + content: [{ type: 'text', text: 'hello from another client' }], + createdAt: '2026-06-11T00:00:00.000Z', + }, + ], + ['turn.started', { turnId: 1 }], + ['turn.step.started', { turnId: 1 }], + ['assistant.delta', { delta: 'received' }], + ['turn.step.completed', { turnId: 1 }], + ['turn.ended', { turnId: 1, reason: 'completed' }], + ]); + + const messages = state.messagesBySession[SESSION]!; + expect(messages[0]).toMatchObject({ + id: 'msg_user_1', + sessionId: SESSION, + role: 'user', + promptId: 'prompt_1', + content: [{ type: 'text', text: 'hello from another client' }], + createdAt: '2026-06-11T00:00:00.000Z', + }); + expect(messages.find((message) => message.role === 'assistant')?.promptId).toBe('prompt_1'); + }); +}); diff --git a/packages/daemon/src/services/gateway/wsBroadcastService.ts b/packages/daemon/src/services/gateway/wsBroadcastService.ts index 37582f9fb..4e90d1e58 100644 --- a/packages/daemon/src/services/gateway/wsBroadcastService.ts +++ b/packages/daemon/src/services/gateway/wsBroadcastService.ts @@ -7,6 +7,7 @@ import { isVolatileEventType, type Event, type SessionCursor } from '@moonshot-a import { IEventService } from '@moonshot-ai/services'; import { IEnvironmentService, ILogService } from '@moonshot-ai/services'; +import { IConnectionRegistry } from './connectionRegistry'; import { InFlightTurnTracker } from './inFlightTurnTracker'; import { ISessionClientsService } from './sessionClients'; import { SessionEventJournal } from './sessionEventJournal'; @@ -47,6 +48,7 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic @IEventService eventService: IEventService, @ILogService private readonly logger: ILogService, @ISessionClientsService private readonly sessionClients: ISessionClientsService, + @IConnectionRegistry private readonly connectionRegistry: IConnectionRegistry, @IEnvironmentService env: IEnvironmentService, ) { super(); @@ -110,7 +112,10 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic } if (this._store.isDisposed) return; - for (const conn of this.sessionClients.getConnections(sid)) { + const targets = isGlobalSessionEvent(evType) + ? this.connectionRegistry.values() + : this.sessionClients.getConnections(sid); + for (const conn of targets) { conn.send(envelope); } } @@ -234,6 +239,10 @@ function extractSessionId(event: Event): string | undefined { return undefined; } +function isGlobalSessionEvent(type: string): boolean { + return type === 'event.session.created'; +} + /** Session ids are ULID-ish, but never trust an id used as a path segment. */ function sanitizeFileName(sid: string): string { return sid.replace(/[^A-Za-z0-9._-]/g, '_'); diff --git a/packages/daemon/test/prompt.e2e.test.ts b/packages/daemon/test/prompt.e2e.test.ts index 60195da53..b6045e878 100644 --- a/packages/daemon/test/prompt.e2e.test.ts +++ b/packages/daemon/test/prompt.e2e.test.ts @@ -600,6 +600,38 @@ describe('Prompt queue and steer routes', () => { }); describe('Prompt lifecycle: WS receives events + synthesized prompt.completed (W7.2)', () => { + it('broadcasts prompt.submitted events to session subscribers', async () => { + const r = await bootDaemon(); + const sid = await createSession(r); + const { ws, received } = await openSubscriber(r, sid); + + const eventBus = r.services.invokeFunction((a) => a.get(IEventService)); + eventBus.publish({ + type: 'prompt.submitted', + agentId: 'main', + sessionId: sid, + promptId: 'prompt_submit_test', + userMessageId: 'msg_submit_test', + status: 'running', + content: [{ type: 'text', text: 'hello from client a' }], + createdAt: '2026-06-11T00:00:00.000Z', + }); + + const submittedFrame = await waitFor( + received, + (f) => f['type'] === 'prompt.submitted', + 2000, + ); + expect(submittedFrame['session_id']).toBe(sid); + expect(submittedFrame['payload']).toMatchObject({ + promptId: 'prompt_submit_test', + userMessageId: 'msg_submit_test', + content: [{ type: 'text', text: 'hello from client a' }], + }); + + ws.close(); + }); + it('synthesizes prompt.completed end-to-end through bus → observer → WS', async () => { const r = await bootDaemon(); const sid = await createSession(r); diff --git a/packages/daemon/test/services.test.ts b/packages/daemon/test/services.test.ts index fccf81d2e..2db44f926 100644 --- a/packages/daemon/test/services.test.ts +++ b/packages/daemon/test/services.test.ts @@ -74,6 +74,34 @@ class FakeSessionClients implements ISessionClientsServiceT { } } +class FakeConnectionRegistry { + readonly _serviceBrand: undefined; + private readonly conns = new Map(); + + constructor(connections: WsConnection[] = []) { + for (const conn of connections) this.conns.set(conn.id, conn); + } + + add(conn: WsConnection): void { + this.conns.set(conn.id, conn); + } + remove(connId: string): void { + this.conns.delete(connId); + } + get(connId: string): WsConnection | undefined { + return this.conns.get(connId); + } + values(): Iterable { + return this.conns.values(); + } + closeAll(): void { + this.conns.clear(); + } + size(): number { + return this.conns.size; + } +} + function fakeConn(id = 'conn_x'): { id: string; sent: unknown[]; send(m: unknown): void } & WsConnection { const sent: unknown[] = []; return { @@ -174,7 +202,7 @@ describe('WSBroadcastService (WS transport pump)', () => { clients.subscribe(c2, 'sid_test'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'fake.x', sessionId: 'sid_test', agentId: 'main' } as unknown as Event); bus.publish({ type: 'fake.y', sessionId: 'sid_test', agentId: 'main' } as unknown as Event); await broadcast._drainForTest('sid_test'); @@ -201,7 +229,7 @@ describe('WSBroadcastService (WS transport pump)', () => { clients.subscribe(cB, 'sid_b'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'e1', sessionId: 'sid_a', agentId: 'main' } as unknown as Event); bus.publish({ type: 'e1', sessionId: 'sid_b', agentId: 'main' } as unknown as Event); bus.publish({ type: 'e2', sessionId: 'sid_a', agentId: 'main' } as unknown as Event); @@ -226,7 +254,7 @@ describe('WSBroadcastService (WS transport pump)', () => { clients.subscribe(onOther, 'sid_other'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'evt', sessionId: 'sid_a', agentId: 'main' } as unknown as Event); await broadcast._drainForTest('sid_a'); expect(onA.sent.length).toBe(1); @@ -235,6 +263,35 @@ describe('WSBroadcastService (WS transport pump)', () => { bus.dispose(); }); + it('broadcasts session.created to every live connection', async () => { + const clients = new FakeSessionClients(); + const subscribed = fakeConn('conn_subscribed'); + const listOnly = fakeConn('conn_list_only'); + clients.subscribe(subscribed, 'sid_new'); + const bus = new EventService(); + const broadcast = new WSBroadcastService( + bus, + testLogger, + clients, + new FakeConnectionRegistry([subscribed, listOnly]), + makeEnv(), + ); + + bus.publish({ + type: 'event.session.created', + sessionId: 'sid_new', + agentId: 'main', + session: { id: 'sid_new' }, + } as unknown as Event); + await broadcast._drainForTest('sid_new'); + + expect(subscribed.sent).toHaveLength(1); + expect(listOnly.sent).toHaveLength(1); + expect((listOnly.sent[0] as { type: string }).type).toBe('event.session.created'); + broadcast.dispose(); + bus.dispose(); + }); + it('drops events without a sessionId / session_id and warns', () => { const clients = new FakeSessionClients(); const c = fakeConn(); @@ -242,7 +299,7 @@ describe('WSBroadcastService (WS transport pump)', () => { const warnSpy = vi.spyOn(testLogger, 'warn'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'no_sid' } as unknown as Event); expect(c.sent.length).toBe(0); @@ -256,7 +313,7 @@ describe('WSBroadcastService (WS transport pump)', () => { const c = fakeConn(); clients.subscribe(c, 'sid_x'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); broadcast.dispose(); bus.publish({ type: 'late', sessionId: 'sid_x', agentId: 'main' } as unknown as Event); await new Promise((r) => setTimeout(r, 10)); @@ -269,7 +326,7 @@ describe('WSBroadcastService (WS transport pump)', () => { const c = fakeConn(); clients.subscribe(c, 'sid_test'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); for (let i = 0; i < 5; i++) { bus.publish({ type: `e${i}`, sessionId: 'sid_test', agentId: 'main' } as unknown as Event); } @@ -284,7 +341,7 @@ describe('WSBroadcastService (WS transport pump)', () => { it('getBufferedSince forces a resync for a cursor ahead of the journal (stale v1 cursor)', async () => { const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); const replay = await broadcast.getBufferedSince('sid_new', { seq: 5 }); expect(replay.events).toEqual([]); expect(replay.resyncRequired).toBe('epoch_changed'); @@ -295,7 +352,7 @@ describe('WSBroadcastService (WS transport pump)', () => { it('getBufferedSince forces a resync on epoch mismatch', async () => { const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'e', sessionId: 'sid_e', agentId: 'main' } as unknown as Event); await broadcast._drainForTest('sid_e'); const replay = await broadcast.getBufferedSince('sid_e', { seq: 0, epoch: 'ep_other' }); @@ -306,7 +363,7 @@ describe('WSBroadcastService (WS transport pump)', () => { it('seq and epoch survive a daemon restart (journal recovery) and serve replay from disk', async () => { const bus1 = new EventService(); - const b1 = new WSBroadcastService(bus1, testLogger, new FakeSessionClients(), makeEnv()); + const b1 = new WSBroadcastService(bus1, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); for (let i = 0; i < 3; i++) { bus1.publish({ type: `e${i}`, sessionId: 'sid_p', agentId: 'main' } as unknown as Event); } @@ -318,7 +375,7 @@ describe('WSBroadcastService (WS transport pump)', () => { await new Promise((r) => setTimeout(r, 50)); const bus2 = new EventService(); - const b2 = new WSBroadcastService(bus2, testLogger, new FakeSessionClients(), makeEnv()); + const b2 = new WSBroadcastService(bus2, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); const after = await b2.getCursor('sid_p'); expect(after.seq).toBe(3); expect(after.epoch).toBe(before.epoch); @@ -341,7 +398,7 @@ describe('WSBroadcastService (WS transport pump)', () => { const c = fakeConn(); clients.subscribe(c, 'sid_v'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'turn.started', @@ -392,7 +449,7 @@ describe('WSBroadcastService (WS transport pump)', () => { it('getSnapshotState clears the in-flight turn after turn.ended', async () => { const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), makeEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); bus.publish({ type: 'turn.started', sessionId: 'sid_t', @@ -490,7 +547,7 @@ describe('ApprovalService (broadcasts + resolve-by-approval_id)', () => { const conn = fakeConn('conn_subscriber'); clients.subscribe(conn, 'sess_1'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, tmpEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), tmpEnv()); const broker = new ApprovalService(testLogger, bus); return { broker, bus, broadcast, clients, conn }; } @@ -611,7 +668,7 @@ describe('QuestionService (broadcasts + dismiss)', () => { const conn = fakeConn('conn_q_subscriber'); clients.subscribe(conn, 's'); const bus = new EventService(); - const broadcast = new WSBroadcastService(bus, testLogger, clients, tmpEnv()); + const broadcast = new WSBroadcastService(bus, testLogger, clients, new FakeConnectionRegistry(), tmpEnv()); const broker = new QuestionService(testLogger, bus); return { broker, bus, broadcast, clients, conn }; } diff --git a/packages/daemon/test/sessions.e2e.test.ts b/packages/daemon/test/sessions.e2e.test.ts index b345bfe68..f2df0fcfb 100644 --- a/packages/daemon/test/sessions.e2e.test.ts +++ b/packages/daemon/test/sessions.e2e.test.ts @@ -32,6 +32,7 @@ import { join } from 'node:path'; import { pino } from 'pino'; import { ErrorCode, sessionSchema, undoSessionResponseSchema } from '@moonshot-ai/protocol'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { WebSocket } from 'ws'; import { IRestGateway, startDaemon, type RunningDaemon } from '../src'; @@ -83,6 +84,57 @@ function envelopeOf(body: unknown): { code: number; msg: string; data: T | nu return body as { code: number; msg: string; data: T | null; request_id: string; details?: unknown }; } +function wsDataToString(data: unknown): string { + if (typeof data === 'string') return data; + if (Buffer.isBuffer(data)) return data.toString('utf8'); + if (data instanceof ArrayBuffer) return Buffer.from(data).toString('utf8'); + return JSON.stringify(data); +} + +async function openSessionListListener(r: RunningDaemon): Promise<{ + ws: WebSocket; + received: Record[]; +}> { + const wsUrl = r.address.replace('http://', 'ws://') + '/api/v1/ws'; + const received: Record[] = []; + const ws = await new Promise((resolve, reject) => { + const sock = new WebSocket(wsUrl); + sock.on('message', (data) => { + try { + received.push(JSON.parse(wsDataToString(data)) as Record); + } catch { + // ignore + } + }); + sock.once('open', () => resolve(sock)); + sock.once('error', reject); + }); + await waitFor(received, (f) => f['type'] === 'server_hello'); + ws.send( + JSON.stringify({ + type: 'client_hello', + id: 'h1', + payload: { client_id: 'session-list-test', subscriptions: [] }, + }), + ); + await waitFor(received, (f) => f['type'] === 'ack' && f['id'] === 'h1'); + return { ws, received }; +} + +async function waitFor( + received: Record[], + pred: (f: Record) => boolean, + timeoutMs = 2000, +): Promise> { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + const found = received.find(pred); + if (found !== undefined) return found; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`timed out waiting for frame; got ${JSON.stringify(received)}`); +} + describe('POST /api/v1/sessions — create', () => { it('returns a Session payload with snake_case + ISO Z timestamps', async () => { const r = await bootDaemon(); @@ -105,6 +157,36 @@ describe('POST /api/v1/sessions — create', () => { expect(session.id.length).toBeGreaterThan(0); }); + it('broadcasts event.session.created to connected clients without a session subscription', async () => { + const r = await bootDaemon(); + const { ws, received } = await openSessionListListener(r); + const cwd = join(tmpDir, 'workspace-create-broadcast'); + + const res = await appOf(r).inject({ + method: 'POST', + url: '/api/v1/sessions', + payload: { metadata: { cwd }, title: 'created via ws test' }, + }); + const env = envelopeOf<{ id: string }>(res.json()); + expect(env.code).toBe(0); + expect(env.data).not.toBeNull(); + + const frame = await waitFor( + received, + (f) => f['type'] === 'event.session.created', + ); + expect(frame['session_id']).toBe(env.data!.id); + expect(frame['payload']).toMatchObject({ + session: { + id: env.data!.id, + title: 'created via ws test', + metadata: { cwd }, + }, + }); + + ws.close(); + }); + it('rejects a body missing metadata.cwd with code 40001 + details', async () => { const r = await bootDaemon(); const res = await appOf(r).inject({ diff --git a/packages/protocol/src/__tests__/events.test.ts b/packages/protocol/src/__tests__/events.test.ts index ac5fdace8..780321b46 100644 --- a/packages/protocol/src/__tests__/events.test.ts +++ b/packages/protocol/src/__tests__/events.test.ts @@ -102,4 +102,54 @@ describe('events / display re-exports', () => { expect(parsed.agentId).toBe('agent_1'); expect(parsed.sessionId).toBe('sess_1'); }); + + it('validates prompt.submitted events', () => { + const parsed = eventSchema.parse({ + type: 'prompt.submitted', + agentId: 'main', + sessionId: 'sess_1', + promptId: 'prompt_1', + userMessageId: 'msg_1', + status: 'running', + content: [{ type: 'text', text: 'hello' }], + createdAt: '2026-06-11T00:00:00.000Z', + }); + + expect(parsed.type).toBe('prompt.submitted'); + expect((parsed as { promptId: string }).promptId).toBe('prompt_1'); + }); + + it('validates event.session.created events', () => { + const parsed = eventSchema.parse({ + type: 'event.session.created', + agentId: 'main', + sessionId: 'sess_1', + session: { + id: 'sess_1', + workspace_id: 'wd_project_123456abcdef', + title: 'Created session', + created_at: '2026-06-11T00:00:00.000Z', + updated_at: '2026-06-11T00:00:00.000Z', + status: 'idle', + metadata: { cwd: '/tmp/project' }, + agent_config: { model: 'kimi-k2' }, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_creation_tokens: 0, + total_cost_usd: 0, + context_tokens: 0, + context_limit: 0, + turn_count: 0, + }, + permission_rules: [], + message_count: 0, + last_seq: 0, + }, + }); + + expect(parsed.type).toBe('event.session.created'); + expect((parsed as { session: { id: string } }).session.id).toBe('sess_1'); + }); }); diff --git a/packages/protocol/src/events.ts b/packages/protocol/src/events.ts index 918adea29..fc0c2ed87 100644 --- a/packages/protocol/src/events.ts +++ b/packages/protocol/src/events.ts @@ -1,6 +1,9 @@ import { z } from 'zod'; import { ToolInputDisplaySchema, type ToolInputDisplay } from './display'; +import { messageContentSchema, type MessageContent } from './message'; +import { sessionSchema, type Session } from './session'; +import { isoDateTimeSchema } from './time'; export interface TokenUsage { readonly inputOther: number; @@ -306,6 +309,11 @@ export interface SessionMetaUpdatedEvent { readonly patch?: Record; } +export interface SessionCreatedEvent { + readonly type: 'event.session.created'; + readonly session: Session; +} + export interface GoalUpdatedEvent { readonly type: 'goal.updated'; readonly snapshot: GoalSnapshot | null; @@ -515,6 +523,15 @@ export interface CronFiredEvent { readonly prompt: string; } +export interface PromptSubmittedEvent { + readonly type: 'prompt.submitted'; + readonly promptId: string; + readonly userMessageId: string; + readonly status: 'running' | 'queued'; + readonly content: readonly MessageContent[]; + readonly createdAt: string; +} + export type ToolListUpdatedReason = 'mcp.connected' | 'mcp.disconnected' | 'mcp.failed'; export interface ToolListUpdatedEvent { @@ -541,6 +558,7 @@ export type AgentEvent = | WarningEvent | AgentStatusUpdatedEvent | SessionMetaUpdatedEvent + | SessionCreatedEvent | GoalUpdatedEvent | SkillActivatedEvent | TurnStartedEvent @@ -569,7 +587,8 @@ export type AgentEvent = | CompactionCompletedEvent | BackgroundTaskStartedEvent | BackgroundTaskTerminatedEvent - | CronFiredEvent; + | CronFiredEvent + | PromptSubmittedEvent; export type Event = AgentEvent & { agentId: string; sessionId: string }; @@ -881,6 +900,11 @@ export const sessionMetaUpdatedEventSchema = z.object({ patch: z.record(z.string(), z.unknown()).optional(), }) satisfies z.ZodType; +export const sessionCreatedEventSchema = z.object({ + type: z.literal('event.session.created'), + session: sessionSchema, +}) satisfies z.ZodType; + export const goalUpdatedEventSchema = z.object({ type: z.literal('goal.updated'), snapshot: goalSnapshotSchema.nullable(), @@ -1090,6 +1114,15 @@ export const cronFiredEventSchema = z.object({ prompt: z.string(), }) satisfies z.ZodType; +export const promptSubmittedEventSchema = z.object({ + type: z.literal('prompt.submitted'), + promptId: z.string(), + userMessageId: z.string(), + status: z.enum(['running', 'queued']), + content: z.array(messageContentSchema), + createdAt: isoDateTimeSchema, +}) satisfies z.ZodType; + export const toolListUpdatedReasonSchema = z.enum([ 'mcp.connected', 'mcp.disconnected', @@ -1120,6 +1153,7 @@ export const agentEventSchema = z.discriminatedUnion('type', [ warningEventSchema, agentStatusUpdatedEventSchema, sessionMetaUpdatedEventSchema, + sessionCreatedEventSchema, goalUpdatedEventSchema, skillActivatedEventSchema, turnStartedEventSchema, @@ -1149,6 +1183,7 @@ export const agentEventSchema = z.discriminatedUnion('type', [ backgroundTaskStartedEventSchema, backgroundTaskTerminatedEventSchema, cronFiredEventSchema, + promptSubmittedEventSchema, ]) satisfies z.ZodType; export const eventSchema = agentEventSchema.and( diff --git a/packages/services/src/index.ts b/packages/services/src/index.ts index 84da4687c..6ac7f18cd 100644 --- a/packages/services/src/index.ts +++ b/packages/services/src/index.ts @@ -149,6 +149,7 @@ export type { SyntheticPromptAbortedEvent, SyntheticPromptCompletedEvent, SyntheticPromptSteeredEvent, + SyntheticPromptSubmittedEvent, } from './prompt/prompt'; export { PromptService } from './prompt/promptService'; diff --git a/packages/services/src/prompt/prompt.ts b/packages/services/src/prompt/prompt.ts index 8008237c2..759fccf85 100644 --- a/packages/services/src/prompt/prompt.ts +++ b/packages/services/src/prompt/prompt.ts @@ -71,6 +71,7 @@ import type { Event } from '@moonshot-ai/agent-core/base/common/event'; import type { PromptListResponse, PromptSubmission, + PromptStatus, PromptSteerResult, PromptSubmitResult, } from '@moonshot-ai/protocol'; @@ -240,6 +241,17 @@ export class PromptAlreadyCompletedError extends Error { } } +export interface SyntheticPromptSubmittedEvent { + readonly type: 'prompt.submitted'; + readonly agentId: string; + readonly sessionId: string; + readonly promptId: string; + readonly userMessageId: string; + readonly status: PromptStatus; + readonly content: PromptSubmission['content']; + readonly createdAt: string; +} + /** * `prompt.completed` synthetic event shape. Matches the agent-core `Event` * type contract (`AgentEvent & { agentId, sessionId }`) so it flows through diff --git a/packages/services/src/prompt/promptService.ts b/packages/services/src/prompt/promptService.ts index 371b889dc..5b4c74afc 100644 --- a/packages/services/src/prompt/promptService.ts +++ b/packages/services/src/prompt/promptService.ts @@ -36,6 +36,7 @@ import { type SyntheticPromptCompletedEvent, type SyntheticPromptAbortedEvent, type SyntheticPromptSteeredEvent, + type SyntheticPromptSubmittedEvent, } from './prompt'; const MAIN_AGENT_ID = 'main'; @@ -317,11 +318,15 @@ export class PromptService const existing = this._active.get(sid); if (existing !== undefined && !existing.completed && !existing.aborted) { this._enqueue(sid, state); - return toPromptItem(state, 'queued'); + const item = toPromptItem(state, 'queued'); + this._publishSubmitted(sid, item); + return item; } await this._startPrompt(sid, state); - return toPromptItem(state, 'running'); + const item = toPromptItem(state, 'running'); + this._publishSubmitted(sid, item); + return item; } async steer(sid: string, promptIds: readonly string[]): Promise { @@ -414,6 +419,20 @@ export class PromptService } } + private _publishSubmitted(sid: string, item: PromptSubmitResult): void { + const event: SyntheticPromptSubmittedEvent = { + type: 'prompt.submitted', + agentId: MAIN_AGENT_ID, + sessionId: sid, + promptId: item.prompt_id, + userMessageId: item.user_message_id, + status: item.status, + content: item.content, + createdAt: item.created_at, + }; + this.eventService.publish(event); + } + async abort(sid: string, pid: string): Promise { await this._requireSession(sid); const state = this._active.get(sid); diff --git a/packages/services/src/session/sessionService.ts b/packages/services/src/session/sessionService.ts index d3232b280..6b934da58 100644 --- a/packages/services/src/session/sessionService.ts +++ b/packages/services/src/session/sessionService.ts @@ -30,6 +30,7 @@ import { } from '@moonshot-ai/protocol'; import { ICoreProcessService } from '../coreProcess/coreProcess'; +import { IEventService } from '../event/event'; import { toProtocolMessage } from '../message/message'; import { IPromptService, type AgentStatePatch } from '../prompt/prompt'; import { @@ -107,6 +108,7 @@ export class SessionService extends Disposable implements ISessionService { constructor( @ICoreProcessService private readonly core: ICoreProcessService, + @IEventService private readonly eventService: IEventService, @IInstantiationService private readonly instantiation: IInstantiationService, ) { @@ -131,7 +133,7 @@ export class SessionService extends Disposable implements ISessionService { } const meta = await this.tryGetMeta(summary.id); const session = toProtocolSession(summary, meta); - this._onDidCreate.fire({ session }); + this.emitCreated(session); return session; } @@ -239,7 +241,7 @@ export class SessionService extends Disposable implements ISessionService { }); const meta = await this.tryGetMeta(summary.id); const session = toProtocolSession(summary, meta); - this._onDidCreate.fire({ session }); + this.emitCreated(session); return session; } @@ -301,10 +303,20 @@ export class SessionService extends Disposable implements ISessionService { }); const meta = await this.tryGetMeta(summary.id); const session = toProtocolSession(summary, meta); - this._onDidCreate.fire({ session }); + this.emitCreated(session); return session; } + private emitCreated(session: Session): void { + this._onDidCreate.fire({ session }); + this.eventService.publish({ + type: 'event.session.created', + agentId: 'main', + sessionId: session.id, + session, + }); + } + async getStatus(id: string): Promise { const all = await this.core.rpc.listSessions({}); const summary = all.find((s) => s.id === id); diff --git a/packages/services/test/prompt-service.test.ts b/packages/services/test/prompt-service.test.ts index a90bc2ccc..a6c85ae81 100644 --- a/packages/services/test/prompt-service.test.ts +++ b/packages/services/test/prompt-service.test.ts @@ -346,6 +346,36 @@ describe('PromptService.submit', () => { ]); }); + it('publishes prompt.submitted when a prompt starts running', async () => { + const { bridge } = makeBridge(); + const { bus, events } = makeBus(); + const impl = newSvc(bridge, bus); + const body = mkBody({ content: [{ type: 'text', text: 'hello from client a' }] }); + + const result = await impl.submit(SID, body); + + const submitted = events.find((event) => event.type === 'prompt.submitted') as + | { + type: 'prompt.submitted'; + sessionId: string; + agentId: string; + promptId: string; + userMessageId: string; + status: string; + content: readonly PromptSubmission['content'][number][]; + } + | undefined; + expect(submitted).toBeDefined(); + expect(submitted?.sessionId).toBe(SID); + expect(submitted?.agentId).toBe('main'); + expect(submitted).toMatchObject({ + promptId: result.prompt_id, + userMessageId: result.user_message_id, + status: 'running', + content: body.content, + }); + }); + it('queues a second prompt when a non-terminal prompt is already active', async () => { const { bridge } = makeBridge(); const { bus } = makeBus(); @@ -360,6 +390,28 @@ describe('PromptService.submit', () => { expect(listed.queued.map((p) => p.prompt_id)).toEqual([second.prompt_id]); }); + it('publishes prompt.submitted when a prompt is queued', async () => { + const { bridge } = makeBridge(); + const { bus, events } = makeBus(); + const impl = newSvc(bridge, bus); + + await impl.submit(SID, mkBody({ content: [{ type: 'text', text: 'one' }] })); + const second = await impl.submit(SID, mkBody({ content: [{ type: 'text', text: 'two' }] })); + + const submitted = events.filter((event) => event.type === 'prompt.submitted') as Array<{ + type: 'prompt.submitted'; + promptId: string; + status: string; + content: readonly PromptSubmission['content'][number][]; + }>; + expect(submitted.map((event) => event.promptId)).toContain(second.prompt_id); + expect(submitted.at(-1)).toMatchObject({ + promptId: second.prompt_id, + status: 'queued', + content: [{ type: 'text', text: 'two' }], + }); + }); + it('starts the next queued prompt after the active prompt completes', async () => { const { bridge, record } = makeBridge(); const { bus, triggerSubscribers } = makeBus(); diff --git a/packages/services/test/session-service.test.ts b/packages/services/test/session-service.test.ts index ca3f730b6..3f66b9017 100644 --- a/packages/services/test/session-service.test.ts +++ b/packages/services/test/session-service.test.ts @@ -232,8 +232,27 @@ function textMessage( let state: FakeBridgeState; let svc: SessionService; let promptStub: ReturnType; +let eventBus: ReturnType; let instantiation: TestInstantiationService; +function makeEventServiceStub(): { + eventService: IEventService; + events: unknown[]; +} { + const events: unknown[] = []; + const emitter = new Emitter(); + return { + events, + eventService: { + _serviceBrand: undefined, + publish: vi.fn((event: unknown) => { + events.push(event); + }) as IEventService['publish'], + onDidPublish: emitter.event as unknown as IEventService['onDidPublish'], + }, + }; +} + function makePromptServiceStub(): { promptService: IPromptService; calls: Array<{ sid: string; patch: Record; source: string; promptId: string | undefined }>; @@ -269,9 +288,11 @@ function makeTestInstantiation(stubs: { beforeEach(() => { state = freshState(); promptStub = makePromptServiceStub(); + eventBus = makeEventServiceStub(); instantiation = makeTestInstantiation({ promptService: promptStub.promptService }); svc = new SessionService( makeFakeBridge(state), + eventBus.eventService, instantiation, ); }); @@ -830,6 +851,16 @@ describe('SessionService per-domain event listeners (Phase C)', () => { expect((events[0] as { session: { id: string } }).session.id).toBe(session.id); }); + it('publishes session.created after creating a session', async () => { + const session = await svc.create({ metadata: { cwd: '/tmp/evt-bus' } }); + expect(eventBus.events).toContainEqual({ + type: 'event.session.created', + sessionId: session.id, + agentId: 'main', + session, + }); + }); + it('onDidCreate detach stops future events', async () => { const events: unknown[] = []; const sub = svc.onDidCreate((e) => { events.push(e); });