feat(kap-server): wire cloud telemetry for engine events

apps/kimi-code/src/cli/sub/web/run.ts

@@ -280,6 +280,10 @@ async function runServerInProcess(
     allowRemoteTerminals: options.allowRemoteTerminals,
     allowedHosts: options.allowedHosts,
     disableAuth: options.dangerousBypassAuth,
+    // Attach the engine's cloud telemetry appender (still gated by the config
+    // `telemetry` toggle). Complements the v1 client registered above, which
+    // only covers host-level events.
+    telemetry: true,
     // Seed the CLI's Kimi identity headers so the engine's outbound
     // requests (model, WebSearch, FetchURL) carry the same User-Agent +
     // X-Msh-* identity as direct CLI runs.

packages/agent-core-v2/src/app/sessionLifecycle/sessionLifecycleService.ts

@@ -7,10 +7,9 @@
  * close/archive — archiving flags the session's `sessionMetadata`, removes
  * its `agentLifecycle` agents, restoring clears the archived flag, and
  * broadcasts through `event`; session start and resume failures are reported
- * through `telemetry`. Because hosts bind their telemetry context only after
- * create()/resume() returns, the created-session announcement binds the
- * session id into telemetry context before emitting `session_started`, and the
- * resume-failure path does the same before `session_load_failed`.
+ * through `telemetry`. Each Session scope receives a telemetry view bound to
+ * its session id, while failures before a scope is available use an ephemeral
+ * context view.
  * Materializes the session's initial metadata on
  * creation by resolving `sessionMetadata`. Bound at App scope. Persisted
  * sessions are discovered through the `sessionIndex` read model, and workspace
@@ -206,7 +205,10 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
       LifecycleScope.Session,
       opts.sessionId,
       {
-        extra: [...sessionContextSeed(ctx)],
+        extra: [
+          ...sessionContextSeed(ctx),
+          [ITelemetryService, this.telemetry.withContext({ sessionId: opts.sessionId })],
+        ],
       },
     ) as ISessionScopeHandle;
     if (additionalDirs.length > 0) {
@@ -250,8 +252,9 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
   private async announceCreated(event: SessionCreatedEvent): Promise<void> {
     await this.hooks.onDidCreateSession.run(event);
     this._onDidCreateSession.fire(event);
-    this.telemetry.setContext({ sessionId: event.sessionId });
-    this.telemetry.track2('session_started', { resumed: event.source === 'resume' });
+    event.handle.accessor
+      .get(ITelemetryService)
+      .track2('session_started', { resumed: event.source === 'resume' });
   }
 
   get(sessionId: string): ISessionScopeHandle | undefined {
@@ -266,10 +269,11 @@ export class SessionLifecycleService extends Disposable implements ISessionLifec
     if (live !== undefined) return Promise.resolve(live);
     const promise = this.doResume(sessionId)
       .catch((error: unknown) => {
-        this.telemetry.setContext({ sessionId });
-        this.telemetry.track2('session_load_failed', {
-          reason: isError2(error) ? error.code : error instanceof Error ? error.name : 'unknown',
-        });
+        this.telemetry
+          .withContext({ sessionId })
+          .track2('session_load_failed', {
+            reason: isError2(error) ? error.code : error instanceof Error ? error.name : 'unknown',
+          });
         throw error;
       })
       .finally(() => this.resuming.delete(sessionId));

packages/agent-core-v2/src/app/telemetry/cloudAppender.ts

@@ -105,10 +105,11 @@ export class CloudAppender implements ITelemetryAppender {
   }
 
   track(event: string, properties?: TelemetryProperties): void {
+    const eventSessionId = properties?.['sessionId'];
     const enriched: EnrichedCloudEvent = {
       event_id: randomUUID().replaceAll('-', ''),
       device_id: this.deviceId,
-      session_id: this.sessionId,
+      session_id: typeof eventSessionId === 'string' ? eventSessionId : this.sessionId,
       event,
       timestamp: Date.now() / 1000,
       properties: cleanTelemetryProperties(sanitizeProperties(properties)),

packages/agent-core-v2/test/app/sessionLifecycle/sessionLifecycle.test.ts

@@ -848,6 +848,32 @@ describe('SessionLifecycleService', () => {
     });
   });
 
+  it('keeps telemetry session context isolated when multiple sessions emit interleaved events', async () => {
+    const svc = build();
+    const first = await svc.create({ sessionId: 'first', workDir: '/tmp/proj' });
+    const second = await svc.create({ sessionId: 'second', workDir: '/tmp/proj' });
+    telemetryRecords.length = 0;
+
+    first.accessor.get(ITelemetryService).track('test_event', { marker: 'first-before' });
+    second.accessor.get(ITelemetryService).track('test_event', { marker: 'second' });
+    first.accessor.get(ITelemetryService).track('test_event', { marker: 'first-after' });
+
+    expect(telemetryRecords).toEqual([
+      {
+        event: 'test_event',
+        properties: { sessionId: 'first', marker: 'first-before' },
+      },
+      {
+        event: 'test_event',
+        properties: { sessionId: 'second', marker: 'second' },
+      },
+      {
+        event: 'test_event',
+        properties: { sessionId: 'first', marker: 'first-after' },
+      },
+    ]);
+  });
+
   it('emits session_started with resumed: true and the bound session id on resume', async () => {
     const workDir = '/tmp/proj';
     const svc = build([

packages/agent-core-v2/test/app/telemetry/cloudAppender.test.ts

@@ -126,6 +126,25 @@ describe('CloudAppender', () => {
     expect(event?.['context_model']).toBe('switched-model');
   });
 
+  it('uses the event sessionId for top-level session_id when it differs from appender context', async () => {
+    const requests: CapturedRequest[] = [];
+    const appender = new CloudAppender(
+      baseOptions({
+        homeDir,
+        sessionId: 'default-session',
+        fetchImpl: makeFetch((req) => {
+          requests.push(req);
+          return okResponse();
+        }),
+      }),
+    );
+
+    appender.track('evt', { sessionId: 'event-session' });
+    await appender.flush();
+
+    expect(requests[0]?.body.events[0]?.['session_id']).toBe('event-session');
+  });
+
   it('sends Authorization header when a token is provided', async () => {
     const requests: CapturedRequest[] = [];
     const appender = new CloudAppender(

packages/kap-server/package.json

@@ -24,6 +24,7 @@
     "@fastify/multipart": "^10.0.0",
     "@fastify/swagger": "^9.7.0",
     "@moonshot-ai/agent-core-v2": "workspace:^",
+    "@moonshot-ai/kimi-code-oauth": "workspace:^",
     "@moonshot-ai/transcript": "workspace:^",
     "bcryptjs": "^2.4.3",
     "fastify": "^5.1.0",

packages/kap-server/src/services/telemetry.ts

@@ -0,0 +1,93 @@
+/**
+ * Server telemetry bootstrap — wires agent-core-v2's `CloudAppender` into the
+ * App-scoped `ITelemetryService` so engine events emitted inside the server
+ * process (`session_started`, turn / tool / permission events, `image_compress`,
+ * …) actually leave the process. Mirrors the v1 `kimi web` host
+ * (`initializeServerTelemetry` in apps/kimi-code): same product app name, the
+ * surface distinguished by `ui_mode = "web"`, the config `telemetry` toggle
+ * honored at startup (a change takes effect on restart).
+ */
+
+import {
+  type CloudAppender,
+  createCloudAppender,
+  IBootstrapService,
+  IConfigService,
+  type IDisposable,
+  IOAuthToolkit,
+  ITelemetryService,
+  type Scope,
+} from '@moonshot-ai/agent-core-v2';
+import { createKimiDeviceId } from '@moonshot-ai/kimi-code-oauth';
+
+// Same product as the CLI; the surface is distinguished by ui_mode (v1
+// convention: CLI_USER_AGENT_PRODUCT / WEB_UI_MODE in apps/kimi-code).
+const SERVER_TELEMETRY_APP_NAME = 'kimi-code-cli';
+const SERVER_TELEMETRY_UI_MODE = 'web';
+const TELEMETRY_DISABLE_ENV = 'KIMI_DISABLE_TELEMETRY';
+const TELEMETRY_DISABLE_ENV_VALUES = new Set(['1', 'true', 't', 'yes', 'y']);
+
+/**
+ * Cap on how long server close waits for its best-effort final flush. A wedged
+ * telemetry endpoint must not hold shutdown hostage.
+ */
+const TELEMETRY_SHUTDOWN_TIMEOUT_MS = 3_000;
+
+export interface ServerTelemetry {
+  /** Present only when telemetry is enabled by both config and environment. */
+  readonly appender?: CloudAppender;
+  readonly registration?: IDisposable;
+}
+
+function isTelemetryDisabledByEnv(core: Scope): boolean {
+  const value = core.accessor.get(IBootstrapService).getEnv(TELEMETRY_DISABLE_ENV);
+  return value !== undefined && TELEMETRY_DISABLE_ENV_VALUES.has(value.trim().toLowerCase());
+}
+
+export async function initializeServerTelemetry(
+  core: Scope,
+  homeDir: string,
+): Promise<ServerTelemetry> {
+  const service = core.accessor.get(ITelemetryService);
+  const config = core.accessor.get(IConfigService);
+  await config.ready;
+  const enabled = config.get('telemetry') !== false;
+  if (!enabled || isTelemetryDisabledByEnv(core)) return {};
+
+  const auth = core.accessor.get(IOAuthToolkit);
+  const appender = createCloudAppender(core.accessor, {
+    deviceId: createKimiDeviceId(homeDir),
+    appName: SERVER_TELEMETRY_APP_NAME,
+    uiMode: SERVER_TELEMETRY_UI_MODE,
+    model: config.get<string>('defaultModel') ?? undefined,
+    getAccessToken: async () => (await auth.getCachedAccessToken()) ?? null,
+  });
+  const registration = service.addAppender(appender);
+  try {
+    // The server is long-lived: flush on a timer, not only at the threshold.
+    appender.startPeriodicFlush();
+  } catch (error) {
+    registration.dispose();
+    throw error;
+  }
+  return { appender, registration };
+}
+
+export async function shutdownServerTelemetry(
+  telemetry: ServerTelemetry,
+  deadlineMs = Date.now() + TELEMETRY_SHUTDOWN_TIMEOUT_MS,
+): Promise<void> {
+  telemetry.registration?.dispose();
+  if (telemetry.appender === undefined) return;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    await Promise.race([
+      telemetry.appender.shutdown(),
+      new Promise<void>((resolve) => {
+        timer = setTimeout(resolve, Math.max(0, deadlineMs - Date.now()));
+      }),
+    ]);
+  } finally {
+    if (timer !== undefined) clearTimeout(timer);
+  }
+}

packages/kap-server/src/start.ts

@@ -61,6 +61,11 @@ import { createSecurityHeadersHook } from './middleware/securityHeaders';
 import { createAuthHook } from './middleware/auth';
 import { GuiStoreService } from './services/guiStore/guiStoreService';
 import { loadSnapshotConfig, SnapshotReader } from './services/snapshot';
+import {
+  initializeServerTelemetry,
+  type ServerTelemetry,
+  shutdownServerTelemetry,
+} from './services/telemetry';
 import { TranscriptService } from './services/transcript/transcriptService';
 import { ModelCatalogRefreshScheduler } from './services/modelCatalog/modelCatalogRefreshScheduler';
 import { createAuthFailureLimiter } from './middleware/rateLimit';
@@ -124,6 +129,14 @@ export interface ServerStartOptions {
    * embedding hosts (the CLI) should pass their own version.
    */
   readonly version?: string;
+  /**
+   * Opt-in cloud telemetry for the engine's `ITelemetryService` events: when
+   * true, a `CloudAppender` is attached at startup (still gated by the config
+   * `telemetry` toggle) and flushed on close. Defaults to false so tests and
+   * embedding hosts that wire their own telemetry never post to the real
+   * endpoint unintentionally; the CLI's `kimi web` host passes true.
+   */
+  readonly telemetry?: boolean;
 }
 
 export interface RunningServer {
@@ -202,7 +215,7 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
   // rooted at `homeDir`, so the Store facades above it (append-log, atomic
   // document, blob) — and in turn session metadata, wire records, blobs, and
   // the session index — all persist to disk.
-  const { app: core } = bootstrap({ homeDir, configPath }, [
+  const { app: core } = bootstrap({ homeDir, configPath, clientVersion: hostVersion }, [
     ...logSeed(logging),
     // Default host identity so outbound requests (model, WebSearch, registry
     // refresh) carry a product User-Agent even when the embedding host did not
@@ -213,6 +226,23 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
     ...(opts.seeds ?? []),
   ]);
 
+  // Attach the cloud telemetry appender BEFORE any session is created:
+  // `session_started` / `session_load_failed` fire inside create()/resume(), so
+  // an appender wired later would drop them to the null appender. Opt-in via
+  // `opts.telemetry` (off by default so tests never post to the real endpoint);
+  // best-effort — telemetry must never block server boot.
+  let telemetry: ServerTelemetry = {};
+  if (opts.telemetry === true) {
+    try {
+      telemetry = await initializeServerTelemetry(core, homeDir);
+    } catch (error) {
+      logger.warn(
+        { err: error instanceof Error ? error.message : String(error) },
+        'telemetry initialization failed; continuing without telemetry',
+      );
+    }
+  }
+
   if (exposureClass !== 'loopback') {
     logger.warn(
       { host, exposureClass },
@@ -292,8 +322,20 @@ export async function startServer(opts: ServerStartOptions = {}): Promise<Runnin
     await app.close();
     authFailureLimiter?.dispose();
     modelCatalogRefreshScheduler.dispose();
-    core.dispose();
-    await registration.release();
+    // Telemetry is best-effort and must never prevent core or instance cleanup.
+    try {
+      await shutdownServerTelemetry(telemetry);
+    } catch (error) {
+      logger.warn(
+        { err: error instanceof Error ? error.message : String(error) },
+        'telemetry shutdown failed; continuing server cleanup',
+      );
+    }
+    try {
+      core.dispose();
+    } finally {
+      await registration.release();
+    }
   };
 
   const connectionRegistry = new ConnectionRegistry();