diff --git a/devlog/_plan/260804_router_intelligence/001_pr_stack_status.md b/devlog/_plan/260804_router_intelligence/001_pr_stack_status.md index 11c8d3fb57..f687d020d1 100644 --- a/devlog/_plan/260804_router_intelligence/001_pr_stack_status.md +++ b/devlog/_plan/260804_router_intelligence/001_pr_stack_status.md @@ -39,8 +39,8 @@ other; closing one is a maintainer decision and neither is stale. | RI | Branch | Base | Head SHA | PR | URL | Status | |---|---|---|---|---|---|---| -| RI-01 | `feat/ri-01-route-decision-traces` | `e44d234f0` | pending | pending | pending | in progress | -| RI-02 | `feat/ri-02-request-history-index` | `feat/ri-01` head | pending | pending | pending | queued | +| RI-01 | `feat/ri-01-route-decision-traces` | `e44d234f0` | `b5a8e7c4c` | #1003 | https://github.com/lidge-jun/opencodex/pull/1003 | MERGED | +| RI-02 | `feat/ri-02-request-history-index` | `dev` (post-#1003 merge) | `03b0eafa7` | #1004 | https://github.com/lidge-jun/opencodex/pull/1004 | OPEN | | RI-03 | `feat/ri-03-routing-analytics` | `feat/ri-02` head | pending | pending | pending | queued | | RI-04 | `feat/ri-04-policy-profile-core` | `feat/ri-03` head | pending | pending | pending | queued | | RI-05 | `feat/ri-05-capability-aware-routing` | `feat/ri-04` head | pending | pending | pending | queued | @@ -90,4 +90,34 @@ other; closing one is a maintainer decision and neither is stale. ### RI-02..RI-10 -Appended as each PR is implemented. +### RI-02 - feat/ri-02-request-history-index + +- Base SHA: `34d21b1bc` (`dev` after #1003 merge; rebased from RI-01 head + `b5a8e7c4c` when #1003 landed: `7efb6e842` -> `03b0eafa7`) +- Reviewed commit: same as final (author self-review before push) +- Findings (self-review): 4 defects caught pre-push - + 1. `destroyAndRecreate` never reassigned the fresh handle to module `db` + (first-open rebuild crashed); + 2. bun:sqlite named-parameter objects silently failed to bind for + `LIMIT $x` and INSERT statements (datatype mismatch / silent no-op) - + query and insert paths switched to positional parameters; + 3. Windows file locking: an unfinalized prepared statement kept the DB + locked after close (EBUSY in tests) - insert statement now finalizes; + a partially-opened handle on a corrupt file is closed before recreate; + 4. duplicate-replay accounting counted ignored rows in `indexedRows` - + now counts real `INSERT` changes. +- Fixes: all four above; tests cover every one. +- PR: #1004 (OPEN) https://github.com/lidge-jun/opencodex/pull/1004 +- Final commit: recorded after review round (rebase + CodeRabbit/simplify + fixes; new head pushes to #1004) +- Verification: + - `bun x tsc --noEmit`: PASSED (0 errors) + - `bun run test tests/request-history-index.test.ts`: 16/16 pass + (1574 assertions) covering the mandatory matrix: empty/missing/corrupt/ + old-schema/partial-line/replacement/truncation/duplicate-replay/cursor + stability/invalid-cursor/page-bounds/rebuild-equivalence/filters/row-by-id + - Focused regression suites: 269/269 pass across 8 files (incl. RI-01 + tests, request-log, usage-log, combos, combo-management-api, + codex-routing, codex-account-namespaces) + - `bun run privacy:scan`: passed +- Remaining Low findings: none diff --git a/src/cli/observe.ts b/src/cli/observe.ts index 9f07e44aaf..1e2c47df30 100644 --- a/src/cli/observe.ts +++ b/src/cli/observe.ts @@ -14,6 +14,8 @@ import { const USAGE = `Usage: ocx observe logs [--provider ] [--model ] [--status ] [--limit ] [--follow] [--json|--jsonl] + ocx logs rebuild-index + ocx logs index-status ocx observe usage [--range <7d|30d|all>] [--surface ] [--json] ocx observe storage [--json] ocx observe memory [--json] @@ -79,6 +81,39 @@ async function logs(argv: string[], deps: RuntimeApiDeps): Promise { } while (true); } +async function rebuildIndex(argv: string[], deps: RuntimeApiDeps): Promise { + const args = [...argv]; + const wantsJson = takeFlag(args, "--json"); + rejectArgs(args, USAGE); + const { rebuildRequestHistoryIndex } = await import("../routing/history/indexer"); + const meta = await rebuildRequestHistoryIndex(); + if (wantsJson) printData(meta, true); + else { + console.log(`Request-history index rebuilt (${meta.dbPath})`); + console.log(` schema version: ${meta.schemaVersion}`); + console.log(` indexed rows: ${meta.indexedRows}`); + console.log(` source size: ${meta.sourceSize} bytes`); + console.log(` last error: ${meta.lastError ?? "none"}`); + } +} + +async function indexStatus(argv: string[], deps: RuntimeApiDeps): Promise { + const args = [...argv]; + const wantsJson = takeFlag(args, "--json"); + rejectArgs(args, USAGE); + const { requestHistoryIndexStatus } = await import("../routing/history/indexer"); + const meta = await requestHistoryIndexStatus(); + if (wantsJson) printData(meta, true); + else { + console.log(`Request-history index (${meta.dbPath})`); + console.log(` schema version: ${meta.schemaVersion}`); + console.log(` indexed rows: ${meta.indexedRows}`); + console.log(` source size: ${meta.sourceSize} bytes`); + console.log(` indexed offset: ${meta.indexedOffset} bytes`); + console.log(` last error: ${meta.lastError ?? "none"}`); + } +} + async function usage(argv: string[], deps: RuntimeApiDeps): Promise { const args = [...argv]; const wantsJson = takeFlag(args, "--json"); @@ -103,7 +138,12 @@ async function simple(path: string, argv: string[], deps: RuntimeApiDeps): Promi export async function handleObserveCommand(argv: string[], deps: RuntimeApiDeps = {}): Promise { return runCliAction(async () => { const [sub = "logs", ...rest] = argv; - if (sub === "logs") await logs(rest, deps); + if (sub === "logs") { + const action = rest[0]; + if (action === "rebuild-index") await rebuildIndex(rest.slice(1), deps); + else if (action === "index-status") await indexStatus(rest.slice(1), deps); + else await logs(rest, deps); + } else if (sub === "usage") await usage(rest, deps); else if (sub === "storage") await simple("/api/storage", rest, deps); else if (sub === "memory") await simple("/api/system/memory", rest, deps); diff --git a/src/routing/history/cursor.ts b/src/routing/history/cursor.ts new file mode 100644 index 0000000000..b56c0ad068 --- /dev/null +++ b/src/routing/history/cursor.ts @@ -0,0 +1,43 @@ +/** + * Opaque keyset cursor for request-history pagination (RI-02, ADR-9). + * + * Ordering is `timestamp DESC, request_id DESC`; the cursor encodes the last + * returned row's `(timestamp, requestId)` pair as base64url JSON. Cursors are + * opaque to clients: any decode failure or shape mismatch yields `null` and + * the API answers `400 invalid_cursor` instead of guessing. + */ + +export interface HistoryCursor { + t: number; + i: string; +} + +export class InvalidCursorError extends Error { + readonly code = "invalid_cursor" as const; + + constructor() { + super("invalid_cursor"); + this.name = "InvalidCursorError"; + } +} + +export function encodeHistoryCursor(cursor: HistoryCursor): string { + return Buffer.from(JSON.stringify(cursor)).toString("base64url"); +} + +export function decodeHistoryCursor(raw: string | null | undefined): HistoryCursor | null { + if (typeof raw !== "string" || raw.length === 0 || raw.length > 4096) return null; + let parsed: unknown; + try { + parsed = JSON.parse(Buffer.from(raw, "base64url").toString("utf-8")); + } catch { + return null; + } + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return null; + const record = parsed as Record; + const t = record.t; + const i = record.i; + if (typeof t !== "number" || !Number.isFinite(t)) return null; + if (typeof i !== "string" || i.length === 0 || i.length > 256) return null; + return { t, i }; +} diff --git a/src/routing/history/indexer.ts b/src/routing/history/indexer.ts new file mode 100644 index 0000000000..6ccb7547f9 --- /dev/null +++ b/src/routing/history/indexer.ts @@ -0,0 +1,569 @@ +/** + * Derived request-history index (RI-02). + * + * `usage.jsonl` stays canonical; this module maintains a rebuildable SQLite + * projection (ADR-1, ADR-8). On every open/query it verifies schema version, + * file identity, integrity, and byte offset, then appends whatever complete + * JSONL rows arrived since the last index. Missing/corrupt/stale index or a + * replaced/truncated source triggers an automatic full rebuild; canonical + * history is never touched. + */ + +import { Database } from "bun:sqlite"; +import { + closeSync, + existsSync, + fstatSync, + openSync, + readSync, + unlinkSync, +} from "node:fs"; +import { getConfigDir } from "../../config"; +import { recordOwnedConfigPath } from "../../lib/config-ownership"; +import { + currentUsageLogRevision, + normalizeUsageEntryForTest, + usageLogPath, + type PersistedUsageEntry, + type UsageLogRevision, +} from "../../usage/log"; +import { + HISTORY_DDL, + HISTORY_META_KEYS, + HISTORY_SCHEMA_VERSION, + historyIndexPath, +} from "./schema"; +import { + decodeHistoryCursor, + encodeHistoryCursor, + InvalidCursorError, + type HistoryCursor, +} from "./cursor"; + +export interface RequestHistoryIndexMeta { + schemaVersion: number; + dbPath: string; + sourceSize: number; + sourceMtimeMs: number; + indexedOffset: number; + indexedRows: number; + builtAtMs: number; + lastError: string | null; +} + +export interface RequestHistoryFilters { + provider?: string; + model?: string; + requestedModel?: string; + status?: number; + conversationId?: string; + surface?: string; + inboundProtocol?: string; + apiKeyId?: string; + profileId?: string; + fallback?: boolean; + from?: number; + to?: number; +} + +export interface RequestHistoryPage { + rows: PersistedUsageEntry[]; + nextCursor?: string; + hasMore: boolean; + meta: RequestHistoryIndexMeta; +} + +export const REQUEST_HISTORY_MAX_PAGE_SIZE = 100; +export const REQUEST_HISTORY_DEFAULT_PAGE_SIZE = 50; +export const REQUEST_HISTORY_INSERT_BATCH = 500; + +let db: Database | null = null; +let dbPath = ""; +let openPromise: Promise | null = null; + +function indexDbPath(): string { + const dir = getConfigDir(); + const path = historyIndexPath(dir); + recordOwnedConfigPath(dir, path); + return path; +} + +function metaValue(dbHandle: Database, key: string): string | null { + const row = dbHandle.query("SELECT value FROM schema_meta WHERE key = ?").get(key) as + | { value: string } + | undefined; + return row?.value ?? null; +} + +function setMeta(dbHandle: Database, key: string, value: string | number): void { + dbHandle.query( + "INSERT INTO schema_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", + ).run(key, String(value)); +} + +function readIndexedMeta(dbHandle: Database): Omit { + const asNumber = (key: string): number => { + const value = metaValue(dbHandle, key); + const parsed = value === null ? NaN : Number(value); + return Number.isFinite(parsed) ? parsed : 0; + }; + return { + schemaVersion: asNumber(HISTORY_META_KEYS.schemaVersion) || HISTORY_SCHEMA_VERSION, + sourceSize: asNumber(HISTORY_META_KEYS.sourceSize), + sourceMtimeMs: asNumber(HISTORY_META_KEYS.sourceMtimeMs), + indexedOffset: asNumber(HISTORY_META_KEYS.indexedOffset), + indexedRows: asNumber(HISTORY_META_KEYS.indexedRows), + builtAtMs: asNumber(HISTORY_META_KEYS.builtAtMs), + lastError: metaValue(dbHandle, HISTORY_META_KEYS.lastError), + }; +} + +function metaFor(dbHandle: Database): RequestHistoryIndexMeta { + return { dbPath, ...readIndexedMeta(dbHandle) }; +} + +function sourceIdentity(): UsageLogRevision | null { + return currentUsageLogRevision(); +} + +function readStoredMetaField(dbHandle: Database, key: string): string { + return metaValue(dbHandle, key) ?? ""; +} + +function sourceIdentityMatches(dbHandle: Database, revision: UsageLogRevision | null): boolean { + const stored = readIndexedMeta(dbHandle); + if (revision === null) return stored.sourceSize === 0; + const storedPath = readStoredMetaField(dbHandle, HISTORY_META_KEYS.sourcePath); + if (!storedPath) return false; + const storedDev = Number(readStoredMetaField(dbHandle, HISTORY_META_KEYS.sourceDev)); + const storedIno = Number(readStoredMetaField(dbHandle, HISTORY_META_KEYS.sourceIno)); + const storedBirthtimeMs = Number(readStoredMetaField(dbHandle, HISTORY_META_KEYS.sourceBirthtimeMs)); + return storedPath === revision.path + && storedDev === Number(revision.dev) + && storedIno === Number(revision.ino) + && storedBirthtimeMs === Number(revision.birthtimeMs); +} + +/** Extract the `requests` row columns from a canonical persisted entry. */ +function extractRow(entry: PersistedUsageEntry): Array { + const attempts = entry.attempts; + return [ + entry.requestId, + entry.timestamp, + entry.provider, + entry.model, + entry.requestedModel ?? null, + entry.status, + entry.surface ?? null, + entry.inboundProtocol ?? null, + entry.apiKeyId ?? null, + entry.conversationId ?? null, + entry.routeDecision?.routeKind ?? null, + entry.routeDecision?.profile?.id ?? null, + entry.routeDecision?.profile?.revision ?? null, + (attempts?.length ?? 0) > 1 ? 1 : 0, + entry.durationMs, + entry.firstOutputMs ?? null, + entry.usageStatus, + entry.usage ? JSON.stringify(entry.usage) : null, + entry.totalTokens ?? null, + entry.errorCode ?? null, + entry.terminalStatus ?? null, + entry.closeReason ?? null, + attempts?.length ?? 1, + entry.routeDecision ? JSON.stringify(entry.routeDecision) : null, + JSON.stringify(entry), + ]; +} + +const ROW_INSERT = ` +INSERT OR IGNORE INTO requests ( + request_id, timestamp, provider, model, requested_model, status, surface, + inbound_protocol, api_key_id, conversation_id, route_kind, profile_id, + profile_revision, fallback, duration_ms, first_output_ms, usage_status, + usage_json, total_tokens, error_code, terminal_status, close_reason, + attempt_count, decision_json, row_json +) VALUES ( + ?, ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ? +)`; + +function readCompleteTail(fd: number, fromOffset: number, size: number): { text: string; nextOffset: number } { + if (size <= fromOffset) return { text: "", nextOffset: fromOffset }; + const length = size - fromOffset; + const buf = Buffer.allocUnsafe(length); + let offset = 0; + while (offset < length) { + const read = readSync(fd, buf, offset, length - offset, fromOffset + offset); + if (read === 0) throw new Error("usage log changed while indexing"); + offset += read; + } + const newline = buf.lastIndexOf(0x0a); + if (newline < 0) return { text: "", nextOffset: fromOffset }; + return { + text: buf.subarray(0, newline).toString("utf-8"), + nextOffset: fromOffset + newline + 1, + }; +} + +function parsedEntryFromLine(line: string): PersistedUsageEntry | null { + if (!line.trim()) return null; + try { + const parsed = JSON.parse(line) as PersistedUsageEntry; + if (parsed && typeof parsed === "object" + && typeof parsed.requestId === "string" + && typeof parsed.timestamp === "number" + && typeof parsed.provider === "string") { + return parsed; + } + } catch { + /* skip partial / hand-edited lines, same as every other reader */ + } + return null; +} + +function ingestText(dbHandle: Database, text: string): number { + if (!text) return 0; + const lines = text.split(/\r?\n/); + let inserted = 0; + let pending: Array> = []; + const insert = dbHandle.prepare(ROW_INSERT); + try { + const commitBatch = () => { + dbHandle.transaction((rows: Array>) => { + for (const row of rows) { + // `changes` counts real inserts only; INSERT OR IGNORE replays add 0. + inserted += insert.run(...row).changes; + } + })(pending); + pending = []; + }; + for (const line of lines) { + const entry = parsedEntryFromLine(line); + if (!entry) continue; + pending.push(extractRow(entry)); + if (pending.length >= REQUEST_HISTORY_INSERT_BATCH) commitBatch(); + } + if (pending.length > 0) commitBatch(); + } finally { + // Windows file locks: an unterminated prepared statement keeps the DB + // file busy after close (verified on Bun 1.3.14). Finalize always. + insert.finalize(); + } + return inserted; +} + +function ingestSourceTail(dbHandle: Database, path: string, fromOffset: number): number { + let fd: number | undefined; + try { + fd = openSync(path, "r"); + const stat = fstatSync(fd); + const { text, nextOffset } = readCompleteTail(fd, fromOffset, Number(stat.size)); + if (text.length === 0 && nextOffset === fromOffset) return 0; + const inserted = ingestText(dbHandle, text); + const current = readIndexedMeta(dbHandle); + setMeta(dbHandle, HISTORY_META_KEYS.indexedOffset, nextOffset); + setMeta(dbHandle, HISTORY_META_KEYS.indexedRows, current.indexedRows + inserted); + setMeta(dbHandle, HISTORY_META_KEYS.sourceSize, Number(stat.size)); + setMeta(dbHandle, HISTORY_META_KEYS.sourceMtimeMs, Number(stat.mtimeMs)); + setMeta(dbHandle, HISTORY_META_KEYS.builtAtMs, Date.now()); + return inserted; + } finally { + if (fd !== undefined) closeSync(fd); + } +} + +function resetAndCreateSchema(dbHandle: Database): void { + dbHandle.exec("DROP TABLE IF EXISTS requests"); + dbHandle.exec("DROP TABLE IF EXISTS schema_meta"); + dbHandle.exec(HISTORY_DDL); + setMeta(dbHandle, HISTORY_META_KEYS.schemaVersion, HISTORY_SCHEMA_VERSION); + setMeta(dbHandle, HISTORY_META_KEYS.indexedOffset, 0); + setMeta(dbHandle, HISTORY_META_KEYS.indexedRows, 0); + setMeta(dbHandle, HISTORY_META_KEYS.builtAtMs, Date.now()); + setMeta(dbHandle, HISTORY_META_KEYS.lastError, "rebuilt"); +} + +function recordSourceMeta(dbHandle: Database, revision: UsageLogRevision | null): void { + setMeta(dbHandle, HISTORY_META_KEYS.sourcePath, revision?.path ?? ""); + setMeta(dbHandle, HISTORY_META_KEYS.sourceDev, revision?.dev ?? 0); + setMeta(dbHandle, HISTORY_META_KEYS.sourceIno, revision?.ino ?? 0); + setMeta(dbHandle, HISTORY_META_KEYS.sourceBirthtimeMs, revision?.birthtimeMs ?? 0); + setMeta(dbHandle, HISTORY_META_KEYS.sourceSize, revision?.size ?? 0); + setMeta(dbHandle, HISTORY_META_KEYS.sourceMtimeMs, revision?.mtimeMs ?? 0); +} + +function isHealthy(dbHandle: Database): boolean { + try { + const row = dbHandle.query("PRAGMA quick_check").get() as { quick_check?: string } | undefined; + return row?.quick_check === "ok"; + } catch { + return false; + } +} + +function destroyAndRecreate(path: string, reason: string): Database { + if (db) { + try { db.close(); } catch { /* already closed */ } + db = null; + } + // Windows: a partially-opened handle from a failed `new Database` can hold + // the file briefly after the throw. Retry the unlink before recreating. + for (let attempt = 0; attempt < 5; attempt++) { + try { + for (const suffix of ["-wal", "-shm"] as const) { + try { unlinkSync(`${path}${suffix}`); } catch { /* sidecar may not exist */ } + } + unlinkSync(path); + break; + } catch { + if (attempt === 4) break; + Bun.sleepSync(50); + } + } + const fresh = new Database(path, { create: true }); + fresh.exec("PRAGMA journal_mode = WAL"); + fresh.exec("PRAGMA busy_timeout = 5000"); + resetAndCreateSchema(fresh); + setMeta(fresh, HISTORY_META_KEYS.lastError, reason); + db = fresh; + return fresh; +} + +function openIndexDb(): Database { + const path = indexDbPath(); + if (db) return db; + let handle: Database | undefined; + try { + handle = new Database(path, { create: true }); + handle.exec("PRAGMA journal_mode = WAL"); + handle.exec("PRAGMA busy_timeout = 5000"); + handle.exec(HISTORY_DDL); + if (metaValue(handle, HISTORY_META_KEYS.schemaVersion) === null) { + // Fresh database: record the schema version so the next refresh treats + // it as current instead of destroying the file we just created. + setMeta(handle, HISTORY_META_KEYS.schemaVersion, HISTORY_SCHEMA_VERSION); + setMeta(handle, HISTORY_META_KEYS.indexedOffset, 0); + setMeta(handle, HISTORY_META_KEYS.indexedRows, 0); + setMeta(handle, HISTORY_META_KEYS.builtAtMs, Date.now()); + setMeta(handle, HISTORY_META_KEYS.lastError, "created"); + } + } catch { + // A partially-opened handle on a corrupt file can hold the OS lock on + // Windows; close it before the destructive recreate. + if (handle) { + try { handle.close(); } catch { /* already unusable */ } + } + handle = destroyAndRecreate(path, "unreadable database recreated"); + } + dbPath = path; + db = handle; + return handle; +} + +function ensureSchemaAndIdentity(dbHandle: Database): "ready" | "rebuilt" { + try { + const storedVersion = metaValue(dbHandle, HISTORY_META_KEYS.schemaVersion); + if (storedVersion !== String(HISTORY_SCHEMA_VERSION)) { + destroyAndRecreate(dbPath, `schema version ${storedVersion ?? "missing"} -> ${HISTORY_SCHEMA_VERSION}`); + return "rebuilt"; + } + if (!isHealthy(dbHandle)) { + destroyAndRecreate(dbPath, "integrity check failed; index rebuilt"); + return "rebuilt"; + } + const revision = sourceIdentity(); + if (!sourceIdentityMatches(dbHandle, revision)) { + destroyAndRecreate(dbPath, "source identity changed; index rebuilt"); + return "rebuilt"; + } + return "ready"; + } catch { + // A file that opens but is not actually SQLite (or is mid-corruption) + // throws on the first statement; treat it as corrupt and rebuild. + destroyAndRecreate(dbPath, "index unreadable; rebuilt"); + return "rebuilt"; + } +} + +function fullRebuild(dbHandle: Database, reason: string): void { + resetAndCreateSchema(dbHandle); + setMeta(dbHandle, HISTORY_META_KEYS.lastError, reason); + const path = usageLogPath(); + const revision = sourceIdentity(); + recordSourceMeta(dbHandle, revision); + if (!existsSync(path)) { + setMeta(dbHandle, HISTORY_META_KEYS.sourceSize, 0); + setMeta(dbHandle, HISTORY_META_KEYS.indexedOffset, 0); + setMeta(dbHandle, HISTORY_META_KEYS.indexedRows, 0); + setMeta(dbHandle, HISTORY_META_KEYS.builtAtMs, Date.now()); + return; + } + const inserted = ingestSourceTail(dbHandle, path, 0); + const current = readIndexedMeta(dbHandle); + setMeta(dbHandle, HISTORY_META_KEYS.indexedRows, inserted); + setMeta(dbHandle, HISTORY_META_KEYS.indexedOffset, Math.max(current.indexedOffset, 0)); + setMeta(dbHandle, HISTORY_META_KEYS.builtAtMs, Date.now()); +} + +async function refreshLocked(): Promise { + openIndexDb(); + const state = ensureSchemaAndIdentity(db!); + const handle = db!; + const revision = sourceIdentity(); + if (state === "rebuilt") { + fullRebuild(db!, "rebuilt after identity/schema mismatch"); + return metaFor(db!); + } + const current = readIndexedMeta(handle); + if (revision === null) { + // Source gone: the derived index must not outlive its canonical ledger. + if (current.indexedRows > 0) fullRebuild(handle, "source ledger missing; index reset"); + return metaFor(handle); + } + const tailNextOffset = current.indexedOffset; + if (Number(revision.size) < tailNextOffset) { + // Truncated source: offsets no longer make sense. + fullRebuild(handle, "source truncated; index rebuilt"); + return metaFor(handle); + } + if (tailNextOffset < Number(revision.size)) { + ingestSourceTail(handle, revision.path, tailNextOffset); + } + return metaFor(handle); +} + +/** + * Open (and refresh) the index. Single-flight: concurrent callers share one + * refresh. Never throws for missing/corrupt index or ledger state; those are + * repaired or reflected in the returned meta. + */ +export function openRequestHistoryIndex(): Promise { + if (!openPromise) { + openPromise = refreshLocked().finally(() => { + openPromise = null; + }); + } + return openPromise; +} + +export function closeRequestHistoryIndex(): void { + if (db) { + try { db.close(); } catch { /* ignore */ } + db = null; + } + openPromise = null; +} + +/** Force a full rebuild from the canonical ledger (CLI / tests). */ +export async function rebuildRequestHistoryIndex(): Promise { + const handle = openIndexDb(); + fullRebuild(handle, "manual rebuild requested"); + return metaFor(handle); +} + +type HistoryQueryRow = { + row_json: string; + timestamp?: number; + request_id?: string; +}; + +function queryRows( + handle: Database, + filters: RequestHistoryFilters, + cursor: HistoryCursor | null, + limit: number, +): { rows: Array; fetched: number } { + const where: string[] = []; + const values: Array = []; + const add = (clause: string, value: string | number) => { + where.push(clause); + values.push(value); + }; + if (filters.provider !== undefined) add("provider = ?", filters.provider); + if (filters.model !== undefined) add("model = ?", filters.model); + if (filters.requestedModel !== undefined) add("requested_model = ?", filters.requestedModel); + if (filters.status !== undefined) add("status = ?", filters.status); + if (filters.conversationId !== undefined) add("conversation_id = ?", filters.conversationId); + if (filters.surface !== undefined) add("surface = ?", filters.surface); + if (filters.inboundProtocol !== undefined) add("inbound_protocol = ?", filters.inboundProtocol); + if (filters.apiKeyId !== undefined) add("api_key_id = ?", filters.apiKeyId); + if (filters.profileId !== undefined) add("profile_id = ?", filters.profileId); + if (filters.fallback !== undefined) add("fallback = ?", filters.fallback ? 1 : 0); + if (filters.from !== undefined) add("timestamp >= ?", filters.from); + if (filters.to !== undefined) add("timestamp <= ?", filters.to); + if (cursor) { + where.push("(timestamp < ? OR (timestamp = ? AND request_id < ?))"); + values.push(cursor.t, cursor.t, cursor.i); + } + const whereSql = where.length > 0 ? ` WHERE ${where.join(" AND ")}` : ""; + const rows = handle.query( + `SELECT row_json, timestamp, request_id FROM requests${whereSql} ORDER BY timestamp DESC, request_id DESC LIMIT ?`, + ).all(...values, limit + 1) as Array; + return { rows, fetched: rows.length }; +} + +function requireDb(): Database { + const handle = db; + if (!handle) throw new Error("request-history index is not open"); + return handle; +} + +function hydrateRow(row: HistoryQueryRow | undefined): PersistedUsageEntry | null { + if (!row) return null; + try { + const parsed = JSON.parse(row.row_json) as PersistedUsageEntry; + if (parsed && typeof parsed === "object" && typeof parsed.requestId === "string") { + return normalizeUsageEntryForTest(parsed); + } + } catch { + /* defensive: a damaged row is skipped, canonical ledger unaffected */ + } + return null; +} + +export async function queryRequestHistory( + filters: RequestHistoryFilters, + rawCursor: string | null | undefined, + pageSize: number | undefined, +): Promise { + const meta = await openRequestHistoryIndex(); + const cursor = decodeHistoryCursor(rawCursor); + if (rawCursor !== null && rawCursor !== undefined && cursor === null) { + throw new InvalidCursorError(); + } + const limit = Math.min( + Math.max(1, Math.trunc(pageSize ?? REQUEST_HISTORY_DEFAULT_PAGE_SIZE)), + REQUEST_HISTORY_MAX_PAGE_SIZE, + ); + const handle = requireDb(); + const { rows, fetched } = queryRows(handle, filters, cursor, limit); + const hasMore = fetched > limit; + const pageRows = rows.slice(0, limit); + const entries: PersistedUsageEntry[] = []; + for (const row of pageRows) { + const entry = hydrateRow(row); + if (entry) entries.push(entry); + } + let nextCursor: string | undefined; + if (hasMore && pageRows.length > 0) { + const last = pageRows[pageRows.length - 1]!; + nextCursor = encodeHistoryCursor({ t: last.timestamp!, i: last.request_id! }); + } + return { rows: entries, ...(nextCursor ? { nextCursor } : {}), hasMore, meta }; +} + +export async function requestHistoryRowById(requestId: string): Promise { + await openRequestHistoryIndex(); + const handle = requireDb(); + const row = handle.query("SELECT row_json FROM requests WHERE request_id = ?").get(requestId) as + | { row_json: string } + | undefined; + return hydrateRow(row); +} + +export async function requestHistoryIndexStatus(): Promise { + return openRequestHistoryIndex(); +} diff --git a/src/routing/history/schema.ts b/src/routing/history/schema.ts new file mode 100644 index 0000000000..7ae3e69ed3 --- /dev/null +++ b/src/routing/history/schema.ts @@ -0,0 +1,72 @@ +/** + * Schema contract for the derived request-history index (RI-02). + * + * `usage.jsonl` remains the canonical append-only evidence ledger; + * `routing-history.sqlite` is a disposable, rebuildable query projection + * (ADR-1/ADR-8 in devlog/_plan/260804_router_intelligence/000_master_plan.md). + */ + +export const HISTORY_SCHEMA_VERSION = 1; +export const HISTORY_DB_FILENAME = "routing-history.sqlite"; + +export const HISTORY_META_KEYS = { + schemaVersion: "schema_version", + sourcePath: "source_path", + sourceDev: "source_dev", + sourceIno: "source_ino", + sourceBirthtimeMs: "source_birthtime_ms", + sourceSize: "source_size", + sourceMtimeMs: "source_mtime_ms", + indexedOffset: "indexed_offset", + indexedRows: "indexed_rows", + builtAtMs: "built_at_ms", + lastError: "last_error", +} as const; + +export const HISTORY_DDL = ` +CREATE TABLE IF NOT EXISTS schema_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS requests ( + request_id TEXT PRIMARY KEY, + timestamp INTEGER NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL, + requested_model TEXT, + status INTEGER NOT NULL, + surface TEXT, + inbound_protocol TEXT, + api_key_id TEXT, + conversation_id TEXT, + route_kind TEXT, + profile_id TEXT, + profile_revision TEXT, + fallback INTEGER NOT NULL DEFAULT 0, + duration_ms INTEGER NOT NULL, + first_output_ms INTEGER, + usage_status TEXT, + usage_json TEXT, + total_tokens INTEGER, + error_code TEXT, + terminal_status TEXT, + close_reason TEXT, + attempt_count INTEGER NOT NULL DEFAULT 1, + decision_json TEXT, + row_json TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_requests_ts ON requests(timestamp DESC, request_id DESC); +CREATE INDEX IF NOT EXISTS idx_requests_provider ON requests(provider); +CREATE INDEX IF NOT EXISTS idx_requests_model ON requests(model); +CREATE INDEX IF NOT EXISTS idx_requests_requested_model ON requests(requested_model); +CREATE INDEX IF NOT EXISTS idx_requests_status ON requests(status); +CREATE INDEX IF NOT EXISTS idx_requests_conversation ON requests(conversation_id); +CREATE INDEX IF NOT EXISTS idx_requests_api_key ON requests(api_key_id); +CREATE INDEX IF NOT EXISTS idx_requests_profile ON requests(profile_id); +`; + +export function historyIndexPath(configDir: string): string { + return `${configDir.replace(/[\\/]+$/, "")}/${HISTORY_DB_FILENAME}`; +} diff --git a/src/server/management-api.ts b/src/server/management-api.ts index 3faed8f930..63129d5dfe 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -59,6 +59,7 @@ import { applySystemEnvToggle } from "./system-env"; import type { ManagementApiDeps } from "./management/context"; import { handleConfigRoutes } from "./management/config-routes"; import { handleLogsUsageRoutes } from "./management/logs-usage-routes"; +import { handleRequestHistoryRoutes } from "./management/request-history-routes"; import { handleProviderRoutes } from "./management/provider-routes"; import { handleModelRoutes } from "./management/model-routes"; import { handleAgentSettingsRoutes } from "./management/agent-settings-routes"; @@ -138,6 +139,7 @@ export async function handleManagementAPI( try { routed = (await handleConfigRoutes(ctx)) ?? (await handleLogsUsageRoutes(ctx)) + ?? (await handleRequestHistoryRoutes(ctx)) ?? (await handleProviderRoutes(ctx)) ?? (await handleModelRoutes(ctx)) ?? (await handleIntegrationRoutes(ctx)) diff --git a/src/server/management/request-history-routes.ts b/src/server/management/request-history-routes.ts new file mode 100644 index 0000000000..db318031cb --- /dev/null +++ b/src/server/management/request-history-routes.ts @@ -0,0 +1,133 @@ +/** + * Cursor-paginated request-history API (RI-02). + * + * - `GET /api/request-history` - keyset-paginated rows with filters + * - `GET /api/request-history/:requestId` - one canonical row + * + * The index is a derived projection of `usage.jsonl`; every response carries + * an `index` status block so callers can see schema version, indexed rows and + * any repair the indexer performed. + */ + +import { + queryRequestHistory, + requestHistoryRowById, + REQUEST_HISTORY_MAX_PAGE_SIZE, +} from "../../routing/history/indexer"; +import { InvalidCursorError } from "../../routing/history/cursor"; +import { requestLogEntryFromPersistedUsage } from "../request-log"; +import { requestLogDto } from "./shared"; +import { jsonResponse } from "../auth-cors"; +import type { ManagementContext } from "./context"; + +function parseQueryInt(raw: string | null): number | undefined | "invalid" { + if (raw === null) return undefined; + const trimmed = raw.trim(); + if (trimmed.length === 0) return "invalid"; + const value = Number(trimmed); + return Number.isInteger(value) ? value : "invalid"; +} + +export async function handleRequestHistoryRoutes(ctx: ManagementContext): Promise { + const { url, req, config } = ctx; + if (!url.pathname.startsWith("/api/request-history")) return null; + + if (url.pathname === "/api/request-history" && req.method === "GET") { + const statusRaw = parseQueryInt(url.searchParams.get("status")); + if (statusRaw === "invalid") { + return jsonResponse({ error: { code: "invalid_status", message: "status must be an integer from 100 to 599" } }, 400, req, config); + } + const status = statusRaw; + if (status !== undefined && (status < 100 || status > 599)) { + return jsonResponse({ error: { code: "invalid_status", message: "status must be an integer from 100 to 599" } }, 400, req, config); + } + const fromRaw = parseQueryInt(url.searchParams.get("from")); + if (fromRaw === "invalid") { + return jsonResponse({ error: { code: "invalid_from", message: "from must be an integer timestamp" } }, 400, req, config); + } + const toRaw = parseQueryInt(url.searchParams.get("to")); + if (toRaw === "invalid") { + return jsonResponse({ error: { code: "invalid_to", message: "to must be an integer timestamp" } }, 400, req, config); + } + const from = fromRaw; + const to = toRaw; + if (from !== undefined && to !== undefined && from > to) { + return jsonResponse({ error: { code: "invalid_range", message: "from must not be after to" } }, 400, req, config); + } + const limitRaw = url.searchParams.get("limit"); + const limitParsed = limitRaw === null ? undefined : parseQueryInt(limitRaw); + if (limitParsed === "invalid") { + return jsonResponse( + { error: { code: "invalid_limit", message: `limit must be an integer from 1 to ${REQUEST_HISTORY_MAX_PAGE_SIZE}` } }, + 400, + req, + config, + ); + } + const limit = limitParsed; + if (limit !== undefined && (limit < 1 || limit > REQUEST_HISTORY_MAX_PAGE_SIZE)) { + return jsonResponse( + { error: { code: "invalid_limit", message: `limit must be an integer from 1 to ${REQUEST_HISTORY_MAX_PAGE_SIZE}` } }, + 400, + req, + config, + ); + } + const cursor = url.searchParams.get("cursor"); + try { + const page = await queryRequestHistory({ + provider: url.searchParams.get("provider")?.trim() || undefined, + model: url.searchParams.get("model")?.trim() || undefined, + requestedModel: url.searchParams.get("requestedModel")?.trim() || undefined, + status, + conversationId: url.searchParams.get("conversationId")?.trim() || undefined, + surface: url.searchParams.get("surface")?.trim() || undefined, + inboundProtocol: url.searchParams.get("inboundProtocol")?.trim() || undefined, + apiKeyId: url.searchParams.get("apiKeyId")?.trim() || undefined, + profileId: url.searchParams.get("profileId")?.trim() || undefined, + fallback: url.searchParams.get("fallback") === "true" + ? true + : url.searchParams.get("fallback") === "false" ? false : undefined, + from, + to, + }, cursor, limit); + return jsonResponse({ + entries: page.rows.map(row => requestLogDto(requestLogEntryFromPersistedUsage(row))), + ...(page.nextCursor ? { nextCursor: page.nextCursor } : {}), + hasMore: page.hasMore, + index: { + schemaVersion: page.meta.schemaVersion, + indexedRows: page.meta.indexedRows, + sourceSize: page.meta.sourceSize, + sourceMtimeMs: page.meta.sourceMtimeMs, + builtAtMs: page.meta.builtAtMs, + lastError: page.meta.lastError, + }, + }, 200, req, config); + } catch (err) { + if (err instanceof InvalidCursorError) { + return jsonResponse({ error: { code: "invalid_cursor", message: "invalid cursor" } }, 400, req, config); + } + throw err; + } + } + + if (url.pathname.startsWith("/api/request-history/") && req.method === "GET") { + let requestId: string; + try { + requestId = decodeURIComponent(url.pathname.slice("/api/request-history/".length)); + } catch { + return jsonResponse({ error: { code: "not_found", message: "unknown request" } }, 404, req, config); + } + if (!requestId || requestId.includes("/")) { + return jsonResponse({ error: { code: "not_found", message: "unknown request" } }, 404, req, config); + } + const entry = await requestHistoryRowById(requestId); + if (!entry) { + return jsonResponse({ error: { code: "not_found", message: "unknown request" } }, 404, req, config); + } + return jsonResponse(requestLogDto(requestLogEntryFromPersistedUsage(entry)), 200, req, config); + } + + return null; +} diff --git a/tests/request-history-index.test.ts b/tests/request-history-index.test.ts new file mode 100644 index 0000000000..a1e3573c69 --- /dev/null +++ b/tests/request-history-index.test.ts @@ -0,0 +1,312 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, rmSync, truncateSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { handleManagementAPI } from "../src/server/management-api"; +import { ManagementRequest } from "./helpers/management-auth"; +import { + appendUsageEntry, + resetUsageReadCacheForTests, + type PersistedUsageEntry, +} from "../src/usage/log"; +import { + closeRequestHistoryIndex, + queryRequestHistory, + rebuildRequestHistoryIndex, + requestHistoryRowById, + REQUEST_HISTORY_MAX_PAGE_SIZE, +} from "../src/routing/history/indexer"; +import { InvalidCursorError } from "../src/routing/history/cursor"; +import { HISTORY_DB_FILENAME } from "../src/routing/history/schema"; +import { getConfigDir } from "../src/config"; +import type { OcxConfig } from "../src/types"; + +let testDir = ""; +let previousHome: string | undefined; + +function entry( + requestId: string, + timestamp: number, + provider = "a", + model = "m1", + overrides: Partial = {}, +): PersistedUsageEntry { + return { + requestId, + timestamp, + provider, + model, + status: 200, + durationMs: 10, + usageStatus: "reported", + ...overrides, + }; +} + +function seedRows(count: number, startTimestamp = 1000, provider = "a"): PersistedUsageEntry[] { + const rows: PersistedUsageEntry[] = []; + for (let index = 0; index < count; index++) { + rows.push(entry(`req-${index}`, startTimestamp + index, provider, `m${index % 3}`)); + } + return rows; +} + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-history-")); + process.env.OPENCODEX_HOME = testDir; + resetUsageReadCacheForTests(); + closeRequestHistoryIndex(); +}); + +afterEach(() => { + closeRequestHistoryIndex(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) rmSync(testDir, { recursive: true, force: true }); +}); + +function config(): OcxConfig { + return { + port: 10100, + defaultProvider: "a", + providers: { a: { adapter: "openai-chat", baseUrl: "https://a.example/v1", apiKey: "ka", models: ["m1"] } }, + }; +} + +async function apiGet(path: string): Promise { + const req = new ManagementRequest(`http://localhost${path}`, { method: "GET" }); + const response = await handleManagementAPI(req, new URL(req.url), config(), { + refreshCodexCatalog: async () => {}, + }); + expect(response).not.toBeNull(); + return response!; +} + +describe("request-history index (RI-02)", () => { + test("missing database and empty history produce an empty page", async () => { + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows).toEqual([]); + expect(page.hasMore).toBe(false); + expect(page.meta.indexedRows).toBe(0); + expect(page.meta.schemaVersion).toBe(1); + expect(existsSync(join(getConfigDir(), HISTORY_DB_FILENAME))).toBe(true); + }); + + test("indexes appended rows incrementally", async () => { + for (const row of seedRows(5)) appendUsageEntry(row); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(5); + expect(page.meta.indexedRows).toBe(5); + const offsetAfterInitial = page.meta.indexedOffset; + expect(page.meta.lastError).not.toMatch(/identity changed/i); + // New appends are picked up without a rebuild. + appendUsageEntry(entry("req-late", 9000)); + const after = await queryRequestHistory({}, undefined, 10); + expect(after.rows.length).toBe(6); + expect(after.meta.indexedRows).toBe(6); + expect(after.meta.indexedOffset).toBeGreaterThan(offsetAfterInitial); + expect(after.meta.lastError).not.toMatch(/identity changed/i); + }); + + test("large history indexes fully and paginates without duplicates or misses", async () => { + const rows = seedRows(1500, 10_000); + for (const row of rows) appendUsageEntry(row); + const seen = new Set(); + let cursor: string | undefined; + let pages = 0; + do { + const page = await queryRequestHistory({}, cursor, 100); + for (const row of page.rows) { + expect(seen.has(row.requestId)).toBe(false); + seen.add(row.requestId); + } + pages += 1; + cursor = page.nextCursor; + expect(page.hasMore).toBe(cursor !== undefined); + if (!page.hasMore) break; + } while (pages < 100); + expect(seen.size).toBe(1500); + expect(pages).toBe(15); + }); + + test("corrupt database is repaired by a full rebuild without losing canonical rows", async () => { + for (const row of seedRows(8)) appendUsageEntry(row); + await queryRequestHistory({}, undefined, 10); + closeRequestHistoryIndex(); + const dbFile = join(getConfigDir(), HISTORY_DB_FILENAME); + writeFileSync(dbFile, "this is not a sqlite file at all"); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(8); + expect(page.meta.lastError).toContain("rebuilt"); + }); + + test("old schema version triggers a rebuild", async () => { + for (const row of seedRows(4)) appendUsageEntry(row); + await queryRequestHistory({}, undefined, 10); + const { Database } = await import("bun:sqlite"); + const db = new Database(join(getConfigDir(), HISTORY_DB_FILENAME)); + db.query("UPDATE schema_meta SET value = '999' WHERE key = 'schema_version'").run(); + db.close(); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(4); + expect(page.meta.schemaVersion).toBe(1); + }); + + test("partial final JSONL line is skipped until it completes", async () => { + for (const row of seedRows(3)) appendUsageEntry(row); + // Append a partial line without a trailing newline. + const { appendFileSync } = await import("node:fs"); + const { usageLogPath } = await import("../src/usage/log"); + appendFileSync(usageLogPath(), '{"requestId":"req-partial","timestamp":', "utf-8"); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(3); + expect(page.meta.indexedRows).toBe(3); + // Completing the line makes it indexable on the next refresh. + appendFileSync(usageLogPath(), '9999,"provider":"a","model":"m1","status":200,"durationMs":1,"usageStatus":"reported"}\n', "utf-8"); + const after = await queryRequestHistory({}, undefined, 10); + expect(after.rows.length).toBe(4); + expect(after.rows.some(row => row.requestId === "req-partial")).toBe(true); + }); + + test("duplicate replay is ignored", async () => { + for (let index = 0; index < 3; index++) appendUsageEntry(entry(`dup-${index}`, 1000 + index)); + // Re-append the same request ids (as if the file were replayed). + for (let index = 0; index < 3; index++) appendUsageEntry(entry(`dup-${index}`, 1000 + index)); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(3); + expect(page.meta.indexedRows).toBe(3); + }); + + test("JSONL truncation triggers a rebuild that mirrors the truncated ledger", async () => { + for (const row of seedRows(10)) appendUsageEntry(row); + await queryRequestHistory({}, undefined, 10); + const { usageLogPath } = await import("../src/usage/log"); + truncateSync(usageLogPath(), 0); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(0); + expect(page.meta.indexedRows).toBe(0); + }); + + test("JSONL replacement (new file identity) rebuilds from the new ledger", async () => { + for (const row of seedRows(5, 500)) appendUsageEntry(row); + await queryRequestHistory({}, undefined, 10); + const { usageLogPath } = await import("../src/usage/log"); + // Replace the file wholesale (new inode on most platforms). + rmSync(usageLogPath(), { force: true }); + for (const row of seedRows(7, 7000, "b")) appendUsageEntry(row); + const page = await queryRequestHistory({}, undefined, 10); + expect(page.rows.length).toBe(7); + expect(page.rows.every(row => row.provider === "b")).toBe(true); + }); + + test("filters: provider, model, status, conversationId, surface, date range", async () => { + appendUsageEntry(entry("f1", 1000, "a", "m1", { conversationId: "conv-1", surface: "claude" })); + appendUsageEntry(entry("f2", 2000, "b", "m2", { status: 429, conversationId: "conv-2" })); + appendUsageEntry(entry("f3", 3000, "a", "m2", { surface: "grok" })); + const byProvider = await queryRequestHistory({ provider: "a" }, undefined, 10); + expect(byProvider.rows.map(row => row.requestId).sort()).toEqual(["f1", "f3"]); + const byStatus = await queryRequestHistory({ status: 429 }, undefined, 10); + expect(byStatus.rows.map(row => row.requestId)).toEqual(["f2"]); + const byConversation = await queryRequestHistory({ conversationId: "conv-1" }, undefined, 10); + expect(byConversation.rows.map(row => row.requestId)).toEqual(["f1"]); + const bySurface = await queryRequestHistory({ surface: "grok" }, undefined, 10); + expect(bySurface.rows.map(row => row.requestId)).toEqual(["f3"]); + const byRange = await queryRequestHistory({ from: 1500, to: 2500 }, undefined, 10); + expect(byRange.rows.map(row => row.requestId)).toEqual(["f2"]); + }); + + test("row-by-id returns the canonical entry and unknown ids 404 through the API", async () => { + appendUsageEntry(entry("target-id", 1234)); + const row = await requestHistoryRowById("target-id"); + expect(row?.requestId).toBe("target-id"); + expect(await requestHistoryRowById("missing")).toBeNull(); + + const found = await apiGet("/api/request-history/target-id"); + expect(found.status).toBe(200); + const body = await found.json() as { requestId?: string }; + expect(body.requestId).toBe("target-id"); + + const missing = await apiGet("/api/request-history/missing"); + expect(missing.status).toBe(404); + + const malformed = await apiGet("/api/request-history/%"); + expect(malformed.status).toBe(404); + }); + + test("API list endpoint returns entries, cursor, hasMore and index status", async () => { + for (const row of seedRows(5)) appendUsageEntry(row); + const response = await apiGet("/api/request-history?limit=2"); + expect(response.status).toBe(200); + const body = await response.json() as { + entries: Array<{ requestId?: string }>; + nextCursor?: string; + hasMore: boolean; + index: { schemaVersion: number; indexedRows: number }; + }; + expect(body.entries.length).toBe(2); + expect(body.hasMore).toBe(true); + expect(typeof body.nextCursor).toBe("string"); + expect(body.index.schemaVersion).toBe(1); + expect(body.index.indexedRows).toBe(5); + }); + + test("invalid cursor returns 400 invalid_cursor; invalid limit returns 400", async () => { + for (const row of seedRows(3)) appendUsageEntry(row); + const badCursor = await apiGet("/api/request-history?cursor=not-a-cursor"); + expect(badCursor.status).toBe(400); + const badCursorBody = await badCursor.json() as { error?: { code?: string } }; + expect(badCursorBody.error?.code).toBe("invalid_cursor"); + + const badLimit = await apiGet("/api/request-history?limit=9999"); + expect(badLimit.status).toBe(400); + const badLimitBody = await badLimit.json() as { error?: { code?: string } }; + expect(badLimitBody.error?.code).toBe("invalid_limit"); + + const badStatus = await apiGet("/api/request-history?status=42"); + expect(badStatus.status).toBe(400); + const badStatusText = await apiGet("/api/request-history?status=abc"); + expect(badStatusText.status).toBe(400); + const badStatusTextBody = await badStatusText.json() as { error?: { code?: string } }; + expect(badStatusTextBody.error?.code).toBe("invalid_status"); + + const badFrom = await apiGet("/api/request-history?from=abc"); + expect(badFrom.status).toBe(400); + + const badRange = await apiGet("/api/request-history?from=2000&to=1000"); + expect(badRange.status).toBe(400); + const badRangeBody = await badRange.json() as { error?: { code?: string } }; + expect(badRangeBody.error?.code).toBe("invalid_range"); + + await expect(queryRequestHistory({}, "garbage-cursor", 10)).rejects.toBeInstanceOf(InvalidCursorError); + }); + + test("page size is bounded at the indexer level", async () => { + for (const row of seedRows(120)) appendUsageEntry(row); + const page = await queryRequestHistory({}, undefined, 9999); + expect(page.rows.length).toBe(REQUEST_HISTORY_MAX_PAGE_SIZE); + expect(page.hasMore).toBe(true); + }); + + test("index rebuild equivalence: rebuilt rows match the ledger exactly", async () => { + const rows = seedRows(25, 42); + for (const row of rows) appendUsageEntry(row); + await queryRequestHistory({}, undefined, 10); + const canonicalIds = rows.map(row => row.requestId).sort(); + const rebuilt = await rebuildRequestHistoryIndex(); + expect(rebuilt.indexedRows).toBe(25); + const page = await queryRequestHistory({}, undefined, 100); + expect(page.rows.map(row => row.requestId).sort()).toEqual(canonicalIds); + }); + + test("cursor stays stable while appends arrive between pages", async () => { + for (const row of seedRows(6, 1000)) appendUsageEntry(row); + const first = await queryRequestHistory({}, undefined, 3); + // New rows with HIGHER timestamps must not shift the keyset window. + for (const row of seedRows(3, 9000)) appendUsageEntry(row); + const second = await queryRequestHistory({}, first.nextCursor, 3); + const ids = [...first.rows, ...second.rows].map(row => row.requestId); + expect(ids).toEqual(["req-5", "req-4", "req-3", "req-2", "req-1", "req-0"]); + expect(second.hasMore).toBe(false); + }); +}); diff --git a/tests/route-decision-trace.test.ts b/tests/route-decision-trace.test.ts index b7f4aa8516..cb6fd6a599 100644 --- a/tests/route-decision-trace.test.ts +++ b/tests/route-decision-trace.test.ts @@ -288,6 +288,7 @@ describe("route decision traces (RI-01)", () => { expect(serialized).not.toContain("apiKey"); expect(serialized).not.toContain("baseUrl"); expect(serialized).not.toContain("https://a.example/v1"); + expect(serialized).not.toContain("prompt"); }); test("trace round-trips through usage.jsonl and request-log hydration", () => {