Skip to content

fix: reconcile Hermes attribution per bucket without global equality #197

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pitimon merged 2 commits into from
Aug 24, 2026
Merged

fix: reconcile Hermes attribution per bucket without global equality #197

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
65a6b24
fix: reconcile Hermes attribution per bucket without global equality
Aug 24, 2026
409650a
fix: include conversation_count in hermes reconciliation field compar…
Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 16 additions & 16 deletions openwiki-facts/source-facts.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,67 +255,67 @@
},
{
"name": "parseHermesIncremental",
"evidence": "src/lib/rollout.js:3699"
"evidence": "src/lib/rollout.js:3757"
},
{
"name": "parseKiroCliIncremental",
"evidence": "src/lib/rollout.js:4403"
"evidence": "src/lib/rollout.js:4461"
},
{
"name": "parseKimiIncremental",
"evidence": "src/lib/rollout.js:5048"
"evidence": "src/lib/rollout.js:5106"
},
{
"name": "parseKimiCodeIncremental",
"evidence": "src/lib/rollout.js:5239"
"evidence": "src/lib/rollout.js:5297"
},
{
"name": "parseCodebuddyIncremental",
"evidence": "src/lib/rollout.js:5463"
"evidence": "src/lib/rollout.js:5521"
},
{
"name": "parseRoocodeIncremental",
"evidence": "src/lib/rollout.js:5966"
"evidence": "src/lib/rollout.js:6024"
},
{
"name": "parseZedIncremental",
"evidence": "src/lib/rollout.js:6283"
"evidence": "src/lib/rollout.js:6341"
},
{
"name": "parseGooseIncremental",
"evidence": "src/lib/rollout.js:6600"
"evidence": "src/lib/rollout.js:6658"
},
{
"name": "parseDroidIncremental",
"evidence": "src/lib/rollout.js:6998"
"evidence": "src/lib/rollout.js:7056"
},
{
"name": "parseKilocodeIncremental",
"evidence": "src/lib/rollout.js:7245"
"evidence": "src/lib/rollout.js:7303"
},
{
"name": "parseOmpIncremental",
"evidence": "src/lib/rollout.js:7381"
"evidence": "src/lib/rollout.js:7439"
},
{
"name": "parsePiIncremental",
"evidence": "src/lib/rollout.js:7634"
"evidence": "src/lib/rollout.js:7692"
},
{
"name": "parseCraftIncremental",
"evidence": "src/lib/rollout.js:7918"
"evidence": "src/lib/rollout.js:7976"
},
{
"name": "parseCopilotIncremental",
"evidence": "src/lib/rollout.js:8282"
"evidence": "src/lib/rollout.js:8340"
},
{
"name": "parseGrokBuildIncremental",
"evidence": "src/lib/rollout.js:8748"
"evidence": "src/lib/rollout.js:8806"
},
{
"name": "parseAntigravityIncremental",
"evidence": "src/lib/rollout.js:8920"
"evidence": "src/lib/rollout.js:8978"
}
]
}
Expand Down
126 changes: 92 additions & 34 deletions src/lib/rollout.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3446,6 +3446,7 @@ const HERMES_TOKEN_TOTAL_FIELDS = [
"reasoning_output_tokens",
"total_tokens",
"billable_total_tokens",
"conversation_count",
];

function hermesRowTotals(row) {
Expand Down Expand Up @@ -3570,12 +3571,16 @@ async function appendHermesReconciliationRows({ queuePath, queueStatePath, canon
}
}

// One-time, fail-closed correction for the v0.39.48 adoption gate. That gate
// One-time, per-bucket correction for the v0.39.48 adoption gate. That gate
// intentionally avoided replaying legacy aggregate snapshots, which preserved
// the grand total but left already-counted mixed-model usage on sessions.model.
// Rebuild only when the authoritative Hermes model rows have exactly the same
// token total as the existing Hermes accumulator; otherwise leave data intact
// and let the normal incremental path catch up before retrying next sync.
// Compares each canonical (model, half-hour) bucket against the existing
// accumulator instead of requiring global grand-total equality: production
// installs cannot satisfy exact equality (source retention deletes ingested
// sessions, live sessions grow between reads). Differing canonical buckets are
// appended as latest-wins correction rows after the consumed upload offset;
// existing buckets with no canonical counterpart that predate retained source
// coverage are zeroed out explicitly.
async function reconcileHermesMixedModelUsage({ dbPaths, cursors, queuePath, queueStatePath, hermesState, updatedAt, sqliteOptions }) {
if (!cursors || typeof cursors !== "object") return { status: "invalid_cursor" };
const migrations = cursors.migrations && typeof cursors.migrations === "object" ? cursors.migrations : {};
Expand Down Expand Up @@ -3604,13 +3609,22 @@ async function reconcileHermesMixedModelUsage({ dbPaths, cursors, queuePath, que
}

const hourlyState = normalizeHourlyState(cursors.hourly);
const existingTotals = initTotals();
for (const [key, bucket] of Object.entries(hourlyState.buckets || {})) {
if (parseBucketKey(key).source === "hermes") addTotals(existingTotals, bucket?.totals);
// Retained source coverage boundary: the oldest session start still present in
// any authoritative database. Existing buckets whose hour starts before this
// boundary cannot have a surviving source row, so they are provably orphans.
let retentionBoundaryEpochSec = null;
for (const { rows } of rowsBySource) {
for (const row of rows) {
const startedAt = Number(row.started_at);
if (Number.isFinite(startedAt) && startedAt > 0) {
retentionBoundaryEpochSec = retentionBoundaryEpochSec == null
? startedAt
: Math.min(retentionBoundaryEpochSec, startedAt);
}
}
}

const canonicalBuckets = {};
const authoritativeTotals = initTotals();
for (const { rows } of rowsBySource) {
for (const row of rows) {
const totals = hermesRowTotals(row);
Expand All @@ -3621,46 +3635,82 @@ async function reconcileHermesMixedModelUsage({ dbPaths, cursors, queuePath, que
const key = bucketKey("hermes", model, hourStart);
const bucket = canonicalBuckets[key] ||= { totals: initTotals(), queuedKey: null };
addTotals(bucket.totals, totals);
addTotals(authoritativeTotals, totals);
}
}

if (!sameHermesTokenTotals(existingTotals, authoritativeTotals)) {
cursors.migrations ||= {};
cursors.migrations[HERMES_MIXED_MODEL_RECONCILIATION_MIGRATION_KEY] = {
status: "blocked_total_mismatch",
updatedAt,
existingTotalTokens: existingTotals.total_tokens,
authoritativeTotalTokens: authoritativeTotals.total_tokens,
};
return cursors.migrations[HERMES_MIXED_MODEL_RECONCILIATION_MIGRATION_KEY];
function bucketHourStartToEpochSec(hourStart) {
const parsedMs = Date.parse(`${hourStart}Z`.replace(/Z+$/, "Z"));
return Number.isFinite(parsedMs) ? parsedMs / 1000 : null;
}

for (const key of Object.keys(hourlyState.buckets || {})) {
if (parseBucketKey(key).source === "hermes") delete hourlyState.buckets[key];
}
for (const key of Object.keys(hourlyState.groupQueued || {})) {
if (key.startsWith("hermes|")) delete hourlyState.groupQueued[key];
}
const canonicalRows = [];
const existingHermesKeys = new Set(Object.keys(hourlyState.buckets || {})
.filter((key) => parseBucketKey(key).source === "hermes"));

const correctionRows = [];
let bucketsCorrected = 0;
let bucketsSkipped = 0;
let orphansZeroed = 0;
let stalePlacementsZeroed = 0;

// Corrections for canonical buckets that differ from the existing accumulator.
for (const [key, bucket] of Object.entries(canonicalBuckets)) {
const { model, hourStart } = parseBucketKey(key);
const existingBucket = hourlyState.buckets[key];
if (existingBucket && sameHermesTokenTotals(existingBucket?.totals, bucket.totals)) {
bucketsSkipped += 1;
continue;
}
bucket.queuedKey = totalsKey(bucket.totals);
hourlyState.buckets[key] = bucket;
canonicalRows.push(JSON.stringify({
correctionRows.push(JSON.stringify({
source: "hermes",
model,
hour_start: hourStart,
...bucket.totals,
}));
bucketsCorrected += 1;
}

// Zero out existing buckets with no canonical counterpart. Outside retained
// coverage they are provably orphans (source rows deleted after ingestion).
// Inside coverage they are stale placements: the incremental path spread a
// session's cumulative totals across observation-time buckets while the
// canonical view places them at the session's current last_seen bucket.
// Latest-wins upserts make the zero row retire the stale placement without
// rewriting queue history; unfinished sessions re-emit growth at their new
// bucket on the next incremental sync.
for (const key of existingHermesKeys) {
if (canonicalBuckets[key]) continue;
const { model, hourStart } = parseBucketKey(key);
const existingBucketTotal = Number(hourlyState.buckets[key]?.totals?.total_tokens || 0);
if (existingBucketTotal === 0) continue;
const hourStartSec = bucketHourStartToEpochSec(hourStart);
const provablyOrphaned = hourStartSec != null
&& retentionBoundaryEpochSec != null
&& hourStartSec < retentionBoundaryEpochSec;
const zeroTotals = initTotals();
hourlyState.buckets[key] = { totals: zeroTotals, queuedKey: totalsKey(zeroTotals) };
correctionRows.push(JSON.stringify({
source: "hermes",
model,
hour_start: hourStart,
...zeroTotals,
}));
if (provablyOrphaned) {
orphansZeroed += 1;
} else {
stalePlacementsZeroed += 1;
}
}

await appendHermesReconciliationRows({
queuePath,
queueStatePath,
canonicalRows,
updatedAt,
});
if (correctionRows.length > 0) {
await appendHermesReconciliationRows({
queuePath,
queueStatePath,
canonicalRows: correctionRows,
updatedAt,
});
}

const nextHermesState = hermesState && typeof hermesState === "object" ? hermesState : {};
for (const { profileName, rows } of rowsBySource) {
Expand All @@ -3677,13 +3727,21 @@ async function reconcileHermesMixedModelUsage({ dbPaths, cursors, queuePath, que
);
}
}
for (const key of Object.keys(hourlyState.groupQueued || {})) {
if (key.startsWith("hermes|")) delete hourlyState.groupQueued[key];
}
cursors.hourly = hourlyState;
cursors.migrations ||= {};
cursors.migrations[HERMES_MIXED_MODEL_RECONCILIATION_MIGRATION_KEY] = {
status: "applied",
appliedAt: updatedAt,
totalTokens: authoritativeTotals.total_tokens,
bucketsRebuilt: canonicalRows.length,
totalTokens: Object.values(canonicalBuckets).reduce(
(sum, bucket) => sum + Number(bucket.totals.total_tokens || 0), 0),
bucketsRebuilt: Object.keys(canonicalBuckets).length,
bucketsCorrected,
bucketsSkipped,
orphansZeroed,
stalePlacementsZeroed,
};
return cursors.migrations[HERMES_MIXED_MODEL_RECONCILIATION_MIGRATION_KEY];
}
Expand Down
Loading