fix(stats): preserved service-tier across incremental parseSessionFile offsets
Stats sync depends on service_tier_change entries to derive priority premium requests since 69043d307 stopped folding priority-tier into per-message premiumRequests. parseSessionFile(path, fromOffset) initialized currentServiceTier to undefined per call, so an incremental resume past the tier change attributed subsequent priority OpenAI replies with premiumRequests: 0. When fromOffset > 0, parseSessionFile now replays bytes[0..start] purely to seed currentServiceTier from the latest service_tier_change before parsing the unprocessed tail. Bytes are already loaded, so no extra I/O.
This commit is contained in:
@@ -2,6 +2,10 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed incremental `parseSessionFile(path, fromOffset)` losing the active service tier when resuming past a `service_tier_change` entry, so priority OpenAI replies appended after the offset are now credited with `premiumRequests: 1` (regression introduced by 13f59162e which stopped folding priority-tier into per-message premium counts)
|
||||
|
||||
## [15.0.1] - 2026-05-14
|
||||
### Breaking Changes
|
||||
|
||||
|
||||
@@ -167,6 +167,18 @@ function parseSessionEntriesLenient(bytes: Uint8Array): { entries: SessionEntry[
|
||||
/**
|
||||
* Parse a session file and extract all assistant message stats.
|
||||
* Uses incremental reading with offset tracking.
|
||||
*
|
||||
* Service-tier carry-over: `currentServiceTier` is a session-scoped piece of
|
||||
* state derived from `service_tier_change` entries that affects whether
|
||||
* subsequent OpenAI assistant replies count as premium requests. Incremental
|
||||
* syncs that resume past the most-recent tier change would otherwise lose
|
||||
* that state and silently record `premiumRequests = 0` for priority traffic
|
||||
* (the coding-agent stopped folding the tier into `usage.premiumRequests`
|
||||
* after 13f59162e — the parser is now the sole source of truth). When
|
||||
* `fromOffset > 0` we therefore replay the bytes preceding `fromOffset`
|
||||
* purely to recover the latest service-tier value before parsing the
|
||||
* unprocessed tail. The full file is already in memory, so the extra pass
|
||||
* costs one in-memory JSONL scan with no I/O.
|
||||
*/
|
||||
export interface ParseSessionResult {
|
||||
stats: MessageStats[];
|
||||
@@ -174,7 +186,6 @@ export interface ParseSessionResult {
|
||||
userLinks: UserMessageLink[];
|
||||
newOffset: number;
|
||||
}
|
||||
|
||||
export async function parseSessionFile(sessionPath: string, fromOffset = 0): Promise<ParseSessionResult> {
|
||||
let bytes: Uint8Array;
|
||||
try {
|
||||
@@ -193,6 +204,15 @@ export async function parseSessionFile(sessionPath: string, fromOffset = 0): Pro
|
||||
const unprocessed = bytes.subarray(start);
|
||||
const { entries, read } = parseSessionEntriesLenient(unprocessed);
|
||||
let currentServiceTier: ServiceTier | undefined;
|
||||
if (start > 0) {
|
||||
// Replay the already-ingested prefix to recover the last tier value.
|
||||
// Only `service_tier_change` entries are consulted; assistant/user
|
||||
// stats from the prefix have been emitted by an earlier sync pass.
|
||||
const { entries: prefixEntries } = parseSessionEntriesLenient(bytes.subarray(0, start));
|
||||
for (const prev of prefixEntries) {
|
||||
if (isServiceTierChange(prev)) currentServiceTier = prev.serviceTier ?? undefined;
|
||||
}
|
||||
}
|
||||
for (const entry of entries) {
|
||||
if (isServiceTierChange(entry)) {
|
||||
currentServiceTier = entry.serviceTier ?? undefined;
|
||||
|
||||
@@ -6,6 +6,7 @@ import * as path from "node:path";
|
||||
import { getAgentDir, getSessionsDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
|
||||
import { syncAllSessions } from "../src/aggregator";
|
||||
import { closeDb, getOverallStats, getRecentRequests } from "../src/db";
|
||||
import { parseSessionFile } from "../src/parser";
|
||||
|
||||
const originalConfigDir = process.env.PI_CONFIG_DIR;
|
||||
const originalAgentDir = getAgentDir();
|
||||
@@ -184,4 +185,29 @@ describe("priority service-tier premium-request backfill", () => {
|
||||
expect(request?.entryId).toBe("c1");
|
||||
expect(request?.usage.premiumRequests).toBe(1);
|
||||
});
|
||||
|
||||
it("carries the active service tier across incremental parseSessionFile calls", async () => {
|
||||
// Session opens with priority, then a reply lands after we've already
|
||||
// advanced `fromOffset` past the tier-change entry. The parser must
|
||||
// replay the prefix and still attribute the reply as a premium request.
|
||||
const sessionFile = await writeSession("--tmp--proj", "04.jsonl", {
|
||||
lines: [
|
||||
{ type: "session", version: 1, id: "s4", timestamp: new Date().toISOString(), cwd: "/tmp/proj" },
|
||||
{ type: "service_tier_change", id: "stc", timestamp: new Date().toISOString(), serviceTier: "priority" },
|
||||
assistantEntry({ id: "d1", provider: "openai" }),
|
||||
],
|
||||
});
|
||||
|
||||
// Locate the byte offset immediately past the `service_tier_change`
|
||||
// line so the second sync's `fromOffset` lands between the tier entry
|
||||
// and the assistant reply — the exact window where the regression hid.
|
||||
const bytes = await fs.readFile(sessionFile);
|
||||
const tierLineEnd = bytes.indexOf(0x0a, bytes.indexOf(Buffer.from("service_tier_change"))) + 1;
|
||||
expect(tierLineEnd).toBeGreaterThan(0);
|
||||
|
||||
const second = await parseSessionFile(sessionFile, tierLineEnd);
|
||||
expect(second.stats).toHaveLength(1);
|
||||
expect(second.stats[0]?.entryId).toBe("d1");
|
||||
expect(second.stats[0]?.usage.premiumRequests).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user