diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ed38822..d846966b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ ### Added (CLI) - **Codex throughput tracking**: per-model Tok/s in the dashboard and report, active time excludes tool wait. (#805, thanks @ihearttokyo) +- `codeburn sync push --attribution` (opt-in): sends git attribution spans — the session→commit correlation from `codeburn yield` (`codeburn.session.attribution` and `codeburn.commit` span types with normalized repo remote, commit SHAs, merged/reverted state, and PR links). Nothing new is sent without the flag; local-only repos and Windows filesystem paths are never emitted as repo identities, and sessions whose project path no longer resolves never inherit the push-time working directory's repo. See docs/sync/README.md "Git attribution". ### Fixed (CLI) - **`--project` / `--exclude` now apply to the headline totals, not just the detail panels.** The durable headline unions the carry-forward daily cache with today's live parse, and the cached days were sliced to the requested provider but never to the requested project — so the Overview panel counted excluded projects while By Project / By Activity / By Model (built from the name-filtered parse) left them out, and the two could not be reconciled. Cost, calls, sessions and savings are now sliced out of the per-project day stats the cache has carried since v15. Tokens, models and categories have no per-project split in the cache, so under a project filter they come from the (project-filtered) live parse instead; cached days — or provider slices — carried from before v15 have no project split at all, so they cannot be attributed to a filtered project, and the terminal overview now states how much was set aside rather than folding it into the total. (#864) diff --git a/docs/sync/README.md b/docs/sync/README.md index 5f1343ed..64540b83 100644 --- a/docs/sync/README.md +++ b/docs/sync/README.md @@ -48,6 +48,9 @@ codeburn sync push --since 30d # Preview what would be sent codeburn sync push --dry-run + +# Also push git attribution (opt-in — see "Git attribution" below) +codeburn sync push --attribution ``` ### `codeburn sync status` @@ -95,6 +98,36 @@ Each AI interaction becomes one OTLP span with these attributes: A pseudonymous `device_id` distinguishes your machines without revealing hostnames. +### Git attribution (opt-in: `--attribution`) + +`codeburn sync push --attribution` additionally sends the session→commit correlation that `codeburn yield` computes locally, so the backend can join AI usage to git activity without git hooks. Two extra span types are emitted: + +**`codeburn.session.attribution`** — one per session with joinable evidence: + +| Field | Example | Description | +|---|---|---| +| `ai.session_id` | `abc123…` | Session (shares the usage spans' traceId) | +| `ai.project` | `my-app` | Project name | +| `git.repo` | `github.com/acme/widget` | Normalized `origin` remote (credentials and ports stripped) | +| `git.pr_links` | `["…/pull/12"]` | PR URLs captured for the session | +| `git.commit_count` | `2` | Number of attributed commits | + +**`codeburn.commit`** — one per commit attributed to a session: + +| Field | Example | Description | +|---|---|---| +| `git.sha` | `4f2a…` | Commit SHA | +| `git.in_main` | `true` | Whether the commit landed in the main branch | +| `git.was_reverted` | `false` | Whether a later commit reverted it | + +Attribution is **inferred** (timestamp-window correlation, the same heuristic as `codeburn yield`); the resource attribute `codeburn.attribution_methodology: timestamp-window` marks it as such. State transitions (a commit merging to main, or being reverted) are re-sent automatically on later pushes — receivers should upsert commits by `(git.repo, git.sha)` and session spans by `ai.session_id` (latest state wins). When a commit migrates to a later-parsed session with a tighter window, the losing session re-emits with `git.commit_count: 0` (a retraction), so summing `git.commit_count` across upserted session rows never double-counts. Retractions fire only when the commit was won by another session — commits that merely age out of the `--since` window are not retracted, so a previously-synced count stays correct. Session spans also re-emit when an ongoing session's window grows, keeping the span end time current. + +With `--attribution`, normalized repo remote URLs, commit SHAs, commit timestamps (span start times), PR URLs, and the merged/reverted booleans leave your machine — plus the same pseudonymous `codeburn.device_id` resource attribute the usage spans carry. PR links are rebuilt client-side from scheme + host + path only (userinfo, query strings, and fragments are dropped; https, `/org/repo/pull/N` path, bounded length, max 20 per session), and the repo identity itself passes a strict hostname/path allow-list before sending — malformed or transport-helper remotes (`ext::…`, `codecommit::…`) are rejected outright rather than parsed. Precisely what is and is not sent: + +- **Commits**: only from repos with a network `origin` remote, and only for sessions whose own project path resolved to that repo. Local-only repos, `file://` remotes, and Windows filesystem paths are never emitted as repo identities. A session whose project path no longer resolves never inherits the repo of the directory you happen to push from. +- **PR links**: sent whenever a session captured them, even when the session's repo could not be identified — the PR URL itself names the repo, so this adds no information beyond the link the session already recorded. +- Without the flag, none of this is sent. + ### What is NOT sent - **Prompts** — your actual messages to AI are never included @@ -102,7 +135,7 @@ A pseudonymous `device_id` distinguishes your machines without revealing hostnam - **Bash commands** — may contain secrets, never sent - **Your name/email** — identity is derived server-side from your login token -There is no flag to override this. Privacy is structural, not configurable. +There is no flag to override this. Privacy is structural, not configurable. The only additive opt-in is `--attribution` (repo remotes, commit SHAs, and PR URLs — never code or prompts), described above. ## Authentication diff --git a/src/sync/cli.ts b/src/sync/cli.ts index 174b616c..f6f28db6 100644 --- a/src/sync/cli.ts +++ b/src/sync/cli.ts @@ -22,7 +22,8 @@ import { } 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' +import { collectUnsentCalls, collectUnsentAttribution, sendBatches, sendAttributionBatches, batchCalls, MAX_PER_PUSH, MAX_ATTRIBUTION_PER_PUSH, type PushResult } from './push.js' +import { batchAttributionItems } from './otlp.js' export function registerSyncCommands(program: Command): void { const sync = program @@ -226,7 +227,8 @@ export function registerSyncCommands(program: Command): void { .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 }) => { + .option('--attribution', 'Also push git attribution spans (session→commit correlation from `codeburn yield`, plus PR links). Sends normalized repo remotes and commit SHAs to the endpoint.') + .action(async (opts: { since: string; dryRun?: boolean; attribution?: boolean }) => { const config = readSyncConfig() if (!config) { process.stderr.write('Sync not configured. Run `codeburn sync setup ` first.\n') @@ -277,6 +279,18 @@ export function registerSyncCommands(program: Command): void { // Flatten + filter against sent-ledger const { allCalls, unsent } = collectUnsentCalls(projects) + // Attribution records (opt-in): session→commit correlation computed + // locally from the same parsed projects. Reuses the yield engine. + let attributionUnsent: Awaited>['unsent'] = [] + let attributionTotal = 0 + if (opts.attribution) { + const { computeAttributionRecords } = await import('../yield.js') + const records = computeAttributionRecords(projects, range, process.cwd()) + const collected = collectUnsentAttribution(records) + attributionUnsent = collected.unsent + attributionTotal = collected.allItems.length + } + 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) @@ -285,10 +299,19 @@ export function registerSyncCommands(program: Command): void { 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`) } + if (opts.attribution) { + const toPushAttr = attributionUnsent.slice(0, MAX_ATTRIBUTION_PER_PUSH) + const commits = toPushAttr.filter(i => i.kind === 'commit').length + const sessions = toPushAttr.filter(i => i.kind === 'session').length + process.stderr.write(`[dry-run] Attribution: ${attributionTotal} facts total, would push ${toPushAttr.length} (${sessions} sessions, ${commits} commits)\n`) + if (attributionUnsent.length > MAX_ATTRIBUTION_PER_PUSH) { + process.stderr.write(`[dry-run] ${attributionUnsent.length - MAX_ATTRIBUTION_PER_PUSH} more attribution facts exceed the ${MAX_ATTRIBUTION_PER_PUSH} safety limit — a second push would be needed\n`) + } + } return } - if (unsent.length === 0) { + if (unsent.length === 0 && attributionUnsent.length === 0) { process.stderr.write(`Nothing to push (${allCalls.length} calls already synced).\n`) updateLastSync() return @@ -302,15 +325,18 @@ export function registerSyncCommands(program: Command): void { // 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`), - }) + let result: PushResult = { outcome: 'complete', totalSent: 0, totalRejected: 0, totalCostSent: 0 } + if (toPush.length > 0) { + const batches = batchCalls(toPush, discoveryDoc.max_batch_size) + 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') @@ -323,11 +349,50 @@ export function registerSyncCommands(program: Command): void { process.stderr.write(`Server error (HTTP ${result.httpStatus}). Remaining calls will be sent on the next push.\n`) } + // Attribution spans ride the same endpoint after the usage push + // completes. Skipped when the usage push hit rate limits or server + // errors — the endpoint is already unhappy; both retry on next push. + let attrResult: PushResult | null = null + if (opts.attribution && attributionUnsent.length > 0) { + if (result.outcome === 'complete') { + // Safety valve, mirroring the usage-call cap + const attrToPush = attributionUnsent.slice(0, MAX_ATTRIBUTION_PER_PUSH) + if (attributionUnsent.length > MAX_ATTRIBUTION_PER_PUSH) { + process.stderr.write(`${attributionUnsent.length} attribution facts exceed the ${MAX_ATTRIBUTION_PER_PUSH} safety limit. Pushing first ${MAX_ATTRIBUTION_PER_PUSH}; run again to continue.\n`) + } + const attrBatches = batchAttributionItems(attrToPush, discoveryDoc.max_batch_size) + attrResult = await sendAttributionBatches({ + endpoint, + accessToken: tokens.access_token, + batches: attrBatches, + log: msg => process.stderr.write(`${msg}\n`), + }) + if (attrResult.outcome === 'auth-rejected') { + process.stderr.write('Auth rejected by server during attribution push. Run `codeburn sync setup` to re-authenticate.\n') + process.exit(1) + } + if (attrResult.outcome === 'rate-limited') { + process.stderr.write(`Rate limited during attribution push — gave up after repeated retries. Remaining facts will be sent on the next push.\n`) + } + if (attrResult.outcome === 'server-error') { + process.stderr.write(`Server error (HTTP ${attrResult.httpStatus}) during attribution push. Remaining facts will be sent on the next push.\n`) + } + } else { + process.stderr.write(`Skipping attribution push (${attributionUnsent.length} facts) — will retry on next push.\n`) + } + } + // Update lastSync updateLastSync() // Summary process.stderr.write(`\nSynced ${result.totalSent} calls ($${result.totalCostSent.toFixed(2)}) to ${config.baseUrl}\n`) + if (attrResult) { + const attrSuffix = attrResult.outcome !== 'complete' + ? ` (push incomplete — remainder retries next push)` + : attrResult.totalRejected > 0 ? `, ${attrResult.totalRejected} rejected (will retry)` : '' + process.stderr.write(` Attribution: ${attrResult.totalSent} facts synced${attrSuffix}\n`) + } if (result.totalRejected > 0) { process.stderr.write(` ${result.totalRejected} spans rejected (will retry on next push)\n`) } @@ -337,7 +402,7 @@ export function registerSyncCommands(program: Command): void { // Non-zero exit when the push did not complete, so cron/scripts can // detect it. Ledgered progress is kept; next push resumes. - if (result.outcome !== 'complete') { + if (result.outcome !== 'complete' || (attrResult !== null && attrResult.outcome !== 'complete')) { process.exitCode = 1 } } catch (err) { diff --git a/src/sync/otlp.ts b/src/sync/otlp.ts index 8a12c7e6..8df36c59 100644 --- a/src/sync/otlp.ts +++ b/src/sync/otlp.ts @@ -8,6 +8,7 @@ import { createHash } from 'crypto' import { hostname, userInfo } from 'os' import type { ParsedApiCall } from '../types.js' +import type { SessionAttributionRecord } from '../yield.js' export interface OtlpSpan { traceId: string @@ -141,3 +142,175 @@ export function batchCalls(calls: CallWithSession[], maxBatchSize: number): Call } return batches } + +// --- Attribution spans (sync push --attribution) --- + +export const SESSION_ATTRIBUTION_SPAN_NAME = 'codeburn.session.attribution' +export const COMMIT_ATTRIBUTION_SPAN_NAME = 'codeburn.commit' + +/** + * A single ledger-able attribution unit: either one session-level record + * (repo + PR links + commit count) or one attributed commit. The dedup key + * encodes the mutable state (inMain/wasReverted for commits; repo, PR links, + * and commit set for sessions), so a state TRANSITION mints a new key and the + * updated fact is re-sent on the next push — the receiver upserts by + * (repo, sha) / (session). Identical states dedupe via the sent-ledger. + */ +export type AttributionItem = { + kind: 'session' | 'commit' + dedupKey: string + /** Span start: commit author time for commits, session start for sessions. ISO 8601. */ + timestamp: string + /** Span end for session items (session lastTimestamp). Absent for commits. */ + endTimestamp?: string + sessionId: string + project: string + repo: string | null + // session kind + prLinks?: string[] + commitCount?: number + // commit kind + sha?: string + inMain?: boolean + wasReverted?: boolean +} + +function stateHash(parts: string[]): string { + return createHash('sha256').update(parts.join('\u001e')).digest('hex').slice(0, 16) +} + +/** Deterministic dedup key for a commit attribution fact (state included). */ +export function commitAttributionKey(sessionId: string, sha: string, inMain: boolean, wasReverted: boolean): string { + return `attr:c:${sessionId}:${sha}:${inMain ? 1 : 0}${wasReverted ? 1 : 0}` +} + +/** Deterministic dedup key for a session attribution fact (state included). */ +export function sessionAttributionKey(record: SessionAttributionRecord): string { + const commitStates = record.commits + .map(c => `${c.sha}:${c.inMain ? 1 : 0}${c.wasReverted ? 1 : 0}`) + .sort() + // Project and both window timestamps are part of the state: an ongoing + // session whose window grew (or whose project resolution changed) re-emits + // with the corrected span times instead of freezing at first send. + return `attr:s:${record.sessionId}:${stateHash([ + record.repo ?? '', + record.project, + record.firstTimestamp, + record.lastTimestamp, + ...record.prLinks, + ...commitStates, + ])}` +} + +/** Ledger-key prefix for a session's attribution facts (any state). */ +export function sessionAttributionKeyPrefix(sessionId: string): string { + return `attr:s:${sessionId}:` +} + +/** Flatten attribution records into ledger-able items (one session item + one per commit). */ +export function flattenAttributionRecords(records: SessionAttributionRecord[]): AttributionItem[] { + const items: AttributionItem[] = [] + for (const record of records) { + items.push({ + kind: 'session', + dedupKey: sessionAttributionKey(record), + timestamp: record.firstTimestamp, + endTimestamp: record.lastTimestamp, + sessionId: record.sessionId, + project: record.project, + repo: record.repo, + prLinks: record.prLinks, + commitCount: record.commits.length, + }) + for (const commit of record.commits) { + items.push({ + kind: 'commit', + dedupKey: commitAttributionKey(record.sessionId, commit.sha, commit.inMain, commit.wasReverted), + timestamp: commit.timestamp, + sessionId: record.sessionId, + project: record.project, + repo: record.repo, + sha: commit.sha, + inMain: commit.inMain, + wasReverted: commit.wasReverted, + }) + } + } + return items +} + +/** + * Build an OTLP payload from attribution items. Spans share the session's + * traceId with the usage spans (`deriveTraceId(sessionId)`), so a receiver + * can correlate cost and attribution without any extra key. + */ +export function buildAttributionOtlpPayload(items: AttributionItem[]): OtlpPayload { + const deviceId = getDeviceId() + + const spans: OtlpSpan[] = items.map(item => { + const startNano = toUnixNano(item.timestamp) + // Clamp like the usage builder: end is never 0 (malformed timestamp) and + // never earlier than start + 1ms (out-of-order session timestamps). + const minEndNano = BigInt(startNano) + 1_000_000n + const rawEndNano = item.endTimestamp ? BigInt(toUnixNano(item.endTimestamp)) : 0n + const endNano = (rawEndNano > minEndNano ? rawEndNano : minEndNano).toString() + + const attributes: OtlpAttribute[] = [ + { key: 'ai.session_id', value: { stringValue: item.sessionId } }, + { key: 'ai.project', value: { stringValue: item.project } }, + ] + if (item.repo) { + attributes.push({ key: 'git.repo', value: { stringValue: item.repo } }) + } + + if (item.kind === 'commit') { + attributes.push( + { key: 'git.sha', value: { stringValue: item.sha ?? '' } }, + { key: 'git.in_main', value: { boolValue: item.inMain ?? false } }, + { key: 'git.was_reverted', value: { boolValue: item.wasReverted ?? false } }, + ) + } else { + attributes.push({ key: 'git.commit_count', value: { intValue: String(item.commitCount ?? 0) } }) + if (item.prLinks && item.prLinks.length > 0) { + attributes.push({ + key: 'git.pr_links', + value: { arrayValue: { values: item.prLinks.map(u => ({ stringValue: u })) } }, + }) + } + } + + return { + traceId: deriveTraceId(item.sessionId), + spanId: deriveSpanId(item.dedupKey), + name: item.kind === 'commit' ? COMMIT_ATTRIBUTION_SPAN_NAME : SESSION_ATTRIBUTION_SPAN_NAME, + startTimeUnixNano: startNano, + endTimeUnixNano: endNano, + attributes, + } + }) + + return { + resourceSpans: [{ + resource: { + attributes: [ + { key: 'codeburn.device_id', value: { stringValue: deviceId } }, + // Honesty marker: this attribution is inferred (timestamp-window + // correlation), not declared. Receivers should label it as such. + { key: 'codeburn.attribution_methodology', value: { stringValue: 'timestamp-window' } }, + ], + }, + scopeSpans: [{ + spans, + }], + }], + } +} + +/** Split attribution items into batches of maxBatchSize. */ +export function batchAttributionItems(items: AttributionItem[], maxBatchSize: number): AttributionItem[][] { + const batches: AttributionItem[][] = [] + for (let i = 0; i < items.length; i += maxBatchSize) { + batches.push(items.slice(i, i + maxBatchSize)) + } + return batches +} diff --git a/src/sync/push.ts b/src/sync/push.ts index 0444c718..7a22420e 100644 --- a/src/sync/push.ts +++ b/src/sync/push.ts @@ -8,7 +8,16 @@ import type { ProjectSummary } from '../types.js' import { assertHttps } from './discovery.js' import { ledgerKeySet, appendToLedger, type LedgerEntry } from './ledger.js' -import { buildOtlpPayload, batchCalls, type CallWithSession } from './otlp.js' +import { + buildOtlpPayload, + batchCalls, + buildAttributionOtlpPayload, + flattenAttributionRecords, + type CallWithSession, + type AttributionItem, + type OtlpPayload, +} from './otlp.js' +import type { SessionAttributionRecord } from '../yield.js' /** * Safety valve, not a routine cap — pushes now loop until all batches are @@ -91,6 +100,23 @@ export function parseRetryAfterMs(value: string | null): number | null { * retry on the next push. */ export async function sendBatches(opts: SendBatchesOptions): Promise { + return sendBatchesCore({ + ...opts, + buildPayload: buildOtlpPayload, + toOutbound: c => ({ key: c.call.deduplicationKey, ts: c.call.timestamp, costUSD: c.call.costUSD }), + }) +} + +/** How sendBatchesCore ledgers and prices a batch item. */ +type OutboundItem = { key: string; ts: string; costUSD: number } + +type SendBatchesCoreOptions = Omit & { + batches: T[][] + buildPayload: (batch: T[]) => OtlpPayload + toOutbound: (item: T) => OutboundItem +} + +async function sendBatchesCore(opts: SendBatchesCoreOptions): Promise { assertHttps(opts.endpoint, 'Traces endpoint') const log = opts.log ?? (() => {}) const sleep = opts.sleep ?? ((ms: number) => new Promise(r => setTimeout(r, ms))) @@ -107,7 +133,7 @@ export async function sendBatches(opts: SendBatchesOptions): Promise // Retry loop for the current batch (429 only) for (;;) { - const payload = buildOtlpPayload(batch) + const payload = opts.buildPayload(batch) const response = await fetch(opts.endpoint, { method: 'POST', @@ -158,13 +184,11 @@ export async function sendBatches(opts: SendBatchesOptions): Promise 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, - })) + const outbound = batch.map(opts.toOutbound) + const entries: LedgerEntry[] = outbound.map(o => ({ key: o.key, ts: o.ts })) appendToLedger(entries) totalSent += batch.length - totalCostSent += batch.reduce((s, c) => s + c.call.costUSD, 0) + totalCostSent += outbound.reduce((s, o) => s + o.costUSD, 0) } break // batch done (success or partial) — move to next batch } @@ -173,4 +197,56 @@ export async function sendBatches(opts: SendBatchesOptions): Promise return { outcome: 'complete', totalSent, totalRejected, totalCostSent, totalWaitMs } } +/** + * Safety valve for attribution items, mirroring MAX_PER_PUSH: bounds a first + * `--since all --attribution` push over a long history. Remaining facts are + * sent on the next push (the ledger tracks progress). + */ +export const MAX_ATTRIBUTION_PER_PUSH = 10_000 + +/** Flatten attribution records into items and filter out already-sent ones. */ +export function collectUnsentAttribution(records: SessionAttributionRecord[]): { + allItems: AttributionItem[] + unsent: AttributionItem[] +} { + const sent = ledgerKeySet() + + // Empty records (no commits, no PR links) exist only to RETRACT a session + // span whose commits migrated to another session. Send one only when a + // PRIOR state for that session was already ledgered — a session that was + // never sent has nothing to retract. + const sessionsWithPriorState = new Set() + for (const key of sent) { + if (key.startsWith('attr:s:')) { + const sessionId = key.slice('attr:s:'.length, key.lastIndexOf(':')) + sessionsWithPriorState.add(sessionId) + } + } + const sendable = records.filter(r => + r.commits.length > 0 || r.prLinks.length > 0 || sessionsWithPriorState.has(r.sessionId), + ) + + const allItems = flattenAttributionRecords(sendable) + const unsent = allItems.filter(i => !sent.has(i.dedupKey)) + return { allItems, unsent } +} + +export interface SendAttributionBatchesOptions extends Omit { + batches: AttributionItem[][] +} + +/** + * Send attribution batches through the same retry/ledger pipeline as usage + * batches. Items are ledgered by their state-encoding dedup keys, so an + * identical attribution fact is sent once and a state transition (commit + * merged to main, commit reverted) re-sends the updated fact. + */ +export async function sendAttributionBatches(opts: SendAttributionBatchesOptions): Promise { + return sendBatchesCore({ + ...opts, + buildPayload: buildAttributionOtlpPayload, + toOutbound: item => ({ key: item.dedupKey, ts: item.timestamp, costUSD: 0 }), + }) +} + export { batchCalls } diff --git a/src/yield.ts b/src/yield.ts index ce3cbd05..79239b99 100644 --- a/src/yield.ts +++ b/src/yield.ts @@ -2,7 +2,7 @@ import { execFileSync } from 'child_process' import { realpathSync } from 'fs' import { resolve } from 'path' import { parseAllSessions } from './parser.js' -import type { DateRange, SessionSummary } from './types.js' +import type { DateRange, ProjectSummary, SessionSummary } from './types.js' export type YieldCategory = 'productive' | 'reverted' | 'abandoned' | 'ambiguous' @@ -133,7 +133,98 @@ function getMainBranch(cwd: string): string { return 'main' } -type CommitInfo = { +/** + * Normalize a git remote URL to a host-scoped repo identity (`host/org/repo`) + * usable as a server-side join key. Handles the three common transports: + * + * git@github.com:org/repo.git -> github.com/org/repo + * ssh://git@github.com:22/org/repo.git -> github.com/org/repo + * https://user:tok@github.com/org/repo.git -> github.com/org/repo + * + * Credentials and ports are stripped (a token embedded in an https remote must + * never leave the machine), the host is lowercased (path case is preserved), + * and a trailing `.git` / `/` is removed. Local paths and `file://` remotes + * return null — a repo with no network remote has no server-side identity. + */ +/** Max length of an emitted repo identity (`host/org/repo`). */ +const MAX_REPO_IDENTITY_LENGTH = 200 + +/** + * Positive validation (allow-list) of a composed repo identity — the final + * gate EVERY branch passes through before anything is returned. The host must + * look like a hostname and every path segment like a repo path segment, so no + * upstream parsing quirk (transport-helper remotes like `ext::…` or + * `codecommit::…`, credentials that survived a malformed URL, oversized + * strings) can reach the wire. Rejecting is always safe: an unrecognizable + * remote simply has no server-side identity. + */ +function isValidRepoIdentity(host: string, segments: string[]): boolean { + if (!/^[a-z0-9]([a-z0-9.-]*[a-z0-9])?$/.test(host)) return false + if (segments.length === 0) return false + return segments.every(s => /^[A-Za-z0-9._~-]+$/.test(s)) +} + +export function normalizeRemoteUrl(url: string): string | null { + const trimmed = url.trim() + if (!trimmed) return null + + // Windows drive-letter paths (`C:\Users\...`, `C:/Users/...`) are local + // filesystem paths, not scp-like remotes — without this check the scp-like + // branch would parse `C:` as a host and emit the user's local path as a + // repo identity. + if (/^[a-zA-Z]:[\\/]/.test(trimmed)) return null + + let host: string + let path: string + + if (/^[a-zA-Z][a-zA-Z0-9+.-]*:\/\//.test(trimmed)) { + let parsed: URL + try { + parsed = new URL(trimmed) + } catch { + return null + } + if (parsed.protocol === 'file:') return null + if (!parsed.hostname) return null + host = parsed.hostname + path = parsed.pathname + } else { + // scp-like syntax: [user@]host:path. Credentials (userinfo) are split off + // at the FIRST `@` BEFORE any host matching — expressing userinfo as an + // optional regex group lets backtracking abandon the group and re-parse a + // credential prefix as `host:path`, dumping the token into the path + // (e.g. `x-access-token:ghp_…@github.com/org/repo`). Host must be at + // least 2 chars: a single-character "host" is a Windows drive-relative + // path (`C:repo`), never a real remote host. + const at = trimmed.indexOf('@') + const rest = at >= 0 ? trimmed.slice(at + 1) : trimmed + const scpLike = /^([^:/\\]{2,}):(?!\/\/)(.+)$/.exec(rest) + if (!scpLike) return null + host = scpLike[1] + path = scpLike[2] + } + + const cleanPath = path + .replace(/\/+/g, '/') // collapse doubled slashes → one join key, not two + .replace(/^\/+/, '') + .replace(/\/+$/, '') + .replace(/\.git$/i, '') // case-insensitive: Repo.GIT joins with repo.git + if (!cleanPath) return null + + const identity = `${host.toLowerCase()}/${cleanPath}` + if (identity.length > MAX_REPO_IDENTITY_LENGTH) return null + if (!isValidRepoIdentity(host.toLowerCase(), cleanPath.split('/'))) return null + + return identity +} + +/** `git remote get-url origin`, normalized. Null when absent or local-only. */ +function getRepoRemote(gitDir: string): string | null { + const url = runGit(['remote', 'get-url', 'origin'], gitDir) + return url ? normalizeRemoteUrl(url) : null +} + +export type CommitInfo = { sha: string timestamp: Date inMain: boolean @@ -296,34 +387,39 @@ function categorizeSession( return { category: 'abandoned', commitCount: commits.length } } -export async function computeYield(range: DateRange, cwd: string, provider: string = 'all'): Promise { - const projects = await parseAllSessions(range, provider) - - const summary: YieldSummary = { - productive: { cost: 0, sessions: 0 }, - reverted: { cost: 0, sessions: 0 }, - abandoned: { cost: 0, sessions: 0 }, - ambiguous: { cost: 0, sessions: 0 }, - total: { cost: 0, sessions: 0 }, - details: [], - } +type RepoGroup = { + commits: CommitInfo[] + sessions: SessionSummary[] + projectNames: string[] + /** Parallel to `sessions`: true when the session's identity came from its + * OWN project path; false when it inherited the cwd-fallback identity. + * The attribution (sync) path must never egress fallback-derived repos. */ + ownIdentity: boolean[] + /** A directory to run further git queries in (remote lookup); null when the group has no git identity. */ + gitDir: string | null +} +/** + * Group sessions by canonical repository identity and load each group's + * commits for the range. Shared by `computeYield` (categorization) and + * `computeAttributionRecords` (sync). Grouping semantics are unchanged from + * the original computeYield implementation: each commit is awarded at most + * once across the whole repo; monorepo subdirectories and worktrees collapse + * to one group; a project whose path is missing or not a git repo falls back + * to the cwd repo (or an empty commit list when cwd is not a repo either). + */ +function buildRepoGroups( + projects: ProjectSummary[], + range: DateRange, + cwd: string, +): Map { const repoIdentityCache = new Map() - // Get all commits in the date range for correlation const cwdIdentity = resolveRepoIdentity(cwd, repoIdentityCache) const cwdCommits = cwdIdentity ? getCommitsInRange(cwd, range.start, range.end, getMainBranch(cwd)) : [] - // Group sessions by canonical repository identity before attributing so that - // each commit is awarded at most once across the whole repo. Two monorepo - // subdirectory sessions, or two worktrees of one repo, resolve to the same - // git-common-dir and share ONE group; keying on the raw path would double - // count. A project whose path is missing or not a git repo falls back to the - // cwd repo (or, when cwd is not a repo either, an empty commit list) exactly - // as before. - type RepoGroup = { commits: CommitInfo[]; sessions: SessionSummary[]; projectNames: string[] } const repoGroups = new Map() for (const project of projects) { const projectIdentity = project.projectPath @@ -342,15 +438,35 @@ export async function computeYield(range: DateRange, cwd: string, provider: stri : getCommitsInRange(identity.gitDir, range.start, range.end, getMainBranch(identity.gitDir)), sessions: [], projectNames: [], + ownIdentity: [], + gitDir: identity?.gitDir ?? null, } repoGroups.set(groupKey, group) } for (const session of project.sessions) { group.sessions.push(session) group.projectNames.push(project.project) + group.ownIdentity.push(projectIdentity !== null) } } + return repoGroups +} + +export async function computeYield(range: DateRange, cwd: string, provider: string = 'all'): Promise { + const projects = await parseAllSessions(range, provider) + + const summary: YieldSummary = { + productive: { cost: 0, sessions: 0 }, + reverted: { cost: 0, sessions: 0 }, + abandoned: { cost: 0, sessions: 0 }, + ambiguous: { cost: 0, sessions: 0 }, + total: { cost: 0, sessions: 0 }, + details: [], + } + + const repoGroups = buildRepoGroups(projects, range, cwd) + for (const group of repoGroups.values()) { const attributions = attributeCommits(group.sessions, group.commits) for (const [index, session] of group.sessions.entries()) { @@ -446,3 +562,160 @@ export function buildYieldJsonReport( })), } } + +// --- Sync attribution (codeburn sync push --attribution) --- + +export type CommitAttribution = { + sha: string + /** Commit author time, ISO 8601. */ + timestamp: string + inMain: boolean + wasReverted: boolean +} + +/** + * Per-session git attribution record — the sync-facing projection of the + * yield timestamp-window correlation. One record per session that produced + * server-joinable evidence: attributed commits (requires a normalized remote) + * and/or PR links. + */ +export type SessionAttributionRecord = { + sessionId: string + project: string + /** Normalized origin remote (`host/org/repo`), the server-side join key. Null when only prLinks are available. */ + repo: string | null + /** GitHub PR URLs captured for the session (already normalized upstream). */ + prLinks: string[] + /** Commits attributed to this session. Empty when repo is null (SHAs without a repo identity cannot be joined). */ + commits: CommitAttribution[] + /** Session window, ISO 8601 — lets the receiver reason about attribution recency. */ + firstTimestamp: string + lastTimestamp: string +} + +/** Max PR links retained per session attribution record. */ +export const MAX_PR_LINKS_PER_SESSION = 20 + +/** + * Shape-check PR links before they leave the machine. Upstream parsers only + * verify truthiness, so arbitrary strings can land in `session.prLinks`. + * Keep only https URLs shaped like a PR (`/org/repo/pull/N` — GitHub and + * GitHub Enterprise), bounded in length, capped per session, sorted. + * + * Links are REBUILT from `origin + pathname`, never passed through verbatim: + * userinfo (`https://alice:token@…`), query strings (copy-pasted GitHub + * links routinely carry `?notification_referrer_id=…`), and fragments are + * all dropped. Rebuilt links that collapse to the same URL dedupe. + */ +export function sanitizePrLinks(links: string[]): string[] { + const valid = new Set() + for (const link of links) { + if (typeof link !== 'string' || link.length === 0 || link.length > 512) continue + let url: URL + try { + url = new URL(link) + } catch { + continue + } + if (url.protocol !== 'https:') continue + if (!/^\/[^/]+\/[^/]+\/pull\/\d+$/.test(url.pathname)) continue + const rebuilt = `${url.origin}${url.pathname}` + if (rebuilt.length > 256) continue + valid.add(rebuilt) + } + return [...valid].sort().slice(0, MAX_PR_LINKS_PER_SESSION) +} + +/** + * Compute per-session attribution records for sync. Reuses the exact yield + * repo grouping + tightest-window commit attribution (`methodology: + * timestamp-window`), then joins in each repo group's normalized origin + * remote and the session's sanitized PR links. + * + * Inclusion rules: + * - Only sessions whose OWN project path resolved to a repo participate in + * commit attribution; cwd-fallback sessions never carry a repo or commits + * (privacy gate — see below) but still emit a record when they have PR links. + * - Commits require a normalized remote; a SHA without a repo identity has + * no server-side meaning. + * - A session with no commits and no PR links is emitted ONLY when it lost a + * commit to a tighter-window session in THIS computation (`lostCandidacy`) + * — a retraction candidate. A session that merely aged its commits out of + * the range lost them to nobody, and emitting an empty record for it would + * permanently retract a still-correct server-side count. + * + * Takes already-parsed projects (sync push has them in hand) instead of + * re-parsing like computeYield does. + */ +export function computeAttributionRecords( + projects: ProjectSummary[], + range: DateRange, + cwd: string, +): SessionAttributionRecord[] { + const repoGroups = buildRepoGroups(projects, range, cwd) + const records: SessionAttributionRecord[] = [] + + for (const group of repoGroups.values()) { + const remote = group.gitDir ? getRepoRemote(group.gitDir) : null + + // Privacy gate: only sessions whose identity came from their OWN project + // path participate in commit attribution. A session whose project path no + // longer resolves (deleted/renamed dir, non-repo session) inherits the + // cwd-fallback identity in buildRepoGroups — attributing it here would + // egress whatever repo the user happens to be pushing from, with commits + // that session never touched. Fallback sessions get no repo and no + // commits; they still emit a record when they carry PR links (which are + // session-native and safe). Excluding them from the competition also + // prevents a fallback window from stealing a commit that belongs to a + // genuine session. + const ownSessions = group.sessions.filter((_, i) => group.ownIdentity[i]) + const attributions = attributeCommits(ownSessions, group.commits) + // Keyed by object reference: session objects are unique per group entry, + // whereas sessionId strings could collide across projects. + const attributionBySession = new Map() + const lostCandidacyBySession = new Map() + for (const [i, session] of ownSessions.entries()) { + attributionBySession.set(session, attributions[i]?.commits ?? []) + lostCandidacyBySession.set(session, attributions[i]?.lostCandidacy ?? false) + } + + for (const [index, session] of group.sessions.entries()) { + if (!session.firstTimestamp) continue + + const isOwn = group.ownIdentity[index] === true + const sessionRemote = isOwn ? remote : null + const attributedCommits = sessionRemote + ? (attributionBySession.get(session) ?? []) + : [] + const prLinks = sanitizePrLinks(session.prLinks ?? []) + // Empty sessions are retraction candidates ONLY when they lost a commit + // to a tighter-window session in THIS run: that commit's server-side + // attribution is migrating, so the loser must re-emit commit_count=0. + // An empty session whose commits merely aged out of the --since range + // (rolling window, or a narrower window than a previous push) lost them + // to NOBODY — emitting a retraction for it would permanently zero a + // still-correct server-side count, because the original state key stays + // ledgered and is never re-sent. + const lostToTighterSession = sessionRemote !== null && + (lostCandidacyBySession.get(session) ?? false) + if (attributedCommits.length === 0 && prLinks.length === 0 && !lostToTighterSession) continue + + records.push({ + sessionId: session.sessionId, + project: group.projectNames[index] ?? session.project, + repo: sessionRemote, + prLinks, + commits: attributedCommits.map(c => ({ + sha: c.sha, + timestamp: c.timestamp.toISOString(), + inMain: c.inMain, + wasReverted: c.wasReverted, + })), + firstTimestamp: session.firstTimestamp, + lastTimestamp: session.lastTimestamp ?? session.firstTimestamp, + }) + } + } + + return records +} diff --git a/tests/fixtures/mock-idp.ts b/tests/fixtures/mock-idp.ts index e873b6b6..d70b9f57 100644 --- a/tests/fixtures/mock-idp.ts +++ b/tests/fixtures/mock-idp.ts @@ -34,6 +34,8 @@ export interface MockIdp { revokedTokens: string[] /** Authorization codes that have been exchanged */ exchangedCodes: string[] + /** OTLP trace batches received at POST /v1/traces */ + tracesRequests: Array<{ auth: string | undefined; body: unknown }> } export async function startMockIdp(opts: MockIdpOptions = {}): Promise { @@ -52,6 +54,7 @@ export async function startMockIdp(opts: MockIdpOptions = {}): Promise issuedTokens: { access: [], refresh: [] }, revokedTokens: [], exchangedCodes: [], + tracesRequests: [], close: async () => {}, } @@ -59,6 +62,20 @@ export async function startMockIdp(opts: MockIdpOptions = {}): Promise const url = new URL(req.url ?? '/', `http://127.0.0.1:${state.port}`) const path = url.pathname + // --- OTLP traces collector (records batches for push tests) --- + if (path === '/v1/traces' && req.method === 'POST') { + let body = '' + req.on('data', chunk => { body += chunk }) + req.on('end', () => { + let parsed: unknown = null + try { parsed = JSON.parse(body || '{}') } catch { /* keep null */ } + state.tracesRequests.push({ auth: req.headers.authorization, body: parsed }) + res.writeHead(200, { 'Content-Type': 'application/json' }) + res.end('{}') + }) + return + } + // --- Discovery doc --- if (path === '/.well-known/codeburn-export.json') { res.writeHead(200, { 'Content-Type': 'application/json' }) diff --git a/tests/sync-attribution-cli.test.ts b/tests/sync-attribution-cli.test.ts new file mode 100644 index 00000000..295350fc --- /dev/null +++ b/tests/sync-attribution-cli.test.ts @@ -0,0 +1,157 @@ +/** + * CLI-level tests for `codeburn sync push --attribution`. + * + * Drives the real commander action against a mock IdP + collector: + * - --dry-run --attribution: NO telemetry reaches the traces endpoint + * - push WITHOUT the flag: no attribution span names on the wire + * - push WITH the flag: attribution spans arrive alongside usage spans + */ + +import { execFileSync } from 'node:child_process' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach, vi } from 'vitest' +import { Command } from 'commander' + +import { startMockIdp, type MockIdp } from './fixtures/mock-idp.js' +import type { ProjectSummary, SessionSummary, ParsedApiCall, TokenUsage } from '../src/types.js' + +const { parseAllSessionsMock } = vi.hoisted(() => ({ parseAllSessionsMock: vi.fn() })) +vi.mock('../src/parser.js', () => ({ parseAllSessions: parseAllSessionsMock })) + +function git(cwd: string, args: string[], env: Record = {}): string { + return execFileSync('git', args, { cwd, encoding: 'utf-8', env: { ...process.env, ...env } }).trim() +} + +function makeUsage(): TokenUsage { + return { inputTokens: 10, outputTokens: 5, cacheCreationInputTokens: 0, cacheReadInputTokens: 0, cachedInputTokens: 0, reasoningTokens: 0, webSearchRequests: 0 } +} + +function makeCall(key: string, ts: string): ParsedApiCall { + return { + provider: 'test', model: 'test-model', usage: makeUsage(), costUSD: 0.01, + tools: [], mcpTools: [], skills: [], subagentTypes: [], hasAgentSpawn: false, + hasPlanMode: false, speed: 'standard', timestamp: ts, bashCommands: [], + deduplicationKey: key, + } +} + +function makeSession(id: string, first: string, last: string, calls: ParsedApiCall[]): SessionSummary { + return { + sessionId: id, project: 'app', firstTimestamp: first, lastTimestamp: last, + totalCostUSD: 0.01, totalSavingsUSD: 0, totalInputTokens: 10, totalOutputTokens: 5, + totalReasoningTokens: 0, totalCacheReadTokens: 0, totalCacheWriteTokens: 0, + apiCalls: calls.length, + turns: [{ userMessage: 'x', assistantCalls: calls, timestamp: first, sessionId: id, category: 'coding', retries: 0, hasEdits: false }], + modelBreakdown: {}, toolBreakdown: {}, mcpBreakdown: {}, bashBreakdown: {}, + categoryBreakdown: {} as SessionSummary['categoryBreakdown'], skillBreakdown: {}, subagentBreakdown: {}, + } +} + +let idp: MockIdp +let tmpHome: string +let repoDir: string +const originalHome = process.env.HOME +const originalXdg = process.env.XDG_CACHE_HOME +const originalStore = process.env.CODEBURN_SYNC_TOKEN_STORE + +async function runPush(args: string[]): Promise { + const { registerSyncCommands } = await import('../src/sync/cli.js') + const program = new Command() + program.exitOverride() // throw instead of process.exit on commander errors + registerSyncCommands(program) + await program.parseAsync(['node', 'codeburn', 'sync', 'push', ...args]) +} + +beforeAll(async () => { + idp = await startMockIdp({ rotateTokens: false }) + process.env.CODEBURN_SYNC_TOKEN_STORE = 'file' + + // Real git repo with a remote — a recent commit inside the session window + repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-cli-repo-')) + git(repoDir, ['init', '-b', 'main']) + git(repoDir, ['config', 'user.email', 't@e.com']) + git(repoDir, ['config', 'user.name', 'T']) + git(repoDir, ['remote', 'add', 'origin', 'git@github.com:acme/cli-widget.git']) + await writeFile(join(repoDir, 'f.txt'), 'x\n') + const commitIso = new Date(Date.now() - 60 * 60 * 1000).toISOString() + git(repoDir, ['add', '.']) + git(repoDir, ['commit', '-m', 'feat: recent'], { GIT_AUTHOR_DATE: commitIso, GIT_COMMITTER_DATE: commitIso }) +}) + +afterAll(async () => { + await idp.close() + await rm(repoDir, { recursive: true, force: true }) + if (originalStore === undefined) delete process.env.CODEBURN_SYNC_TOKEN_STORE + else process.env.CODEBURN_SYNC_TOKEN_STORE = originalStore +}) + +beforeEach(async () => { + tmpHome = await mkdtemp(join(tmpdir(), 'codeburn-attr-cli-')) + process.env.HOME = tmpHome + process.env.XDG_CACHE_HOME = join(tmpHome, '.cache') + idp.tracesRequests.length = 0 + + // Configure sync against the mock IdP + store the refresh token + const { writeSyncConfig } = await import('../src/sync/config.js') + const { createCredentialStore } = await import('../src/sync/credentials.js') + writeSyncConfig({ baseUrl: idp.baseUrl, clientId: 'mock-client-id', tracesPath: '/v1/traces', issuer: idp.baseUrl }) + createCredentialStore().store('mock-refresh-token-v1') + + // One session in the last hour, working in the real repo + const now = Date.now() + const first = new Date(now - 90 * 60 * 1000).toISOString() + const last = new Date(now - 30 * 60 * 1000).toISOString() + const session = makeSession('cli-sess-1', first, last, [makeCall(`call-${now}`, first)]) + parseAllSessionsMock.mockResolvedValue([ + { project: 'app', projectPath: repoDir, sessions: [session] } as ProjectSummary, + ]) + + // The push action reads process.cwd() for the attribution cwd — run from + // a neutral non-repo dir so nothing can come from a cwd fallback. + vi.spyOn(process, 'cwd').mockReturnValue(tmpHome) +}) + +afterEach(async () => { + vi.restoreAllMocks() + process.exitCode = 0 + process.env.HOME = originalHome + if (originalXdg === undefined) delete process.env.XDG_CACHE_HOME + else process.env.XDG_CACHE_HOME = originalXdg + await rm(tmpHome, { recursive: true, force: true }) +}) + +const spanNames = (): string[] => + idp.tracesRequests.flatMap(r => { + const body = r.body as { resourceSpans?: Array<{ scopeSpans: Array<{ spans: Array<{ name: string }> }> }> } + return (body.resourceSpans ?? []).flatMap(rs => rs.scopeSpans.flatMap(ss => ss.spans.map(s => s.name))) + }) + +describe('sync push --attribution (CLI level)', () => { + it('--dry-run --attribution sends nothing to the traces endpoint', async () => { + await runPush(['--dry-run', '--attribution']) + expect(idp.tracesRequests).toHaveLength(0) + }) + + it('push WITHOUT --attribution never emits attribution span names', async () => { + await runPush([]) + expect(idp.tracesRequests.length).toBeGreaterThan(0) + const names = spanNames() + expect(names.length).toBeGreaterThan(0) + expect(names).not.toContain('codeburn.session.attribution') + expect(names).not.toContain('codeburn.commit') + expect(JSON.stringify(idp.tracesRequests)).not.toContain('git.sha') + }) + + it('push WITH --attribution emits usage + attribution spans', async () => { + await runPush(['--attribution']) + const names = spanNames() + expect(names).toContain('test/test-model') // usage span + expect(names).toContain('codeburn.session.attribution') // session span + expect(names).toContain('codeburn.commit') // commit span + const wire = JSON.stringify(idp.tracesRequests) + expect(wire).toContain('github.com/acme/cli-widget') + }) +}) diff --git a/tests/sync-attribution.test.ts b/tests/sync-attribution.test.ts new file mode 100644 index 00000000..d4b8f6c1 --- /dev/null +++ b/tests/sync-attribution.test.ts @@ -0,0 +1,753 @@ +/** + * Tests for sync git attribution (sync push --attribution). + * + * Covers: remote URL normalization, session→commit attribution record + * computation (reusing the yield engine), state-encoding dedup keys, + * attribution OTLP span construction, and the send/ledger pipeline. + */ + +import { execFileSync } from 'node:child_process' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { createServer, type Server } from 'node:http' + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' + +import type { ProjectSummary, SessionSummary } from '../src/types.js' +import { + normalizeRemoteUrl, + computeAttributionRecords, + sanitizePrLinks, + MAX_PR_LINKS_PER_SESSION, + type SessionAttributionRecord, +} from '../src/yield.js' +import { + flattenAttributionRecords, + commitAttributionKey, + sessionAttributionKey, + buildAttributionOtlpPayload, + batchAttributionItems, + deriveTraceId, + SESSION_ATTRIBUTION_SPAN_NAME, + COMMIT_ATTRIBUTION_SPAN_NAME, + type OtlpAttribute, +} from '../src/sync/otlp.js' + +// ── Git fixtures (mirrors yield-repo-grouping.test.ts) ──────────────── + +function git(cwd: string, args: string[], env: Record = {}): string { + return execFileSync('git', args, { + cwd, + encoding: 'utf-8', + env: { ...process.env, ...env }, + }).trim() +} + +function initRepo(dir: string): void { + git(dir, ['init', '-b', 'main']) + git(dir, ['config', 'user.email', 'test@example.com']) + git(dir, ['config', 'user.name', 'Test']) +} + +function commitAt(dir: string, message: string, iso: string): void { + git(dir, ['add', '.']) + git(dir, ['commit', '-m', message], { + GIT_AUTHOR_DATE: iso, + GIT_COMMITTER_DATE: iso, + }) +} + +function makeSession(overrides: Partial): SessionSummary { + return { + sessionId: 'session', + project: 'app', + firstTimestamp: '2026-01-01T10:00:00.000Z', + lastTimestamp: '2026-01-01T11:00:00.000Z', + totalCostUSD: 1, + totalSavingsUSD: 0, + totalInputTokens: 0, + totalOutputTokens: 0, + totalReasoningTokens: 0, + totalCacheReadTokens: 0, + totalCacheWriteTokens: 0, + apiCalls: 1, + turns: [], + modelBreakdown: {}, + toolBreakdown: {}, + mcpBreakdown: {}, + bashBreakdown: {}, + categoryBreakdown: {} as SessionSummary['categoryBreakdown'], + skillBreakdown: {}, + subagentBreakdown: {}, + ...overrides, + } +} + +const range = { + start: new Date('2026-01-01T00:00:00.000Z'), + end: new Date('2026-01-02T00:00:00.000Z'), +} + +// ── normalizeRemoteUrl ──────────────────────────────────────────────── + +describe('normalizeRemoteUrl', () => { + it('normalizes scp-like ssh remotes', () => { + expect(normalizeRemoteUrl('git@github.com:acme/widget.git')).toBe('github.com/acme/widget') + expect(normalizeRemoteUrl('git@GitHub.com:Acme/Widget')).toBe('github.com/Acme/Widget') + }) + + it('normalizes ssh:// remotes, dropping user and port', () => { + expect(normalizeRemoteUrl('ssh://git@github.com/acme/widget.git')).toBe('github.com/acme/widget') + expect(normalizeRemoteUrl('ssh://git@gitlab.example.com:2222/group/sub/repo.git')).toBe('gitlab.example.com/group/sub/repo') + }) + + it('normalizes https remotes and strips embedded credentials', () => { + expect(normalizeRemoteUrl('https://github.com/acme/widget.git')).toBe('github.com/acme/widget') + expect(normalizeRemoteUrl('https://user:s3cret-token@github.com/acme/widget.git')).toBe('github.com/acme/widget') + expect(normalizeRemoteUrl('https://github.com/acme/widget/')).toBe('github.com/acme/widget') + }) + + it('returns null for local paths and file:// remotes', () => { + expect(normalizeRemoteUrl('/home/dev/repos/widget')).toBeNull() + expect(normalizeRemoteUrl('file:///home/dev/repos/widget')).toBeNull() + expect(normalizeRemoteUrl('../relative/repo')).toBeNull() + expect(normalizeRemoteUrl('')).toBeNull() + }) + + it('rejects Windows drive-letter paths (never a remote identity)', () => { + expect(normalizeRemoteUrl('C:/Users/alice/private/repo')).toBeNull() + expect(normalizeRemoteUrl('C:\\Users\\alice\\private\\repo')).toBeNull() + expect(normalizeRemoteUrl('c:/repo')).toBeNull() + expect(normalizeRemoteUrl('Z:\\work\\nda-client-repo')).toBeNull() + // Drive-relative (no slash after colon) — single-char host rejection + expect(normalizeRemoteUrl('C:repo')).toBeNull() + expect(normalizeRemoteUrl('c:relative\\path')).toBeNull() + }) + + it('rejects other adversarial forms without over-rejecting real remotes', () => { + // Single-character "host" is never a real remote host + expect(normalizeRemoteUrl('a:path/to/repo')).toBeNull() + expect(normalizeRemoteUrl('git@C:/foo')).toBeNull() + // Dotless intranet hosts remain valid (2+ chars) + expect(normalizeRemoteUrl('gitserver:team/repo.git')).toBe('gitserver/team/repo') + expect(normalizeRemoteUrl('git@gitbox:org/repo.git')).toBe('gitbox/org/repo') + }) + + it('never leaks credentials via scp-branch backtracking on malformed remotes', () => { + // Credential-prefixed remotes: the userinfo split happens BEFORE host + // matching, so a token can never be re-parsed as host:path. + expect(normalizeRemoteUrl('x-access-token:ghp_LIVETOKEN_abcdefghijklmnop@github.com/acme/private-repo.git')).toBeNull() + expect(normalizeRemoteUrl('oauth2:glpat-TOKEN@gitlab.com/org/repo.git')).toBeNull() + // One dropped slash: not a URL, must not fall through as host "https" + expect(normalizeRemoteUrl('https:/user:ghp_TOKEN@github.com/org/repo.git')).toBeNull() + // Multiple @: split at the first, residual @ fails the allow-list + expect(normalizeRemoteUrl('a@b@github.com:org/repo.git')).toBeNull() + expect(normalizeRemoteUrl('user@host:path@with-at')).toBeNull() + }) + + it('rejects transport-helper remotes and enforces shape + length on the identity', () => { + // git-remote-ext: embeds a local SSH key path + expect(normalizeRemoteUrl('ext::ssh -i /Users/me/.ssh/id_ed25519_work git@github.com %S /acme/private.git')).toBeNull() + expect(normalizeRemoteUrl('ext::sh -c whatever')).toBeNull() + // git-remote-codecommit: embeds an AWS profile name + expect(normalizeRemoteUrl('codecommit::us-east-1://MyAwsProfile@MyRepo')).toBeNull() + expect(normalizeRemoteUrl('codecommit::us-east-1://MyRepo')).toBeNull() + // Length bound + expect(normalizeRemoteUrl(`git@github.com:org/${'a'.repeat(300)}.git`)).toBeNull() + // Path segments must be repo-shaped (no spaces, colons, @) + expect(normalizeRemoteUrl('gitserver:has space/repo.git')).toBeNull() + // Legit multi-segment (GitLab subgroup) paths survive the allow-list + expect(normalizeRemoteUrl('https://gitlab.example.com/group/sub/repo.git')).toBe('gitlab.example.com/group/sub/repo') + }) + + it('normalizes .GIT case-insensitively and collapses doubled slashes to one join key', () => { + expect(normalizeRemoteUrl('git@github.com:acme/Repo.GIT')).toBe('github.com/acme/Repo') + expect(normalizeRemoteUrl('https://github.com/acme/Repo.Git')).toBe('github.com/acme/Repo') + expect(normalizeRemoteUrl('https://github.com/acme//repo.git')).toBe('github.com/acme/repo') + expect(normalizeRemoteUrl('git@github.com:acme//repo.git')).toBe('github.com/acme/repo') + }) +}) + +// ── computeAttributionRecords ───────────────────────────────────────── + +describe('computeAttributionRecords', () => { + it('attributes commits with normalized remote, inMain, and timestamps', async () => { + const repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-repo-')) + try { + initRepo(repoDir) + git(repoDir, ['remote', 'add', 'origin', 'git@github.com:acme/widget.git']) + await writeFile(join(repoDir, 'file.txt'), 'hello\n') + commitAt(repoDir, 'feat: shipped', '2026-01-01T10:30:00Z') + const sha = git(repoDir, ['rev-parse', 'HEAD']) + + const session = makeSession({ + sessionId: 'sess-a', + prLinks: ['https://github.com/acme/widget/pull/12'], + }) + const projects = [ + { project: 'app', projectPath: repoDir, sessions: [session] } as ProjectSummary, + ] + + const records = computeAttributionRecords(projects, range, repoDir) + + expect(records).toHaveLength(1) + const record = records[0]! + expect(record.sessionId).toBe('sess-a') + expect(record.repo).toBe('github.com/acme/widget') + expect(record.prLinks).toEqual(['https://github.com/acme/widget/pull/12']) + expect(record.commits).toHaveLength(1) + expect(record.commits[0]).toMatchObject({ sha, inMain: true, wasReverted: false }) + expect(new Date(record.commits[0]!.timestamp).toISOString()).toBe('2026-01-01T10:30:00.000Z') + expect(record.firstTimestamp).toBe('2026-01-01T10:00:00.000Z') + expect(record.lastTimestamp).toBe('2026-01-01T11:00:00.000Z') + } finally { + await rm(repoDir, { recursive: true, force: true }) + } + }) + + it('omits empty sessions that lost nothing — commits aged out of range are NOT retracted', async () => { + const repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-empty-')) + try { + initRepo(repoDir) + git(repoDir, ['remote', 'add', 'origin', 'git@github.com:acme/widget.git']) + await writeFile(join(repoDir, 'file.txt'), 'hello\n') + // Commit outside every session window: no session competes for it, so + // nobody "lost" it. Even if this session previously synced a commit + // (now outside the --since range), emitting an empty record here would + // permanently zero a still-correct server-side count. + commitAt(repoDir, 'feat: unrelated', '2026-01-01T20:00:00Z') + + const session = makeSession({ sessionId: 'sess-idle' }) + const projects = [ + { project: 'app', projectPath: repoDir, sessions: [session] } as ProjectSummary, + ] + + expect(computeAttributionRecords(projects, range, repoDir)).toEqual([]) + + // …and therefore nothing can be sent, even with prior ledger state for + // this session (simulating an earlier wider---since push). + const { writeLedger } = await import('../src/sync/ledger.js') + writeLedger([{ key: 'attr:s:sess-idle:0123456789abcdef', ts: '2026-01-01T10:00:00.000Z' }]) + const { collectUnsentAttribution } = await import('../src/sync/push.js') + expect(collectUnsentAttribution(computeAttributionRecords(projects, range, repoDir)).unsent).toEqual([]) + } finally { + await rm(repoDir, { recursive: true, force: true }) + } + }) + + it('emits a retraction candidate for a session that lost its commit to a tighter window', async () => { + const repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-lost-')) + try { + initRepo(repoDir) + git(repoDir, ['remote', 'add', 'origin', 'git@github.com:acme/widget.git']) + await writeFile(join(repoDir, 'file.txt'), 'hello\n') + commitAt(repoDir, 'feat: contested', '2026-01-01T10:30:00Z') + + const tight = makeSession({ + sessionId: 'sess-tight', + firstTimestamp: '2026-01-01T10:15:00.000Z', + lastTimestamp: '2026-01-01T10:45:00.000Z', + }) + const broadLoser = makeSession({ sessionId: 'sess-broad-loser' }) + const projects = [ + { project: 'app', projectPath: repoDir, sessions: [tight, broadLoser] } as ProjectSummary, + ] + + const records = computeAttributionRecords(projects, range, repoDir) + const loser = records.find(r => r.sessionId === 'sess-broad-loser')! + // The loser IS emitted (retraction candidate: lostCandidacy) with zero + // commits — the sync layer decides whether a prior state warrants + // actually sending it. + expect(loser).toBeDefined() + expect(loser.commits).toEqual([]) + expect(loser.repo).toBe('github.com/acme/widget') + } finally { + await rm(repoDir, { recursive: true, force: true }) + } + }) + + it('drops commits when the repo has no remote, but keeps PR-linked sessions', async () => { + const repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-noremote-')) + try { + initRepo(repoDir) // no origin remote + await writeFile(join(repoDir, 'file.txt'), 'hello\n') + commitAt(repoDir, 'feat: local only', '2026-01-01T10:30:00Z') + + const withPr = makeSession({ + sessionId: 'sess-pr', + prLinks: ['https://github.com/acme/widget/pull/7'], + firstTimestamp: '2026-01-01T10:15:00.000Z', + lastTimestamp: '2026-01-01T10:45:00.000Z', + }) + const withoutPr = makeSession({ sessionId: 'sess-nopr' }) + const projects = [ + { project: 'app', projectPath: repoDir, sessions: [withPr, withoutPr] } as ProjectSummary, + ] + + const records = computeAttributionRecords(projects, range, repoDir) + + // sess-pr wins the commit window but has no repo identity, so commits + // are dropped; the PR link alone justifies the record. sess-nopr has + // nothing joinable and is omitted. + expect(records).toHaveLength(1) + expect(records[0]!.sessionId).toBe('sess-pr') + expect(records[0]!.repo).toBeNull() + expect(records[0]!.commits).toEqual([]) + expect(records[0]!.prLinks).toEqual(['https://github.com/acme/widget/pull/7']) + } finally { + await rm(repoDir, { recursive: true, force: true }) + } + }) + + it('never egresses the cwd repo for sessions whose project path did not resolve (fallback)', async () => { + // The reviewer's repro: push from inside a private repo while a session's + // project path no longer resolves. The fallback identity must NOT leak + // the cwd repo's remote or commits into that session's attribution. + const cwdRepo = await mkdtemp(join(tmpdir(), 'codeburn-attr-privatecwd-')) + try { + initRepo(cwdRepo) + git(cwdRepo, ['remote', 'add', 'origin', 'git@github.com:secret-org/nda-client-repo.git']) + await writeFile(join(cwdRepo, 'file.txt'), 'confidential\n') + commitAt(cwdRepo, 'feat: private work', '2026-01-01T10:30:00Z') + + // Session A: project path is gone (deleted dir) — falls back to cwd + const orphanNoPr = makeSession({ sessionId: 'orphan-nopr', ...{ firstTimestamp: '2026-01-01T10:15:00.000Z', lastTimestamp: '2026-01-01T10:45:00.000Z' } }) + // Session B: also fallback, but carries a PR link (session-native, safe) + const orphanWithPr = makeSession({ + sessionId: 'orphan-pr', + prLinks: ['https://github.com/acme/widget/pull/9'], + firstTimestamp: '2026-01-01T12:00:00.000Z', + lastTimestamp: '2026-01-01T12:30:00.000Z', + }) + // Session C: genuinely belongs to the cwd repo (own path resolves) + const genuine = makeSession({ sessionId: 'genuine-cwd', firstTimestamp: '2026-01-01T10:00:00.000Z', lastTimestamp: '2026-01-01T11:00:00.000Z' }) + + const projects = [ + { project: 'ghost', projectPath: join(cwdRepo, 'no-such-dir-anymore-xyz'), sessions: [orphanNoPr] }, + { project: 'ghost2', projectPath: '', sessions: [orphanWithPr] }, + { project: 'real', projectPath: cwdRepo, sessions: [genuine] }, + ] as ProjectSummary[] + + const records = computeAttributionRecords(projects, range, cwdRepo) + + // orphan-nopr: nothing joinable -> no record at all + expect(records.find(r => r.sessionId === 'orphan-nopr')).toBeUndefined() + // orphan-pr: PR link only — no repo, no commits + const pr = records.find(r => r.sessionId === 'orphan-pr')! + expect(pr.repo).toBeNull() + expect(pr.commits).toEqual([]) + // genuine cwd session keeps full attribution + const own = records.find(r => r.sessionId === 'genuine-cwd')! + expect(own.repo).toBe('github.com/secret-org/nda-client-repo') + expect(own.commits).toHaveLength(1) + // The private repo identity appears ONLY on the genuine record + const leaked = records.filter(r => r.sessionId !== 'genuine-cwd' && JSON.stringify(r).includes('secret-org')) + expect(leaked).toEqual([]) + } finally { + await rm(cwdRepo, { recursive: true, force: true }) + } + }) + + it('fallback sessions cannot steal a commit from a genuine session', async () => { + const cwdRepo = await mkdtemp(join(tmpdir(), 'codeburn-attr-steal-')) + try { + initRepo(cwdRepo) + git(cwdRepo, ['remote', 'add', 'origin', 'git@github.com:acme/widget.git']) + await writeFile(join(cwdRepo, 'file.txt'), 'x\n') + commitAt(cwdRepo, 'feat: mine', '2026-01-01T10:30:00Z') + + // Fallback session has the TIGHTER window (would win under old logic); + // genuine session has the broader window. + const fallbackTight = makeSession({ + sessionId: 'fallback-tight', + prLinks: ['https://github.com/acme/widget/pull/2'], + firstTimestamp: '2026-01-01T10:25:00.000Z', + lastTimestamp: '2026-01-01T10:35:00.000Z', + }) + const genuineBroad = makeSession({ sessionId: 'genuine-broad' }) + + const projects = [ + { project: 'ghost', projectPath: '', sessions: [fallbackTight] }, + { project: 'real', projectPath: cwdRepo, sessions: [genuineBroad] }, + ] as ProjectSummary[] + + const records = computeAttributionRecords(projects, range, cwdRepo) + + expect(records.find(r => r.sessionId === 'fallback-tight')!.commits).toEqual([]) + expect(records.find(r => r.sessionId === 'genuine-broad')!.commits).toHaveLength(1) + } finally { + await rm(cwdRepo, { recursive: true, force: true }) + } + }) + + it('awards each commit to a single session (tightest window)', async () => { + const repoDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-overlap-')) + try { + initRepo(repoDir) + git(repoDir, ['remote', 'add', 'origin', 'https://github.com/acme/widget.git']) + await writeFile(join(repoDir, 'file.txt'), 'hello\n') + commitAt(repoDir, 'feat: shared window', '2026-01-01T10:30:00Z') + + const tight = makeSession({ + sessionId: 'sess-tight', + firstTimestamp: '2026-01-01T10:15:00.000Z', + lastTimestamp: '2026-01-01T10:45:00.000Z', + }) + const broad = makeSession({ sessionId: 'sess-broad' }) + const projects = [ + { project: 'app', projectPath: repoDir, sessions: [tight, broad] } as ProjectSummary, + ] + + const records = computeAttributionRecords(projects, range, repoDir) + + // Both sessions get records (the loser is a retraction candidate), + // but the commit is awarded exactly once — to the tighter window. + expect(records).toHaveLength(2) + const tightRecord = records.find(r => r.sessionId === 'sess-tight')! + const broadRecord = records.find(r => r.sessionId === 'sess-broad')! + expect(tightRecord.commits).toHaveLength(1) + expect(broadRecord.commits).toEqual([]) + } finally { + await rm(repoDir, { recursive: true, force: true }) + } + }) +}) + +// ── Dedup keys and flattening ───────────────────────────────────────── + +function makeRecord(overrides: Partial = {}): SessionAttributionRecord { + return { + sessionId: 'sess-1', + project: 'app', + repo: 'github.com/acme/widget', + prLinks: ['https://github.com/acme/widget/pull/3'], + commits: [ + { sha: 'a'.repeat(40), timestamp: '2026-01-01T10:30:00.000Z', inMain: true, wasReverted: false }, + ], + firstTimestamp: '2026-01-01T10:00:00.000Z', + lastTimestamp: '2026-01-01T11:00:00.000Z', + ...overrides, + } +} + +describe('attribution dedup keys', () => { + it('encodes commit state so a state transition mints a new key', () => { + const before = commitAttributionKey('sess-1', 'abc123', false, false) + const merged = commitAttributionKey('sess-1', 'abc123', true, false) + const reverted = commitAttributionKey('sess-1', 'abc123', true, true) + expect(new Set([before, merged, reverted]).size).toBe(3) + // Same state = same key (ledger dedupes repeats) + expect(commitAttributionKey('sess-1', 'abc123', true, false)).toBe(merged) + }) + + it('session key is stable for identical state and changes with commit state', () => { + const record = makeRecord() + expect(sessionAttributionKey(record)).toBe(sessionAttributionKey(makeRecord())) + + const mutated = makeRecord({ + commits: [{ sha: 'a'.repeat(40), timestamp: '2026-01-01T10:30:00.000Z', inMain: true, wasReverted: true }], + }) + expect(sessionAttributionKey(mutated)).not.toBe(sessionAttributionKey(record)) + }) + + it('session key changes when the window or project changes (ongoing sessions re-emit)', () => { + const base = makeRecord() + const grown = makeRecord({ lastTimestamp: '2026-01-01T12:00:00.000Z' }) + const renamed = makeRecord({ project: 'app-renamed' }) + expect(sessionAttributionKey(grown)).not.toBe(sessionAttributionKey(base)) + expect(sessionAttributionKey(renamed)).not.toBe(sessionAttributionKey(base)) + }) + + it('flattens one session item plus one item per commit', () => { + const items = flattenAttributionRecords([makeRecord()]) + expect(items).toHaveLength(2) + expect(items[0]).toMatchObject({ kind: 'session', commitCount: 1, endTimestamp: '2026-01-01T11:00:00.000Z' }) + expect(items[1]).toMatchObject({ kind: 'commit', sha: 'a'.repeat(40), inMain: true, wasReverted: false }) + expect(items.map(i => i.dedupKey)).toEqual([ + sessionAttributionKey(makeRecord()), + commitAttributionKey('sess-1', 'a'.repeat(40), true, false), + ]) + }) +}) + +// ── PR link sanitization ────────────────────────────────────────────── + +describe('sanitizePrLinks', () => { + it('keeps only https URLs shaped like org/repo/pull/N', () => { + expect(sanitizePrLinks([ + 'https://github.com/acme/widget/pull/12', + 'https://ghe.corp.example.com/team/svc/pull/3', // GHE hosts allowed + 'http://github.com/acme/widget/pull/12', // not https + 'javascript:alert(1)', // not a URL shape we accept + 'https://github.com/acme/widget/issues/12', // not a PR path + 'https://github.com/acme/widget/pull/12/files', // extra path segment + 'not a url at all', + '', + 'https://github.com/acme/widget/pull/notanumber', + ])).toEqual([ + 'https://ghe.corp.example.com/team/svc/pull/3', + 'https://github.com/acme/widget/pull/12', + ]) + }) + + it('rebuilds links from origin + pathname: userinfo, query, and fragment never survive', () => { + expect(sanitizePrLinks([ + 'https://alice:ghp_TOKEN@github.com/acme/widget/pull/5', + 'https://github.com/acme/widget/pull/7?notification_referrer_id=xyz', + 'https://github.com/acme/widget/pull/6#pullrequestreview-123', + ])).toEqual([ + 'https://github.com/acme/widget/pull/5', + 'https://github.com/acme/widget/pull/6', + 'https://github.com/acme/widget/pull/7', + ]) + }) + + it('dedupes links that collapse to the same rebuilt URL', () => { + expect(sanitizePrLinks([ + 'https://github.com/acme/widget/pull/9', + 'https://github.com/acme/widget/pull/9?ref=a', + 'https://github.com/acme/widget/pull/9#comment', + ])).toEqual(['https://github.com/acme/widget/pull/9']) + }) + + it('drops oversized inputs and caps the count per session', () => { + // A long referrer query is stripped, so the link survives... + const longQuery = `https://github.com/acme/widget/pull/1?x=${'a'.repeat(300)}` + expect(sanitizePrLinks([longQuery])).toEqual(['https://github.com/acme/widget/pull/1']) + // ...but pathological inputs beyond the input bound are dropped outright + const huge = `https://github.com/acme/widget/pull/1?x=${'a'.repeat(600)}` + expect(sanitizePrLinks([huge])).toEqual([]) + // And a rebuilt link that is itself oversized is dropped + const longPath = `https://github.com/${'o'.repeat(150)}/${'r'.repeat(80)}/pull/1` + expect(sanitizePrLinks([longPath])).toEqual([]) + + const many = Array.from({ length: 30 }, (_, i) => `https://github.com/acme/widget/pull/${i + 1}`) + expect(sanitizePrLinks(many)).toHaveLength(MAX_PR_LINKS_PER_SESSION) + }) +}) + +// ── OTLP payload ────────────────────────────────────────────────────── + +function attrMap(attributes: OtlpAttribute[]): Record { + return Object.fromEntries(attributes.map(a => [a.key, a.value])) +} + +describe('buildAttributionOtlpPayload', () => { + it('builds session and commit spans sharing the session traceId', () => { + const items = flattenAttributionRecords([makeRecord()]) + const payload = buildAttributionOtlpPayload(items) + + const resource = payload.resourceSpans[0]! + const resourceAttrs = attrMap(resource.resource.attributes) + expect(resourceAttrs['codeburn.attribution_methodology']).toEqual({ stringValue: 'timestamp-window' }) + expect(resourceAttrs['codeburn.device_id']).toBeDefined() + + const spans = resource.scopeSpans[0]!.spans + expect(spans).toHaveLength(2) + + const sessionSpan = spans.find(s => s.name === SESSION_ATTRIBUTION_SPAN_NAME)! + const commitSpan = spans.find(s => s.name === COMMIT_ATTRIBUTION_SPAN_NAME)! + expect(sessionSpan.traceId).toBe(deriveTraceId('sess-1')) + expect(commitSpan.traceId).toBe(deriveTraceId('sess-1')) + expect(sessionSpan.spanId).not.toBe(commitSpan.spanId) + + const sessionAttrs = attrMap(sessionSpan.attributes) + expect(sessionAttrs['ai.session_id']).toEqual({ stringValue: 'sess-1' }) + expect(sessionAttrs['ai.project']).toEqual({ stringValue: 'app' }) + expect(sessionAttrs['git.repo']).toEqual({ stringValue: 'github.com/acme/widget' }) + expect(sessionAttrs['git.commit_count']).toEqual({ intValue: '1' }) + expect(sessionAttrs['git.pr_links']).toEqual({ + arrayValue: { values: [{ stringValue: 'https://github.com/acme/widget/pull/3' }] }, + }) + // Session span carries the real window as its duration + expect(sessionSpan.startTimeUnixNano).toBe((BigInt(new Date('2026-01-01T10:00:00.000Z').getTime()) * 1_000_000n).toString()) + expect(sessionSpan.endTimeUnixNano).toBe((BigInt(new Date('2026-01-01T11:00:00.000Z').getTime()) * 1_000_000n).toString()) + + const commitAttrs = attrMap(commitSpan.attributes) + expect(commitAttrs['git.sha']).toEqual({ stringValue: 'a'.repeat(40) }) + expect(commitAttrs['git.in_main']).toEqual({ boolValue: true }) + expect(commitAttrs['git.was_reverted']).toEqual({ boolValue: false }) + expect(commitAttrs['git.repo']).toEqual({ stringValue: 'github.com/acme/widget' }) + }) + + it('omits git.repo when null and pr_links when empty', () => { + const items = flattenAttributionRecords([makeRecord({ repo: null, prLinks: [], commits: [] })]) + const payload = buildAttributionOtlpPayload(items) + const spans = payload.resourceSpans[0]!.scopeSpans[0]!.spans + expect(spans).toHaveLength(1) + const attrs = attrMap(spans[0]!.attributes) + expect(attrs['git.repo']).toBeUndefined() + expect(attrs['git.pr_links']).toBeUndefined() + expect(attrs['git.commit_count']).toEqual({ intValue: '0' }) + }) + + it('batches items by maxBatchSize', () => { + const items = flattenAttributionRecords([makeRecord(), makeRecord({ sessionId: 'sess-2' })]) + expect(batchAttributionItems(items, 3).map(b => b.length)).toEqual([3, 1]) + }) + + it('clamps span end time: never 0, never earlier than start + 1ms', () => { + // Session window ends BEFORE it starts (out-of-order provider timestamps) + const outOfOrder = flattenAttributionRecords([makeRecord({ + firstTimestamp: '2026-01-01T11:00:00.000Z', + lastTimestamp: '2026-01-01T10:00:00.000Z', + })]) + const span1 = buildAttributionOtlpPayload(outOfOrder).resourceSpans[0]!.scopeSpans[0]!.spans[0]! + expect(BigInt(span1.endTimeUnixNano)).toBe(BigInt(span1.startTimeUnixNano) + 1_000_000n) + + // Malformed end timestamp (toUnixNano -> 0) + const malformed = flattenAttributionRecords([makeRecord({ lastTimestamp: 'not-a-date' })]) + const span2 = buildAttributionOtlpPayload(malformed).resourceSpans[0]!.scopeSpans[0]!.spans[0]! + expect(span2.endTimeUnixNano).not.toBe('0') + expect(BigInt(span2.endTimeUnixNano)).toBe(BigInt(span2.startTimeUnixNano) + 1_000_000n) + }) +}) + +// ── Send + ledger pipeline ──────────────────────────────────────────── + +type MockResponse = { status: number; body?: unknown; headers?: Record } + +function startMockOtlp(responses: MockResponse[]): Promise<{ + url: string + server: Server + requests: Array<{ auth: string | undefined; body: unknown }> +}> { + const requests: Array<{ auth: string | undefined; body: unknown }> = [] + let idx = 0 + + return new Promise(resolve => { + const server = createServer((req, res) => { + let raw = '' + req.on('data', c => { raw += c }) + req.on('end', () => { + requests.push({ auth: req.headers.authorization, body: JSON.parse(raw || '{}') }) + const r = responses[Math.min(idx, responses.length - 1)]! + idx++ + res.writeHead(r.status, { 'Content-Type': 'application/json', ...r.headers }) + res.end(r.body !== undefined ? JSON.stringify(r.body) : '{}') + }) + }) + server.listen(0, '127.0.0.1', () => { + const addr = server.address() as { port: number } + resolve({ url: `http://127.0.0.1:${addr.port}/v1/traces`, server, requests }) + }) + }) +} + +let tmpDir: string +const originalHome = process.env.HOME +const originalXdgCache = process.env.XDG_CACHE_HOME + +beforeEach(async () => { + tmpDir = await mkdtemp(join(tmpdir(), 'codeburn-attr-push-')) + process.env.HOME = tmpDir + process.env.XDG_CACHE_HOME = join(tmpDir, '.cache') +}) + +afterEach(async () => { + process.env.HOME = originalHome + if (originalXdgCache === undefined) delete process.env.XDG_CACHE_HOME + else process.env.XDG_CACHE_HOME = originalXdgCache + await rm(tmpDir, { recursive: true, force: true }) +}) + +describe('sendAttributionBatches + collectUnsentAttribution', () => { + it('sends attribution spans, ledgers dedup keys, and filters them on the next collect', async () => { + const { sendAttributionBatches, collectUnsentAttribution } = await import('../src/sync/push.js') + const { readLedger } = await import('../src/sync/ledger.js') + + const record = makeRecord() + const first = collectUnsentAttribution([record]) + expect(first.unsent).toHaveLength(2) + + const mock = await startMockOtlp([{ status: 200 }]) + try { + const result = await sendAttributionBatches({ + endpoint: mock.url, + accessToken: 'token-1', + batches: [first.unsent], + }) + + expect(result.outcome).toBe('complete') + expect(result.totalSent).toBe(2) + expect(result.totalCostSent).toBe(0) + expect(mock.requests).toHaveLength(1) + expect(mock.requests[0]!.auth).toBe('Bearer token-1') + + const body = mock.requests[0]!.body as { resourceSpans: Array<{ scopeSpans: Array<{ spans: Array<{ name: string }> }> }> } + const names = body.resourceSpans[0]!.scopeSpans[0]!.spans.map(s => s.name).sort() + expect(names).toEqual([COMMIT_ATTRIBUTION_SPAN_NAME, SESSION_ATTRIBUTION_SPAN_NAME]) + + const ledgered = readLedger().map(e => e.key).sort() + expect(ledgered).toEqual(first.unsent.map(i => i.dedupKey).sort()) + + // Identical state on the next push: nothing unsent + expect(collectUnsentAttribution([record]).unsent).toEqual([]) + + // State transition (commit reverted): the changed facts re-send + const mutated = makeRecord({ + commits: [{ sha: 'a'.repeat(40), timestamp: '2026-01-01T10:30:00.000Z', inMain: true, wasReverted: true }], + }) + const after = collectUnsentAttribution([mutated]) + expect(after.unsent.map(i => i.kind).sort()).toEqual(['commit', 'session']) + } finally { + mock.server.close() + } + }) + + it('retracts a session span when its commit migrates to a tighter-window session', async () => { + const { sendAttributionBatches, collectUnsentAttribution } = await import('../src/sync/push.js') + + const sha = 'b'.repeat(40) + const commit = { sha, timestamp: '2026-01-01T10:30:00.000Z', inMain: true, wasReverted: false } + + // Push 1: session A (broad window) owns the commit + const push1 = collectUnsentAttribution([makeRecord({ sessionId: 'sess-A', prLinks: [], commits: [commit] })]) + expect(push1.unsent).toHaveLength(2) + const mock = await startMockOtlp([{ status: 200 }]) + try { + await sendAttributionBatches({ endpoint: mock.url, accessToken: 't', batches: [push1.unsent] }) + + // Push 2: a later-parsed tighter session B now wins the commit; A is empty + const push2 = collectUnsentAttribution([ + makeRecord({ sessionId: 'sess-A', prLinks: [], commits: [] }), // loser: retraction candidate + makeRecord({ sessionId: 'sess-B', prLinks: [], commits: [commit] }), // winner + ]) + + // A re-emits with commit_count 0 (retraction), B emits session + commit + const kinds = push2.unsent.map(i => `${i.sessionId}:${i.kind}`).sort() + expect(kinds).toEqual(['sess-A:session', 'sess-B:commit', 'sess-B:session']) + const retraction = push2.unsent.find(i => i.sessionId === 'sess-A')! + expect(retraction.commitCount).toBe(0) + expect(retraction.dedupKey).not.toBe(push1.unsent.find(i => i.kind === 'session')!.dedupKey) + + // A session that was NEVER sent stays excluded when empty + const neverSent = collectUnsentAttribution([ + makeRecord({ sessionId: 'sess-never', prLinks: [], commits: [] }), + ]) + expect(neverSent.unsent).toEqual([]) + } finally { + mock.server.close() + } + }) + + it('does not ledger on server error', async () => { + const { sendAttributionBatches } = await import('../src/sync/push.js') + const { readLedger } = await import('../src/sync/ledger.js') + + const items = flattenAttributionRecords([makeRecord()]) + const mock = await startMockOtlp([{ status: 500 }]) + try { + const result = await sendAttributionBatches({ + endpoint: mock.url, + accessToken: 'token-1', + batches: [items], + }) + expect(result.outcome).toBe('server-error') + expect(readLedger()).toEqual([]) + } finally { + mock.server.close() + } + }) +})