fix(agent-core-v2): track session index mirror give-up event once per failure episode

packages/agent-core-v2/src/app/sessionIndex/sessionIndexMirrorService.ts

@@ -37,6 +37,7 @@ export class SessionIndexMirror extends Disposable implements ISessionIndexMirro
   private readonly timer = this._register(new IntervalTimer({ unref: true }));
   private flushing: Promise<void> | undefined;
   private consecutiveFailures = 0;
+  private giveUpTracked = false;
   private disposed = false;
   private overflowLogged = false;
 
@@ -169,6 +170,7 @@ export class SessionIndexMirror extends Disposable implements ISessionIndexMirro
         if (this.pendingMap.get(id) === summary) this.pendingMap.delete(id);
       }
       this.consecutiveFailures = 0;
+      this.giveUpTracked = false;
     } catch (error) {
       this.consecutiveFailures += 1;
       this.log.warn('failed to flush session index mirror chunk', {
@@ -180,10 +182,13 @@ export class SessionIndexMirror extends Disposable implements ISessionIndexMirro
         this.log.warn('session index mirror giving up until the next record; reconciliation will heal', {
           pending: this.pendingMap.size,
         });
-        this.telemetry.track2('session_index_mirror_give_up', {
-          pending_count: this.pendingMap.size,
-          consecutive_failures: this.consecutiveFailures,
-        });
+        if (!this.giveUpTracked) {
+          this.giveUpTracked = true;
+          this.telemetry.track2('session_index_mirror_give_up', {
+            pending_count: this.pendingMap.size,
+            consecutive_failures: this.consecutiveFailures,
+          });
+        }
       }
     }
   }

packages/agent-core-v2/test/app/sessionIndex/sessionIndexMirror.test.ts

@@ -31,6 +31,7 @@ import { IQueryStore } from '#/persistence/interface/queryStore';
 
 import { stubBootstrap } from '../bootstrap/stubs';
 import { stubFlag } from '../flag/stubs';
+import { recordingTelemetry, type TelemetryRecord } from '../telemetry/stubs';
 import { stubLog } from '../../_base/log/stubs';
 
 const WORKSPACE = 'wd_test';
@@ -84,12 +85,15 @@ describe('SessionIndexMirror', () => {
     await queryStore.setCheckpoint(SESSION_INDEX_MANIFEST, { seq: GENERATION });
   }
 
-  function build(flagEnabled = true): ISessionIndexMirror {
+  function build(
+    flagEnabled = true,
+    telemetry: ITelemetryService = noopTelemetryService,
+  ): ISessionIndexMirror {
     const host = createScopedTestHost([
       stubPair(IBootstrapService, stubBootstrap(homeDir)),
       stubPair(ILogService, stubLog()),
       stubPair(IFlagService, stubFlag(flagEnabled)),
-      stubPair(ITelemetryService, noopTelemetryService),
+      stubPair(ITelemetryService, telemetry),
     ]);
     disposeHost = () => {
       host.dispose();
@@ -211,4 +215,57 @@ describe('SessionIndexMirror', () => {
       id: 'a',
     });
   });
+
+  it('tracks the give-up event once per consecutive failure episode', async () => {
+    const records: TelemetryRecord[] = [];
+    build(true, recordingTelemetry(records));
+    await publishGeneration();
+
+    const realBatch = queryStore.batch.bind(queryStore);
+    queryStore.batch = async () => {
+      throw new Error('injected flush failure');
+    };
+    const giveUps = (): TelemetryRecord[] =>
+      records.filter((record) => record.event === 'session_index_mirror_give_up');
+
+    mirror.record(summary('a'));
+    for (let i = 0; i < 8; i++) await mirror.drain();
+    expect(giveUps()).toHaveLength(1);
+    expect(giveUps()[0]?.properties).toMatchObject({
+      pending_count: 1,
+      consecutive_failures: 5,
+    });
+
+    queryStore.batch = realBatch;
+    await mirror.drain();
+    expect(mirror.pending()).toEqual([]);
+
+    queryStore.batch = async () => {
+      throw new Error('injected flush failure');
+    };
+    mirror.record(summary('a', { updatedAt: 9 }));
+    for (let i = 0; i < 6; i++) await mirror.drain();
+    expect(giveUps()).toHaveLength(2);
+  });
+
+  it('tracks the give-up event when unpublished flushes precede a throwing one', async () => {
+    const records: TelemetryRecord[] = [];
+    build(true, recordingTelemetry(records));
+
+    mirror.record(summary('a'));
+    for (let i = 0; i < 5; i++) await mirror.drain();
+    expect(records).toEqual([]);
+
+    await publishGeneration();
+    queryStore.batch = async () => {
+      throw new Error('injected flush failure');
+    };
+    for (let i = 0; i < 3; i++) await mirror.drain();
+    const giveUps = records.filter((record) => record.event === 'session_index_mirror_give_up');
+    expect(giveUps).toHaveLength(1);
+    expect(giveUps[0]?.properties).toMatchObject({
+      pending_count: 1,
+      consecutive_failures: 6,
+    });
+  });
 });