import * as fs from 'fs' import * as path from 'path' import { getOrCreateInstallId } from './telemetry/install-id' import { sanitizeString } from './telemetry/sanitize' import { syncOnce } from './telemetry/sync' /** * Hardcoded telemetry endpoint + publishable key. * * These ship inside every published binary. No env var override, no runtime * fallback. All enabled installs send events to Agent Flow's Supabase project. * Forks that republish under a different name must edit these constants and * rebuild. * * Safe to commit: publishable keys are designed to be public. Postgres RLS * denies the anon role everything; the only write path is the telemetry-ingest * edge function, which runs under the secret key and validates every event. */ export const TELEMETRY_ENDPOINT = 'https://dxwtgqdkyunfhbywqmrz.supabase.co' export const TELEMETRY_PUBLISHABLE_KEY = 'sb_publishable_AgJ_DIUH9zm8E0yHC9KsRw_WsIv4qc8' /** * Progressive sync schedule. After init(), fire syncs at these offsets: * - 2s (captures session_start that the relay emits right after init) * - +2min * - +3min * - then every 5min * * Short sessions get flushed quickly; long sessions settle into steady cadence. */ const FIRST_SYNC_DELAY_MS = 2 * 1000 const SYNC_SCHEDULE_MS = [2 * 60 * 1000, 3 * 60 * 1000] const SYNC_REPEAT_MS = 5 * 60 * 1000 const FALSY_VALUES = new Set(['false', '0', 'disabled', '']) export interface TelemetryEvent { event_type: 'session_start' | 'session_end' | 'error' session_id: string agent_flow_version: string os: string arch: string source?: string duration_s?: number event_count?: number error_class?: string /** Comma-separated distinct model IDs observed during the session * (e.g., `"claude-opus-4-7,gpt-5"`). session_end only. */ models?: string /** Which runtimes were being watched: `"claude"`, `"codex"`, or `"claude,codex"`. * session_end only. */ runtimes?: string } export interface TelemetryClientOptions { /** Directory for events.jsonl and .cursor. Usually `~/.agent-flow/telemetry`. */ logDir: string /** Path to the stable install UUID. Usually `~/.agent-flow/installation-id`. */ installIdPath: string /** Override for tests. Defaults to `process.env`. */ env?: NodeJS.ProcessEnv /** Override the endpoint for tests. Defaults to the hardcoded constant. */ endpoint?: string /** Override the key for tests. Defaults to the hardcoded constant. */ apiKey?: string } export interface TelemetryClient { /** Resolve install ID and start the sync timer when telemetry is enabled. */ init(): Promise /** Append an event to the JSONL log. No-op when disabled. */ emit(event: TelemetryEvent): void /** Current enabled state. Re-evaluated from env on every call. */ isEnabled(): boolean /** Stop the sync timer and do a final flush. */ dispose(): Promise } /** * Pure function: given an env record, return true if telemetry should emit. * * Rules: * - `DO_NOT_TRACK` truthy → disabled (wins over everything) * - `AGENT_FLOW_TELEMETRY` falsy (`false`, `0`, `disabled`, ``) → disabled * - Otherwise enabled (including when AGENT_FLOW_TELEMETRY is unset) */ export function isTelemetryEnabled(env: NodeJS.ProcessEnv): boolean { const dnt = env.DO_NOT_TRACK if (dnt !== undefined && dnt !== '' && dnt !== '0' && dnt.toLowerCase() !== 'false') { return false } const flag = env.AGENT_FLOW_TELEMETRY if (flag !== undefined && FALSY_VALUES.has(flag.toLowerCase())) { return false } return true } export function createTelemetryClient(opts: TelemetryClientOptions): TelemetryClient { const logDir = opts.logDir const jsonlPath = path.join(logDir, 'events.jsonl') const cursorPath = path.join(logDir, '.cursor') const endpoint = opts.endpoint ?? TELEMETRY_ENDPOINT const apiKey = opts.apiKey ?? TELEMETRY_PUBLISHABLE_KEY const getEnv = () => opts.env ?? process.env let installId = '' let syncTimer: NodeJS.Timeout | null = null let disposed = false /** Promise for the currently-running sync, or null if idle. Prevents two * syncOnce calls from racing on the cursor file — the scheduled sync and * dispose's final sync used to both read cursor=N before either advanced * it, POSTing overlapping batches and double-inserting. */ let syncInFlight: Promise | null = null function enabled(): boolean { return isTelemetryEnabled(getEnv()) } function serialize(event: TelemetryEvent): string { return JSON.stringify({ v: 1, ts: new Date().toISOString(), event_type: event.event_type, installation_id: installId, session_id: sanitizeString(event.session_id), agent_flow_version: sanitizeString(event.agent_flow_version), os: sanitizeString(event.os, 16), arch: sanitizeString(event.arch, 16), source: sanitizeString(event.source ?? 'npx', 32), duration_s: event.duration_s ?? null, event_count: event.event_count ?? null, error_class: event.error_class ? sanitizeString(event.error_class, 64) : null, models: event.models ? sanitizeString(event.models, 128) : null, runtimes: event.runtimes ? sanitizeString(event.runtimes, 32) : null, }) } function append(event: TelemetryEvent) { if (!fs.existsSync(logDir)) fs.mkdirSync(logDir, { recursive: true }) fs.appendFileSync(jsonlPath, serialize(event) + '\n', { mode: 0o600 }) } function fireSync() { if (syncInFlight) return // another sync is already running — skip, not queue syncInFlight = syncOnce({ jsonlPath, cursorPath, endpoint, apiKey }) .catch(() => { // Swallow — next tick retries from the same cursor position. }) .finally(() => { syncInFlight = null }) } function scheduleNext(idx: number) { const delay = idx < SYNC_SCHEDULE_MS.length ? SYNC_SCHEDULE_MS[idx] : SYNC_REPEAT_MS syncTimer = setTimeout(() => { fireSync() scheduleNext(idx + 1) }, delay) syncTimer.unref?.() } function startSync() { if (syncTimer) return // First sync on a short delay so the relay can write session_start first. syncTimer = setTimeout(() => { fireSync() scheduleNext(0) }, FIRST_SYNC_DELAY_MS) syncTimer.unref?.() } function stopSync() { if (syncTimer) { clearTimeout(syncTimer); syncTimer = null } } return { async init() { // When disabled, write nothing to disk — no install-id, no telemetry/ dir. if (!enabled()) return installId = getOrCreateInstallId(opts.installIdPath) startSync() }, emit(event: TelemetryEvent) { if (!enabled()) return // Lazy install-id init in case env toggled between init() and emit(). if (!installId) installId = getOrCreateInstallId(opts.installIdPath) append(event) }, isEnabled() { return enabled() }, async dispose() { if (disposed) return disposed = true stopSync() // Wait for any in-flight scheduled sync to finish so our final flush // starts from an up-to-date cursor. if (syncInFlight) { try { await syncInFlight } catch { /* best effort */ } } if (enabled()) { try { await syncOnce({ jsonlPath, cursorPath, endpoint, apiKey }) } catch { /* best effort */ } } }, } }