smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge PR #3735: fix(stats): avoid macOS stats sync worker abort (@roboomp)

Browse Source 
# Conflicts:
#	packages/stats/src/parser.ts
#	packages/stats/test/parser-large-session.test.ts
This commit is contained in:
can1357
2026-06-29 16:48:59 +02:00
parent b4f75abb30 7d5544493e
commit a402486d9a
14 changed files with 367 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stats/CHANGELOG.md
+6
View File
@@ -2,6 +2,12 @@
## [Unreleased]
### Fixed
- Kept stats session sync on the serial parser path for `workers: 1` and macOS defaults, avoiding Bun worker re-entry aborts when launching `/stats` ([#3733](https://github.com/can1357/oh-my-pi/issues/3733)).
- Replaced the native `Bun.JSONL.parseChunk` session parser path with a lenient JS line scanner, avoiding Bun aborts on large stats session files ([#3733](https://github.com/can1357/oh-my-pi/issues/3733)).
- Skipped the stats sync worker smoke probe on darwin so `omp --smoke-test` (and the macOS signing/notarization pre-launch run) no longer re-enters the Bun-worker abort surface the serial macOS sync now avoids ([#3733](https://github.com/can1357/oh-my-pi/issues/3733)).
## [16.2.3] - 2026-06-28
### Added
packages/stats/src/aggregator.ts
+59 -35
View File
@@ -23,7 +23,7 @@ import {
setFileOffset,
updateUserMessageLinks,
} from "./db";
import { getSessionEntry, listAllSessionFiles, type ParseSessionResult } from "./parser";
import { getSessionEntry, listAllSessionFiles, type ParseSessionResult, parseSessionFile } from "./parser";
import type { SyncWorkerRequest, SyncWorkerResponse } from "./sync-worker";
// Coding-agent binary/bundle workers route through the CLI entrypoint with a
// hidden argv mode, so the compiled binary and npm bundle only need one
@@ -68,6 +68,9 @@ export interface SyncOptions {
}
function defaultWorkerCount(): number {
// Bun 1.3.x can abort the macOS process when stats sync workers re-enter
// the compiled `omp` binary. Keep macOS on the documented serial path.
if (process.platform === "darwin") return 1;
// `navigator.hardwareConcurrency` is the portable answer in Bun; fall
// back to a small fixed pool if it's somehow unavailable.
const hw = typeof navigator !== "undefined" ? (navigator.hardwareConcurrency ?? 0) : 0;
@@ -149,10 +152,16 @@ function dispatch(handle: WorkerHandle, request: SyncWorkerRequest): Promise<Par
* spawn path on a fresh install (no session files = early return), so a
* dedicated probe is the only reliable signal.
*
* Resolves with the worker's `import.meta.url` (caller-visible diagnostics);
* rejects on transport error, error response, or timeout.
* No-op on darwin: `syncAllSessions` keeps macOS on the serial parser path
* (see {@link defaultWorkerCount}) so the worker spawn surface is unreachable
* from the CLI, and probing it under the hardened runtime in
* `scripts/ci-macos-sign.sh` would re-enter the Bun-worker abort surface that
* motivated the darwin serial default in the first place.
*
* Rejects on transport error, error response, or timeout.
*/
export async function smokeTestSyncWorker({ timeoutMs = 5_000 }: { timeoutMs?: number } = {}): Promise<void> {
if (process.platform === "darwin") return;
const worker = createSyncWorker();
const { promise, resolve, reject } = Promise.withResolvers<void>();
const timer = setTimeout(() => reject(new Error(`sync worker did not pong within ${timeoutMs}ms`)), timeoutMs);
@@ -183,11 +192,11 @@ export async function smokeTestSyncWorker({ timeoutMs = 5_000 }: { timeoutMs?: n
/**
* Sync all session files to the database.
*
* Parsing fans out across a worker pool (one in-flight job per worker)
* while DB writes and offset bookkeeping stay on the calling thread so the
* single SQLite handle stays uncontended. `onProgress` fires once per
* completed file (skipped files included so the bar walks at a steady
* rate).
* `workers: 1` parses inline. Larger pools fan parsing out across workers
* (one in-flight job per worker) while DB writes and offset bookkeeping stay on
* the calling thread so the single SQLite handle stays uncontended.
* `onProgress` fires once per completed file (skipped files included so the
* bar walks at a steady rate).
*/
export async function syncAllSessions(opts?: SyncOptions): Promise<{ processed: number; files: number }> {
await initDb();
@@ -200,10 +209,6 @@ export async function syncAllSessions(opts?: SyncOptions): Promise<{ processed:
let completed = 0;
let cursor = 0;
const poolSize = Math.max(1, Math.min(files.length, opts?.workers ?? defaultWorkerCount()));
const handles: WorkerHandle[] = [];
for (let i = 0; i < poolSize; i++) handles.push(spawnWorker());
const report = (sessionFile: string) => {
completed++;
opts?.onProgress?.({
@@ -214,34 +219,53 @@ export async function syncAllSessions(opts?: SyncOptions): Promise<{ processed:
});
};
const processFile = async (
sessionFile: string,
parse: (sessionFile: string, fromOffset: number) => Promise<ParseSessionResult>,
): Promise<void> => {
let fileStats: fs.Stats;
try {
fileStats = await fs.promises.stat(sessionFile);
} catch {
report(sessionFile);
return;
}
const lastModified = fileStats.mtimeMs;
const stored = getFileOffset(sessionFile);
if (stored && stored.lastModified >= lastModified) {
report(sessionFile);
return;
}
const fromOffset = stored?.offset ?? 0;
const result = await parse(sessionFile, fromOffset);
const inserted = applyParseResult(sessionFile, lastModified, result);
if (inserted > 0) {
totalProcessed += inserted;
filesProcessed++;
}
report(sessionFile);
};
const requestedWorkers = Math.max(1, Math.floor(opts?.workers ?? defaultWorkerCount()));
if (requestedWorkers === 1) {
for (const sessionFile of files) {
await processFile(sessionFile, parseSessionFile);
}
return { processed: totalProcessed, files: filesProcessed };
}
const poolSize = Math.min(files.length, requestedWorkers);
const handles: WorkerHandle[] = [];
for (let i = 0; i < poolSize; i++) handles.push(spawnWorker());
async function drain(handle: WorkerHandle): Promise<void> {
while (true) {
const idx = cursor++;
if (idx >= files.length) return;
const sessionFile = files[idx];
let fileStats: fs.Stats;
try {
fileStats = await fs.promises.stat(sessionFile);
} catch {
report(sessionFile);
continue;
}
const lastModified = fileStats.mtimeMs;
const stored = getFileOffset(sessionFile);
if (stored && stored.lastModified >= lastModified) {
report(sessionFile);
continue;
}
const fromOffset = stored?.offset ?? 0;
const result = await dispatch(handle, { sessionFile, fromOffset });
const inserted = applyParseResult(sessionFile, lastModified, result);
if (inserted > 0) {
totalProcessed += inserted;
filesProcessed++;
}
report(sessionFile);
await processFile(sessionFile, (file, fromOffset) => dispatch(handle, { sessionFile: file, fromOffset }));
}
}
packages/stats/src/parser.ts
+40 -41
View File
@@ -164,54 +164,53 @@ function extractStats(
}
const LF = 0x0a;
const CR = 0x0d;
const jsonLineDecoder = new TextDecoder();
function parseJsonLine(bytes: Uint8Array, start: number, end: number): SessionEntry | null {
while (end > start && bytes[end - 1] === CR) end--;
if (end <= start) return null;
try {
return JSON.parse(jsonLineDecoder.decode(bytes.subarray(start, end))) as SessionEntry;
} catch {
return null;
}
}
function visitSessionEntriesLenient(bytes: Uint8Array, visit: (entry: SessionEntry) => void): number {
let cursor = 0;
let read = 0;
while (cursor < bytes.length) {
const newline = bytes.indexOf(LF, cursor);
const hasNewline = newline !== -1;
const lineEnd = hasNewline ? newline : bytes.length;
const entry = parseJsonLine(bytes, cursor, lineEnd);
if (entry) {
visit(entry);
read = hasNewline ? newline + 1 : lineEnd;
} else if (hasNewline) {
read = newline + 1;
} else {
break;
}
cursor = hasNewline ? newline + 1 : lineEnd;
}
return read;
}
function parseSessionEntriesLenient(bytes: Uint8Array): { entries: SessionEntry[]; read: number } {
const entries: SessionEntry[] = [];
let cursor = 0;
while (cursor < bytes.length) {
const { values, error, read, done } = Bun.JSONL.parseChunk(bytes, cursor, bytes.length);
for (const value of values as SessionEntry[]) {
entries.push(value);
}
if (error) {
const nextNewline = bytes.indexOf(LF, Math.max(read, cursor));
if (nextNewline === -1) break;
cursor = nextNewline + 1;
continue;
}
if (read <= cursor) break;
cursor = read;
if (done) break;
}
return { entries, read: cursor };
const read = visitSessionEntriesLenient(bytes, entry => entries.push(entry));
return { entries, read };
}
function scanLastServiceTier(bytes: Uint8Array): ServiceTier | undefined {
let cursor = 0;
let currentServiceTier: ServiceTier | undefined;
while (cursor < bytes.length) {
const { values, error, read, done } = Bun.JSONL.parseChunk(bytes, cursor, bytes.length);
for (const value of values as SessionEntry[]) {
if (isServiceTierChange(value)) currentServiceTier = value.serviceTier ?? undefined;
}
if (error) {
const nextNewline = bytes.indexOf(LF, Math.max(read, cursor));
if (nextNewline === -1) break;
cursor = nextNewline + 1;
continue;
}
if (read <= cursor) break;
cursor = read;
if (done) break;
}
visitSessionEntriesLenient(bytes, entry => {
if (isServiceTierChange(entry)) currentServiceTier = entry.serviceTier ?? undefined;
});
return currentServiceTier;
}
/**
packages/stats/test/agent-type.test.ts
+5 -33
View File
@@ -1,43 +1,15 @@
import { Database } from "bun:sqlite";
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { describe, expect, it } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { getOverviewStats } from "@oh-my-pi/omp-stats/aggregator";
import { closeDb, getStatsByAgentType, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import { getStatsByAgentType, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import { classifyAgentType } from "@oh-my-pi/omp-stats/parser";
import type { AgentType, MessageStats } from "@oh-my-pi/omp-stats/types";
import {
getAgentDir,
getConfigRootDir,
getSessionsDir,
getStatsDbPath,
setAgentDir,
TempDir,
} from "@oh-my-pi/pi-utils";
import { getConfigRootDir, getSessionsDir, getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-agent-type-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-agent-type-");
interface Tokens {
input: number;
packages/stats/test/behavior-backfill.test.ts
+4 -25
View File
@@ -1,34 +1,13 @@
import { Database } from "bun:sqlite";
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { describe, expect, it } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { syncAllSessions } from "@oh-my-pi/omp-stats/aggregator";
import { closeDb, getBehaviorOverall, getFileOffset, initDb } from "@oh-my-pi/omp-stats/db";
import { getAgentDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getAgentDir, getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-behavior-backfill-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-behavior-backfill-");
async function writeSessionFile(): Promise<string> {
const sessionDir = path.join(getAgentDir(), "sessions", "--tmp--behavior-backfill");
packages/stats/test/db-cost.test.ts
+4 -26
View File
@@ -1,34 +1,12 @@
import { Database } from "bun:sqlite";
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import * as os from "node:os";
import * as path from "node:path";
import { describe, expect, it } from "bun:test";
import { closeDb, getRecentRequests, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import type { MessageStats } from "@oh-my-pi/omp-stats/types";
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { getAgentDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-db-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-db-");
function createCodexGptStats(entryId: string): MessageStats {
return {
packages/stats/test/db-range.test.ts
+4 -27
View File
@@ -1,33 +1,10 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import * as os from "node:os";
import * as path from "node:path";
import { describe, expect, it } from "bun:test";
import { getDashboardStats } from "@oh-my-pi/omp-stats/aggregator";
import { closeDb, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import { initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import type { MessageStats } from "@oh-my-pi/omp-stats/types";
import { getAgentDir, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-db-range-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-db-range-");
function makeMessage(timestamp: number, entryId: string): MessageStats {
return {
packages/stats/test/fork-dedup.test.ts
+4 -36
View File
@@ -1,46 +1,14 @@
import { Database } from "bun:sqlite";
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { describe, expect, it } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { syncAllSessions } from "@oh-my-pi/omp-stats/aggregator";
import { closeDb, getOverallStats, getRecentRequests, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import type { MessageStats } from "@oh-my-pi/omp-stats/types";
import { getAgentDir, getSessionsDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getSessionsDir, getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const XDG_KEYS = ["XDG_DATA_HOME", "XDG_STATE_HOME", "XDG_CACHE_HOME"] as const;
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
const originalXdg: Record<string, string | undefined> = {};
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-fork-dedup-");
for (const key of XDG_KEYS) {
originalXdg[key] = process.env[key];
delete process.env[key];
}
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
for (const key of XDG_KEYS) {
const prior = originalXdg[key];
if (prior === undefined) delete process.env[key];
else process.env[key] = prior;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-fork-dedup-");
interface AssistantOptions {
entryId: string;
packages/stats/test/gain-aggregator.test.ts
+5 -26
View File
@@ -1,34 +1,13 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { describe, expect, it } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { closeDb, initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import { initDb, insertMessageStats } from "@oh-my-pi/omp-stats/db";
import { dedupeProjects, getGainDashboardStats, normalizeProjectPath } from "@oh-my-pi/omp-stats/gain-aggregator";
import type { MessageStats } from "@oh-my-pi/omp-stats/types";
import { getAgentDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-gain-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-gain-");
function makeMessage(sessionFile: string, folder: string, entryId: string, timestamp: number): MessageStats {
return {
packages/stats/test/helpers/temp-agent.ts
+70
View File
@@ -0,0 +1,70 @@
/**
* Shared test isolation for stats Bun tests.
*
* The default profile's stats.db is redirected to `$XDG_DATA_HOME/omp/stats.db`
* by {@link DirResolver} whenever `agentDirOverride === defaultAgent`. Tests
* that only set `PI_CONFIG_DIR` + `setAgentDir(<home>/<config>/agent)` resolve
* to that default and silently share `stats.db` across files when an XDG
* variable is set (e.g. CI's `XDG_DATA_HOME`), producing the cross-test row
* pollution that fails `db-range`, `behavior-backfill`, `priority-premium-*`,
* and `agent-type` runs.
*
* `installStatsTestIsolation` snapshots and clears `XDG_*_HOME` plus
* `PI_CONFIG_DIR` for the test, points the agent directory at a fresh
* `TempDir`, closes the stats DB handle, and tears everything back down in the
* matching `afterEach`.
*/
import { afterEach, beforeEach } from "bun:test";
import * as os from "node:os";
import * as path from "node:path";
import { closeDb } from "@oh-my-pi/omp-stats/db";
import { getAgentDir, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
const XDG_KEYS = ["XDG_DATA_HOME", "XDG_STATE_HOME", "XDG_CACHE_HOME"] as const;
export interface StatsTestIsolation {
/** Active per-test `TempDir`. Null between tests. */
current(): TempDir | null;
}
export function installStatsTestIsolation(prefix: string): StatsTestIsolation {
const originalAgentDir = getAgentDir();
let originalConfigDir: string | undefined;
const originalXdg: Record<string, string | undefined> = {};
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync(prefix);
originalConfigDir = process.env.PI_CONFIG_DIR;
for (const key of XDG_KEYS) {
originalXdg[key] = process.env[key];
delete process.env[key];
}
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
for (const key of XDG_KEYS) {
const prior = originalXdg[key];
if (prior === undefined) delete process.env[key];
else process.env[key] = prior;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
return {
current() {
return tempDir;
},
};
}
packages/stats/test/parser-large-session.test.ts
+57 -21
View File
@@ -1,32 +1,56 @@
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { afterEach, describe, expect, it, vi } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { parseSessionFile } from "@oh-my-pi/omp-stats/parser";
import { getAgentDir, getSessionsDir, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getSessionsDir } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-large-session-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
installStatsTestIsolation("@pi-stats-large-session-");
afterEach(() => {
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
vi.restoreAllMocks();
});
async function writeLargeSessionFile(): Promise<string> {
const sessionDir = path.join(getSessionsDir(), "--tmp--large-session");
await fs.mkdir(sessionDir, { recursive: true });
const sessionFile = path.join(sessionDir, "session.jsonl");
const timestamp = new Date().toISOString();
const payload = "x".repeat(16 * 1024);
const lines: string[] = [];
for (let i = 0; i < 256; i++) {
lines.push(
JSON.stringify({
type: "message",
id: `assistant-${i}`,
parentId: null,
timestamp,
message: {
role: "assistant",
content: [{ type: "text", text: payload }],
api: "openai-responses",
provider: "openai",
model: "gpt-5.4",
usage: {
input: 1,
output: 2,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 3,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now() + i,
duration: 10,
ttft: 5,
},
}),
);
}
await Bun.write(sessionFile, `${lines.join("\n")}\n`);
return sessionFile;
}
describe("large session parsing", () => {
it("parses a JSONL chunk with more entries than the JavaScript argument limit", async () => {
const dir = path.join(getSessionsDir(), "--tmp--large-session");
@@ -43,4 +67,16 @@ describe("large session parsing", () => {
expect(result.userStats).toEqual([]);
expect(result.userLinks).toEqual([]);
});
it("parses multi-megabyte JSONL without entering Bun.JSONL.parseChunk", async () => {
const sessionFile = await writeLargeSessionFile();
vi.spyOn(Bun.JSONL, "parseChunk").mockImplementation(() => {
throw new Error("native JSONL parser unavailable");
});
const result = await parseSessionFile(sessionFile);
expect(result.stats).toHaveLength(256);
expect(result.newOffset).toBeGreaterThan(4 * 1024 * 1024);
});
});
packages/stats/test/priority-premium-requests.test.ts
+4 -25
View File
@@ -1,35 +1,14 @@
import { Database } from "bun:sqlite";
import { afterEach, beforeEach, describe, expect, it } from "bun:test";
import { describe, expect, it } from "bun:test";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { syncAllSessions } from "@oh-my-pi/omp-stats/aggregator";
import { closeDb, getOverallStats, getRecentRequests } from "@oh-my-pi/omp-stats/db";
import { parseSessionFile } from "@oh-my-pi/omp-stats/parser";
import { getAgentDir, getSessionsDir, getStatsDbPath, setAgentDir, TempDir } from "@oh-my-pi/pi-utils";
import { getSessionsDir, getStatsDbPath } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
const originalConfigDir = process.env.PI_CONFIG_DIR;
const originalAgentDir = getAgentDir();
let tempDir: TempDir | null = null;
beforeEach(() => {
tempDir = TempDir.createSync("@pi-stats-priority-");
const configDir = path.relative(os.homedir(), tempDir.join("config"));
process.env.PI_CONFIG_DIR = configDir;
setAgentDir(path.join(os.homedir(), configDir, "agent"));
});
afterEach(() => {
closeDb();
if (originalConfigDir === undefined) {
delete process.env.PI_CONFIG_DIR;
} else {
process.env.PI_CONFIG_DIR = originalConfigDir;
}
setAgentDir(originalAgentDir);
tempDir?.removeSync();
tempDir = null;
});
installStatsTestIsolation("@pi-stats-priority-");
interface SessionLines {
lines: Array<Record<string, unknown>>;
packages/stats/test/smoke-worker-darwin.test.ts
+21
View File
@@ -0,0 +1,21 @@
import { afterEach, describe, expect, it, vi } from "bun:test";
import { smokeTestSyncWorker } from "@oh-my-pi/omp-stats/aggregator";
import { installStatsTestIsolation } from "./helpers/temp-agent";
installStatsTestIsolation("@pi-stats-smoke-darwin-");
afterEach(() => {
vi.restoreAllMocks();
});
describe("smokeTestSyncWorker", () => {
it("skips the worker spawn on darwin so omp --smoke-test stays off the macOS abort surface", async () => {
vi.spyOn(process, "platform", "get").mockReturnValue("darwin");
const workerSpy = vi.spyOn(globalThis, "Worker").mockImplementation(() => {
throw new Error("worker should not be created on darwin");
});
await expect(smokeTestSyncWorker()).resolves.toBeUndefined();
expect(workerSpy).not.toHaveBeenCalled();
});
});
packages/stats/test/sync-serial.test.ts
+84
View File
@@ -0,0 +1,84 @@
import { afterEach, describe, expect, it, vi } from "bun:test";
import * as fs from "node:fs/promises";
import * as path from "node:path";
import { syncAllSessions } from "@oh-my-pi/omp-stats/aggregator";
import { getOverallStats } from "@oh-my-pi/omp-stats/db";
import { getSessionsDir } from "@oh-my-pi/pi-utils";
import { installStatsTestIsolation } from "./helpers/temp-agent";
installStatsTestIsolation("@pi-stats-sync-serial-");
afterEach(() => {
vi.restoreAllMocks();
});
async function writeSessionFile(): Promise<void> {
const sessionDir = path.join(getSessionsDir(), "--tmp--sync-serial");
await fs.mkdir(sessionDir, { recursive: true });
const timestamp = new Date().toISOString();
const sessionFile = path.join(sessionDir, "session.jsonl");
const assistant = {
type: "message",
id: "assistant-1",
parentId: null,
timestamp,
message: {
role: "assistant",
content: [{ type: "text", text: "ok" }],
api: "openai-responses",
provider: "openai",
model: "gpt-5.4",
usage: {
input: 1,
output: 2,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 3,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "stop",
timestamp: Date.now(),
duration: 10,
ttft: 5,
},
};
await Bun.write(sessionFile, `${JSON.stringify(assistant)}\n`);
}
describe("stats sync serial mode", () => {
it("honors workers: 1 without spawning a worker", async () => {
await writeSessionFile();
const workerSpy = vi.spyOn(globalThis, "Worker");
const synced = await syncAllSessions({ workers: 1 });
const overall = getOverallStats();
expect(synced.files).toBe(1);
expect(overall.totalRequests).toBe(1);
expect(workerSpy).not.toHaveBeenCalled();
});
it("uses the serial parser by default on macOS", async () => {
await writeSessionFile();
vi.spyOn(process, "platform", "get").mockReturnValue("darwin");
const workerSpy = vi.spyOn(globalThis, "Worker");
const synced = await syncAllSessions();
const overall = getOverallStats();
expect(synced.files).toBe(1);
expect(overall.totalRequests).toBe(1);
expect(workerSpy).not.toHaveBeenCalled();
});
it("spawns a worker pool when callers explicitly request workers: 2 with a single file", async () => {
await writeSessionFile();
const workerProbe = new Error("worker probe");
const workerSpy = vi.spyOn(globalThis, "Worker").mockImplementation(() => {
throw workerProbe;
});
await expect(syncAllSessions({ workers: 2 })).rejects.toBe(workerProbe);
expect(workerSpy).toHaveBeenCalled();
});
});