feat: add FilesystemSink for telemetry audit mode

Author: Hweinstock

src/cli/telemetry/__tests__/filesystem-sink.test.ts

@@ -0,0 +1,161 @@
+import type { ResourceAttributes } from '../schemas/common-attributes';
+import { FilesystemSink } from '../sinks/filesystem-sink';
+import { mkdir, readFile, rm } from 'fs/promises';
+import { randomUUID } from 'node:crypto';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterAll, beforeEach, describe, expect, it } from 'vitest';
+
+const sessionId = randomUUID();
+const testDir = join(tmpdir(), `agentcore-fs-sink-${randomUUID()}`);
+const outputDir = join(testDir, 'telemetry');
+
+const resource: ResourceAttributes = {
+  'service.name': 'agentcore-cli',
+  'service.version': '1.0.0',
+  'agentcore-cli.installation_id': randomUUID(),
+  'agentcore-cli.session_id': sessionId,
+  'agentcore-cli.mode': 'cli',
+  'os.type': 'Linux',
+  'os.version': '6.1.0',
+  'host.arch': 'x64',
+  'node.version': process.version,
+};
+
+function createSink(opts: { dir?: string; log?: (msg: string) => void } = {}) {
+  return new FilesystemSink({ outputDir: opts.dir ?? outputDir, resource, log: opts.log });
+}
+
+function expectFilePath(sink: FilesystemSink): string {
+  const p = sink.filePath;
+  if (p === undefined) throw new Error('expected filePath to be defined');
+  return p;
+}
+
+describe('FilesystemSink', () => {
+  beforeEach(async () => {
+    await rm(testDir, { recursive: true, force: true });
+    await mkdir(testDir, { recursive: true });
+  });
+  afterAll(() => rm(testDir, { recursive: true, force: true }));
+
+  it('filePath is undefined before any records', () => {
+    const sink = createSink();
+    expect(sink.filePath).toBeUndefined();
+  });
+
+  it('derives file name from first record command_group', () => {
+    const sink = createSink();
+    sink.record(100, { command_group: 'deploy', command: 'deploy' });
+    expect(sink.filePath).toBe(join(outputDir, `deploy-${sessionId}.json`));
+  });
+
+  it('uses "unknown" as entrypoint when command_group is absent', () => {
+    const sink = createSink();
+    sink.record(100, { command: 'orphan' });
+    expect(sink.filePath).toBe(join(outputDir, `unknown-${sessionId}.json`));
+  });
+
+  it('entrypoint stays fixed after first record', () => {
+    const sink = createSink();
+    sink.record(100, { command_group: 'deploy', command: 'deploy' });
+    sink.record(200, { command_group: 'status', command: 'status' });
+    expect(sink.filePath).toBe(join(outputDir, `deploy-${sessionId}.json`));
+  });
+
+  it('flush writes buffered entries as a JSON array', async () => {
+    const sink = createSink();
+    sink.record(42, { command_group: 'deploy', command: 'deploy', exit_reason: 'success' });
+
+    await sink.flush();
+
+    const path = expectFilePath(sink);
+    const written = JSON.parse(await readFile(path, 'utf-8'));
+    expect(written).toHaveLength(1);
+    expect(written[0]).toMatchObject({
+      metric_name: 'cli.command_run',
+      resource,
+      attributes: { command_group: 'deploy', command: 'deploy', exit_reason: 'success', duration_ms: 42 },
+    });
+  });
+
+  it('accumulates multiple records into a single file', async () => {
+    const sink = createSink();
+    sink.record(10, { command_group: 'add', command: 'add.agent' });
+    sink.record(20, { command_group: 'add', command: 'add.memory' });
+
+    await sink.flush();
+
+    const path = expectFilePath(sink);
+    const written = JSON.parse(await readFile(path, 'utf-8'));
+    expect(written).toHaveLength(2);
+    expect(written[0].attributes.duration_ms).toBe(10);
+    expect(written[1].attributes.duration_ms).toBe(20);
+  });
+
+  it('creates output directory if it does not exist', async () => {
+    const nested = join(testDir, 'deep', 'nested', 'telemetry');
+    const sink = createSink({ dir: nested });
+    sink.record(1, { command_group: 'status', command: 'status' });
+
+    await sink.flush();
+
+    const path = expectFilePath(sink);
+    const written = JSON.parse(await readFile(path, 'utf-8'));
+    expect(written).toHaveLength(1);
+  });
+
+  it('flush is a no-op when no records exist', async () => {
+    const sink = createSink();
+    await expect(sink.flush()).resolves.toBeUndefined();
+  });
+
+  it('shutdown writes entries and logs audit message', async () => {
+    const logged: string[] = [];
+    const sink = createSink({ log: msg => logged.push(msg) });
+    sink.record(99, { command_group: 'invoke', command: 'invoke' });
+
+    await sink.shutdown();
+
+    const path = expectFilePath(sink);
+    const written = JSON.parse(await readFile(path, 'utf-8'));
+    expect(written).toHaveLength(1);
+    expect(logged).toHaveLength(1);
+    expect(logged[0]).toContain('[audit mode]');
+    expect(logged[0]).toContain(path);
+  });
+
+  it('shutdown does not log when no records were written', async () => {
+    const logged: string[] = [];
+    const sink = createSink({ log: msg => logged.push(msg) });
+
+    await sink.shutdown();
+
+    expect(logged).toHaveLength(0);
+  });
+
+  it('shutdown after flush with no new records still logs', async () => {
+    const logged: string[] = [];
+    const sink = createSink({ log: msg => logged.push(msg) });
+    sink.record(1, { command_group: 'deploy', command: 'deploy' });
+    await sink.flush();
+
+    await sink.shutdown();
+
+    expect(logged).toHaveLength(1);
+    expect(logged[0]).toContain('[audit mode]');
+  });
+
+  it('subsequent flushes include all records', async () => {
+    const sink = createSink();
+    sink.record(1, { command_group: 'deploy', command: 'deploy' });
+    await sink.flush();
+
+    sink.record(2, { command_group: 'deploy', command: 'deploy' });
+    await sink.flush();
+
+    const path = expectFilePath(sink);
+    const written = JSON.parse(await readFile(path, 'utf-8'));
+    expect(written).toHaveLength(2);
+  });
+});

src/cli/telemetry/index.ts

@@ -3,4 +3,5 @@ export type { TelemetryPreference } from './config.js';
 export { TelemetryClient, CANCELLED } from './client.js';
 export { type MetricSink, CompositeSink } from './sinks/metric-sink.js';
 export { OtelMetricSink, type OtelMetricSinkConfig } from './sinks/otel-metric-sink.js';
+export { FilesystemSink, type FilesystemSinkConfig } from './sinks/filesystem-sink.js';
 export { classifyError, isUserError } from './error-classification.js';

src/cli/telemetry/sinks/filesystem-sink.ts

@@ -0,0 +1,63 @@
+import type { ResourceAttributes } from '../schemas/common-attributes.js';
+import type { MetricSink } from './metric-sink.js';
+import { mkdir, writeFile } from 'fs/promises';
+import { join } from 'path';
+
+export interface FilesystemSinkConfig {
+  outputDir: string;
+  resource: ResourceAttributes;
+  log?: (message: string) => void;
+}
+
+interface MetricEntry {
+  metric_name: 'cli.command_run';
+  resource: ResourceAttributes;
+  attributes: Record<string, string | number>;
+}
+
+export class FilesystemSink implements MetricSink {
+  private readonly entries: MetricEntry[] = [];
+  private readonly outputDir: string;
+  private readonly resource: ResourceAttributes;
+  private readonly sessionId: string;
+  private readonly log: (message: string) => void;
+  private entrypoint: string | undefined;
+
+  constructor(config: FilesystemSinkConfig) {
+    this.outputDir = config.outputDir;
+    this.resource = config.resource;
+    this.sessionId = config.resource['agentcore-cli.session_id'];
+    this.log = config.log ?? (msg => console.error(msg));
+  }
+
+  get filePath(): string | undefined {
+    if (this.entrypoint === undefined) return undefined;
+    return join(this.outputDir, `${this.entrypoint}-${this.sessionId}.json`);
+  }
+
+  record(value: number, attrs: Record<string, string | number>): void {
+    this.entrypoint ??= String(attrs.command_group ?? 'unknown');
+    this.entries.push({
+      metric_name: 'cli.command_run',
+      resource: this.resource,
+      attributes: { ...attrs, duration_ms: value },
+    });
+  }
+
+  async flush(): Promise<void> {
+    await this.writeEntries();
+  }
+
+  async shutdown(): Promise<void> {
+    await this.writeEntries();
+    if (this.filePath !== undefined) {
+      this.log(`[audit mode] Telemetry written to ${this.filePath}`);
+    }
+  }
+
+  private async writeEntries(): Promise<void> {
+    if (this.entries.length === 0 || this.filePath === undefined) return;
+    await mkdir(this.outputDir, { recursive: true });
+    await writeFile(this.filePath, JSON.stringify(this.entries, null, 2));
+  }
+}