fix: make CloudAppender shutdown durable

.changeset/cloud-appender-durable-shutdown.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/agent-core-v2": patch
+---
+
+Make CloudAppender shutdown durable: serialize flush() to prevent concurrent buffer races, add deadline-bounded shutdown() with AbortController that hands unsent events to durable storage and replays v2 spool data before completing.

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

@@ -1,11 +1,13 @@
 /**
- * `telemetry` domain — `CloudAppender`, an `ITelemetryAppender` that
+ * `telemetry` domain (L1) — `CloudAppender`, an `ITelemetryAppender` that
  * batches events, drops non-primitive properties, redacts PII from string
  * values, enriches events with common context, and posts them to the
  * telemetry endpoint through `CloudTransport`, which persists failed events
- * through the `storage` byte layer. Reads host facts (`clientIdentity`, env,
- * platform/arch) from `IBootstrapService`; `createCloudAppender` assembles
- * one from a `ServicesAccessor` so hosts only supply identity facts.
+ * through the `storage` byte layer (`IFileSystemStorageService`). Reads host
+ * facts (env, platform/arch) from `IBootstrapService`;
+ * `createCloudAppender` assembles one from a `ServicesAccessor` so hosts only
+ * supply identity facts.
+ *
  * App-scoped; independent of `@moonshot-ai/kimi-telemetry`.
  */
 
@@ -74,6 +76,7 @@ export function createCloudAppender(
 
 const DEFAULT_FLUSH_THRESHOLD = 50;
 const DEFAULT_FLUSH_INTERVAL_MS = 30_000;
+const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000;
 
 export class CloudAppender implements ITelemetryAppender {
   private readonly transport: CloudTransport;
@@ -85,6 +88,10 @@ export class CloudAppender implements ITelemetryAppender {
   private buffer: EnrichedCloudEvent[] = [];
   private flushTimer: ReturnType<typeof setInterval> | null = null;
 
+  private flushInFlight: Promise<void> | null = null;
+  private shutdownController: AbortController | null = null;
+  private shutDown = false;
+
   constructor(options: CloudAppenderOptions) {
     this.deviceId = options.deviceId;
     this.sessionId = options.sessionId ?? null;
@@ -105,6 +112,7 @@ export class CloudAppender implements ITelemetryAppender {
   }
 
   track(event: string, properties?: TelemetryProperties): void {
+    if (this.shutDown) return;
     const eventSessionId = properties?.['sessionId'];
     const enriched: EnrichedCloudEvent = {
       event_id: randomUUID().replaceAll('-', ''),
@@ -137,15 +145,70 @@ export class CloudAppender implements ITelemetryAppender {
   }
 
   async flush(): Promise<void> {
+    const prev = this.flushInFlight;
+    const flushPromise = (async () => {
+      if (prev !== null) {
+        await prev.catch(() => {});
+      }
+      if (this.buffer.length === 0) return;
+      await this.doFlush();
+    })();
+    this.flushInFlight = flushPromise;
+    await flushPromise;
+  }
+
+  private async doFlush(): Promise<void> {
     if (this.buffer.length === 0) return;
     const events = this.buffer;
     this.buffer = [];
-    await this.transport.send(events);
+    const signal = this.shutdownController?.signal;
+    await this.transport.send(events, signal);
   }
 
-  async shutdown(): Promise<void> {
+  async shutdown(deadlineMs?: number): Promise<void> {
+    if (this.shutDown) return;
+    this.shutDown = true;
+
     this.stopPeriodicFlush();
-    await this.flush();
+
+    const deadline = deadlineMs ?? Date.now() + DEFAULT_SHUTDOWN_TIMEOUT_MS;
+    this.shutdownController = new AbortController();
+    const signal = this.shutdownController.signal;
+    const remainingMs = Math.max(0, deadline - Date.now());
+
+    let deadlineTimer: ReturnType<typeof setTimeout> | undefined;
+    if (remainingMs > 0) {
+      deadlineTimer = setTimeout(() => {
+        this.shutdownController?.abort(new Error('shutdown deadline expired'));
+      }, remainingMs);
+      deadlineTimer.unref?.();
+    } else {
+      this.shutdownController.abort(new Error('shutdown deadline already expired'));
+    }
+
+    try {
+      // Bound shutdown even when a pre-existing flush is already running:
+      // a send started before shutdownController existed has no abort signal,
+      // so a hung request could block the server close past the deadline.
+      await Promise.race([
+        this.flush().catch(() => {}),
+        new Promise<void>((resolve) => {
+          const t = setTimeout(() => resolve(), remainingMs);
+          t.unref?.();
+        }),
+      ]);
+
+      if (this.buffer.length > 0) {
+        await this.transport.saveToDisk(this.buffer).catch(() => {});
+        this.buffer = [];
+      }
+
+      await this.transport.retryDiskEvents(signal).catch(() => {});
+    } finally {
+      if (deadlineTimer !== undefined) {
+        clearTimeout(deadlineTimer);
+      }
+    }
   }
 
   startPeriodicFlush(): void {

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

@@ -1,6 +1,6 @@
 /**
- * `telemetry` domain — `CloudTransport`, the HTTP transport for cloud
- * telemetry. Posts enriched events to the telemetry endpoint with Bearer
+ * `telemetry` domain (L1) — `CloudTransport`, the HTTP transport behind
+ * `CloudAppender`. Posts enriched events to the telemetry endpoint with Bearer
  * auth, retry, and a byte-store fallback for failed events, persisted through
  * the `storage` byte layer (`IFileSystemStorageService`) under the `telemetry` scope.
  * App-scoped; independent of `@moonshot-ai/kimi-telemetry`.
@@ -139,10 +139,12 @@ export class CloudTransport {
     await this.storage.write(TELEMETRY_SCOPE, key, textEncoder.encode(text));
   }
 
-  async retryDiskEvents(): Promise<void> {
+  async retryDiskEvents(signal?: AbortSignal): Promise<void> {
+    if (signal?.aborted === true) return;
     const keys = await this.storage.list(TELEMETRY_SCOPE, FAILED_PREFIX);
     const now = this.now();
     for (const key of keys) {
+      if (signal?.aborted === true) return;
       if (!key.startsWith(FAILED_PREFIX) || !key.endsWith(JSONL_SUFFIX)) continue;
       const createdAt = parseFailedTimestamp(key);
       if (createdAt === undefined || now - createdAt > DISK_EVENT_MAX_AGE_MS) {
@@ -163,10 +165,11 @@ export class CloudTransport {
       }
 
       try {
-        await this.sendHttp(payload);
+        await this.sendHttp(payload, signal);
         await this.storage.delete(TELEMETRY_SCOPE, key);
       } catch (error) {
         if (error instanceof TransientCloudError) continue;
+        if (signal?.aborted === true) return;
       }
     }
   }

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

@@ -11,7 +11,7 @@ import {
 import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService';
 import { CloudAppender, type CloudAppenderOptions } from '#/app/telemetry/cloudAppender';
 
-import { stubBootstrap, stubClientIdentity } from '../bootstrap/stubs';
+import { stubBootstrap } from '../bootstrap/stubs';
 
 interface CapturedRequest {
   readonly url: string;
@@ -50,7 +50,7 @@ function baseOptions(
   const { homeDir: dir = '', storage, ...rest } = overrides;
   return {
     storage: storage ?? new FileStorageService(dir),
-    bootstrap: { ...stubBootstrap(), clientIdentity: { ...stubClientIdentity, version: '1.0.0' } },
+    bootstrap: { ...stubBootstrap(), clientVersion: '1.0.0' },
     deviceId: 'dev',
     appName: 'test-app',
     sleep: async () => {},
@@ -292,4 +292,127 @@ describe('CloudAppender', () => {
       resetUnexpectedErrorHandler();
     }
   });
+
+  describe('shutdown durability', () => {
+    it('serializes concurrent flush calls without losing events', async () => {
+      const requests: CapturedRequest[] = [];
+      const appender = new CloudAppender(
+        baseOptions({
+          homeDir,
+          fetchImpl: makeFetch((req) => {
+            requests.push(req);
+            return okResponse();
+          }),
+        }),
+      );
+
+      // Track multiple events
+      appender.track('e1');
+      appender.track('e2');
+      appender.track('e3');
+
+      // Fire multiple concurrent flushes — they must serialize, not race
+      await Promise.all([appender.flush(), appender.flush(), appender.flush()]);
+
+      // All events must have been sent (possibly in multiple batches, but
+      // the total event count must be 3)
+      const totalEvents = requests.reduce(
+        (sum, req) => sum + req.body.events.length,
+        0,
+      );
+      expect(totalEvents).toBe(3);
+    });
+
+    it('shutdown is idempotent — calling it twice only flushes once', async () => {
+      let sends = 0;
+      const appender = new CloudAppender(
+        baseOptions({
+          homeDir,
+          fetchImpl: makeFetch(() => {
+            sends += 1;
+            return okResponse();
+          }),
+        }),
+      );
+
+      appender.track('e1');
+      await appender.shutdown();
+      await appender.shutdown(); // Second call should be a no-op
+      expect(sends).toBe(1);
+    });
+
+    it('shutdown respects the deadline and hands unsent events to disk', async () => {
+      const appender = new CloudAppender(
+        baseOptions({
+          homeDir,
+          // Fetch that never resolves within the deadline
+          fetchImpl: makeFetch(
+            () => new Promise((resolve) => setTimeout(() => resolve(okResponse()), 10_000)),
+          ),
+        }),
+      );
+
+      appender.track('e1');
+      appender.track('e2');
+
+      // Shutdown with a very short deadline (already expired)
+      await appender.shutdown(Date.now());
+
+      // Events should have been saved to disk
+      const files = readdirSync(join(homeDir, 'telemetry')).filter((f) =>
+        f.startsWith('failed_'),
+      );
+      expect(files.length).toBeGreaterThanOrEqual(1);
+    });
+
+    it('shutdown replays spool data from disk', async () => {
+      let sends = 0;
+      let shouldFail = true;
+      const appender = new CloudAppender(
+        baseOptions({
+          homeDir,
+          fetchImpl: makeFetch(() => {
+            if (shouldFail) return statusResponse(500);
+            sends += 1;
+            return okResponse();
+          }),
+        }),
+      );
+
+      // First flush fails → events go to disk
+      appender.track('disk_event');
+      await appender.flush();
+      expect(
+        readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')),
+      ).toHaveLength(1);
+
+      // Now make fetch succeed and call shutdown — it should replay disk events
+      shouldFail = false;
+      await appender.shutdown();
+
+      // Disk file should be cleaned up after successful replay
+      expect(
+        readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')),
+      ).toHaveLength(0);
+      expect(sends).toBe(1); // The replayed event was sent
+    });
+
+    it('track() after shutdown is silently ignored', async () => {
+      let sends = 0;
+      const appender = new CloudAppender(
+        baseOptions({
+          homeDir,
+          fetchImpl: makeFetch(() => {
+            sends += 1;
+            return okResponse();
+          }),
+        }),
+      );
+
+      appender.track('before_shutdown');
+      await appender.shutdown();
+      appender.track('after_shutdown'); // Should be ignored
+      expect(sends).toBe(1);
+    });
+  });
 });

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

@@ -79,15 +79,5 @@ export async function shutdownServerTelemetry(
 ): 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);
-  }
+  await telemetry.appender.shutdown(deadlineMs);
 }