From ca39a69b6befc6e87acbcbcc49d4ec7741a366f6 Mon Sep 17 00:00:00 2001 From: Andrew Lee Date: Sun, 12 Jul 2026 16:05:34 +0000 Subject: [PATCH] feat(sync): push AI usage telemetry to a team OTLP backend (preview) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an opt-in `codeburn sync` command group: setup, push, status, logout, reset. Developers authenticate via standard OIDC browser login (Authorization Code + PKCE, provider-agnostic — Cognito, Auth0, Okta, Keycloak, etc.) and push per-call usage metadata to a team-operated OTLP/HTTP endpoint. Privacy is structural: the payload builder has no field that can carry prompts, responses, file contents, or bash commands. Only metadata leaves the machine (provider, model, tokens, cost, project, tool names, timestamps). There is deliberately no flag to include more. Setup derives everything from a single URL: GET {base}/.well-known/codeburn-export.json → issuer, client_id, traces_path, max_batch_size Modules: - discovery.ts discovery doc fetch/validation (version-gated) - auth.ts OIDC discovery, PKCE, localhost callback (:19876-78), token exchange/refresh/revoke - credentials.ts refresh token in OS keychain (macOS security / Linux libsecret / Windows DPAPI / 0600 file fallback). All subprocess calls use execFileSync arg arrays; tokens flow via stdin or env var — never shell-interpolated - config.ts ~/.config/codeburn/sync.json - ledger.ts client-side sent-ledger for exact dedup (no watermark races), 6-month prune - otlp.ts ParsedApiCall[] → ExportTraceServiceRequest JSON with deterministic IDs: span=SHA256(dedupKey)[:8], trace=SHA256(sessionId)[:16] — every retry idempotent - push.ts orchestration with typed outcomes; pushes run to completion — 429 Retry-After honored (≤120s/wait, 3 retries/batch) before deferring; partial_success is batch-atomic (nothing ledgered, whole batch retries); 50K safety valve, no routine cap - cli.ts command layer; --dry-run reports exact counts/cost without sending Preview feature: protocol may change between releases. AI-Origin: human --- package-lock.json | 48 ++++++ package.json | 1 + src/main.ts | 2 + src/sync/auth.ts | 297 ++++++++++++++++++++++++++++++++++++ src/sync/cli.ts | 328 ++++++++++++++++++++++++++++++++++++++++ src/sync/config.ts | 66 ++++++++ src/sync/credentials.ts | 211 ++++++++++++++++++++++++++ src/sync/discovery.ts | 93 ++++++++++++ src/sync/index.ts | 8 + src/sync/ledger.ts | 75 +++++++++ src/sync/otlp.ts | 139 +++++++++++++++++ src/sync/push.ts | 171 +++++++++++++++++++++ 12 files changed, 1439 insertions(+) create mode 100644 src/sync/auth.ts create mode 100644 src/sync/cli.ts create mode 100644 src/sync/config.ts create mode 100644 src/sync/credentials.ts create mode 100644 src/sync/discovery.ts create mode 100644 src/sync/index.ts create mode 100644 src/sync/ledger.ts create mode 100644 src/sync/otlp.ts create mode 100644 src/sync/push.ts diff --git a/package-lock.json b/package-lock.json index e649491..16b2e23 100644 --- a/package-lock.json +++ b/package-lock.json @@ -27,6 +27,7 @@ "@types/node": "^22.19.17", "@types/react": "^19.2.14", "@types/selfsigned": "^2.0.4", + "playwright": "^1.61.1", "tsup": "^8.4.0", "tsx": "^4.19.0", "typescript": "^5.8.0", @@ -2834,6 +2835,53 @@ "node": ">=16.0.0" } }, + "node_modules/playwright": { + "version": "1.61.1", + "resolved": "https://registry.npmjs.org/playwright/-/playwright-1.61.1.tgz", + "integrity": "sha512-DWnY5o3YbLWK4GovuAVwpqL+1VwGNdUGrRr++8j8PtQQzvAVZUIMjKQ90fY689sEJZJBbZVw1rXaOKSTitkzPQ==", + "dev": true, + "license": "Apache-2.0", + "dependencies": { + "playwright-core": "1.61.1" + }, + "bin": { + "playwright": "cli.js" + }, + "engines": { + "node": ">=18" + }, + "optionalDependencies": { + "fsevents": "2.3.2" + } + }, + "node_modules/playwright-core": { + "version": "1.61.1", + "resolved": "https://registry.npmjs.org/playwright-core/-/playwright-core-1.61.1.tgz", + "integrity": "sha512-h7Qlt6m4REp25qvIdvbDtVmD4LqVXfpRxhORv9L0jzETM05p4fuPJ3dKyuSXQxDSbXnmS79HAgi9589lGSpLkg==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "playwright-core": "cli.js" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/playwright/node_modules/fsevents": { + "version": "2.3.2", + "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.2.tgz", + "integrity": "sha512-xiqMQR4xAeHTuB9uWm+fFRcIOgKBMiOBP+eXiyT7jsgVCq1bkVygt00oASowB7EdtpOHaaPgKt812P9ab+DDKA==", + "dev": true, + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": "^8.16.0 || ^10.6.0 || >=11.0.0" + } + }, "node_modules/postcss": { "version": "8.5.15", "resolved": "https://registry.npmjs.org/postcss/-/postcss-8.5.15.tgz", diff --git a/package.json b/package.json index f3f59c0..9fda978 100644 --- a/package.json +++ b/package.json @@ -63,6 +63,7 @@ "@types/node": "^22.19.17", "@types/react": "^19.2.14", "@types/selfsigned": "^2.0.4", + "playwright": "^1.61.1", "tsup": "^8.4.0", "tsx": "^4.19.0", "typescript": "^5.8.0", diff --git a/src/main.ts b/src/main.ts index a4e435d..6b4abbc 100644 --- a/src/main.ts +++ b/src/main.ts @@ -26,6 +26,7 @@ import { formatDateRangeLabel, parseDateRangeFlags, parseDayFlag, parseDaysFlag, import { runOptimize } from './optimize.js' import { registerActCommands } from './act/cli.js' import { registerGuardCommands } from './guard/cli.js' +import { registerSyncCommands } from './sync/cli.js' import { runContextCommand } from './context-tree.js' import { renderCompare } from './compare.js' import { @@ -1597,5 +1598,6 @@ program registerActCommands(program) registerGuardCommands(program) +registerSyncCommands(program) program.parse() diff --git a/src/sync/auth.ts b/src/sync/auth.ts new file mode 100644 index 0000000..3ce7af2 --- /dev/null +++ b/src/sync/auth.ts @@ -0,0 +1,297 @@ +/** + * codeburn sync — OIDC authentication. + * + * Authorization Code + PKCE flow with localhost callback server. + */ + +import { createHash, randomBytes } from 'crypto' +import { createServer, type Server } from 'http' +import { URL } from 'url' + +export interface OidcConfig { + authorization_endpoint: string + token_endpoint: string + revocation_endpoint?: string + scopes_supported?: string[] +} + +export interface TokenResponse { + access_token: string + refresh_token?: string + expires_in?: number + token_type: string +} + +export class AuthError extends Error { + constructor(message: string) { + super(message) + this.name = 'AuthError' + } +} + +// --- OIDC Discovery --- + +export async function fetchOidcConfig(issuer: string): Promise { + const url = `${issuer.replace(/\/$/, '')}/.well-known/openid-configuration` + + let response: Response + try { + response = await fetch(url) + } catch (err) { + throw new AuthError(`Cannot reach OIDC discovery at ${url}: ${(err as Error).message}`) + } + + if (!response.ok) { + throw new AuthError(`OIDC discovery returned HTTP ${response.status}: ${url}`) + } + + let body: Record + try { + body = await response.json() as Record + } catch { + throw new AuthError(`OIDC discovery returned invalid JSON: ${url}`) + } + + const authorization_endpoint = body.authorization_endpoint + const token_endpoint = body.token_endpoint + if (typeof authorization_endpoint !== 'string' || typeof token_endpoint !== 'string') { + throw new AuthError('OIDC discovery missing authorization_endpoint or token_endpoint') + } + + return { + authorization_endpoint, + token_endpoint, + revocation_endpoint: typeof body.revocation_endpoint === 'string' ? body.revocation_endpoint : undefined, + scopes_supported: Array.isArray(body.scopes_supported) + ? (body.scopes_supported as unknown[]).filter((s): s is string => typeof s === 'string') + : undefined, + } +} + +// --- PKCE --- + +export interface PkceChallenge { + code_verifier: string + code_challenge: string + code_challenge_method: 'S256' +} + +export function generatePkce(): PkceChallenge { + // RFC 7636: 43-128 chars, unreserved characters + const code_verifier = randomBytes(32).toString('base64url') + const code_challenge = createHash('sha256').update(code_verifier).digest('base64url') + return { code_verifier, code_challenge, code_challenge_method: 'S256' } +} + +// --- Auth URL --- + +export const CALLBACK_PORTS = [19876, 19877, 19878] as const + +export interface AuthUrlParams { + authorization_endpoint: string + client_id: string + redirect_uri: string + scopes: string[] + state: string + pkce: PkceChallenge +} + +export function buildAuthUrl(params: AuthUrlParams): string { + const url = new URL(params.authorization_endpoint) + url.searchParams.set('response_type', 'code') + url.searchParams.set('client_id', params.client_id) + url.searchParams.set('redirect_uri', params.redirect_uri) + url.searchParams.set('scope', params.scopes.join(' ')) + url.searchParams.set('state', params.state) + url.searchParams.set('code_challenge', params.pkce.code_challenge) + url.searchParams.set('code_challenge_method', params.pkce.code_challenge_method) + return url.toString() +} + +export function resolveScopes(requestedScopes: string[], idpScopesSupported?: string[]): string[] { + const scopes = [...requestedScopes] + // Add offline_access only if the IdP advertises it + if (idpScopesSupported?.includes('offline_access') && !scopes.includes('offline_access')) { + scopes.push('offline_access') + } + return scopes +} + +// --- Callback Server --- + +export interface CallbackResult { + code: string + port: number +} + +export function startCallbackServer( + expectedState: string, + timeoutMs: number = 300_000, // 5 minutes +): { promise: Promise; port: number; server: Server } { + let resolvedPort = 0 + let server: Server + + const promise = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + server?.close() + reject(new AuthError('Login timed out after 5 minutes. Please try again.')) + }, timeoutMs) + + server = createServer((req, res) => { + const url = new URL(req.url ?? '/', `http://127.0.0.1`) + const code = url.searchParams.get('code') + const state = url.searchParams.get('state') + const error = url.searchParams.get('error') + + if (error) { + res.writeHead(400, { 'Content-Type': 'text/html' }) + res.end('

Login failed

You can close this tab.

') + clearTimeout(timer) + server.close() + reject(new AuthError(`IdP returned error: ${error}`)) + return + } + + if (state !== expectedState) { + res.writeHead(400, { 'Content-Type': 'text/plain' }) + res.end('Invalid state parameter') + return // don't close — might be a stale request + } + + if (!code) { + res.writeHead(400, { 'Content-Type': 'text/plain' }) + res.end('Missing authorization code') + return + } + + res.writeHead(200, { 'Content-Type': 'text/html' }) + res.end('

✓ Login successful

You can close this tab.

') + clearTimeout(timer) + server.close() + resolve({ code, port: resolvedPort }) + }) + + // Try ports in order + const tryListen = (ports: readonly number[], idx: number) => { + if (idx >= ports.length) { + clearTimeout(timer) + reject(new AuthError(`All callback ports (${ports.join(', ')}) are in use. Close other codeburn instances and retry.`)) + return + } + const port = ports[idx]! + server.once('error', (err: NodeJS.ErrnoException) => { + if (err.code === 'EADDRINUSE') { + tryListen(ports, idx + 1) + } else { + clearTimeout(timer) + reject(new AuthError(`Callback server error: ${err.message}`)) + } + }) + server.listen(port, '127.0.0.1', () => { + resolvedPort = port + }) + } + + tryListen(CALLBACK_PORTS, 0) + }) + + // Return immediately so caller can read the port + return { promise, port: resolvedPort, server: server! } +} + +// --- Token Exchange --- + +export async function exchangeCode( + tokenEndpoint: string, + code: string, + codeVerifier: string, + redirectUri: string, + clientId: string, +): Promise { + const body = new URLSearchParams({ + grant_type: 'authorization_code', + code, + code_verifier: codeVerifier, + redirect_uri: redirectUri, + client_id: clientId, + }) + + const response = await fetch(tokenEndpoint, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: body.toString(), + }) + + if (!response.ok) { + let detail = '' + try { detail = await response.text() } catch {} + throw new AuthError(`Token exchange failed (HTTP ${response.status}): ${detail}`) + } + + const data = await response.json() as Record + if (typeof data.access_token !== 'string') { + throw new AuthError('Token response missing access_token') + } + + return { + access_token: data.access_token, + refresh_token: typeof data.refresh_token === 'string' ? data.refresh_token : undefined, + expires_in: typeof data.expires_in === 'number' ? data.expires_in : undefined, + token_type: typeof data.token_type === 'string' ? data.token_type : 'Bearer', + } +} + +export async function refreshToken( + tokenEndpoint: string, + refreshTokenValue: string, + clientId: string, +): Promise { + const body = new URLSearchParams({ + grant_type: 'refresh_token', + refresh_token: refreshTokenValue, + client_id: clientId, + }) + + const response = await fetch(tokenEndpoint, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: body.toString(), + }) + + if (!response.ok) { + let detail = '' + try { detail = await response.text() } catch {} + if (response.status === 400 || response.status === 401) { + throw new AuthError('Sync auth expired. Run `codeburn sync setup` to re-authenticate.') + } + throw new AuthError(`Token refresh failed (HTTP ${response.status}): ${detail}`) + } + + const data = await response.json() as Record + if (typeof data.access_token !== 'string') { + throw new AuthError('Refresh response missing access_token') + } + + return { + access_token: data.access_token, + refresh_token: typeof data.refresh_token === 'string' ? data.refresh_token : undefined, + expires_in: typeof data.expires_in === 'number' ? data.expires_in : undefined, + token_type: typeof data.token_type === 'string' ? data.token_type : 'Bearer', + } +} + +export async function revokeToken( + revocationEndpoint: string, + token: string, + clientId: string, +): Promise { + try { + await fetch(revocationEndpoint, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams({ token, client_id: clientId }).toString(), + }) + } catch { + // Best-effort — don't fail logout if revocation endpoint is unavailable + } +} diff --git a/src/sync/cli.ts b/src/sync/cli.ts new file mode 100644 index 0000000..10bd941 --- /dev/null +++ b/src/sync/cli.ts @@ -0,0 +1,328 @@ +/** + * codeburn sync — CLI commands. + * + * Registers: sync setup | push | status | logout | reset + */ + +import type { Command } from 'commander' +import { randomBytes } from 'crypto' + +import { fetchDiscoveryDoc } from './discovery.js' +import { + fetchOidcConfig, + generatePkce, + buildAuthUrl, + resolveScopes, + startCallbackServer, + exchangeCode, + refreshToken, + revokeToken, + CALLBACK_PORTS, +} from './auth.js' +import { createCredentialStore } from './credentials.js' +import { readSyncConfig, writeSyncConfig, deleteSyncConfig, updateLastSync } from './config.js' +import { collectUnsentCalls, sendBatches, batchCalls, MAX_PER_PUSH } from './push.js' + +export function registerSyncCommands(program: Command): void { + const sync = program + .command('sync') + .description('Sync AI usage telemetry to a remote OTLP endpoint') + + // --- setup --- + sync + .command('setup ') + .description('Configure sync with a remote endpoint (one-time)') + .action(async (url: string) => { + const baseUrl = url.replace(/\/$/, '') + process.stderr.write(`Fetching discovery doc from ${baseUrl}...\n`) + + // 1. Fetch codeburn discovery doc + const discovery = await fetchDiscoveryDoc(baseUrl) + process.stderr.write(` Issuer: ${discovery.issuer}\n`) + process.stderr.write(` Client: ${discovery.client_id}\n`) + + // 2. Fetch OIDC configuration from the issuer + const oidc = await fetchOidcConfig(discovery.issuer) + process.stderr.write(` Auth endpoint: ${oidc.authorization_endpoint}\n`) + + // 3. Resolve scopes + const scopes = resolveScopes(discovery.scopes, oidc.scopes_supported) + + // 4. Generate PKCE + const pkce = generatePkce() + const state = randomBytes(16).toString('hex') + + // 5. Start callback server + const { promise: callbackPromise, server } = startCallbackServer(state) + + // Wait briefly for server to bind, then get the port + await new Promise(resolve => setTimeout(resolve, 100)) + const addr = server.address() + const port = typeof addr === 'object' && addr ? addr.port : CALLBACK_PORTS[0] + const redirectUri = `http://127.0.0.1:${port}/callback` + + // 6. Build auth URL and open browser + const authUrl = buildAuthUrl({ + authorization_endpoint: oidc.authorization_endpoint, + client_id: discovery.client_id, + redirect_uri: redirectUri, + scopes, + state, + pkce, + }) + + process.stderr.write(`\nOpening browser for login...\n`) + process.stderr.write(`If the browser doesn't open, visit:\n ${authUrl}\n\n`) + + // Open browser (best-effort, platform-specific). + // execFileSync with args array — authUrl comes from the remote discovery + // doc so it must never be shell-interpolated. Scheme is also validated. + try { + if (!/^https:\/\//.test(authUrl)) { + throw new Error('auth URL must be https') + } + const { execFileSync } = await import('child_process') + if (process.platform === 'darwin') { + execFileSync('open', [authUrl], { stdio: 'ignore' }) + } else if (process.platform === 'win32') { + // `start` is a cmd builtin; empty first arg is the window title + execFileSync('cmd', ['/c', 'start', '', authUrl], { stdio: 'ignore' }) + } else { + execFileSync('xdg-open', [authUrl], { stdio: 'ignore' }) + } + } catch { + // Browser open failed — user sees the URL above + } + + // 7. Wait for callback + process.stderr.write(`Waiting for login (5 min timeout)...\n`) + const callback = await callbackPromise + + // 8. Exchange code for tokens + const tokenRedirectUri = `http://127.0.0.1:${callback.port}/callback` + const tokens = await exchangeCode( + oidc.token_endpoint, + callback.code, + pkce.code_verifier, + tokenRedirectUri, + discovery.client_id, + ) + + if (!tokens.refresh_token) { + process.stderr.write(`Warning: IdP did not return a refresh token. You may need to re-authenticate frequently.\n`) + } + + // 9. Store credentials + const store = createCredentialStore() + if (tokens.refresh_token) { + store.store(tokens.refresh_token) + } + + // 10. Write config + writeSyncConfig({ + baseUrl, + clientId: discovery.client_id, + tracesPath: discovery.traces_path, + issuer: discovery.issuer, + }) + + process.stderr.write(`\n✓ Sync configured successfully.\n`) + process.stderr.write(` Endpoint: ${baseUrl}\n`) + process.stderr.write(` Token stored in: ${store.method()}\n`) + process.stderr.write(`\nRun \`codeburn sync push\` to send telemetry data.\n`) + }) + + // --- status --- + sync + .command('status') + .description('Show sync configuration and auth status') + .action(async () => { + const config = readSyncConfig() + if (!config) { + process.stderr.write('Sync not configured. Run `codeburn sync setup ` first.\n') + process.exit(1) + } + + const store = createCredentialStore() + const token = store.retrieve() + + process.stdout.write(`Endpoint: ${config.baseUrl}\n`) + process.stdout.write(`Traces path: ${config.tracesPath}\n`) + process.stdout.write(`Issuer: ${config.issuer}\n`) + process.stdout.write(`Auth: ${token ? 'configured' : 'missing (run sync setup)'}\n`) + process.stdout.write(`Token storage: ${store.method()}\n`) + process.stdout.write(`Last sync: ${config.lastSync ?? 'never'}\n`) + }) + + // --- logout --- + sync + .command('logout') + .description('Remove stored credentials and revoke token') + .action(async () => { + const config = readSyncConfig() + const store = createCredentialStore() + const token = store.retrieve() + + // Revoke if we have a token and know the revocation endpoint + if (token && config) { + try { + const oidc = await fetchOidcConfig(config.issuer) + if (oidc.revocation_endpoint) { + await revokeToken(oidc.revocation_endpoint, token, config.clientId) + process.stderr.write('Token revoked at IdP.\n') + } + } catch { + // Best-effort revocation + } + } + + store.delete() + deleteSyncConfig() + process.stderr.write('Sync credentials and config removed.\n') + }) + + // --- reset --- + sync + .command('reset') + .description('Clear the sent-ledger (next push re-sends all calls in window)') + .option('--confirm', 'Required to confirm reset') + .action(async (opts: { confirm?: boolean }) => { + if (!opts.confirm) { + process.stderr.write('This will clear the sent-ledger, causing the next push to re-send all data.\n') + process.stderr.write('Run with --confirm to proceed.\n') + process.exit(1) + } + + const { join } = await import('path') + const { homedir } = await import('os') + const { unlinkSync, existsSync } = await import('fs') + + const ledgerPath = join(homedir(), '.cache', 'codeburn', 'sync-ledger.json') + if (existsSync(ledgerPath)) { + unlinkSync(ledgerPath) + process.stderr.write('Ledger cleared. Next push will re-send all calls in window.\n') + } else { + process.stderr.write('No ledger file found (nothing to reset).\n') + } + }) + + // --- push (placeholder for Step 2) --- + sync + .command('push') + .description('Push unsent telemetry data to the configured endpoint') + .option('--since ', 'Time window: today, 7d, 30d, month, all (max 6 months)', '7d') + .option('--dry-run', 'Show what would be sent without sending') + .action(async (opts: { since: string; dryRun?: boolean }) => { + const config = readSyncConfig() + if (!config) { + process.stderr.write('Sync not configured. Run `codeburn sync setup ` first.\n') + process.exit(1) + } + + const store = createCredentialStore() + const rt = store.retrieve() + if (!rt) { + process.stderr.write('No auth token found. Run `codeburn sync setup` to authenticate.\n') + process.exit(1) + } + + // Refresh token + try { + const oidc = await fetchOidcConfig(config.issuer) + const tokens = await refreshToken(oidc.token_endpoint, rt, config.clientId) + + // Store rotated token if present + if (tokens.refresh_token && tokens.refresh_token !== rt) { + store.store(tokens.refresh_token) + } + + if (opts.dryRun) { + process.stderr.write(`[dry-run] Auth: valid (Bearer token obtained)\n`) + } + + // Collect data + const { parseAllSessions } = await import('../parser.js') + const { getDateRange } = await import('../cli-date.js') + + // Map --since to a parser period. Strict: unknown values are an error. + const sinceToPeriod: Record = { + 'today': 'today', + '7d': 'week', 'week': 'week', + '30d': '30days', '30days': '30days', + 'month': 'month', + 'all': 'all', // up to 6 months (parser retention limit) + } + const period = sinceToPeriod[opts.since] + if (!period) { + process.stderr.write(`Unknown --since value "${opts.since}". Valid: today, 7d, 30d, month, all.\n`) + process.exit(1) + } + const { range } = getDateRange(period) + const projects = await parseAllSessions(range) + + // Flatten + filter against sent-ledger + const { allCalls, unsent } = collectUnsentCalls(projects) + + if (opts.dryRun) { + const toPushCount = Math.min(unsent.length, MAX_PER_PUSH) + const cost = unsent.slice(0, MAX_PER_PUSH).reduce((s, c) => s + c.call.costUSD, 0) + process.stderr.write(`[dry-run] Window: ${opts.since} — ${allCalls.length} calls total, ${allCalls.length - unsent.length} already synced\n`) + process.stderr.write(`[dry-run] Would push ${toPushCount} calls ($${cost.toFixed(2)}) to ${config.baseUrl}${config.tracesPath}\n`) + if (unsent.length > MAX_PER_PUSH) { + process.stderr.write(`[dry-run] ${unsent.length - MAX_PER_PUSH} more calls exceed the ${MAX_PER_PUSH} safety limit — a second push would be needed\n`) + } + return + } + + if (unsent.length === 0) { + process.stderr.write(`Nothing to push (${allCalls.length} calls already synced).\n`) + updateLastSync() + return + } + + // Safety valve (not a routine cap — pushes run to completion) + const toPush = unsent.slice(0, MAX_PER_PUSH) + if (unsent.length > MAX_PER_PUSH) { + process.stderr.write(`${unsent.length} unsent calls exceed the ${MAX_PER_PUSH} safety limit. Pushing first ${MAX_PER_PUSH}; run again to continue.\n`) + } + + // Batch and send (loops until done; waits out 429 rate limits) + const discoveryDoc = await fetchDiscoveryDoc(config.baseUrl) + const batches = batchCalls(toPush, discoveryDoc.max_batch_size) + const endpoint = `${config.baseUrl}${config.tracesPath}` + + const result = await sendBatches({ + endpoint, + accessToken: tokens.access_token, + batches, + log: msg => process.stderr.write(`${msg}\n`), + }) + + if (result.outcome === 'auth-rejected') { + process.stderr.write('Auth rejected by server. Run `codeburn sync setup` to re-authenticate.\n') + process.exit(1) + } + if (result.outcome === 'rate-limited') { + process.stderr.write(`Rate limited — gave up after repeated retries. Remaining calls will be sent on the next push.\n`) + } + if (result.outcome === 'server-error') { + process.stderr.write(`Server error (HTTP ${result.httpStatus}). Remaining calls will be sent on the next push.\n`) + } + + // Update lastSync + updateLastSync() + + // Summary + process.stderr.write(`\nSynced ${result.totalSent} calls ($${result.totalCostSent.toFixed(2)}) to ${config.baseUrl}\n`) + if (result.totalRejected > 0) { + process.stderr.write(` ${result.totalRejected} spans rejected (will retry on next push)\n`) + } + if (unsent.length > MAX_PER_PUSH) { + process.stderr.write(` ${unsent.length - MAX_PER_PUSH} calls remaining (safety limit). Run \`codeburn sync push\` again.\n`) + } + } catch (err) { + process.stderr.write(`${(err as Error).message}\n`) + process.exit(1) + } + }) +} diff --git a/src/sync/config.ts b/src/sync/config.ts new file mode 100644 index 0000000..6ddfcec --- /dev/null +++ b/src/sync/config.ts @@ -0,0 +1,66 @@ +/** + * codeburn sync — config file management. + * + * Stores non-secret sync configuration at ~/.config/codeburn/sync.json + */ + +import { existsSync, mkdirSync, readFileSync, writeFileSync, unlinkSync } from 'fs' +import { join } from 'path' +import { homedir } from 'os' + +export interface SyncConfig { + baseUrl: string + clientId: string + tracesPath: string + issuer: string + lastSync?: string +} + +function configDir(): string { + return join(homedir(), '.config', 'codeburn') +} + +function configPath(): string { + return join(configDir(), 'sync.json') +} + +export function readSyncConfig(): SyncConfig | null { + const path = configPath() + if (!existsSync(path)) return null + + try { + const raw = readFileSync(path, 'utf-8') + const data = JSON.parse(raw) as Record + + if (typeof data.baseUrl !== 'string' || typeof data.clientId !== 'string') { + return null + } + + return { + baseUrl: data.baseUrl, + clientId: data.clientId, + tracesPath: typeof data.tracesPath === 'string' ? data.tracesPath : '/v1/traces', + issuer: typeof data.issuer === 'string' ? data.issuer : '', + lastSync: typeof data.lastSync === 'string' ? data.lastSync : undefined, + } + } catch { + return null + } +} + +export function writeSyncConfig(config: SyncConfig): void { + const dir = configDir() + mkdirSync(dir, { recursive: true }) + writeFileSync(configPath(), JSON.stringify(config, null, 2) + '\n') +} + +export function updateLastSync(): void { + const config = readSyncConfig() + if (!config) return + config.lastSync = new Date().toISOString() + writeSyncConfig(config) +} + +export function deleteSyncConfig(): void { + try { unlinkSync(configPath()) } catch { /* may not exist */ } +} diff --git a/src/sync/credentials.ts b/src/sync/credentials.ts new file mode 100644 index 0000000..a560581 --- /dev/null +++ b/src/sync/credentials.ts @@ -0,0 +1,211 @@ +/** + * codeburn sync — OS credential storage. + * + * Stores refresh tokens in the OS keychain. + * Falls back to a 0600 file when no keychain is available. + */ + +import { execFileSync } from 'child_process' +import { existsSync, mkdirSync, readFileSync, writeFileSync, unlinkSync, chmodSync } from 'fs' +import { join } from 'path' +import { homedir } from 'os' + +const SERVICE_NAME = 'codeburn-sync' +const ACCOUNT_NAME = 'refresh-token' + +export type StorageMethod = 'keychain' | 'secret-tool' | 'dpapi' | 'file' + +export interface CredentialStore { + store(token: string): void + retrieve(): string | null + delete(): void + method(): StorageMethod +} + +// --- macOS Keychain --- + +class KeychainStore implements CredentialStore { + store(token: string): void { + // Delete existing first (add-generic-password fails if entry exists) + try { + execFileSync('security', ['delete-generic-password', '-s', SERVICE_NAME, '-a', ACCOUNT_NAME], { stdio: 'pipe' }) + } catch { /* may not exist */ } + + // Token passed as arg to execFileSync (no shell interpolation). + // Note: still briefly visible in process args on macOS; `security` has no + // stdin mode for -w with non-interactive use, so this is the best available. + execFileSync('security', ['add-generic-password', '-s', SERVICE_NAME, '-a', ACCOUNT_NAME, '-w', token], { stdio: 'pipe' }) + } + + retrieve(): string | null { + try { + const result = execFileSync( + 'security', + ['find-generic-password', '-s', SERVICE_NAME, '-a', ACCOUNT_NAME, '-w'], + { encoding: 'utf-8', stdio: ['pipe', 'pipe', 'pipe'] } + ) + return result.trim() || null + } catch { + return null + } + } + + delete(): void { + try { + execFileSync('security', ['delete-generic-password', '-s', SERVICE_NAME, '-a', ACCOUNT_NAME], { stdio: 'pipe' }) + } catch { /* may not exist */ } + } + + method(): StorageMethod { return 'keychain' } +} + +// --- Linux libsecret --- + +class SecretToolStore implements CredentialStore { + store(token: string): void { + // Token passed via stdin — never appears in argv or shell string + execFileSync( + 'secret-tool', + ['store', `--label=${SERVICE_NAME}`, 'service', SERVICE_NAME, 'account', ACCOUNT_NAME], + { input: token, stdio: ['pipe', 'pipe', 'pipe'] } + ) + } + + retrieve(): string | null { + try { + const result = execFileSync( + 'secret-tool', + ['lookup', 'service', SERVICE_NAME, 'account', ACCOUNT_NAME], + { encoding: 'utf-8', stdio: ['pipe', 'pipe', 'pipe'] } + ) + return result.trim() || null + } catch { + return null + } + } + + delete(): void { + try { + execFileSync('secret-tool', ['clear', 'service', SERVICE_NAME, 'account', ACCOUNT_NAME], { stdio: 'pipe' }) + } catch { /* may not exist */ } + } + + method(): StorageMethod { return 'secret-tool' } +} + +// --- Windows DPAPI --- + +class DpapiStore implements CredentialStore { + private filePath: string + + constructor() { + this.filePath = join(homedir(), '.config', 'codeburn', '.sync-token-dpapi') + } + + store(token: string): void { + // Token passed via environment variable — never in argv or command string + const ps = `$s = ConvertTo-SecureString $env:CODEBURN_SYNC_TOKEN -AsPlainText -Force; ConvertFrom-SecureString $s` + const encrypted = execFileSync('powershell', ['-NoProfile', '-Command', ps], { + encoding: 'utf-8', + stdio: ['pipe', 'pipe', 'pipe'], + env: { ...process.env, CODEBURN_SYNC_TOKEN: token }, + }).trim() + + const dir = join(homedir(), '.config', 'codeburn') + mkdirSync(dir, { recursive: true }) + writeFileSync(this.filePath, encrypted, { mode: 0o600 }) + } + + retrieve(): string | null { + if (!existsSync(this.filePath)) return null + try { + const encrypted = readFileSync(this.filePath, 'utf-8').trim() + const ps = `$s = ConvertTo-SecureString $env:CODEBURN_SYNC_BLOB; [Runtime.InteropServices.Marshal]::PtrToStringAuto([Runtime.InteropServices.Marshal]::SecureStringToBSTR($s))` + const result = execFileSync('powershell', ['-NoProfile', '-Command', ps], { + encoding: 'utf-8', + stdio: ['pipe', 'pipe', 'pipe'], + env: { ...process.env, CODEBURN_SYNC_BLOB: encrypted }, + }) + return result.trim() || null + } catch { + return null + } + } + + delete(): void { + try { unlinkSync(this.filePath) } catch { /* may not exist */ } + } + + method(): StorageMethod { return 'dpapi' } +} + +// --- File Fallback --- + +class FileStore implements CredentialStore { + private filePath: string + + constructor() { + this.filePath = join(homedir(), '.config', 'codeburn', '.sync-token') + } + + store(token: string): void { + const dir = join(homedir(), '.config', 'codeburn') + mkdirSync(dir, { recursive: true }) + writeFileSync(this.filePath, token, { mode: 0o600 }) + // Ensure permissions (writeFile mode doesn't always work on existing files) + try { chmodSync(this.filePath, 0o600) } catch {} + } + + retrieve(): string | null { + if (!existsSync(this.filePath)) return null + try { + return readFileSync(this.filePath, 'utf-8').trim() || null + } catch { + return null + } + } + + delete(): void { + try { unlinkSync(this.filePath) } catch { /* may not exist */ } + } + + method(): StorageMethod { return 'file' } +} + +// --- Factory --- + +function isCommandAvailable(cmd: string): boolean { + try { + const probe = process.platform === 'win32' ? 'where' : 'which' + execFileSync(probe, [cmd], { stdio: 'pipe' }) + return true + } catch { + return false + } +} + +export function createCredentialStore(): CredentialStore { + if (process.platform === 'darwin') { + return new KeychainStore() + } + + if (process.platform === 'win32') { + return new DpapiStore() + } + + // Linux: try secret-tool, fall back to file + if (isCommandAvailable('secret-tool')) { + // Also verify the keyring daemon is running + try { + execFileSync('secret-tool', ['lookup', 'service', '__codeburn_probe__', 'account', '__probe__'], { stdio: 'pipe' }) + return new SecretToolStore() + } catch (err) { + // Exit code 1 = not found (keyring works). Other errors = keyring not running. + if ((err as { status?: number }).status === 1) { + return new SecretToolStore() + } + } + } + + return new FileStore() +} diff --git a/src/sync/discovery.ts b/src/sync/discovery.ts new file mode 100644 index 0000000..60f5ffe --- /dev/null +++ b/src/sync/discovery.ts @@ -0,0 +1,93 @@ +/** + * codeburn sync — discovery document parser. + * + * Fetches and validates {baseUrl}/.well-known/codeburn-export.json + */ + +export interface CodeburnDiscoveryDoc { + version: number + issuer: string + client_id: string + scopes: string[] + traces_path: string + max_batch_size: number +} + +export class DiscoveryError extends Error { + constructor(message: string) { + super(message) + this.name = 'DiscoveryError' + } +} + +const SUPPORTED_VERSION = 1 + +export function parseDiscoveryDoc(raw: unknown): CodeburnDiscoveryDoc { + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) { + throw new DiscoveryError('Discovery doc must be a JSON object') + } + + const doc = raw as Record + + // Version check + const version = typeof doc.version === 'number' ? doc.version : 1 + if (version > SUPPORTED_VERSION) { + throw new DiscoveryError( + `This endpoint requires codeburn sync v${version}. Please update codeburn.` + ) + } + + // Required fields + const issuer = doc.issuer + if (typeof issuer !== 'string' || !issuer) { + throw new DiscoveryError('Discovery doc missing required field: issuer') + } + + const client_id = doc.client_id + if (typeof client_id !== 'string' || !client_id) { + throw new DiscoveryError('Discovery doc missing required field: client_id') + } + + // Optional with defaults + const scopes = Array.isArray(doc.scopes) + ? doc.scopes.filter((s): s is string => typeof s === 'string') + : ['openid'] + + const traces_path = typeof doc.traces_path === 'string' ? doc.traces_path : '/v1/traces' + + const max_batch_size = typeof doc.max_batch_size === 'number' && doc.max_batch_size > 0 + ? doc.max_batch_size + : 1000 + + return { version, issuer, client_id, scopes, traces_path, max_batch_size } +} + +export async function fetchDiscoveryDoc(baseUrl: string): Promise { + const url = `${baseUrl.replace(/\/$/, '')}/.well-known/codeburn-export.json` + + let response: Response + try { + response = await fetch(url) + } catch (err) { + throw new DiscoveryError(`Cannot reach ${url}: ${(err as Error).message}`) + } + + if (response.status === 404) { + throw new DiscoveryError( + `Server does not support codeburn sync.\n${url} returned 404.` + ) + } + + if (!response.ok) { + throw new DiscoveryError(`${url} returned HTTP ${response.status}`) + } + + let body: unknown + try { + body = await response.json() + } catch { + throw new DiscoveryError(`${url} returned invalid JSON`) + } + + return parseDiscoveryDoc(body) +} diff --git a/src/sync/index.ts b/src/sync/index.ts new file mode 100644 index 0000000..8aa19e4 --- /dev/null +++ b/src/sync/index.ts @@ -0,0 +1,8 @@ +export { registerSyncCommands } from './cli.js' +export { fetchDiscoveryDoc, parseDiscoveryDoc, type CodeburnDiscoveryDoc } from './discovery.js' +export { fetchOidcConfig, generatePkce, buildAuthUrl, resolveScopes, exchangeCode, refreshToken } from './auth.js' +export { createCredentialStore, type CredentialStore, type StorageMethod } from './credentials.js' +export { readSyncConfig, writeSyncConfig, deleteSyncConfig, type SyncConfig } from './config.js' +export { readLedger, appendToLedger, ledgerKeySet, clearLedger } from './ledger.js' +export { buildOtlpPayload, batchCalls, deriveSpanId, deriveTraceId, getDeviceId } from './otlp.js' +export { collectUnsentCalls, sendBatches, parseRetryAfterMs, MAX_PER_PUSH, type PushResult, type PushOutcome } from './push.js' diff --git a/src/sync/ledger.ts b/src/sync/ledger.ts new file mode 100644 index 0000000..eb4d7a3 --- /dev/null +++ b/src/sync/ledger.ts @@ -0,0 +1,75 @@ +/** + * codeburn sync — sent-ledger. + * + * Client-side deduplication: tracks which calls have been successfully pushed. + * Push logic: window minus ledger = what to send. + */ + +import { existsSync, mkdirSync, readFileSync, writeFileSync, unlinkSync } from 'fs' +import { join } from 'path' +import { homedir } from 'os' + +export interface LedgerEntry { + key: string // deduplicationKey + ts: string // call timestamp (for pruning) +} + +const SIX_MONTHS_MS = 180 * 24 * 60 * 60 * 1000 + +function ledgerPath(): string { + return join(homedir(), '.cache', 'codeburn', 'sync-ledger.json') +} + +export function readLedger(): LedgerEntry[] { + const path = ledgerPath() + if (!existsSync(path)) return [] + try { + const raw = readFileSync(path, 'utf-8') + const entries = JSON.parse(raw) as unknown + if (!Array.isArray(entries)) return [] + return entries.filter( + (e): e is LedgerEntry => typeof e === 'object' && e !== null && typeof e.key === 'string' + ) + } catch { + return [] + } +} + +export function writeLedger(entries: LedgerEntry[]): void { + const dir = join(homedir(), '.cache', 'codeburn') + mkdirSync(dir, { recursive: true }) + writeFileSync(ledgerPath(), JSON.stringify(entries)) +} + +/** Append new entries after a successful push. Also prunes entries older than 6 months. */ +export function appendToLedger(newEntries: LedgerEntry[]): void { + const existing = readLedger() + const cutoff = new Date(Date.now() - SIX_MONTHS_MS).toISOString() + + // Prune old + dedupe + const keySet = new Set(existing.map(e => e.key)) + const pruned = existing.filter(e => !e.ts || e.ts > cutoff) + + for (const entry of newEntries) { + if (!keySet.has(entry.key)) { + pruned.push(entry) + keySet.add(entry.key) + } + } + + writeLedger(pruned) +} + +/** Get the set of already-sent deduplication keys for fast lookup. */ +export function ledgerKeySet(): Set { + return new Set(readLedger().map(e => e.key)) +} + +/** Clear the ledger (for sync reset). Returns the number of entries removed. */ +export function clearLedger(): number { + const path = ledgerPath() + if (!existsSync(path)) return 0 + const count = readLedger().length + unlinkSync(path) + return count +} diff --git a/src/sync/otlp.ts b/src/sync/otlp.ts new file mode 100644 index 0000000..b470431 --- /dev/null +++ b/src/sync/otlp.ts @@ -0,0 +1,139 @@ +/** + * codeburn sync — OTLP payload builder. + * + * Converts ParsedApiCall[] into an ExportTraceServiceRequest (OTLP/HTTP JSON). + * Span and trace IDs are derived deterministically from deduplicationKey/sessionId. + */ + +import { createHash } from 'crypto' +import { hostname, userInfo } from 'os' +import type { ParsedApiCall } from '../types.js' + +export interface OtlpSpan { + traceId: string + spanId: string + name: string + startTimeUnixNano: string + endTimeUnixNano: string + attributes: OtlpAttribute[] +} + +export interface OtlpAttribute { + key: string + value: OtlpValue +} + +export type OtlpValue = + | { stringValue: string } + | { intValue: string } + | { doubleValue: number } + | { boolValue: boolean } + | { arrayValue: { values: OtlpValue[] } } + +export interface OtlpPayload { + resourceSpans: Array<{ + resource: { attributes: OtlpAttribute[] } + scopeSpans: Array<{ + spans: OtlpSpan[] + }> + }> +} + +// --- Device ID (pseudonymous, stable) --- + +let cachedDeviceId: string | null = null + +export function getDeviceId(): string { + if (cachedDeviceId) return cachedDeviceId + const raw = `${hostname()}:${userInfo().username}` + cachedDeviceId = createHash('sha256').update(raw).digest('hex').slice(0, 16) + return cachedDeviceId +} + +// --- Span/Trace ID derivation (deterministic) --- + +export function deriveSpanId(deduplicationKey: string): string { + return createHash('sha256').update(deduplicationKey).digest('hex').slice(0, 16) +} + +export function deriveTraceId(sessionId: string): string { + return createHash('sha256').update(sessionId).digest('hex').slice(0, 32) +} + +// --- Timestamp conversion --- + +function toUnixNano(isoTimestamp: string): string { + const ms = new Date(isoTimestamp).getTime() + if (isNaN(ms)) return '0' + return (BigInt(ms) * 1_000_000n).toString() +} + +// --- Payload construction --- + +export interface CallWithSession { + call: ParsedApiCall + sessionId: string + project: string +} + +export function buildOtlpPayload(calls: CallWithSession[]): OtlpPayload { + const deviceId = getDeviceId() + + const spans: OtlpSpan[] = calls.map(({ call, sessionId, project }) => { + const startNano = toUnixNano(call.timestamp) + // End time = start + 1ms (we don't have real duration, but OTLP requires both) + const endNano = (BigInt(startNano) + 1_000_000n).toString() + + const attributes: OtlpAttribute[] = [ + { key: 'ai.provider', value: { stringValue: call.provider } }, + { key: 'ai.model', value: { stringValue: call.model } }, + { key: 'ai.input_tokens', value: { intValue: String(call.usage.inputTokens) } }, + { key: 'ai.output_tokens', value: { intValue: String(call.usage.outputTokens) } }, + { key: 'ai.cost_usd', value: { doubleValue: call.costUSD } }, + { key: 'ai.project', value: { stringValue: project } }, + { key: 'ai.speed', value: { stringValue: call.speed } }, + ] + + if (call.tools.length > 0) { + attributes.push({ + key: 'ai.tools', + value: { arrayValue: { values: call.tools.map(t => ({ stringValue: t })) } }, + }) + } + + // cost_estimated = true when provider reports char-based estimates + const isEstimated = call.provider === 'kiro' || call.usage.inputTokens === 0 + attributes.push({ key: 'ai.cost_estimated', value: { boolValue: isEstimated } }) + + return { + traceId: deriveTraceId(sessionId), + spanId: deriveSpanId(call.deduplicationKey), + name: `${call.provider}/${call.model}`, + startTimeUnixNano: startNano, + endTimeUnixNano: endNano, + attributes, + } + }) + + return { + resourceSpans: [{ + resource: { + attributes: [ + { key: 'codeburn.device_id', value: { stringValue: deviceId } }, + ], + }, + scopeSpans: [{ + spans, + }], + }], + } +} + +/** Split calls into batches of maxBatchSize. */ +export function batchCalls(calls: CallWithSession[], maxBatchSize: number): CallWithSession[][] { + const batches: CallWithSession[][] = [] + for (let i = 0; i < calls.length; i += maxBatchSize) { + batches.push(calls.slice(i, i + maxBatchSize)) + } + return batches +} diff --git a/src/sync/push.ts b/src/sync/push.ts new file mode 100644 index 0000000..ffa1102 --- /dev/null +++ b/src/sync/push.ts @@ -0,0 +1,171 @@ +/** + * codeburn sync — push orchestration. + * + * Extracted from the CLI action so the flatten/filter/batch/send/ledger + * pipeline is unit-testable without a full CLI invocation. + */ + +import type { ProjectSummary } from '../types.js' +import { ledgerKeySet, appendToLedger, type LedgerEntry } from './ledger.js' +import { buildOtlpPayload, batchCalls, type CallWithSession } from './otlp.js' + +/** + * Safety valve, not a routine cap — pushes now loop until all batches are + * sent (429s are waited out). This only bounds a single push in pathological + * cases (e.g. corrupted ledger causing a full re-send of years of data). + */ +export const MAX_PER_PUSH = 50_000 + +/** Flatten parsed projects into individual calls and filter out already-sent ones. */ +export function collectUnsentCalls(projects: ProjectSummary[]): { + allCalls: CallWithSession[] + unsent: CallWithSession[] +} { + const allCalls: CallWithSession[] = [] + for (const project of projects) { + for (const session of project.sessions) { + for (const turn of session.turns) { + for (const call of turn.assistantCalls) { + allCalls.push({ + call, + sessionId: session.sessionId, + project: project.project, + }) + } + } + } + } + + const sent = ledgerKeySet() + const unsent = allCalls.filter(c => !sent.has(c.call.deduplicationKey)) + return { allCalls, unsent } +} + +export type PushOutcome = 'complete' | 'auth-rejected' | 'rate-limited' | 'server-error' + +export interface PushResult { + outcome: PushOutcome + totalSent: number + totalRejected: number + totalCostSent: number + retryAfter?: string + httpStatus?: number + /** Total milliseconds spent waiting on 429 Retry-After */ + totalWaitMs?: number +} + +export interface SendBatchesOptions { + endpoint: string + accessToken: string + batches: CallWithSession[][] + log?: (msg: string) => void + /** Injectable sleep for tests. Defaults to real setTimeout. */ + sleep?: (ms: number) => Promise + /** Max wait per 429 (caps Retry-After). Default 120s. */ + maxWaitMs?: number + /** Consecutive 429 retries per batch before giving up. Default 3. */ + max429Retries?: number +} + +/** Parse Retry-After header: delta-seconds or HTTP-date. Returns ms, or null. */ +export function parseRetryAfterMs(value: string | null): number | null { + if (!value) return null + if (/^-?\d+$/.test(value.trim())) { + const seconds = Number(value) + return seconds >= 0 ? seconds * 1000 : null + } + const date = Date.parse(value) + if (!isNaN(date)) return Math.max(0, date - Date.now()) + return null +} + +/** + * Send batches sequentially until all are sent. Ledgers each fully-accepted + * batch. Partially-rejected batches are NOT ledgered (OTLP doesn't identify + * which spans were rejected; deterministic span IDs make full-batch retry safe). + * + * 429 responses are honored: waits Retry-After (capped at maxWaitMs, default + * backoff 5s when absent) and retries the same batch, up to max429Retries + * consecutive times before giving up. Stops on 401/5xx — unsent batches + * retry on the next push. + */ +export async function sendBatches(opts: SendBatchesOptions): Promise { + const log = opts.log ?? (() => {}) + const sleep = opts.sleep ?? ((ms: number) => new Promise(r => setTimeout(r, ms))) + const maxWaitMs = opts.maxWaitMs ?? 120_000 + const max429Retries = opts.max429Retries ?? 3 + + let totalSent = 0 + let totalRejected = 0 + let totalCostSent = 0 + let totalWaitMs = 0 + + for (const batch of opts.batches) { + let attempts429 = 0 + + // Retry loop for the current batch (429 only) + for (;;) { + const payload = buildOtlpPayload(batch) + + const response = await fetch(opts.endpoint, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${opts.accessToken}`, + }, + body: JSON.stringify(payload), + }) + + if (response.status === 401) { + return { outcome: 'auth-rejected', totalSent, totalRejected, totalCostSent, totalWaitMs, httpStatus: 401 } + } + + if (response.status === 429) { + const retryAfterHeader = response.headers.get('Retry-After') + attempts429++ + if (attempts429 > max429Retries) { + return { + outcome: 'rate-limited', totalSent, totalRejected, totalCostSent, totalWaitMs, + retryAfter: retryAfterHeader ?? undefined, httpStatus: 429, + } + } + const waitMs = Math.min(parseRetryAfterMs(retryAfterHeader) ?? 5000, maxWaitMs) + log(` Rate limited — waiting ${Math.round(waitMs / 1000)}s before retrying (attempt ${attempts429}/${max429Retries})`) + totalWaitMs += waitMs + await sleep(waitMs) + continue + } + + if (!response.ok) { + return { outcome: 'server-error', totalSent, totalRejected, totalCostSent, totalWaitMs, httpStatus: response.status } + } + + // Check for partial success + let rejected = 0 + try { + const body = await response.json() as { partialSuccess?: { rejectedSpans?: number } } + rejected = body?.partialSuccess?.rejectedSpans ?? 0 + } catch { /* empty response = full success */ } + + if (rejected > 0) { + // OTLP partial_success doesn't identify WHICH spans were rejected. + // Ledger nothing — the whole batch retries on the next push. + totalRejected += rejected + log(` Batch: ${rejected}/${batch.length} spans rejected — whole batch will retry on next push`) + } else { + const entries: LedgerEntry[] = batch.map(c => ({ + key: c.call.deduplicationKey, + ts: c.call.timestamp, + })) + appendToLedger(entries) + totalSent += batch.length + totalCostSent += batch.reduce((s, c) => s + c.call.costUSD, 0) + } + break // batch done (success or partial) — move to next batch + } + } + + return { outcome: 'complete', totalSent, totalRejected, totalCostSent, totalWaitMs } +} + +export { batchCalls }