fix(analytics): avoid bridge double counting in native collectors

This commit is contained in:
Tam Nhu Tran
2026-04-26 18:19:41 -04:00
parent fe7cc6f9ba
commit e95b354297
6 changed files with 75 additions and 19 deletions
@@ -16,6 +16,10 @@ interface CodexTokenSnapshot {
outputTokens: number;
}
function isCliProxyBackedProvider(value: string | undefined): boolean {
return value === 'cliproxy' || value === 'ccs_runtime';
}
function isObject(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null;
}
@@ -123,7 +127,7 @@ async function parseRolloutFile(
!isObject(parsed.payload) ||
parsed.payload.type !== 'token_count' ||
!sessionId ||
(modelProvider === 'cliproxy' && !includeCliproxySessions)
(isCliProxyBackedProvider(modelProvider) && !includeCliproxySessions)
) {
continue;
}
@@ -20,6 +20,8 @@ interface DroidSessionMetadata {
version?: string;
}
const DROID_COST_BATCH_SIZE = 1000;
function isObject(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null;
}
@@ -149,11 +151,19 @@ export async function scanDroidNativeUsageEntries(
const costsDbPath = path.join(factoryDir, 'costs.db');
if (!fs.existsSync(costsDbPath)) return [];
const rows =
(await query(
const rows: Array<Record<string, unknown>> = [];
let offset = 0;
while (true) {
const batch = await query(
costsDbPath,
'SELECT session_id, timestamp, input_tokens, output_tokens FROM costs ORDER BY timestamp ASC;'
)) ?? [];
`SELECT session_id, timestamp, input_tokens, output_tokens FROM costs ` +
`ORDER BY timestamp ASC LIMIT ${DROID_COST_BATCH_SIZE} OFFSET ${offset};`
);
if (!batch.length) break;
rows.push(...batch);
if (batch.length < DROID_COST_BATCH_SIZE) break;
offset += batch.length;
}
if (!rows.length) return [];
const metadata = loadSessionMetadata(factoryDir);
@@ -168,11 +178,11 @@ export async function scanDroidNativeUsageEntries(
if (!sessionId || !timestamp || inputTokens === null || outputTokens === null) continue;
const session = metadata.get(sessionId);
if (
session &&
!options.includeCcsManagedSessions &&
isCcsManagedSelector(session.rawSelector)
) {
if (!session) {
continue;
}
if (!options.includeCcsManagedSessions && isCcsManagedSelector(session.rawSelector)) {
continue;
}
@@ -181,11 +191,11 @@ export async function scanDroidNativeUsageEntries(
outputTokens,
cacheCreationTokens: 0,
cacheReadTokens: 0,
model: session?.model ?? 'unknown-droid-model',
model: session.model,
sessionId,
timestamp,
projectPath: session?.projectPath ?? '/',
version: session?.version,
projectPath: session.projectPath,
version: session.version,
target: 'droid',
});
}
+5 -6
View File
@@ -13,10 +13,7 @@ function isCommandMissing(error: unknown): boolean {
return nodeError.code === 'ENOENT' || /not found/i.test(nodeError.message);
}
export async function querySqliteJson(
dbPath: string,
sql: string
): Promise<SqliteJsonRow[] | null> {
export async function querySqliteJson(dbPath: string, sql: string): Promise<SqliteJsonRow[]> {
if (!fs.existsSync(dbPath)) {
return [];
}
@@ -34,8 +31,10 @@ export async function querySqliteJson(
return Array.isArray(parsed) ? (parsed as SqliteJsonRow[]) : [];
} catch (error) {
if (isCommandMissing(error)) {
return null;
throw new Error('sqlite3 command not available');
}
return [];
const message = error instanceof Error ? error.message : String(error);
throw new Error(`sqlite3 query failed for ${dbPath}: ${message}`);
}
}
@@ -173,4 +173,15 @@ describe('codex native usage collector', () => {
expect(entries).toHaveLength(0);
});
it('also skips ccs_runtime-backed codex bridge sessions by default', async () => {
writeCodexRollout(tempRoot, { modelProvider: 'ccs_runtime' });
const entries = await scanCodexNativeUsageEntries({
env: { CODEX_HOME: tempRoot },
homeDir: tempRoot,
});
expect(entries).toHaveLength(0);
});
});
@@ -131,4 +131,21 @@ describe('droid native usage collector', () => {
expect(entries).toHaveLength(0);
});
it('surfaces sqlite failures instead of flattening them into an empty result', async () => {
writeDroidGlobalSettings(tempRoot);
writeDroidSessionFixture(tempRoot);
const querySqliteJson: DroidSqliteQuery = async () => {
throw new Error('sqlite blew up');
};
await expect(
scanDroidNativeUsageEntries({
env: { CCS_HOME: tempRoot },
homeDir: tempRoot,
querySqliteJson,
})
).rejects.toThrow('sqlite blew up');
});
});
@@ -280,4 +280,19 @@ describe('usage aggregator native runtime integration', () => {
});
expect(daily[0].modelsUsed).not.toContain('gpt-5');
});
it('ignores ccs_runtime-backed codex bridge rollouts to avoid double counting', async () => {
writeCodexFixture('ccs_runtime');
aggregator.clearUsageCache();
const daily = await aggregator.getCachedDailyData();
expect(daily).toHaveLength(1);
expect(daily[0]).toMatchObject({
inputTokens: 150,
outputTokens: 50,
cacheReadTokens: 5,
});
expect(daily[0].modelsUsed).not.toContain('gpt-5');
});
});