From 04c902884b96ac2a1de4cae3b3819307c28c8496 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Tue, 21 Jul 2026 11:38:56 +0800 Subject: [PATCH 1/2] feat(kap-server): broadcast agent.created/agent.disposed session events - emit durable agent.created / agent.disposed facts from the SessionEventBroadcaster lifecycle callbacks, ahead of the agent's own events, and let them bypass per-subscription agent allowlists - add the agent.created / agent.disposed wire types and zod schemas - stamp disposedAt on the transcript roster entry via TranscriptStore.markDisposed so REST consumers can tell a dead agent from a live one --- .../kap-server/src/protocol/events-zod.ts | 10 +++ .../src/services/transcript/coreBinding.ts | 1 + .../kap-server/src/transport/ws/v1/events.ts | 10 +++ .../ws/v1/sessionEventBroadcaster.ts | 40 ++++++++--- .../test/services/transcript.test.ts | 10 ++- .../test/sessionEventBroadcaster.test.ts | 69 ++++++++++++++++++- .../transcript/src/store/transcriptStore.ts | 9 ++- packages/transcript/src/wire/schema.ts | 1 + packages/transcript/test/store.test.ts | 24 +++++++ 9 files changed, 160 insertions(+), 14 deletions(-) diff --git a/packages/kap-server/src/protocol/events-zod.ts b/packages/kap-server/src/protocol/events-zod.ts index f53c6259ef..a6963f3a76 100644 --- a/packages/kap-server/src/protocol/events-zod.ts +++ b/packages/kap-server/src/protocol/events-zod.ts @@ -540,6 +540,14 @@ export const sessionMetaUpdatedEventSchema = z.object({ patch: z.record(z.string(), z.unknown()).optional(), }); +export const agentCreatedEventSchema = z.object({ + type: z.literal('agent.created'), +}); + +export const agentDisposedEventSchema = z.object({ + type: z.literal('agent.disposed'), +}); + export const sessionCreatedEventSchema = z.object({ type: z.literal('event.session.created'), session: sessionSchema, @@ -904,6 +912,8 @@ export const agentEventSchema = z.discriminatedUnion('type', [ errorEventSchema, warningEventSchema, agentStatusUpdatedEventSchema, + agentCreatedEventSchema, + agentDisposedEventSchema, sessionMetaUpdatedEventSchema, sessionCreatedEventSchema, workspaceCreatedEventSchema, diff --git a/packages/kap-server/src/services/transcript/coreBinding.ts b/packages/kap-server/src/services/transcript/coreBinding.ts index 56c38a19a2..a708ee3ac5 100644 --- a/packages/kap-server/src/services/transcript/coreBinding.ts +++ b/packages/kap-server/src/services/transcript/coreBinding.ts @@ -235,6 +235,7 @@ export function bindSessionTranscript( agentDisposables.delete(agentId); subscribedAgents.delete(agentId); projectors.delete(agentId); + store.markDisposed(agentId, new Date().toISOString()); }), ); diff --git a/packages/kap-server/src/transport/ws/v1/events.ts b/packages/kap-server/src/transport/ws/v1/events.ts index abe6d6adc5..22efbc7c8f 100644 --- a/packages/kap-server/src/transport/ws/v1/events.ts +++ b/packages/kap-server/src/transport/ws/v1/events.ts @@ -33,6 +33,14 @@ export interface AgentStatusUpdatedEvent { readonly phase?: AgentPhase; } +export interface AgentCreatedEvent { + readonly type: 'agent.created'; +} + +export interface AgentDisposedEvent { + readonly type: 'agent.disposed'; +} + export interface SessionMetaUpdatedEvent { readonly type: 'session.meta.updated'; readonly title?: string; @@ -160,6 +168,8 @@ export interface BackgroundTaskTerminatedEvent { export type AgentEvent = | DomainEvent | AgentStatusUpdatedEvent + | AgentCreatedEvent + | AgentDisposedEvent | SessionMetaUpdatedEvent | SessionCreatedEvent | WorkspaceCreatedEvent diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index 72d3dbcd22..200f7311d4 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -10,7 +10,10 @@ * 1. Subscribes to every agent's `IEventBus` via * `IAgentLifecycleService` reach-down-via-handle (and `onDidCreate` / * `onDidDispose` for late agents); `record` emissions are persisted and not - * broadcast (see step 3). Also subscribes to the session's + * broadcast (see step 3). The same lifecycle callbacks fan durable + * `agent.created` / `agent.disposed` facts out at session granularity + * (they bypass per-subscription agent allowlists but never leave the + * session). 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`. @@ -827,12 +830,27 @@ export class SessionEventBroadcaster { }; for (const handle of agents.list()) subscribeAgent(handle); state.lifecycleDisposables.push( - agents.onDidCreate((handle) => subscribeAgent(handle)), + agents.onDidCreate((handle) => { + subscribeAgent(handle); + // Session-grained lifecycle fact, ahead of any of the agent's own + // events: `onDidCreate` fires before the agent's eager services + // ignite, so this enqueue lands first in the queue. + this.enqueueDurable(state, { + type: 'agent.created', + agentId: handle.id, + sessionId, + }); + }), agents.onDidDispose((agentId) => { const d = state.agentDisposables.get(agentId); if (d !== undefined) { d.dispose(); state.agentDisposables.delete(agentId); + this.enqueueDurable(state, { + type: 'agent.disposed', + agentId, + sessionId, + }); } // A removed agent can no longer contribute work; drop its fold and // re-evaluate the aggregate (its turn.ended normally lands first, but @@ -1246,14 +1264,19 @@ function isGlobalEvent(type: string): boolean { ); } +function isAgentLifecycleEvent(type: string): boolean { + return type === 'agent.created' || type === 'agent.disposed'; +} + /** * Per-subscription agent allowlist check — shared by live fan-out and replay. * Returns `true` when the envelope should be delivered to a subscriber carrying * `filter`: * - `filter === undefined` → receive every agent (legacy session-grained * behavior); - * - global events (session/workspace/config) are not agent - * events and always pass; + * - global events (session/workspace/config) and agent lifecycle events + * (`agent.created` / `agent.disposed`) are not per-agent stream content + * and always pass; * - events without a string `agentId` (should not happen on the v1 wire, * where the broadcaster stamps every event) pass defensively rather than * being dropped; @@ -1262,6 +1285,7 @@ function isGlobalEvent(type: string): boolean { function matchesAgentFilter(envelope: EventEnvelope, filter: AgentFilter): boolean { if (filter === undefined) return true; if (isGlobalEvent(envelope.type)) return true; + if (isAgentLifecycleEvent(envelope.type)) return true; const payload = envelope.payload; const agentId = typeof payload === 'object' && payload !== null @@ -1367,8 +1391,8 @@ function sessionMetaUpdatedPayload( const title = typeof candidate.title === 'string' ? candidate.title : undefined; const patch = typeof candidate.patch === 'object' && - candidate.patch !== null && - !Array.isArray(candidate.patch) + candidate.patch !== null && + !Array.isArray(candidate.patch) ? candidate.patch : undefined; if (title === undefined && patch === undefined) return undefined; @@ -1399,8 +1423,8 @@ function sessionCreatedPayload( : undefined; const session = typeof candidate.session === 'object' && - candidate.session !== null && - !Array.isArray(candidate.session) + candidate.session !== null && + !Array.isArray(candidate.session) ? (candidate.session as SessionCreatedEvent['session']) : undefined; if (sessionId === undefined || session === undefined) return undefined; diff --git a/packages/kap-server/test/services/transcript.test.ts b/packages/kap-server/test/services/transcript.test.ts index d17588bbdf..1b6f551ec4 100644 --- a/packages/kap-server/test/services/transcript.test.ts +++ b/packages/kap-server/test/services/transcript.test.ts @@ -1120,16 +1120,22 @@ describe('bindSessionTranscript', () => { ); const sub = agents.add('sub-1'); + agents.add('main'); sub.bus.emit(ev({ type: 'turn.started', turnId: 0, origin: { kind: 'user' }, prompt: 'scan' })); sub.bus.emit(ev({ type: 'turn.ended', turnId: 0, reason: 'completed' })); expect(store.getAgent('sub-1')?.getItems()).toHaveLength(1); // Disposal kills the projector but must not drop already-served history: // the service's backfill cache dedupes per agent, so removing the - // transcript would rebuild an empty shell on the next read. + // transcript would rebuild an empty shell on the next read. The roster + // entry stays and carries its end timestamp so REST / fresh-reset + // consumers can tell the dead agent from a live one. agents.remove('sub-1'); expect(store.getAgent('sub-1')?.getItems()).toHaveLength(1); - expect(store.agents().map((a) => a.agentId)).toContain('sub-1'); + const descriptor = store.agents().find((a) => a.agentId === 'sub-1'); + expect(descriptor).toBeDefined(); + expect(typeof descriptor?.disposedAt).toBe('string'); + expect(store.agents().find((a) => a.agentId === 'main')?.disposedAt).toBeUndefined(); binding.dispose(); }); diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts index 551b39b855..fa9551e60b 100644 --- a/packages/kap-server/test/sessionEventBroadcaster.test.ts +++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts @@ -441,12 +441,75 @@ describe('SessionEventBroadcaster', () => { await bc.subscribe('s1', target); const late = lc.addAgent('main'); // created after subscribe + // Let the lifecycle dispatch drain before driving the turn — a + // synchronous emit would fold its activity ahead of the queued + // work_changed task and reorder it in front of agent.created (see the + // note above about the production interleaving). + await bc.getCursor('s1'); late.bus.emit(agentEvent('turn.started', { turnId: 7 })); await bc.getCursor('s1'); - // work_changed(busy) (seq 1) is emitted ahead of turn.started (seq 2); - // the volatile agent.status.updated phase frame rides alongside. - expect(envelopes.filter((e) => e.volatile !== true).map((e) => e.seq)).toEqual([1, 2]); + // agent.created (seq 1) leads; work_changed(busy) (seq 2) is emitted ahead + // of turn.started (seq 3); the volatile agent.status.updated phase frame + // rides alongside. + expect(envelopes.filter((e) => e.volatile !== true).map((e) => e.seq)).toEqual([1, 2, 3]); + expect(envelopes[0]).toMatchObject({ type: 'agent.created' }); + expect((envelopes[0]!.payload as { agentId: string }).agentId).toBe('main'); + }); + + it('broadcasts agent.disposed only for agents this state attached', async () => { + const lc = new FakeLifecycle(); + lc.addAgent('main'); + lc.addAgent('agent-0'); + sessions.set('s1', lc); + const { target, envelopes } = collectingTarget(); + await bc.subscribe('s1', target); + + lc.removeAgent('agent-0'); + // The creation-failure path fires onDidDispose for an agent that was + // never created (and never attached) — clients must not hear about it. + lc.removeAgent('ghost'); + await bc.getCursor('s1'); + + const disposed = envelopes.filter((e) => e.type === 'agent.disposed'); + expect(disposed).toHaveLength(1); + expect((disposed[0]!.payload as { agentId: string }).agentId).toBe('agent-0'); + expect(disposed[0]!.volatile).toBeUndefined(); // durable + }); + + it('delivers lifecycle events past the agent allowlist (session-grained)', async () => { + const lc = new FakeLifecycle(); + lc.addAgent('main'); + sessions.set('s1', lc); + const { target, envelopes } = collectingTarget(); + await bc.subscribe('s1', target, new Set(['main'])); + + lc.addAgent('agent-0'); // outside the allowlist + lc.removeAgent('agent-0'); + await bc.getCursor('s1'); + + const types = envelopes.map((e) => e.type); + expect(types).toContain('agent.created'); + expect(types).toContain('agent.disposed'); + }); + + it('journals lifecycle events for replay', async () => { + const lc = new FakeLifecycle(); + lc.addAgent('main'); + sessions.set('s1', lc); + const { target } = collectingTarget(); + await bc.subscribe('s1', target); + + lc.addAgent('agent-0'); + lc.removeAgent('agent-0'); + await bc.getCursor('s1'); + + const result = await bc.getBufferedSince('s1', { seq: 0 }); + expect(result.resyncRequired).toBe(false); + expect(result.events.map((e) => e.envelope.type)).toEqual([ + 'agent.created', + 'agent.disposed', + ]); }); it('getSnapshotState returns the in-flight turn', async () => { diff --git a/packages/transcript/src/store/transcriptStore.ts b/packages/transcript/src/store/transcriptStore.ts index cf77077ae4..245552b067 100644 --- a/packages/transcript/src/store/transcriptStore.ts +++ b/packages/transcript/src/store/transcriptStore.ts @@ -18,6 +18,7 @@ export interface AgentDescriptor { readonly parentAgentId?: AgentId; readonly label?: string; readonly createdAt?: string; + readonly disposedAt?: string; } export type RosterListener = (agents: readonly AgentDescriptor[]) => void; @@ -27,7 +28,7 @@ export class TranscriptStore { readonly #descriptors = new Map(); readonly #rosterListeners = new Set(); - constructor(readonly sessionId: string) {} + constructor(readonly sessionId: string) { } /** Lazily create (or fetch) the transcript for an agent. */ ensureAgent(agentId: AgentId, descriptor?: AgentDescriptor): AgentTranscript { @@ -62,6 +63,12 @@ export class TranscriptStore { } } + markDisposed(agentId: AgentId, disposedAt: string): void { + const descriptor = this.#descriptors.get(agentId); + if (descriptor === undefined || descriptor.disposedAt !== undefined) return; + this.describeAgent({ ...descriptor, disposedAt }); + } + agents(): readonly AgentDescriptor[] { return [...this.#descriptors.values()]; } diff --git a/packages/transcript/src/wire/schema.ts b/packages/transcript/src/wire/schema.ts index b14e424cda..1da0da2cb9 100644 --- a/packages/transcript/src/wire/schema.ts +++ b/packages/transcript/src/wire/schema.ts @@ -370,6 +370,7 @@ export const agentDescriptorSchema = z.object({ parentAgentId: agentIdSchema.optional(), label: z.string().optional(), createdAt: z.string().optional(), + disposedAt: z.string().optional(), }); export const transcriptResponseSchema = z.object({ diff --git a/packages/transcript/test/store.test.ts b/packages/transcript/test/store.test.ts index 4a7e5c6b3b..c839412f10 100644 --- a/packages/transcript/test/store.test.ts +++ b/packages/transcript/test/store.test.ts @@ -439,4 +439,28 @@ describe('TranscriptStore', () => { expect(rosters).toEqual([2, 1]); expect(store.agents().map((a) => a.agentId)).toEqual(['main']); }); + + it('markDisposed stamps disposedAt on the existing descriptor only', () => { + const store = new TranscriptStore('s1'); + store.ensureAgent('main', { agentId: 'main', type: 'main' }); + + // Never-announced agents must not gain a roster entry. + store.markDisposed('ghost', '2026-07-20T00:00:00.000Z'); + expect(store.agents().map((a) => a.agentId)).toEqual(['main']); + + const rosters: Array = []; + store.onRosterChange((agents) => rosters.push(agents.map((a) => a.agentId))); + store.markDisposed('main', '2026-07-20T01:00:00.000Z'); + expect(rosters).toEqual([['main']]); + expect(store.agents()[0]).toMatchObject({ + agentId: 'main', + type: 'main', + disposedAt: '2026-07-20T01:00:00.000Z', + }); + + // Idempotent: the first stamp wins and no roster re-emit fires. + store.markDisposed('main', '2026-07-20T02:00:00.000Z'); + expect(store.agents()[0]?.disposedAt).toBe('2026-07-20T01:00:00.000Z'); + expect(rosters).toHaveLength(1); + }); }); From 1c33b6541fdb75bbbf704d33b30f8c96f6a38c0e Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Tue, 21 Jul 2026 13:34:43 +0800 Subject: [PATCH 2/2] chore: add changeset for agent lifecycle events --- .changeset/agent-lifecycle-events.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/agent-lifecycle-events.md diff --git a/.changeset/agent-lifecycle-events.md b/.changeset/agent-lifecycle-events.md new file mode 100644 index 0000000000..939aaf1fca --- /dev/null +++ b/.changeset/agent-lifecycle-events.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Add agent.created and agent.disposed events to the server session event stream, and expose each agent's disposal time in the transcript API.