zclaw_openfang/desktop/src/lib/telemetry-collector.ts

/**
 * Telemetry Collector — 桌面端遥测收集器
 *
 * 收集本地 LLM 调用的 Token 用量统计和审计日志摘要（均无内容），
 * 定期批量上报到 SaaS。
 *
 * 用量缓冲区上限 100 条，审计缓冲区上限 200 条，超限自动 flush。
 * 定时 flush 每 5 分钟。仅在 SaaS 已登录时上报。
 */

import { saasClient } from './saas-client';
import { createLogger } from './logger';

const log = createLogger('TelemetryCollector');

// === Types ===

export interface TelemetryEntry {
  model_id: string;
  input_tokens: number;
  output_tokens: number;
  latency_ms?: number;
  success: boolean;
  error_type?: string;
  timestamp: string;
  connection_mode: string;
}

interface AuditEntry {
  action: string;
  target: string;
  result: string;
  timestamp: string;
}

// === State ===

const USAGE_BUFFER_LIMIT = 100;
const AUDIT_BUFFER_LIMIT = 200;
const FLUSH_INTERVAL_MS = 5 * 60 * 1000;

let usageBuffer: TelemetryEntry[] = [];
let auditBuffer: AuditEntry[] = [];
let flushTimer: ReturnType<typeof setInterval> | null = null;
let deviceId: string | null = null;

// === Public API ===

/**
 * 初始化遥测收集器（在 SaaS 登录后调用）。
 * @param devId 设备 ID（与 saasStore 使用的相同）
 */
export function initTelemetryCollector(devId: string): void {
  deviceId = devId;

  if (flushTimer) {
    clearInterval(flushTimer);
  }

  flushTimer = setInterval(() => {
    flushAll().catch((err: unknown) => {
      log.warn('Scheduled telemetry flush failed:', err);
    });
  }, FLUSH_INTERVAL_MS);

  log.info('Telemetry collector initialized');
}

/**
 * 停止遥测收集器（在 SaaS 登出时调用）。
 * 会尝试 flush 剩余条目。
 */
export function stopTelemetryCollector(): void {
  if (flushTimer) {
    clearInterval(flushTimer);
    flushTimer = null;
  }

  // 尝试最后一次 flush
  flushAll().catch(() => {
    // 登出时不阻塞
  });

  usageBuffer = [];
  auditBuffer = [];
  deviceId = null;
  log.info('Telemetry collector stopped');
}

/**
 * 记录一次 LLM 调用的用量。
 *
 * @param modelId 模型标识
 * @param inputTokens 输入 Token 数
 * @param outputTokens 输出 Token 数
 * @param options 可选参数
 */
export function recordLLMUsage(
  modelId: string,
  inputTokens: number,
  outputTokens: number,
  options?: {
    latencyMs?: number;
    success?: boolean;
    errorType?: string;
    connectionMode?: string;
  },
): void {
  if (!deviceId) return;

  usageBuffer.push({
    model_id: modelId,
    input_tokens: inputTokens,
    output_tokens: outputTokens,
    latency_ms: options?.latencyMs,
    success: options?.success ?? true,
    error_type: options?.errorType,
    timestamp: new Date().toISOString(),
    connection_mode: options?.connectionMode || 'tauri',
  });

  if (usageBuffer.length >= USAGE_BUFFER_LIMIT) {
    flushUsage().catch((err: unknown) => {
      log.warn('Auto-flush usage triggered but failed:', err);
    });
  }
}

/**
 * 记录一条审计日志摘要（仅操作类型，无内容）。
 *
 * @param action 操作类型（如 "hand.trigger", "agent.create"）
 * @param target 操作目标（如 Agent/Hand 名称）
 * @param result 操作结果
 */
export function recordAuditEvent(
  action: string,
  target: string,
  result: 'success' | 'failure' | 'pending',
): void {
  if (!deviceId) return;

  auditBuffer.push({
    action,
    target,
    result,
    timestamp: new Date().toISOString(),
  });

  if (auditBuffer.length >= AUDIT_BUFFER_LIMIT) {
    flushAudit().catch((err: unknown) => {
      log.warn('Auto-flush audit triggered but failed:', err);
    });
  }
}

// === Internal ===

async function flushAll(): Promise<void> {
  await Promise.allSettled([
    flushUsage(),
    flushAudit(),
  ]);
}

async function flushUsage(): Promise<void> {
  if (usageBuffer.length === 0 || !deviceId || !saasClient.isAuthenticated()) {
    return;
  }

  const entries = usageBuffer;
  usageBuffer = [];

  try {
    const appVersion = typeof __APP_VERSION__ !== 'undefined' ? __APP_VERSION__ : 'unknown';

    const result = await saasClient.reportTelemetry({
      device_id: deviceId,
      app_version: appVersion,
      entries,
    });

    log.info(`Usage telemetry flushed: ${result.accepted} accepted, ${result.rejected} rejected`);
  } catch (err: unknown) {
    usageBuffer = [...entries, ...usageBuffer].slice(0, USAGE_BUFFER_LIMIT * 2);
    log.warn('Usage telemetry flush failed, entries re-buffered:', err);
  }
}

async function flushAudit(): Promise<void> {
  if (auditBuffer.length === 0 || !deviceId || !saasClient.isAuthenticated()) {
    return;
  }

  const entries = auditBuffer;
  auditBuffer = [];

  try {
    const result = await saasClient.reportAuditSummary({
      device_id: deviceId,
      entries,
    });

    log.info(`Audit summary flushed: ${result.accepted} accepted`);
  } catch (err: unknown) {
    auditBuffer = [...entries, ...auditBuffer].slice(0, AUDIT_BUFFER_LIMIT * 2);
    log.warn('Audit summary flush failed, entries re-buffered:', err);
  }
}