can1357
00bb61b054
- Passed `bucketMs` to `getModelTimeSeries` and `getModelPerformanceSeries` so SQL bucketing matches the selected range. - Added `range-meta.ts` to centralise bucket sizes, counts, tick formats, and labels per time range. - Updated `ChartsContainer` and `ModelsTable` to use range-aware x-axis formatting and trend labels.
1101 lines
34 KiB
TypeScript
1101 lines
34 KiB
TypeScript
import { Database } from "bun:sqlite";
|
|
import * as fs from "node:fs/promises";
|
|
import { type GeneratedProvider, getBundledModel, type Usage } from "@oh-my-pi/pi-ai";
|
|
import { getConfigRootDir, getStatsDbPath } from "@oh-my-pi/pi-utils";
|
|
import type {
|
|
AggregatedStats,
|
|
BehaviorModelStats,
|
|
BehaviorOverallStats,
|
|
BehaviorTimeSeriesPoint,
|
|
CostTimeSeriesPoint,
|
|
FolderStats,
|
|
MessageStats,
|
|
ModelPerformancePoint,
|
|
ModelStats,
|
|
ModelTimeSeriesPoint,
|
|
TimeSeriesPoint,
|
|
UserMessageLink,
|
|
UserMessageStats,
|
|
} from "./types";
|
|
|
|
type ModelCost = { input: number; output: number; cacheRead: number; cacheWrite: number };
|
|
type UsageCost = Usage["cost"];
|
|
type CostTokens = Pick<Usage, "input" | "output" | "cacheRead" | "cacheWrite">;
|
|
|
|
interface CostBackfillRow {
|
|
id: number;
|
|
provider: string;
|
|
model: string;
|
|
input_tokens: number;
|
|
output_tokens: number;
|
|
cache_read_tokens: number;
|
|
cache_write_tokens: number;
|
|
}
|
|
|
|
let db: Database | null = null;
|
|
|
|
const BACKFILL_COMPLETE = "complete";
|
|
const BACKFILL_PENDING = "pending";
|
|
const USER_MESSAGES_BACKFILL_KEY = "user_messages_v5";
|
|
const USER_MESSAGE_LINKS_REPAIR_KEY = "user_message_links_v1";
|
|
const PRIORITY_PREMIUM_REQUESTS_BACKFILL_KEY = "premium_requests_priority_v1";
|
|
function shouldResetBackfill(value: string | undefined): boolean {
|
|
return value !== BACKFILL_COMPLETE && value !== BACKFILL_PENDING;
|
|
}
|
|
/**
|
|
* Initialize the database and create tables.
|
|
*/
|
|
export async function initDb(): Promise<Database> {
|
|
if (db) return db;
|
|
|
|
// Ensure directory exists
|
|
await fs.mkdir(getConfigRootDir(), { recursive: true });
|
|
|
|
db = new Database(getStatsDbPath());
|
|
db.exec("PRAGMA journal_mode = WAL");
|
|
|
|
// Create tables
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_file TEXT NOT NULL,
|
|
entry_id TEXT NOT NULL,
|
|
folder TEXT NOT NULL,
|
|
model TEXT NOT NULL,
|
|
provider TEXT NOT NULL,
|
|
api TEXT NOT NULL,
|
|
timestamp INTEGER NOT NULL,
|
|
duration INTEGER,
|
|
ttft INTEGER,
|
|
stop_reason TEXT NOT NULL,
|
|
error_message TEXT,
|
|
input_tokens INTEGER NOT NULL,
|
|
output_tokens INTEGER NOT NULL,
|
|
cache_read_tokens INTEGER NOT NULL,
|
|
cache_write_tokens INTEGER NOT NULL,
|
|
total_tokens INTEGER NOT NULL,
|
|
premium_requests REAL NOT NULL,
|
|
cost_input REAL NOT NULL,
|
|
cost_output REAL NOT NULL,
|
|
cost_cache_read REAL NOT NULL,
|
|
cost_cache_write REAL NOT NULL,
|
|
cost_total REAL NOT NULL,
|
|
UNIQUE(session_file, entry_id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_model ON messages(model);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_folder ON messages(folder);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_file);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_timestamp_model_provider ON messages(timestamp, model, provider);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_timestamp_folder ON messages(timestamp, folder);
|
|
CREATE INDEX IF NOT EXISTS idx_messages_stop_reason_timestamp ON messages(stop_reason, timestamp);
|
|
|
|
CREATE TABLE IF NOT EXISTS file_offsets (
|
|
session_file TEXT PRIMARY KEY,
|
|
offset INTEGER NOT NULL,
|
|
last_modified INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_file TEXT NOT NULL,
|
|
entry_id TEXT NOT NULL,
|
|
folder TEXT NOT NULL,
|
|
timestamp INTEGER NOT NULL,
|
|
model TEXT,
|
|
provider TEXT,
|
|
chars INTEGER NOT NULL,
|
|
words INTEGER NOT NULL,
|
|
yelling INTEGER NOT NULL,
|
|
profanity INTEGER NOT NULL,
|
|
anguish INTEGER NOT NULL,
|
|
negation INTEGER NOT NULL DEFAULT 0,
|
|
repetition INTEGER NOT NULL DEFAULT 0,
|
|
blame INTEGER NOT NULL DEFAULT 0,
|
|
UNIQUE(session_file, entry_id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_user_messages_timestamp ON user_messages(timestamp);
|
|
CREATE INDEX IF NOT EXISTS idx_user_messages_timestamp_model ON user_messages(timestamp, model, provider);
|
|
|
|
CREATE TABLE IF NOT EXISTS meta (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
`);
|
|
|
|
const messageColumns = db.prepare("PRAGMA table_info(messages)").all() as { name: string }[];
|
|
if (!messageColumns.some(column => column.name === "premium_requests")) {
|
|
db.exec("ALTER TABLE messages ADD COLUMN premium_requests REAL NOT NULL DEFAULT 0");
|
|
}
|
|
db.exec("UPDATE messages SET premium_requests = 0 WHERE premium_requests IS NULL");
|
|
// Each behavior-metric bump invalidates previously-ingested rows. We detect
|
|
// the stale schema by column name and drop the table; `IF NOT EXISTS` above
|
|
// already produced the new schema, but we want a clean wipe + re-ingest.
|
|
// `backfillUserMessages` then clears `file_offsets` so the next sync
|
|
// re-parses every session under the current metric definitions.
|
|
// v1 -> v2: yelling sentences replace `caps_words`.
|
|
// v2 -> v3: `drama_runs` folded into a single `anguish` signal that
|
|
// also captures elongated interjections, `dude`, and dot runs,
|
|
// gated on a stripped prose-line budget.
|
|
// v3 -> v4: added `negation`, `repetition`, `blame` frustration signals
|
|
// plus profanity dictionary expansion + word-boundary fix.
|
|
// v4 -> v5: column `yelling_sentences` renamed to `yelling` to match
|
|
// the other single-word signal columns.
|
|
const userMessageColumns = db.prepare("PRAGMA table_info(user_messages)").all() as {
|
|
name: string;
|
|
}[];
|
|
const hasStaleColumn =
|
|
userMessageColumns.length > 0 &&
|
|
(userMessageColumns.some(column => column.name === "caps_words") ||
|
|
userMessageColumns.some(column => column.name === "drama_runs") ||
|
|
userMessageColumns.some(column => column.name === "yelling_sentences"));
|
|
const hasV4Columns = userMessageColumns.some(column => column.name === "negation");
|
|
const hasOldUserMessages = userMessageColumns.length > 0;
|
|
if (hasStaleColumn || (hasOldUserMessages && !hasV4Columns)) {
|
|
db.exec("DROP TABLE user_messages");
|
|
db.exec(`
|
|
CREATE TABLE user_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_file TEXT NOT NULL,
|
|
entry_id TEXT NOT NULL,
|
|
folder TEXT NOT NULL,
|
|
timestamp INTEGER NOT NULL,
|
|
model TEXT,
|
|
provider TEXT,
|
|
chars INTEGER NOT NULL,
|
|
words INTEGER NOT NULL,
|
|
yelling INTEGER NOT NULL,
|
|
profanity INTEGER NOT NULL,
|
|
anguish INTEGER NOT NULL,
|
|
negation INTEGER NOT NULL DEFAULT 0,
|
|
repetition INTEGER NOT NULL DEFAULT 0,
|
|
blame INTEGER NOT NULL DEFAULT 0,
|
|
UNIQUE(session_file, entry_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_user_messages_timestamp ON user_messages(timestamp);
|
|
CREATE INDEX IF NOT EXISTS idx_user_messages_timestamp_model ON user_messages(timestamp, model, provider);
|
|
`);
|
|
}
|
|
backfillUserMessages(db);
|
|
repairUserMessageLinks(db);
|
|
backfillPriorityPremiumRequests(db);
|
|
backfillMissingCatalogCosts(db);
|
|
return db;
|
|
}
|
|
|
|
function hasBillableCost(cost: ModelCost): boolean {
|
|
return cost.input !== 0 || cost.output !== 0 || cost.cacheRead !== 0 || cost.cacheWrite !== 0;
|
|
}
|
|
|
|
function getBundledModelCost(provider: string, modelId: string): ModelCost | null {
|
|
const model = getBundledModel(provider as GeneratedProvider, modelId);
|
|
return model?.cost ?? null;
|
|
}
|
|
|
|
function getCatalogCost(provider: string, modelId: string): ModelCost | null {
|
|
const primaryCost = getBundledModelCost(provider, modelId);
|
|
if (primaryCost && hasBillableCost(primaryCost)) {
|
|
return primaryCost;
|
|
}
|
|
|
|
if (provider === "openai-codex") {
|
|
const openAICost = getBundledModelCost("openai", modelId);
|
|
if (openAICost && hasBillableCost(openAICost)) {
|
|
return openAICost;
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function calculateCatalogCost(provider: string, modelId: string, tokens: CostTokens): UsageCost | null {
|
|
const cost = getCatalogCost(provider, modelId);
|
|
if (!cost) return null;
|
|
|
|
const input = (cost.input / 1_000_000) * tokens.input;
|
|
const output = (cost.output / 1_000_000) * tokens.output;
|
|
const cacheRead = (cost.cacheRead / 1_000_000) * tokens.cacheRead;
|
|
const cacheWrite = (cost.cacheWrite / 1_000_000) * tokens.cacheWrite;
|
|
|
|
return {
|
|
input,
|
|
output,
|
|
cacheRead,
|
|
cacheWrite,
|
|
total: input + output + cacheRead + cacheWrite,
|
|
};
|
|
}
|
|
|
|
function resolveStoredCost(stats: MessageStats): UsageCost {
|
|
if (stats.usage.cost.total !== 0) {
|
|
return stats.usage.cost;
|
|
}
|
|
|
|
return calculateCatalogCost(stats.provider, stats.model, stats.usage) ?? stats.usage.cost;
|
|
}
|
|
|
|
function backfillMissingCatalogCosts(database: Database): void {
|
|
const rows = database
|
|
.prepare(`
|
|
SELECT id, provider, model, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens
|
|
FROM messages
|
|
WHERE cost_total = 0 AND total_tokens > 0
|
|
`)
|
|
.all() as CostBackfillRow[];
|
|
|
|
if (rows.length === 0) return;
|
|
|
|
const update = database.prepare(`
|
|
UPDATE messages
|
|
SET cost_input = ?, cost_output = ?, cost_cache_read = ?, cost_cache_write = ?, cost_total = ?
|
|
WHERE id = ?
|
|
`);
|
|
|
|
const applyBackfill = database.transaction(() => {
|
|
for (const row of rows) {
|
|
const cost = calculateCatalogCost(row.provider, row.model, {
|
|
input: row.input_tokens,
|
|
output: row.output_tokens,
|
|
cacheRead: row.cache_read_tokens,
|
|
cacheWrite: row.cache_write_tokens,
|
|
});
|
|
|
|
if (!cost || cost.total === 0) continue;
|
|
|
|
update.run(cost.input, cost.output, cost.cacheRead, cost.cacheWrite, cost.total, row.id);
|
|
}
|
|
});
|
|
|
|
applyBackfill();
|
|
}
|
|
|
|
/**
|
|
* Get the stored offset for a session file.
|
|
*/
|
|
export function getFileOffset(sessionFile: string): { offset: number; lastModified: number } | null {
|
|
if (!db) return null;
|
|
|
|
const stmt = db.prepare("SELECT offset, last_modified FROM file_offsets WHERE session_file = ?");
|
|
const row = stmt.get(sessionFile) as { offset: number; last_modified: number } | undefined;
|
|
|
|
return row ? { offset: row.offset, lastModified: row.last_modified } : null;
|
|
}
|
|
|
|
/**
|
|
* Update the stored offset for a session file.
|
|
*/
|
|
export function setFileOffset(sessionFile: string, offset: number, lastModified: number): void {
|
|
if (!db) return;
|
|
|
|
const stmt = db.prepare(`
|
|
INSERT OR REPLACE INTO file_offsets (session_file, offset, last_modified)
|
|
VALUES (?, ?, ?)
|
|
`);
|
|
stmt.run(sessionFile, offset, lastModified);
|
|
}
|
|
|
|
/**
|
|
* Insert message stats into the database.
|
|
*/
|
|
export function insertMessageStats(stats: MessageStats[]): number {
|
|
if (!db || stats.length === 0) return 0;
|
|
|
|
// Use UPSERT so a re-sync can fix up `premium_requests` for rows persisted
|
|
// before priority service-tier traffic was counted as premium. The guard
|
|
// `WHERE messages.premium_requests < excluded.premium_requests` keeps every
|
|
// other column immutable and never demotes an existing count (e.g. when a
|
|
// later parse drops back to 0 for the same row).
|
|
const stmt = db.prepare(`
|
|
INSERT INTO messages (
|
|
session_file, entry_id, folder, model, provider, api, timestamp,
|
|
duration, ttft, stop_reason, error_message,
|
|
input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, total_tokens, premium_requests,
|
|
cost_input, cost_output, cost_cache_read, cost_cache_write, cost_total
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(session_file, entry_id) DO UPDATE SET
|
|
premium_requests = excluded.premium_requests
|
|
WHERE messages.premium_requests < excluded.premium_requests
|
|
`);
|
|
|
|
let inserted = 0;
|
|
const insert = db.transaction(() => {
|
|
for (const s of stats) {
|
|
const cost = resolveStoredCost(s);
|
|
const result = stmt.run(
|
|
s.sessionFile,
|
|
s.entryId,
|
|
s.folder,
|
|
s.model,
|
|
s.provider,
|
|
s.api,
|
|
s.timestamp,
|
|
s.duration,
|
|
s.ttft,
|
|
s.stopReason,
|
|
s.errorMessage,
|
|
s.usage.input,
|
|
s.usage.output,
|
|
s.usage.cacheRead,
|
|
s.usage.cacheWrite,
|
|
s.usage.totalTokens,
|
|
s.usage.premiumRequests ?? 0,
|
|
cost.input,
|
|
cost.output,
|
|
cost.cacheRead,
|
|
cost.cacheWrite,
|
|
cost.total,
|
|
);
|
|
if (result.changes > 0) inserted++;
|
|
}
|
|
});
|
|
|
|
insert();
|
|
return inserted;
|
|
}
|
|
|
|
/**
|
|
* Build aggregated stats from query results.
|
|
*/
|
|
function buildAggregatedStats(rows: any[]): AggregatedStats {
|
|
if (rows.length === 0) {
|
|
return {
|
|
totalRequests: 0,
|
|
successfulRequests: 0,
|
|
failedRequests: 0,
|
|
errorRate: 0,
|
|
totalInputTokens: 0,
|
|
totalOutputTokens: 0,
|
|
totalCacheReadTokens: 0,
|
|
totalCacheWriteTokens: 0,
|
|
cacheRate: 0,
|
|
totalCost: 0,
|
|
totalPremiumRequests: 0,
|
|
avgDuration: null,
|
|
avgTtft: null,
|
|
avgTokensPerSecond: null,
|
|
firstTimestamp: 0,
|
|
lastTimestamp: 0,
|
|
};
|
|
}
|
|
|
|
const row = rows[0];
|
|
const totalRequests = row.total_requests || 0;
|
|
const failedRequests = row.failed_requests || 0;
|
|
const successfulRequests = totalRequests - failedRequests;
|
|
const totalInputTokens = row.total_input_tokens || 0;
|
|
const totalCacheReadTokens = row.total_cache_read_tokens || 0;
|
|
const totalPremiumRequests = row.total_premium_requests || 0;
|
|
|
|
return {
|
|
totalRequests,
|
|
successfulRequests,
|
|
failedRequests,
|
|
errorRate: totalRequests > 0 ? failedRequests / totalRequests : 0,
|
|
totalInputTokens,
|
|
totalOutputTokens: row.total_output_tokens || 0,
|
|
totalCacheReadTokens,
|
|
totalCacheWriteTokens: row.total_cache_write_tokens || 0,
|
|
cacheRate:
|
|
totalInputTokens + totalCacheReadTokens > 0
|
|
? totalCacheReadTokens / (totalInputTokens + totalCacheReadTokens)
|
|
: 0,
|
|
totalCost: row.total_cost || 0,
|
|
totalPremiumRequests,
|
|
avgDuration: row.avg_duration,
|
|
avgTtft: row.avg_ttft,
|
|
avgTokensPerSecond: row.avg_tokens_per_second,
|
|
firstTimestamp: row.first_timestamp || 0,
|
|
lastTimestamp: row.last_timestamp || 0,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get overall aggregated stats.
|
|
*/
|
|
export function getOverallStats(cutoff?: number): AggregatedStats {
|
|
if (!db) return buildAggregatedStats([]);
|
|
|
|
const hasCutoff = cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
COUNT(*) as total_requests,
|
|
SUM(CASE WHEN stop_reason = 'error' THEN 1 ELSE 0 END) as failed_requests,
|
|
SUM(input_tokens) as total_input_tokens,
|
|
SUM(output_tokens) as total_output_tokens,
|
|
SUM(cache_read_tokens) as total_cache_read_tokens,
|
|
SUM(cache_write_tokens) as total_cache_write_tokens,
|
|
SUM(premium_requests) as total_premium_requests,
|
|
SUM(cost_total) as total_cost,
|
|
AVG(duration) as avg_duration,
|
|
AVG(ttft) as avg_ttft,
|
|
AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second,
|
|
MIN(timestamp) as first_timestamp,
|
|
MAX(timestamp) as last_timestamp
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
`);
|
|
|
|
const rows = hasCutoff ? stmt.all(cutoff) : stmt.all();
|
|
return buildAggregatedStats(rows as any[]);
|
|
}
|
|
/**
|
|
* Get stats grouped by model.
|
|
*/
|
|
export function getStatsByModel(cutoff?: number): ModelStats[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
model,
|
|
provider,
|
|
COUNT(*) as total_requests,
|
|
SUM(CASE WHEN stop_reason = 'error' THEN 1 ELSE 0 END) as failed_requests,
|
|
SUM(input_tokens) as total_input_tokens,
|
|
SUM(output_tokens) as total_output_tokens,
|
|
SUM(cache_read_tokens) as total_cache_read_tokens,
|
|
SUM(cache_write_tokens) as total_cache_write_tokens,
|
|
SUM(premium_requests) as total_premium_requests,
|
|
SUM(cost_total) as total_cost,
|
|
AVG(duration) as avg_duration,
|
|
AVG(ttft) as avg_ttft,
|
|
AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second,
|
|
MIN(timestamp) as first_timestamp,
|
|
MAX(timestamp) as last_timestamp
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY model, provider
|
|
ORDER BY total_requests DESC
|
|
`);
|
|
|
|
const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as any[];
|
|
return rows.map(row => ({
|
|
model: row.model,
|
|
provider: row.provider,
|
|
...buildAggregatedStats([row]),
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Get stats grouped by folder.
|
|
*/
|
|
export function getStatsByFolder(cutoff?: number): FolderStats[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
folder,
|
|
COUNT(*) as total_requests,
|
|
SUM(CASE WHEN stop_reason = 'error' THEN 1 ELSE 0 END) as failed_requests,
|
|
SUM(input_tokens) as total_input_tokens,
|
|
SUM(output_tokens) as total_output_tokens,
|
|
SUM(cache_read_tokens) as total_cache_read_tokens,
|
|
SUM(cache_write_tokens) as total_cache_write_tokens,
|
|
SUM(premium_requests) as total_premium_requests,
|
|
SUM(cost_total) as total_cost,
|
|
AVG(duration) as avg_duration,
|
|
AVG(ttft) as avg_ttft,
|
|
AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second,
|
|
MIN(timestamp) as first_timestamp,
|
|
MAX(timestamp) as last_timestamp
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY folder
|
|
ORDER BY total_requests DESC
|
|
`);
|
|
|
|
const rows = (hasCutoff ? stmt.all(cutoff) : stmt.all()) as any[];
|
|
return rows.map(row => ({
|
|
folder: row.folder,
|
|
...buildAggregatedStats([row]),
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Get time series data.
|
|
*/
|
|
export function getTimeSeries(hours = 24, cutoff?: number | null, bucketMs = 60 * 60 * 1000): TimeSeriesPoint[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== null;
|
|
const seriesCutoff = hasCutoff ? (cutoff ?? Date.now() - hours * 60 * 60 * 1000) : 0;
|
|
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
(timestamp / ?) * ? as bucket,
|
|
COUNT(*) as requests,
|
|
SUM(CASE WHEN stop_reason = 'error' THEN 1 ELSE 0 END) as errors,
|
|
SUM(total_tokens) as tokens,
|
|
SUM(cost_total) as cost
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY bucket
|
|
ORDER BY bucket ASC
|
|
`);
|
|
|
|
const rows = hasCutoff
|
|
? (stmt.all(bucketMs, bucketMs, seriesCutoff) as any[])
|
|
: (stmt.all(bucketMs, bucketMs) as any[]);
|
|
return rows.map(row => ({
|
|
timestamp: row.bucket,
|
|
requests: row.requests,
|
|
errors: row.errors,
|
|
tokens: row.tokens,
|
|
cost: row.cost,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Get daily performance time series data for the last N days.
|
|
*/
|
|
/**
|
|
* Get daily model usage time series data for the last N days.
|
|
*/
|
|
export function getModelTimeSeries(
|
|
days = 14,
|
|
cutoff?: number | null,
|
|
bucketMs = 24 * 60 * 60 * 1000,
|
|
): ModelTimeSeriesPoint[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== null;
|
|
const seriesCutoff = hasCutoff ? (cutoff ?? Date.now() - days * 24 * 60 * 60 * 1000) : 0;
|
|
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
(timestamp / ?) * ? as bucket,
|
|
model,
|
|
provider,
|
|
COUNT(*) as requests
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY bucket, model, provider
|
|
ORDER BY bucket ASC
|
|
`);
|
|
|
|
const rowsRaw = hasCutoff ? stmt.all(bucketMs, bucketMs, seriesCutoff) : stmt.all(bucketMs, bucketMs);
|
|
const rows = rowsRaw as Array<{ bucket: number; model: string; provider: string; requests: number }>;
|
|
return rows.map(row => ({
|
|
timestamp: row.bucket,
|
|
model: row.model,
|
|
provider: row.provider,
|
|
requests: row.requests,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Get daily model performance time series data for the last N days.
|
|
*/
|
|
export function getModelPerformanceSeries(
|
|
days = 14,
|
|
cutoff?: number | null,
|
|
bucketMs = 24 * 60 * 60 * 1000,
|
|
): ModelPerformancePoint[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== null;
|
|
const seriesCutoff = hasCutoff ? (cutoff ?? Date.now() - days * 24 * 60 * 60 * 1000) : 0;
|
|
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
(timestamp / ?) * ? as bucket,
|
|
model,
|
|
provider,
|
|
COUNT(*) as requests,
|
|
AVG(ttft) as avg_ttft,
|
|
AVG(CASE WHEN duration > 0 THEN output_tokens * 1000.0 / duration ELSE NULL END) as avg_tokens_per_second
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY bucket, model, provider
|
|
ORDER BY bucket ASC
|
|
`);
|
|
|
|
const rowsRaw = hasCutoff ? stmt.all(bucketMs, bucketMs, seriesCutoff) : stmt.all(bucketMs, bucketMs);
|
|
const rows = rowsRaw as Array<{
|
|
bucket: number;
|
|
model: string;
|
|
provider: string;
|
|
requests: number;
|
|
avg_ttft: number | null;
|
|
avg_tokens_per_second: number | null;
|
|
}>;
|
|
return rows.map(row => ({
|
|
timestamp: row.bucket,
|
|
model: row.model,
|
|
provider: row.provider,
|
|
requests: row.requests,
|
|
avgTtft: row.avg_ttft,
|
|
avgTokensPerSecond: row.avg_tokens_per_second,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Get total message count.
|
|
*/
|
|
export function getMessageCount(): number {
|
|
if (!db) return 0;
|
|
const stmt = db.prepare("SELECT COUNT(*) as count FROM messages");
|
|
const row = stmt.get() as { count: number };
|
|
return row.count;
|
|
}
|
|
|
|
/**
|
|
* Close the database connection.
|
|
*/
|
|
export function closeDb(): void {
|
|
if (db) {
|
|
db.close();
|
|
db = null;
|
|
}
|
|
}
|
|
|
|
function rowToMessageStats(row: any): MessageStats {
|
|
return {
|
|
id: row.id,
|
|
sessionFile: row.session_file,
|
|
entryId: row.entry_id,
|
|
folder: row.folder,
|
|
model: row.model,
|
|
provider: row.provider,
|
|
api: row.api,
|
|
timestamp: row.timestamp,
|
|
duration: row.duration,
|
|
ttft: row.ttft,
|
|
stopReason: row.stop_reason as any,
|
|
errorMessage: row.error_message,
|
|
usage: {
|
|
input: row.input_tokens,
|
|
output: row.output_tokens,
|
|
cacheRead: row.cache_read_tokens,
|
|
cacheWrite: row.cache_write_tokens,
|
|
totalTokens: row.total_tokens,
|
|
premiumRequests: row.premium_requests ?? 0,
|
|
cost: {
|
|
input: row.cost_input,
|
|
output: row.cost_output,
|
|
cacheRead: row.cost_cache_read,
|
|
cacheWrite: row.cost_cache_write,
|
|
total: row.cost_total,
|
|
},
|
|
},
|
|
};
|
|
}
|
|
|
|
export function getRecentRequests(limit = 100): MessageStats[] {
|
|
if (!db) return [];
|
|
const stmt = db.prepare(`
|
|
SELECT * FROM messages
|
|
ORDER BY timestamp DESC
|
|
LIMIT ?
|
|
`);
|
|
return (stmt.all(limit) as any[]).map(rowToMessageStats);
|
|
}
|
|
|
|
export function getRecentErrors(limit = 100): MessageStats[] {
|
|
if (!db) return [];
|
|
const stmt = db.prepare(`
|
|
SELECT * FROM messages
|
|
WHERE stop_reason = 'error'
|
|
ORDER BY timestamp DESC
|
|
LIMIT ?
|
|
`);
|
|
return (stmt.all(limit) as any[]).map(rowToMessageStats);
|
|
}
|
|
|
|
export function getMessageById(id: number): MessageStats | null {
|
|
if (!db) return null;
|
|
const stmt = db.prepare("SELECT * FROM messages WHERE id = ?");
|
|
const row = stmt.get(id);
|
|
return row ? rowToMessageStats(row) : null;
|
|
}
|
|
|
|
/**
|
|
* Get daily cost time series data for the last N days, broken down by model.
|
|
*/
|
|
export function getCostTimeSeries(days = 90, cutoff?: number | null): CostTimeSeriesPoint[] {
|
|
if (!db) return [];
|
|
|
|
const hasCutoff = cutoff !== null;
|
|
const seriesCutoff = hasCutoff ? (cutoff ?? Date.now() - days * 24 * 60 * 60 * 1000) : 0;
|
|
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
(timestamp / 86400000) * 86400000 as bucket,
|
|
model,
|
|
provider,
|
|
SUM(cost_total) as cost,
|
|
SUM(cost_input) as cost_input,
|
|
SUM(cost_output) as cost_output,
|
|
SUM(cost_cache_read) as cost_cache_read,
|
|
SUM(cost_cache_write) as cost_cache_write,
|
|
COUNT(*) as requests
|
|
FROM messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY bucket, model, provider
|
|
ORDER BY bucket ASC
|
|
`);
|
|
|
|
const rows = hasCutoff ? (stmt.all(seriesCutoff) as any[]) : (stmt.all() as any[]);
|
|
return rows.map(row => ({
|
|
timestamp: row.bucket,
|
|
model: row.model,
|
|
provider: row.provider,
|
|
cost: row.cost,
|
|
costInput: row.cost_input,
|
|
costOutput: row.cost_output,
|
|
costCacheRead: row.cost_cache_read,
|
|
costCacheWrite: row.cost_cache_write,
|
|
requests: row.requests,
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Reset `file_offsets` (and any existing `user_messages` rows) so the next
|
|
* successful sync re-parses every session and re-derives behavioral metrics.
|
|
* Run once per metric-definition bump; the meta sentinel is only marked
|
|
* complete after `syncAllSessions` finishes. Older timestamp sentinel values
|
|
* are treated as pending so a failed compiled-binary sync cannot permanently
|
|
* suppress the backfill.
|
|
*
|
|
* - v1: initial introduction of `user_messages`.
|
|
* - v2: yelling-sentence metric replaces caps-word counts; existing rows are
|
|
* computed under the old definition and must be discarded.
|
|
* - v3: drama runs collapsed into `anguish` (drama + elongated interjections
|
|
* + `dude` + dot runs), scored on a stripped prose body and gated on
|
|
* line count. Existing rows used the narrower definition.
|
|
* - v4: added `negation` / `repetition` / `blame` signals and fixed a
|
|
* latent word-boundary bug in the profanity / anguish regexes that had
|
|
* left those metrics matching nothing in real prose.
|
|
* - v5: renamed `yelling_sentences` column to `yelling` to match the other
|
|
* single-word signal columns (profanity, anguish, negation, ...).
|
|
*
|
|
* Existing `messages` rows are unaffected - `INSERT OR IGNORE` keeps them.
|
|
*/
|
|
function backfillUserMessages(database: Database): void {
|
|
const row = database.prepare("SELECT value FROM meta WHERE key = ?").get(USER_MESSAGES_BACKFILL_KEY) as
|
|
| { value: string }
|
|
| undefined;
|
|
if (!shouldResetBackfill(row?.value)) return;
|
|
|
|
database.exec("DELETE FROM user_messages");
|
|
database.exec("DELETE FROM file_offsets");
|
|
database
|
|
.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)")
|
|
.run(USER_MESSAGES_BACKFILL_KEY, BACKFILL_PENDING);
|
|
}
|
|
|
|
/**
|
|
* One-shot wipe of `file_offsets` to force `parseSessionFile` to re-parse
|
|
* every session from byte zero. We don't touch `user_messages`; the parser
|
|
* now emits a `UserMessageLink` for every assistant->parent pair, and the
|
|
* guarded `updateUserMessageLinks` UPDATE fixes any row whose `model` was
|
|
* left NULL by the old in-pass-only linking logic. Idempotent: gated by a
|
|
* sentinel row in `meta`.
|
|
*/
|
|
function repairUserMessageLinks(database: Database): void {
|
|
const row = database.prepare("SELECT value FROM meta WHERE key = ?").get(USER_MESSAGE_LINKS_REPAIR_KEY) as
|
|
| { value: string }
|
|
| undefined;
|
|
if (!shouldResetBackfill(row?.value)) return;
|
|
|
|
database.exec("DELETE FROM file_offsets");
|
|
database
|
|
.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)")
|
|
.run(USER_MESSAGE_LINKS_REPAIR_KEY, BACKFILL_PENDING);
|
|
}
|
|
|
|
/**
|
|
* One-shot wipe of `file_offsets` so the next sync re-parses every session
|
|
* and re-derives `premium_requests` from recorded `service_tier_change`
|
|
* entries. Earlier ingestions captured priority OpenAI traffic with
|
|
* `premium_requests = 0` because the AI layer only set the field for GitHub
|
|
* Copilot traffic. The parser now folds priority requests into the same
|
|
* counter; combined with the UPSERT in `insertMessageStats`, a single sync
|
|
* pass brings the messages table up to date without touching any other
|
|
* column. Idempotent: gated by a sentinel row in `meta`.
|
|
*/
|
|
function backfillPriorityPremiumRequests(database: Database): void {
|
|
const row = database.prepare("SELECT value FROM meta WHERE key = ?").get(PRIORITY_PREMIUM_REQUESTS_BACKFILL_KEY) as
|
|
| { value: string }
|
|
| undefined;
|
|
if (!shouldResetBackfill(row?.value)) return;
|
|
|
|
database.exec("DELETE FROM file_offsets");
|
|
database
|
|
.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)")
|
|
.run(PRIORITY_PREMIUM_REQUESTS_BACKFILL_KEY, BACKFILL_PENDING);
|
|
}
|
|
|
|
export function markPriorityPremiumRequestsBackfillComplete(): void {
|
|
if (!db) return;
|
|
db.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)").run(
|
|
PRIORITY_PREMIUM_REQUESTS_BACKFILL_KEY,
|
|
BACKFILL_COMPLETE,
|
|
);
|
|
}
|
|
|
|
export function markUserMessagesBackfillComplete(): void {
|
|
if (!db) return;
|
|
db.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)").run(
|
|
USER_MESSAGES_BACKFILL_KEY,
|
|
BACKFILL_COMPLETE,
|
|
);
|
|
}
|
|
|
|
export function markUserMessageLinksRepairComplete(): void {
|
|
if (!db) return;
|
|
db.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)").run(
|
|
USER_MESSAGE_LINKS_REPAIR_KEY,
|
|
BACKFILL_COMPLETE,
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Insert user-message stats. Idempotent via UNIQUE(session_file, entry_id).
|
|
*/
|
|
export function insertUserMessageStats(stats: UserMessageStats[]): number {
|
|
if (!db || stats.length === 0) return 0;
|
|
|
|
const stmt = db.prepare(`
|
|
INSERT OR IGNORE INTO user_messages (
|
|
session_file, entry_id, folder, timestamp, model, provider,
|
|
chars, words, yelling, profanity, anguish,
|
|
negation, repetition, blame
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`);
|
|
|
|
let inserted = 0;
|
|
const insert = db.transaction(() => {
|
|
for (const s of stats) {
|
|
const result = stmt.run(
|
|
s.sessionFile,
|
|
s.entryId,
|
|
s.folder,
|
|
s.timestamp,
|
|
s.model,
|
|
s.provider,
|
|
s.chars,
|
|
s.words,
|
|
s.yelling,
|
|
s.profanity,
|
|
s.anguish,
|
|
s.negation,
|
|
s.repetition,
|
|
s.blame,
|
|
);
|
|
if (result.changes > 0) inserted++;
|
|
}
|
|
});
|
|
insert();
|
|
return inserted;
|
|
}
|
|
|
|
/**
|
|
* Backfill the responding `model`/`provider` on user-message rows that were
|
|
* persisted before their assistant reply was parsed (a side effect of
|
|
* incremental `fromOffset` syncing: the `userByEntryId` map in
|
|
* `parseSessionFile` only spans a single pass). Each row is updated at most
|
|
* once because the `model IS NULL` guard short-circuits subsequent passes.
|
|
*
|
|
* Returns the number of rows actually updated.
|
|
*/
|
|
export function updateUserMessageLinks(links: UserMessageLink[]): number {
|
|
if (!db || links.length === 0) return 0;
|
|
|
|
const stmt = db.prepare(`
|
|
UPDATE user_messages
|
|
SET model = ?, provider = ?
|
|
WHERE session_file = ? AND entry_id = ? AND model IS NULL
|
|
`);
|
|
|
|
let updated = 0;
|
|
const apply = db.transaction(() => {
|
|
for (const link of links) {
|
|
const result = stmt.run(link.model, link.provider, link.sessionFile, link.entryId);
|
|
if (result.changes > 0) updated++;
|
|
}
|
|
});
|
|
apply();
|
|
return updated;
|
|
}
|
|
|
|
const UNKNOWN_MODEL = "unknown";
|
|
|
|
interface BehaviorSeriesRow {
|
|
bucket: number;
|
|
model: string;
|
|
provider: string;
|
|
messages: number;
|
|
yelling: number | null;
|
|
profanity: number | null;
|
|
anguish: number | null;
|
|
negation: number | null;
|
|
repetition: number | null;
|
|
blame: number | null;
|
|
chars: number | null;
|
|
}
|
|
|
|
/**
|
|
* Daily behavioral time series, grouped by responding model+provider.
|
|
*/
|
|
export function getBehaviorTimeSeries(cutoff?: number | null): BehaviorTimeSeriesPoint[] {
|
|
if (!db) return [];
|
|
const hasCutoff = cutoff !== null && cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
(timestamp / 86400000) * 86400000 as bucket,
|
|
COALESCE(model, ?) as model,
|
|
COALESCE(provider, ?) as provider,
|
|
COUNT(*) as messages,
|
|
SUM(yelling) as yelling,
|
|
SUM(profanity) as profanity,
|
|
SUM(anguish) as anguish,
|
|
SUM(negation) as negation,
|
|
SUM(repetition) as repetition,
|
|
SUM(blame) as blame,
|
|
SUM(chars) as chars
|
|
FROM user_messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY bucket, model, provider
|
|
ORDER BY bucket ASC
|
|
`);
|
|
const rows = (
|
|
hasCutoff ? stmt.all(UNKNOWN_MODEL, UNKNOWN_MODEL, cutoff) : stmt.all(UNKNOWN_MODEL, UNKNOWN_MODEL)
|
|
) as BehaviorSeriesRow[];
|
|
return rows.map(row => ({
|
|
timestamp: row.bucket,
|
|
model: row.model,
|
|
provider: row.provider,
|
|
messages: row.messages,
|
|
yelling: row.yelling ?? 0,
|
|
profanity: row.profanity ?? 0,
|
|
anguish: row.anguish ?? 0,
|
|
negation: row.negation ?? 0,
|
|
repetition: row.repetition ?? 0,
|
|
blame: row.blame ?? 0,
|
|
chars: row.chars ?? 0,
|
|
}));
|
|
}
|
|
|
|
interface BehaviorOverallRow {
|
|
total_messages: number;
|
|
total_yelling: number | null;
|
|
total_profanity: number | null;
|
|
total_anguish: number | null;
|
|
total_negation: number | null;
|
|
total_repetition: number | null;
|
|
total_blame: number | null;
|
|
total_chars: number | null;
|
|
first_timestamp: number | null;
|
|
last_timestamp: number | null;
|
|
}
|
|
|
|
/**
|
|
* Overall behavioral totals across the cutoff window.
|
|
*/
|
|
export function getBehaviorOverall(cutoff?: number | null): BehaviorOverallStats {
|
|
const empty: BehaviorOverallStats = {
|
|
totalMessages: 0,
|
|
totalYelling: 0,
|
|
totalProfanity: 0,
|
|
totalAnguish: 0,
|
|
totalNegation: 0,
|
|
totalRepetition: 0,
|
|
totalBlame: 0,
|
|
totalChars: 0,
|
|
firstTimestamp: 0,
|
|
lastTimestamp: 0,
|
|
};
|
|
if (!db) return empty;
|
|
const hasCutoff = cutoff !== null && cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
COUNT(*) as total_messages,
|
|
SUM(yelling) as total_yelling,
|
|
SUM(profanity) as total_profanity,
|
|
SUM(anguish) as total_anguish,
|
|
SUM(negation) as total_negation,
|
|
SUM(repetition) as total_repetition,
|
|
SUM(blame) as total_blame,
|
|
SUM(chars) as total_chars,
|
|
MIN(timestamp) as first_timestamp,
|
|
MAX(timestamp) as last_timestamp
|
|
FROM user_messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
`);
|
|
const row = (hasCutoff ? stmt.get(cutoff) : stmt.get()) as BehaviorOverallRow | undefined;
|
|
if (!row?.total_messages) return empty;
|
|
return {
|
|
totalMessages: row.total_messages,
|
|
totalYelling: row.total_yelling ?? 0,
|
|
totalProfanity: row.total_profanity ?? 0,
|
|
totalAnguish: row.total_anguish ?? 0,
|
|
totalNegation: row.total_negation ?? 0,
|
|
totalRepetition: row.total_repetition ?? 0,
|
|
totalBlame: row.total_blame ?? 0,
|
|
totalChars: row.total_chars ?? 0,
|
|
firstTimestamp: row.first_timestamp ?? 0,
|
|
lastTimestamp: row.last_timestamp ?? 0,
|
|
};
|
|
}
|
|
|
|
interface BehaviorByModelRow {
|
|
model: string;
|
|
provider: string;
|
|
total_messages: number;
|
|
total_yelling: number | null;
|
|
total_profanity: number | null;
|
|
total_anguish: number | null;
|
|
total_negation: number | null;
|
|
total_repetition: number | null;
|
|
total_blame: number | null;
|
|
total_chars: number | null;
|
|
last_timestamp: number | null;
|
|
}
|
|
|
|
/**
|
|
* Per-model behavioral totals over the cutoff window. "Unknown" represents
|
|
* user messages that never received an assistant reply.
|
|
*/
|
|
export function getBehaviorByModel(cutoff?: number | null): BehaviorModelStats[] {
|
|
if (!db) return [];
|
|
const hasCutoff = cutoff !== null && cutoff !== undefined && cutoff > 0;
|
|
const stmt = db.prepare(`
|
|
SELECT
|
|
COALESCE(model, ?) as model,
|
|
COALESCE(provider, ?) as provider,
|
|
COUNT(*) as total_messages,
|
|
SUM(yelling) as total_yelling,
|
|
SUM(profanity) as total_profanity,
|
|
SUM(anguish) as total_anguish,
|
|
SUM(negation) as total_negation,
|
|
SUM(repetition) as total_repetition,
|
|
SUM(blame) as total_blame,
|
|
SUM(chars) as total_chars,
|
|
MAX(timestamp) as last_timestamp
|
|
FROM user_messages
|
|
${hasCutoff ? "WHERE timestamp >= ?" : ""}
|
|
GROUP BY model, provider
|
|
ORDER BY total_messages DESC
|
|
`);
|
|
const rows = (
|
|
hasCutoff ? stmt.all(UNKNOWN_MODEL, UNKNOWN_MODEL, cutoff) : stmt.all(UNKNOWN_MODEL, UNKNOWN_MODEL)
|
|
) as BehaviorByModelRow[];
|
|
return rows.map(row => ({
|
|
model: row.model,
|
|
provider: row.provider,
|
|
totalMessages: row.total_messages,
|
|
totalYelling: row.total_yelling ?? 0,
|
|
totalProfanity: row.total_profanity ?? 0,
|
|
totalAnguish: row.total_anguish ?? 0,
|
|
totalNegation: row.total_negation ?? 0,
|
|
totalRepetition: row.total_repetition ?? 0,
|
|
totalBlame: row.total_blame ?? 0,
|
|
totalChars: row.total_chars ?? 0,
|
|
lastTimestamp: row.last_timestamp ?? 0,
|
|
}));
|
|
}
|