import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test'; import { mkdtempSync, rmSync } from 'fs'; import { tmpdir } from 'os'; import { join } from 'path'; import { postHogCaptureCalls } from '../preload'; import { __resetTelemetryForTests } from '../../src/services/telemetry/telemetry'; import { telemetryBuffer } from '../../src/services/telemetry/buffer'; /** * TelemetryBuffer unit tests. * * posthog-node is mocked globally in tests/preload.ts (bunfig.toml preload). * We verify buffer behaviour by asserting on postHogCaptureCalls — the same * spy array the telemetry-client tests use. Consent is forced on via env vars * so captureEvent() passes the consent gate and forwards to the mock client. * * Phase 2: session_compressed is a PER-SESSION accumulator (keyed by * sessionDbId, flushed at session end via flushSession / drainAllSessions / * safetyFlush). context_injected stays a TIME-WINDOW rollup drained by flush(). */ let tempDir: string; const savedEnv: Record = {}; const ENV_KEYS = [ 'CLAUDE_MEM_DATA_DIR', 'CLAUDE_MEM_TELEMETRY', 'CLAUDE_MEM_TELEMETRY_DEBUG', 'DO_NOT_TRACK', ]; beforeAll(() => { for (const key of ENV_KEYS) savedEnv[key] = process.env[key]; tempDir = mkdtempSync(join(tmpdir(), 'claude-mem-buffer-test-')); process.env.CLAUDE_MEM_DATA_DIR = tempDir; process.env.CLAUDE_MEM_TELEMETRY = '1'; delete process.env.CLAUDE_MEM_TELEMETRY_DEBUG; delete process.env.DO_NOT_TRACK; __resetTelemetryForTests(); }); afterAll(() => { for (const key of ENV_KEYS) { if (savedEnv[key] === undefined) delete process.env[key]; else process.env[key] = savedEnv[key]; } rmSync(tempDir, { recursive: true, force: true }); telemetryBuffer.__resetForTests(); __resetTelemetryForTests(); }); beforeEach(() => { postHogCaptureCalls.length = 0; telemetryBuffer.__resetForTests(); }); afterEach(() => { telemetryBuffer.__resetForTests(); }); // --------------------------------------------------------------------------- // flushSession() — per-session session_compressed rollup (Phase 2) // --------------------------------------------------------------------------- describe('flushSession() — observer_turn_rollup', () => { it('emits exactly ONE rollup for N records of one session with correct sums and rollup_reason', () => { const SID = 42; telemetryBuffer.record('session_compressed', SID, { outcome: 'ok', tokens_input: 1000, tokens_output: 200, cost_usd: 0.01, duration_ms: 800, compression_ms: 400, model: 'claude-sonnet-4-5', }); telemetryBuffer.record('session_compressed', SID, { outcome: 'ok', tokens_input: 2000, tokens_output: 300, cost_usd: 0.02, duration_ms: 1200, compression_ms: 600, model: 'claude-sonnet-4-5', }); telemetryBuffer.record('session_compressed', SID, { outcome: 'error', tokens_input: 500, tokens_output: 100, cost_usd: 0.005, duration_ms: 300, // compression_ms deliberately omitted — must be skipped from avg model: 'claude-haiku-3-5', }); const emitted = telemetryBuffer.flushSession(SID, 'session_end'); expect(emitted).toBe(true); // Exactly one rollup event for the whole session expect(postHogCaptureCalls.length).toBe(1); const call = postHogCaptureCalls[0] as { event: string; properties: Record }; expect(call.event).toBe('observer_turn_rollup'); const p = call.properties; expect(p.count).toBe(3); expect(p.total_tokens_input).toBe(3500); expect(p.total_tokens_output).toBe(600); expect(p.total_cost_usd).toBeCloseTo(0.035, 6); // avg_duration_ms: (800 + 1200 + 300) / 3 = 766.666... expect(p.avg_duration_ms).toBeCloseTo(2300 / 3, 4); // avg_compression_ms: only 2 records had it → (400 + 600) / 2 = 500 expect(p.avg_compression_ms).toBe(500); expect(p.outcomes_ok).toBe(2); expect(p.outcomes_error).toBe(1); expect(p.outcomes_aborted).toBe(0); expect(p.outcomes_invalid_output).toBe(0); expect(p.top_model).toBe('claude-sonnet-4-5'); expect(typeof p.window_start_ts).toBe('number'); expect(p.window_start_ts).toBeGreaterThan(0); // Phase 2 metadata expect(p.rollup_reason).toBe('session_end'); expect(p.window_seq).toBe(0); // sessionDbId must NEVER appear in emitted props (map key only) expect(p.sessionDbId).toBeUndefined(); }); it('sums generation-side observation volume and obs_type_* across the session', () => { const SID = 7; telemetryBuffer.record('session_compressed', SID, { outcome: 'ok', cost_usd: 0.04, count: 5, obs_type_bugfix: 2, obs_type_discovery: 1, obs_type_decision: 0, obs_type_refactor: 1, obs_type_other: 1, }); telemetryBuffer.record('session_compressed', SID, { outcome: 'ok', cost_usd: 0.06, count: 3, obs_type_bugfix: 0, obs_type_discovery: 2, obs_type_decision: 1, obs_type_refactor: 0, obs_type_other: 0, }); expect(telemetryBuffer.flushSession(SID, 'session_end')).toBe(true); expect(postHogCaptureCalls.length).toBe(1); const p = (postHogCaptureCalls[0] as { properties: Record }).properties; // rollup `count` is TURNS (records.length); observations_created is the sum // of per-turn observation counts — distinct concepts. expect(p.count).toBe(2); expect(p.observations_created).toBe(8); expect(p.total_cost_usd).toBeCloseTo(0.1, 6); // cost-per-observation is now derivable from the rollup alone. expect((p.total_cost_usd as number) / (p.observations_created as number)).toBeCloseTo(0.0125, 6); expect(p.obs_type_bugfix).toBe(2); expect(p.obs_type_discovery).toBe(3); expect(p.obs_type_decision).toBe(1); expect(p.obs_type_refactor).toBe(1); expect(p.obs_type_other).toBe(1); }); it('covers all outcome buckets correctly', () => { const SID = 7; telemetryBuffer.record('session_compressed', SID, { outcome: 'ok' }); telemetryBuffer.record('session_compressed', SID, { outcome: 'aborted' }); telemetryBuffer.record('session_compressed', SID, { outcome: 'invalid_output' }); telemetryBuffer.record('session_compressed', SID, { outcome: 'error' }); telemetryBuffer.record('session_compressed', SID, { outcome: 'ok' }); telemetryBuffer.flushSession(SID, 'session_end'); const p = (postHogCaptureCalls[0] as { properties: Record }).properties; expect(p.count).toBe(5); expect(p.outcomes_ok).toBe(2); expect(p.outcomes_error).toBe(1); expect(p.outcomes_aborted).toBe(1); expect(p.outcomes_invalid_output).toBe(1); }); it('omits top_model when no model strings are recorded', () => { const SID = 9; telemetryBuffer.record('session_compressed', SID, { outcome: 'error' }); telemetryBuffer.flushSession(SID, 'session_end'); const p = (postHogCaptureCalls[0] as { properties: Record }).properties; expect(p.top_model).toBeUndefined(); }); it('two sessions accumulate independently and each emits its own rollup', () => { const A = 100; const B = 200; telemetryBuffer.record('session_compressed', A, { outcome: 'ok', tokens_input: 10 }); telemetryBuffer.record('session_compressed', B, { outcome: 'error', tokens_input: 999 }); telemetryBuffer.record('session_compressed', A, { outcome: 'ok', tokens_input: 20 }); // Two live buckets expect(telemetryBuffer.__activeSessionBucketCount()).toBe(2); telemetryBuffer.flushSession(A, 'session_end'); expect(postHogCaptureCalls.length).toBe(1); let p = (postHogCaptureCalls[0] as { properties: Record }).properties; expect(p.count).toBe(2); expect(p.total_tokens_input).toBe(30); expect(p.outcomes_ok).toBe(2); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(1); telemetryBuffer.flushSession(B, 'session_end'); expect(postHogCaptureCalls.length).toBe(2); p = (postHogCaptureCalls[1] as { properties: Record }).properties; expect(p.count).toBe(1); expect(p.total_tokens_input).toBe(999); expect(p.outcomes_error).toBe(1); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(0); }); it('re-flush of an already-flushed (absent) session is a safe no-op', () => { const SID = 55; telemetryBuffer.record('session_compressed', SID, { outcome: 'ok' }); expect(telemetryBuffer.flushSession(SID, 'session_end')).toBe(true); expect(postHogCaptureCalls.length).toBe(1); // Second flush — bucket already removed; emits nothing (guards the // deleteSession/removeSessionImmediate double-teardown pair). expect(telemetryBuffer.flushSession(SID, 'session_end')).toBe(false); expect(postHogCaptureCalls.length).toBe(1); // Flushing a never-seen session is also a no-op. expect(telemetryBuffer.flushSession(99999, 'session_end')).toBe(false); expect(postHogCaptureCalls.length).toBe(1); }); it('drops a session_compressed record with a non-numeric session key', () => { // @ts-expect-error — exercising the runtime guard against a null key telemetryBuffer.record('session_compressed', null, { outcome: 'ok' }); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(0); telemetryBuffer.drainAllSessions('worker_shutdown'); expect(postHogCaptureCalls.length).toBe(0); }); }); // --------------------------------------------------------------------------- // drainAllSessions() — worker_shutdown // --------------------------------------------------------------------------- describe('drainAllSessions() — worker_shutdown', () => { it('flushes ALL active session buckets with rollup_reason worker_shutdown', () => { telemetryBuffer.record('session_compressed', 1, { outcome: 'ok' }); telemetryBuffer.record('session_compressed', 2, { outcome: 'ok' }); telemetryBuffer.record('session_compressed', 3, { outcome: 'error' }); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(3); telemetryBuffer.drainAllSessions('worker_shutdown'); expect(postHogCaptureCalls.length).toBe(3); for (const c of postHogCaptureCalls) { const call = c as { event: string; properties: Record }; expect(call.event).toBe('observer_turn_rollup'); expect(call.properties.rollup_reason).toBe('worker_shutdown'); } // Map drained — memory released before client shutdown. expect(telemetryBuffer.__activeSessionBucketCount()).toBe(0); }); it('is a no-op when there are no active sessions', () => { telemetryBuffer.drainAllSessions('worker_shutdown'); expect(postHogCaptureCalls.length).toBe(0); }); }); // --------------------------------------------------------------------------- // safetyFlush() — over-cap partial rollup with window_seq increment // --------------------------------------------------------------------------- describe('safetyFlush() — over-cap sessions', () => { it('emits a partial rollup for an over-count session, bumps window_seq, and re-arms the bucket', () => { const SID = 314; // Exceed the hard record cap (SAFETY_MAX_RECORDS = 1000). for (let i = 0; i < 1000; i++) { telemetryBuffer.record('session_compressed', SID, { outcome: 'ok', tokens_input: 1 }); } telemetryBuffer.safetyFlush(); // One partial rollup, reason safety_flush, window_seq still 0 (the seq of // the window just emitted), count = 1000. expect(postHogCaptureCalls.length).toBe(1); const p = (postHogCaptureCalls[0] as { properties: Record }).properties; expect(p.rollup_reason).toBe('safety_flush'); expect(p.window_seq).toBe(0); expect(p.count).toBe(1000); expect(p.total_tokens_input).toBe(1000); // Bucket re-armed in place (NOT removed): map stays bounded, session keeps // accumulating into window_seq 1. expect(telemetryBuffer.__activeSessionBucketCount()).toBe(1); telemetryBuffer.record('session_compressed', SID, { outcome: 'error' }); telemetryBuffer.flushSession(SID, 'session_end'); expect(postHogCaptureCalls.length).toBe(2); const p2 = (postHogCaptureCalls[1] as { properties: Record }).properties; expect(p2.window_seq).toBe(1); expect(p2.rollup_reason).toBe('session_end'); expect(p2.count).toBe(1); // Now fully flushed. expect(telemetryBuffer.__activeSessionBucketCount()).toBe(0); }); it('leaves under-cap sessions untouched', () => { telemetryBuffer.record('session_compressed', 1, { outcome: 'ok' }); telemetryBuffer.safetyFlush(); expect(postHogCaptureCalls.length).toBe(0); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(1); }); }); // --------------------------------------------------------------------------- // Consent gate — nothing sent when consent is off // --------------------------------------------------------------------------- describe('consent off ⇒ nothing sent', () => { it('emits no events when DO_NOT_TRACK forces consent off', () => { const prev = process.env.DO_NOT_TRACK; process.env.DO_NOT_TRACK = '1'; __resetTelemetryForTests(); // clear cached consent try { telemetryBuffer.record('session_compressed', 1, { outcome: 'ok' }); telemetryBuffer.flushSession(1, 'session_end'); telemetryBuffer.record('session_compressed', 2, { outcome: 'ok' }); telemetryBuffer.drainAllSessions('worker_shutdown'); expect(postHogCaptureCalls.length).toBe(0); } finally { if (prev === undefined) delete process.env.DO_NOT_TRACK; else process.env.DO_NOT_TRACK = prev; __resetTelemetryForTests(); } }); }); // --------------------------------------------------------------------------- // flush() — context_injected TIME-WINDOW rollup (unchanged path) // --------------------------------------------------------------------------- describe('flush() — context_injected_rollup', () => { it('emits one rollup event with correct token sums and averages', () => { telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 500, observation_count: 12, tokens_saved_vs_naive: 4000, }); telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 1500, observation_count: 30, tokens_saved_vs_naive: 11000, }); telemetryBuffer.record('context_injected', null, { outcome: 'error' }); // no tokens/obs — skipped from sums telemetryBuffer.flush(); expect(postHogCaptureCalls.length).toBe(1); const call = postHogCaptureCalls[0] as { event: string; properties: Record }; expect(call.event).toBe('context_injected_rollup'); const p = call.properties; expect(p.count).toBe(3); expect(p.total_tokens).toBe(2000); expect(p.avg_tokens).toBe(1000); // Injection-side observation accounting folded into the rollup. expect(p.total_observations_injected).toBe(42); expect(p.total_tokens_saved_vs_naive).toBe(15000); expect(p.outcomes_ok).toBe(2); expect(p.outcomes_error).toBe(1); expect(typeof p.window_start_ts).toBe('number'); }); it('does NOT flush per-session session_compressed buckets', () => { telemetryBuffer.record('session_compressed', 1, { outcome: 'ok' }); telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 100 }); telemetryBuffer.flush(); // Only the context_injected rollup — the session bucket survives flush(). expect(postHogCaptureCalls.length).toBe(1); expect((postHogCaptureCalls[0] as { event: string }).event).toBe('context_injected_rollup'); expect(telemetryBuffer.__activeSessionBucketCount()).toBe(1); }); }); // --------------------------------------------------------------------------- // Empty buckets — no captureEvent call // --------------------------------------------------------------------------- describe('flush() — empty buckets', () => { it('emits no events when no records have been buffered', () => { telemetryBuffer.flush(); expect(postHogCaptureCalls.length).toBe(0); }); }); // --------------------------------------------------------------------------- // start() / stop() interval wiring // --------------------------------------------------------------------------- describe('start() / stop() interval wiring', () => { it('is idempotent — calling start() twice does not create two intervals', async () => { telemetryBuffer.start(50); telemetryBuffer.start(50); // second call must be a no-op telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 1 }); await new Promise(resolve => setTimeout(resolve, 80)); telemetryBuffer.stop(); // The interval flushed the time-window record automatically — exactly once. expect(postHogCaptureCalls.length).toBe(1); expect((postHogCaptureCalls[0] as { event: string }).event).toBe('context_injected_rollup'); }); it('stop() clears the interval so no further auto-flushes occur', async () => { telemetryBuffer.start(30); telemetryBuffer.stop(); telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 1 }); await new Promise(resolve => setTimeout(resolve, 60)); expect(postHogCaptureCalls.length).toBe(0); telemetryBuffer.flush(); expect(postHogCaptureCalls.length).toBe(1); }); it('stop() does not flush — caller must drain explicitly', async () => { telemetryBuffer.start(100); telemetryBuffer.record('context_injected', null, { outcome: 'ok', tokens_injected: 1 }); telemetryBuffer.stop(); expect(postHogCaptureCalls.length).toBe(0); telemetryBuffer.flush(); expect(postHogCaptureCalls.length).toBe(1); }); });