mirror of
https://github.com/AgentSeal/codeburn.git
synced 2026-08-04 22:01:25 +00:00
feat(sync): push git attribution spans with --attribution
Expose the yield session-to-commit correlation through codeburn sync so backends can join AI usage to git activity without local git hooks. - yield: export normalizeRemoteUrl (host/org/repo; credentials, ports, and .git stripped) and computeAttributionRecords, which reuses the exact repo-grouping + tightest-window attribution from computeYield (extracted into a shared buildRepoGroups) and joins in the normalized origin remote and session prLinks. - otlp: two new span types sharing the session traceId — codeburn.session.attribution (git.repo, git.pr_links, git.commit_count) and codeburn.commit (git.sha, git.in_main, git.was_reverted). Resource attribute codeburn.attribution_methodology=timestamp-window marks the attribution as inferred. - push: generic send core reused by usage and attribution batches. Dedup keys encode mutable state (inMain/wasReverted), so a state transition re-sends the updated fact while identical states dedupe via the existing sent-ledger. - cli: opt-in --attribution flag on sync push (dry-run aware); commits in repos with no network remote are never sent. AI-Origin: human
This commit is contained in:
Andrew Lee
parent
2de4d100bf
commit
1bf7206842
6 changed files with 941 additions and 41 deletions
|
|
@ -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,32 @@ 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 by `(git.repo, git.sha)`.
|
||||
|
||||
With `--attribution`, normalized repo remote URLs, commit SHAs, commit timestamps, and PR URLs leave your machine. Commits in repos with no network remote are never sent (there is nothing to join them to). Without the flag, none of this is sent.
|
||||
|
||||
### What is NOT sent
|
||||
|
||||
- **Prompts** — your actual messages to AI are never included
|
||||
|
|
@ -102,7 +131,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
|
||||
|
||||
|
|
|
|||
|
|
@ -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, 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 <period>', '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 <url>` 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<ReturnType<typeof collectUnsentAttribution>>['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,15 @@ 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 commits = attributionUnsent.filter(i => i.kind === 'commit').length
|
||||
const sessions = attributionUnsent.filter(i => i.kind === 'session').length
|
||||
process.stderr.write(`[dry-run] Attribution: ${attributionTotal} facts total, would push ${attributionUnsent.length} (${sessions} sessions, ${commits} commits)\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 +321,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 +345,36 @@ 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') {
|
||||
const attrBatches = batchAttributionItems(attributionUnsent, 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)
|
||||
}
|
||||
} 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) {
|
||||
process.stderr.write(` Attribution: ${attrResult.totalSent} facts synced${attrResult.totalRejected > 0 ? `, ${attrResult.totalRejected} rejected (will retry)` : ''}\n`)
|
||||
}
|
||||
if (result.totalRejected > 0) {
|
||||
process.stderr.write(` ${result.totalRejected} spans rejected (will retry on next push)\n`)
|
||||
}
|
||||
|
|
@ -337,7 +384,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) {
|
||||
|
|
|
|||
156
src/sync/otlp.ts
156
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,158 @@ 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()
|
||||
return `attr:s:${record.sessionId}:${stateHash([record.repo ?? '', ...record.prLinks, ...commitStates])}`
|
||||
}
|
||||
|
||||
/** 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)
|
||||
const endNano = item.endTimestamp
|
||||
? toUnixNano(item.endTimestamp)
|
||||
: (BigInt(startNano) + 1_000_000n).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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<PushResult> {
|
||||
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<T> = Omit<SendBatchesOptions, 'batches'> & {
|
||||
batches: T[][]
|
||||
buildPayload: (batch: T[]) => OtlpPayload
|
||||
toOutbound: (item: T) => OutboundItem
|
||||
}
|
||||
|
||||
async function sendBatchesCore<T>(opts: SendBatchesCoreOptions<T>): Promise<PushResult> {
|
||||
assertHttps(opts.endpoint, 'Traces endpoint')
|
||||
const log = opts.log ?? (() => {})
|
||||
const sleep = opts.sleep ?? ((ms: number) => new Promise<void>(r => setTimeout(r, ms)))
|
||||
|
|
@ -107,7 +133,7 @@ export async function sendBatches(opts: SendBatchesOptions): Promise<PushResult>
|
|||
|
||||
// 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<PushResult>
|
|||
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,33 @@ export async function sendBatches(opts: SendBatchesOptions): Promise<PushResult>
|
|||
return { outcome: 'complete', totalSent, totalRejected, totalCostSent, totalWaitMs }
|
||||
}
|
||||
|
||||
/** Flatten attribution records into items and filter out already-sent ones. */
|
||||
export function collectUnsentAttribution(records: SessionAttributionRecord[]): {
|
||||
allItems: AttributionItem[]
|
||||
unsent: AttributionItem[]
|
||||
} {
|
||||
const allItems = flattenAttributionRecords(records)
|
||||
const sent = ledgerKeySet()
|
||||
const unsent = allItems.filter(i => !sent.has(i.dedupKey))
|
||||
return { allItems, unsent }
|
||||
}
|
||||
|
||||
export interface SendAttributionBatchesOptions extends Omit<SendBatchesOptions, 'batches'> {
|
||||
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<PushResult> {
|
||||
return sendBatchesCore({
|
||||
...opts,
|
||||
buildPayload: buildAttributionOtlpPayload,
|
||||
toOutbound: item => ({ key: item.dedupKey, ts: item.timestamp, costUSD: 0 }),
|
||||
})
|
||||
}
|
||||
|
||||
export { batchCalls }
|
||||
|
|
|
|||
203
src/yield.ts
203
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,63 @@ 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.
|
||||
*/
|
||||
export function normalizeRemoteUrl(url: string): string | null {
|
||||
const trimmed = url.trim()
|
||||
if (!trimmed) return null
|
||||
|
||||
let host: string
|
||||
let path: string
|
||||
|
||||
const scpLike = /^(?:[^@/]+@)?([^:/]+):(?!\/\/)(.+)$/.exec(trimmed)
|
||||
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 if (scpLike) {
|
||||
// scp-like syntax: [user@]host:path — not parseable by URL
|
||||
host = scpLike[1]
|
||||
path = scpLike[2]
|
||||
} else {
|
||||
// Bare local path (or something unrecognizable) — no remote identity.
|
||||
return null
|
||||
}
|
||||
|
||||
const cleanPath = path
|
||||
.replace(/^\/+/, '')
|
||||
.replace(/\/+$/, '')
|
||||
.replace(/\.git$/, '')
|
||||
if (!cleanPath) return null
|
||||
|
||||
return `${host.toLowerCase()}/${cleanPath}`
|
||||
}
|
||||
|
||||
/** `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 +352,35 @@ function categorizeSession(
|
|||
return { category: 'abandoned', commitCount: commits.length }
|
||||
}
|
||||
|
||||
export async function computeYield(range: DateRange, cwd: string, provider: string = 'all'): Promise<YieldSummary> {
|
||||
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[]
|
||||
/** 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<string, RepoGroup> {
|
||||
const repoIdentityCache = new Map<string, RepoIdentity | null>()
|
||||
|
||||
// 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<string, RepoGroup>()
|
||||
for (const project of projects) {
|
||||
const projectIdentity = project.projectPath
|
||||
|
|
@ -342,6 +399,7 @@ export async function computeYield(range: DateRange, cwd: string, provider: stri
|
|||
: getCommitsInRange(identity.gitDir, range.start, range.end, getMainBranch(identity.gitDir)),
|
||||
sessions: [],
|
||||
projectNames: [],
|
||||
gitDir: identity?.gitDir ?? null,
|
||||
}
|
||||
repoGroups.set(groupKey, group)
|
||||
}
|
||||
|
|
@ -351,6 +409,23 @@ export async function computeYield(range: DateRange, cwd: string, provider: stri
|
|||
}
|
||||
}
|
||||
|
||||
return repoGroups
|
||||
}
|
||||
|
||||
export async function computeYield(range: DateRange, cwd: string, provider: string = 'all'): Promise<YieldSummary> {
|
||||
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 +521,87 @@ 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
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 PR links.
|
||||
*
|
||||
* Inclusion rules:
|
||||
* - Sessions with neither attributed commits nor PR links are omitted.
|
||||
* - Commits are only included when the repo has a normalized remote; a SHA
|
||||
* without a repo identity has no server-side meaning. Such sessions still
|
||||
* emit a record when they carry PR links (the PR URL embeds the repo).
|
||||
*
|
||||
* 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
|
||||
const attributions = attributeCommits(group.sessions, group.commits)
|
||||
|
||||
for (const [index, session] of group.sessions.entries()) {
|
||||
if (!session.firstTimestamp) continue
|
||||
|
||||
const attributedCommits = remote ? (attributions[index]?.commits ?? []) : []
|
||||
const prLinks = session.prLinks ?? []
|
||||
if (attributedCommits.length === 0 && prLinks.length === 0) continue
|
||||
|
||||
records.push({
|
||||
sessionId: session.sessionId,
|
||||
project: group.projectNames[index] ?? session.project,
|
||||
repo: remote,
|
||||
prLinks: [...prLinks].sort(),
|
||||
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
|
||||
}
|
||||
|
|
|
|||
456
tests/sync-attribution.test.ts
Normal file
456
tests/sync-attribution.test.ts
Normal file
|
|
@ -0,0 +1,456 @@
|
|||
/**
|
||||
* 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,
|
||||
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, string> = {}): 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>): 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()
|
||||
})
|
||||
})
|
||||
|
||||
// ── 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 sessions with no commits and no PR links', 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
|
||||
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([])
|
||||
} 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('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)
|
||||
|
||||
expect(records).toHaveLength(1)
|
||||
expect(records[0]!.sessionId).toBe('sess-tight')
|
||||
expect(records[0]!.commits).toHaveLength(1)
|
||||
} finally {
|
||||
await rm(repoDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
// ── Dedup keys and flattening ─────────────────────────────────────────
|
||||
|
||||
function makeRecord(overrides: Partial<SessionAttributionRecord> = {}): 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('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),
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
// ── OTLP payload ──────────────────────────────────────────────────────
|
||||
|
||||
function attrMap(attributes: OtlpAttribute[]): Record<string, unknown> {
|
||||
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])
|
||||
})
|
||||
})
|
||||
|
||||
// ── Send + ledger pipeline ────────────────────────────────────────────
|
||||
|
||||
type MockResponse = { status: number; body?: unknown; headers?: Record<string, string> }
|
||||
|
||||
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('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()
|
||||
}
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue