diff --git a/.gitignore b/.gitignore index 5e5ef89..2c8d843 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,9 @@ node_modules /build /build-server +# Analytics (local SQLite db — see docs/ANALYTICS.md) +/data + # OS .DS_Store Thumbs.db diff --git a/docs/ANALYTICS.md b/docs/ANALYTICS.md new file mode 100644 index 0000000..982aa0b --- /dev/null +++ b/docs/ANALYTICS.md @@ -0,0 +1,66 @@ +# Analytics + +Server-side recording of session volume, conversation content, tool usage, and failures for the realtime session relay. Backed by a small local SQLite database — no client-side instrumentation, since every session event already passes through our server (see [REALTIME_SESSION.md](REALTIME_SESSION.md)'s connection flow). This fills a seam that doc explicitly anticipated: _"we can add moderation, analytics, avatar-specific processing... in the middle later — without ever touching the browser client."_ + +## The hook point + +`RealtimeSession` (`session.ts`) is the one place every session event in both directions already funnels through. Analytics attaches at choke points that already exist there for other reasons — `start()`, `close()`, `resolveToolCall()`, `sendToBrowser()` — rather than adding new plumbing. See `src/lib/server/core/analytics.ts` for the `AnalyticsRecorder` interface and its `better-sqlite3`-backed implementation. + +## Why `better-sqlite3`, not `node:sqlite` + +Node has an experimental built-in SQLite module, but `better-sqlite3` was chosen instead: it's the mature, widely-used library with a stable synchronous API, which is what keeps every `AnalyticsRecorder` call in `session.ts` a plain fire-and-forget call rather than something that needs `await`/error-handling threaded through the session's existing synchronous methods. + +## The transcript-buffering gotcha + +`transcript.done` (see `protocol.ts`) carries only `{ role }` — no text. Full message text has to be assembled server-side: + +- **Typed user text** (`input_text.send`): already a complete string — logged directly, no buffering. +- **Assistant voice replies**: stream as many `transcript.delta{role:'assistant'}` chunks as OpenAI cares to send, followed by one text-less `transcript.done{role:'assistant'}`. `RealtimeSession` accumulates deltas into `assistantTranscriptBuffer` and flushes+logs the concatenated text on `done`. +- **User voice transcripts**: today, OpenAI's `conversation.item.input_audio_transcription.completed` maps (in `providers/openai.ts`) to a single `transcript.delta{role:'user'}` event that already carries the _entire_ transcript — no corresponding `transcript.done{role:'user'}` is ever emitted by that mapping. So a user-role delta is logged immediately as a complete message, not buffered. + +**This rests on unverified live behavior** — the same caveat class as REALTIME_SESSION.md's "Still unverified" note on this exact OpenAI event. If a live session ever produces more than one `messages` row for a single spoken user turn, that assumption has broken and this logic needs revisiting (see "Verifying changes" below for how to check). + +## A bundled correctness fix: upstream-disconnect now actually closes the session + +Before this change, `handleProviderEvent` only _forwarded_ a provider-originated `session.closed` event (e.g. the upstream WebSocket dropping) to the browser — it never called `this.close()`. That meant `RealtimeSession`'s own state never updated and no session end time was ever recorded for that path. Since "session end time" is a stated goal of this feature, `handleProviderEvent` now also calls `this.close('upstream_disconnected')` when it sees a `session.closed` event (forwarding first, since `close()` moves the browser socket out of `OPEN` and `sendToBrowser` no-ops after that). + +## Schema + +Four tables, `CREATE TABLE IF NOT EXISTS` on startup — no migration framework; a future column addition is a manual `ALTER TABLE` added to `SCHEMA_SQL` in `analytics.ts`: + +- `sessions(id, model, voice, started_at, ended_at, end_reason)` +- `messages(id, session_id, role, text, created_at)` +- `tool_calls(id, session_id, call_id, name, arguments, output, duration_ms, created_at)` +- `events(id, session_id, type, detail, created_at)` — `session_id` is nullable here (only here): a validation error can occur before `session.start` completes, when no `sessions` row exists yet. + +Foreign keys are declared for documentation but explicitly not enforced (`PRAGMA foreign_keys = OFF` in the constructor — `better-sqlite3` defaults this pragma to _on_, so it has to be turned off explicitly, not just left unset) — this is a logging sink, not a system of record, and a stray write should never be silently dropped by a constraint violation. + +## Configuration + +`ANALYTICS_DB_PATH` (optional) — defaults to `./data/analytics.db`. The parent directory is created automatically if missing. `/data` is gitignored. + +## PII and retention + +`analytics.db` stores full raw conversation text (both roles) and tool call arguments/outputs verbatim, in a local, unencrypted SQLite file, with no retention window, no rotation, no redaction, and no per-user consent tracking. That's fine for internal development or a demo of this base template — it is **not** production-safe as shipped. A client project deploying this for real users needs to add a real retention/redaction policy before treating this as production telemetry; don't assume GDPR/CCPA compliance out of the box. + +## What's deliberately not built yet + +- A query/reporting UI — this is a write-only sink today, read it with the `sqlite3` CLI or any SQLite client. +- Migration tooling beyond manual `ALTER TABLE` additions to `SCHEMA_SQL`. +- DB export/backup, log rotation for the file itself. +- A graceful-shutdown/flush hook — no composition root in this codebase has one for anything yet, including the existing provider connections. +- Multi-process-safe writes beyond WAL's built-in single-writer guarantee (matches the existing "in-memory, single-process only" constraint on session storage itself). + +## Verifying changes + +- `npm run check` / `npm run lint` — type/lint correctness. Note `check` uses `bundler` module resolution and will **not** catch a missing `.js` extension on a relative import inside `src/lib/server/core/**`. +- `npm run build` — also runs the standalone `tsc -p tsconfig.server.json` compile (`nodenext` resolution) that `server.js` depends on; this is the step that catches a missing `.js` extension or a file placed outside that config's `include` globs. +- `npm run test:unit` — `analytics.spec.ts` (schema + each recorder method against `:memory:`, plus a WAL-mode check against a real temp file) and the extended `session.spec.ts` (asserts `AnalyticsRecorder` calls happen at the right lifecycle points, using a hand-written `FakeAnalyticsRecorder` in the same no-`vi.fn()` style as this file's other fakes). +- **Live smoke test**: set `.env`, `npm run dev`, open `/demo/realtime`, and hold a real conversation — a typed message, a full voice turn in both directions, and a tool call (the demo project registers `get_current_time` (server) and `flash_page` (client) out of the box). Then inspect `data/analytics.db` directly: + ```bash + sqlite3 data/analytics.db "select * from sessions;" + sqlite3 data/analytics.db "select id, role, substr(text,1,80), created_at from messages order by id;" + sqlite3 data/analytics.db "select call_id, name, duration_ms from tool_calls;" + sqlite3 data/analytics.db "select type, detail from events;" + ``` + Confirm exactly one assistant `messages` row per spoken AI turn with the **full** assembled text (not fragments), exactly one user `messages` row per spoken user turn (the direct check on the unverified single-chunk assumption above), and — kill the dev server or close the tab mid-session once — that `sessions.ended_at`/`end_reason` get populated (the direct check on the bug fix above). Remember `src/lib/server/core/**` isn't part of Vite's HMR — restart `npm run dev` after editing anything here. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8ef3278..19ca58b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -25,5 +25,6 @@ Routes (`src/routes/`) stay thin and delegate into these — a route file is glu | --------------------------------------------------------------------------------------------- | ------------------------------------------ | | Realtime session management (OpenAI Realtime API today, extensible to other models/pipelines) | [REALTIME_SESSION.md](REALTIME_SESSION.md) | | Tool calling (server- and client-executed) and the core/project ownership boundary | [TOOLS.md](TOOLS.md) | +| Analytics recording (sessions, messages, tool calls, errors) to a local SQLite database | [ANALYTICS.md](ANALYTICS.md) | Add a row here (and a corresponding doc) for each new subsystem as it's built — see the "Documentation" section in [AGENTS.md](../AGENTS.md) for when and how. diff --git a/package-lock.json b/package-lock.json index 1e8a06a..528643c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,6 +8,7 @@ "name": "webavatar-base", "version": "0.0.1", "dependencies": { + "better-sqlite3": "^13.0.3", "ws": "^8.21.3" }, "devDependencies": { @@ -16,6 +17,7 @@ "@sveltejs/adapter-node": "^5.5.7", "@sveltejs/kit": "^2.63.0", "@sveltejs/vite-plugin-svelte": "^7.1.2", + "@types/better-sqlite3": "^9.6.0", "@types/node": "^24", "@types/ws": "^8.18.1", "@vitest/browser-playwright": "^4.1.8", @@ -1279,6 +1281,16 @@ "svelte": "^3 || ^4 || ^5 || ^5.0.0-next.0" } }, + "node_modules/@types/better-sqlite3": { + "version": "9.6.0", + "resolved": "https://registry.npmjs.org/@types/better-sqlite3/-/better-sqlite3-9.6.0.tgz", + "integrity": "sha512-ZEEwBSgMu7GYJOynoagg5X9JbxfL6dTJDsgViJIqh67jV44kyOr9RXfmFjLK5rzC4MWssP06t9hu/JwGDnUbCg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/chai": { "version": "5.2.3", "resolved": "https://registry.npmjs.org/@types/chai/-/chai-5.2.3.tgz", @@ -1829,6 +1841,18 @@ "node": "18 || 20 || >=22" } }, + "node_modules/better-sqlite3": { + "version": "13.0.3", + "resolved": "https://registry.npmjs.org/better-sqlite3/-/better-sqlite3-13.0.3.tgz", + "integrity": "sha512-RbOBxmLBG8uvFUc15X9+9SFemKcQ0WBuISBVkpuiaUB2qblC8UWlHEjdWVoZ8AdhSwmoEgsiXKfopX0CQxaACQ==", + "license": "MIT", + "dependencies": { + "node-addon-api": "^8.0.0" + }, + "engines": { + "node": ">=22" + } + }, "node_modules/brace-expansion": { "version": "5.0.9", "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.9.tgz", @@ -2959,6 +2983,15 @@ "dev": true, "license": "MIT" }, + "node_modules/node-addon-api": { + "version": "8.9.2", + "resolved": "https://registry.npmjs.org/node-addon-api/-/node-addon-api-8.9.2.tgz", + "integrity": "sha512-VijLXbi3UACN69I0JVXJsX4tjACjNoQDgv2gTF6sx2wWEi8tkSg2eX8p5gSIFi8z2+DL3oHmY6OyKce38SDolg==", + "license": "MIT", + "engines": { + "node": "^18 || ^20 || >= 21" + } + }, "node_modules/obug": { "version": "2.1.4", "resolved": "https://registry.npmjs.org/obug/-/obug-2.1.4.tgz", diff --git a/package.json b/package.json index 8ded50f..2236567 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ "@sveltejs/adapter-node": "^5.5.7", "@sveltejs/kit": "^2.63.0", "@sveltejs/vite-plugin-svelte": "^7.1.2", + "@types/better-sqlite3": "^9.6.0", "@types/node": "^24", "@types/ws": "^8.18.1", "@vitest/browser-playwright": "^4.1.8", @@ -42,6 +43,7 @@ "vitest-browser-svelte": "^2.1.1" }, "dependencies": { + "better-sqlite3": "^13.0.3", "ws": "^8.21.3" } } diff --git a/server.js b/server.js index e716683..c2fc40d 100644 --- a/server.js +++ b/server.js @@ -7,7 +7,9 @@ // provide OPENAI_API_KEY / REALTIME_TOKEN_SECRET via your deployment's real // secret mechanism. import { createServer } from 'node:http'; +import { join } from 'node:path'; import { handler } from './build/handler.js'; +import { SqliteAnalyticsRecorder } from './build-server/server/core/analytics.js'; import { DEFAULT_SESSION_DEFAULTS } from './build-server/server/core/defaults.js'; import { OpenAiRealtimeProvider } from './build-server/server/core/providers/openai.js'; import { handleRealtimeUpgrade } from './build-server/server/core/ws-handler.js'; @@ -19,6 +21,8 @@ if (!apiKey) { } const provider = new OpenAiRealtimeProvider({ apiKey }); +const dbPath = process.env.ANALYTICS_DB_PATH ?? join(process.cwd(), 'data', 'analytics.db'); +const recorder = new SqliteAnalyticsRecorder(dbPath); // server/core never imports project code directly (see docs/TOOLS.md) — this // composition root is one of the two places (alongside vite.config.ts) that // merges core defaults with the project's tool registry. @@ -26,7 +30,7 @@ const defaults = { ...DEFAULT_SESSION_DEFAULTS, toolRegistry }; const server = createServer(handler); server.on('upgrade', (req, socket, head) => { - handleRealtimeUpgrade(req, socket, head, { provider, defaults }); + handleRealtimeUpgrade(req, socket, head, { provider, defaults, recorder }); }); const port = process.env.PORT ? Number(process.env.PORT) : 3000; diff --git a/src/lib/server/core/analytics.spec.ts b/src/lib/server/core/analytics.spec.ts new file mode 100644 index 0000000..fe518c3 --- /dev/null +++ b/src/lib/server/core/analytics.spec.ts @@ -0,0 +1,151 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { SqliteAnalyticsRecorder } from './analytics.js'; + +describe('SqliteAnalyticsRecorder', () => { + it('creates all four tables on construction', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + const tables = recorder.rawDb + .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") + .all() + .map((row) => (row as { name: string }).name); + expect(tables).toEqual( + expect.arrayContaining(['sessions', 'messages', 'tool_calls', 'events']) + ); + }); + + it('persists a started session and leaves ended_at/end_reason null', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.sessionStarted({ + sessionId: 'session-1', + model: 'test-model', + voice: 'coral', + startedAt: 1000 + }); + + const row = recorder.rawDb.prepare('SELECT * FROM sessions WHERE id = ?').get('session-1'); + expect(row).toMatchObject({ + id: 'session-1', + model: 'test-model', + voice: 'coral', + started_at: 1000, + ended_at: null, + end_reason: null + }); + }); + + it('updates the same session row on sessionEnded, leaving other columns unchanged', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.sessionStarted({ + sessionId: 'session-1', + model: 'test-model', + voice: undefined, + startedAt: 1000 + }); + recorder.sessionEnded({ sessionId: 'session-1', endedAt: 2000, reason: 'client_requested' }); + + const row = recorder.rawDb.prepare('SELECT * FROM sessions WHERE id = ?').get('session-1'); + expect(row).toMatchObject({ + id: 'session-1', + model: 'test-model', + started_at: 1000, + ended_at: 2000, + end_reason: 'client_requested' + }); + }); + + it('logs messages for both roles', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.messageLogged({ sessionId: 'session-1', role: 'user', text: 'hi', loggedAt: 1 }); + recorder.messageLogged({ + sessionId: 'session-1', + role: 'assistant', + text: 'hello there', + loggedAt: 2 + }); + + const rows = recorder.rawDb + .prepare('SELECT session_id, role, text, created_at FROM messages ORDER BY id') + .all(); + expect(rows).toEqual([ + { session_id: 'session-1', role: 'user', text: 'hi', created_at: 1 }, + { session_id: 'session-1', role: 'assistant', text: 'hello there', created_at: 2 } + ]); + }); + + it('round-trips a logged tool call', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.toolCallLogged({ + sessionId: 'session-1', + callId: 'call-1', + name: 'get_current_time', + arguments: '{}', + output: JSON.stringify({ time: '12:00' }), + durationMs: 42, + loggedAt: 3 + }); + + const row = recorder.rawDb.prepare('SELECT * FROM tool_calls WHERE call_id = ?').get('call-1'); + expect(row).toMatchObject({ + session_id: 'session-1', + call_id: 'call-1', + name: 'get_current_time', + arguments: '{}', + output: JSON.stringify({ time: '12:00' }), + duration_ms: 42 + }); + }); + + it('logs an event with a null session_id for a pre-session error', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.eventLogged({ + sessionId: null, + type: 'error', + detail: JSON.stringify({ type: 'error', code: 'invalid_client_event' }), + loggedAt: 4 + }); + + const row = recorder.rawDb.prepare('SELECT * FROM events WHERE type = ?').get('error') as { + session_id: string | null; + detail: string; + }; + expect(row.session_id).toBeNull(); + expect(JSON.parse(row.detail)).toEqual({ type: 'error', code: 'invalid_client_event' }); + }); + + it('logs an event against a real session_id once a session exists', () => { + const recorder = new SqliteAnalyticsRecorder(':memory:'); + recorder.eventLogged({ + sessionId: 'session-1', + type: 'session.closed', + detail: JSON.stringify({ type: 'session.closed', reason: 'upstream_disconnected' }), + loggedAt: 5 + }); + + const row = recorder.rawDb + .prepare('SELECT session_id FROM events WHERE type = ?') + .get('session.closed'); + expect(row).toEqual({ session_id: 'session-1' }); + }); + + describe('WAL mode', () => { + let tmpDir: string; + + afterEach(() => { + if (tmpDir) rmSync(tmpDir, { recursive: true, force: true }); + }); + + it('enables WAL journal mode for a real file-backed database', () => { + // SQLite silently keeps ':memory:' databases in 'memory' journal mode + // regardless of a WAL pragma call, so this needs a real file path to + // mean anything — asserting it against the ':memory:' recorder used + // everywhere else in this file would be misleading. + tmpDir = mkdtempSync(join(tmpdir(), 'analytics-test-')); + const recorder = new SqliteAnalyticsRecorder(join(tmpDir, 'analytics.db')); + expect(recorder.rawDb.pragma('journal_mode', { simple: true })).toBe('wal'); + recorder.close(); + }); + }); +}); diff --git a/src/lib/server/core/analytics.ts b/src/lib/server/core/analytics.ts new file mode 100644 index 0000000..bcedbc2 --- /dev/null +++ b/src/lib/server/core/analytics.ts @@ -0,0 +1,198 @@ +import Database from 'better-sqlite3'; +import { mkdirSync } from 'node:fs'; +import { dirname } from 'node:path'; + +/** + * Recording sink for the realtime session relay (see docs/ANALYTICS.md). + * Synchronous and void-returning to match `better-sqlite3`'s own synchronous + * API — every call site in `session.ts` stays fire-and-forget with no + * `await`/error-handling threaded through the session's hot path. + */ +export interface AnalyticsRecorder { + sessionStarted(params: { + sessionId: string; + model: string; + voice: string | undefined; + startedAt: number; // epoch ms + }): void; + + sessionEnded(params: { sessionId: string; endedAt: number; reason: string }): void; + + messageLogged(params: { + sessionId: string; + role: 'user' | 'assistant'; + text: string; + loggedAt: number; + }): void; + + toolCallLogged(params: { + sessionId: string; + callId: string; + name: string; + arguments: string; // raw JSON string, as received on the wire + output: string; // raw JSON string sent back as tool.result + durationMs: number; + loggedAt: number; + }): void; + + /** + * `sessionId` is nullable: a validation error can occur before + * `session.start` completes, when no `sessions` row exists yet to + * reference. + */ + eventLogged(params: { + sessionId: string | null; + type: 'error' | 'session.closed'; + detail: string; + loggedAt: number; + }): void; +} + +const SCHEMA_SQL = ` +CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + model TEXT NOT NULL, + voice TEXT, + started_at INTEGER NOT NULL, + ended_at INTEGER, + end_reason TEXT +); + +CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id), + role TEXT NOT NULL CHECK (role IN ('user', 'assistant')), + text TEXT NOT NULL, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id); + +CREATE TABLE IF NOT EXISTS tool_calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL REFERENCES sessions(id), + call_id TEXT NOT NULL, + name TEXT NOT NULL, + arguments TEXT NOT NULL, + output TEXT NOT NULL, + duration_ms INTEGER NOT NULL, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_tool_calls_session_id ON tool_calls(session_id); + +CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT REFERENCES sessions(id), + type TEXT NOT NULL CHECK (type IN ('error', 'session.closed')), + detail TEXT NOT NULL, + created_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_events_session_id ON events(session_id); +`; + +/** + * `better-sqlite3`-backed `AnalyticsRecorder`. Schema is created eagerly on + * construction (`CREATE TABLE IF NOT EXISTS` — no migration framework; a + * future column change is a manual addition to `SCHEMA_SQL`). Foreign keys + * are declared for documentation but deliberately not enforced (no `PRAGMA + * foreign_keys = ON`): this is a logging sink, not a system of record, and a + * stray write should never be silently dropped by a constraint violation. + */ +export class SqliteAnalyticsRecorder implements AnalyticsRecorder { + private readonly db: Database.Database; + private readonly insertSession: Database.Statement; + private readonly updateSessionEnd: Database.Statement; + private readonly insertMessage: Database.Statement; + private readonly insertToolCall: Database.Statement; + private readonly insertEvent: Database.Statement; + + constructor(dbPath: string) { + // better-sqlite3 does not create the parent directory itself — on a + // fresh checkout `data/` doesn't exist yet. No-op for ':memory:' + // (dirname resolves to '.'). + mkdirSync(dirname(dbPath), { recursive: true }); + this.db = new Database(dbPath); + this.db.pragma('journal_mode = WAL'); + // better-sqlite3 defaults foreign_keys to ON, which would turn a stray + // write (e.g. a mismatched session_id) into a dropped insert. This is a + // logging sink, not a system of record — explicitly opt out so the + // declared FKs stay documentation only, never enforcement. + this.db.pragma('foreign_keys = OFF'); + this.db.exec(SCHEMA_SQL); + + this.insertSession = this.db.prepare( + 'INSERT INTO sessions (id, model, voice, started_at) VALUES (?, ?, ?, ?)' + ); + this.updateSessionEnd = this.db.prepare( + 'UPDATE sessions SET ended_at = ?, end_reason = ? WHERE id = ?' + ); + this.insertMessage = this.db.prepare( + 'INSERT INTO messages (session_id, role, text, created_at) VALUES (?, ?, ?, ?)' + ); + this.insertToolCall = this.db.prepare( + `INSERT INTO tool_calls (session_id, call_id, name, arguments, output, duration_ms, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)` + ); + this.insertEvent = this.db.prepare( + 'INSERT INTO events (session_id, type, detail, created_at) VALUES (?, ?, ?, ?)' + ); + } + + sessionStarted({ + sessionId, + model, + voice, + startedAt + }: Parameters[0]): void { + this.insertSession.run(sessionId, model, voice ?? null, startedAt); + } + + sessionEnded({ + sessionId, + endedAt, + reason + }: Parameters[0]): void { + this.updateSessionEnd.run(endedAt, reason, sessionId); + } + + messageLogged({ + sessionId, + role, + text, + loggedAt + }: Parameters[0]): void { + this.insertMessage.run(sessionId, role, text, loggedAt); + } + + // `arguments` can't be used as a bare destructured binding name (reserved + // in strict-mode bindings, and class bodies are always strict) — renamed + // to `args` here. + toolCallLogged({ + sessionId, + callId, + name, + arguments: args, + output, + durationMs, + loggedAt + }: Parameters[0]): void { + this.insertToolCall.run(sessionId, callId, name, args, output, durationMs, loggedAt); + } + + eventLogged({ + sessionId, + type, + detail, + loggedAt + }: Parameters[0]): void { + this.insertEvent.run(sessionId, type, detail, loggedAt); + } + + /** Direct row access for tests only — session.ts must only ever go through AnalyticsRecorder. */ + get rawDb(): Database.Database { + return this.db; + } + + close(): void { + this.db.close(); + } +} diff --git a/src/lib/server/core/dev-plugin.ts b/src/lib/server/core/dev-plugin.ts index af93092..8229522 100644 --- a/src/lib/server/core/dev-plugin.ts +++ b/src/lib/server/core/dev-plugin.ts @@ -1,4 +1,6 @@ +import { join } from 'node:path'; import { loadEnv, type Plugin } from 'vite'; +import { SqliteAnalyticsRecorder } from './analytics.js'; import { OpenAiRealtimeProvider } from './providers/openai.js'; import type { SessionDefaults } from './session.js'; import { handleRealtimeUpgrade } from './ws-handler.js'; @@ -43,9 +45,15 @@ export function realtimeDevWsPlugin(options: RealtimeDevWsPluginOptions): Plugin ); } const provider = new OpenAiRealtimeProvider({ apiKey: apiKey ?? '' }); + const dbPath = process.env.ANALYTICS_DB_PATH ?? join(process.cwd(), 'data', 'analytics.db'); + const recorder = new SqliteAnalyticsRecorder(dbPath); server.httpServer?.on('upgrade', (req, socket, head) => { - handleRealtimeUpgrade(req, socket, head, { provider, defaults: options.defaults }); + handleRealtimeUpgrade(req, socket, head, { + provider, + defaults: options.defaults, + recorder + }); }); } }; diff --git a/src/lib/server/core/session.spec.ts b/src/lib/server/core/session.spec.ts index de8f1f4..c5eff6a 100644 --- a/src/lib/server/core/session.spec.ts +++ b/src/lib/server/core/session.spec.ts @@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type WebSocket from 'ws'; import type { ClientEvent, ServerEvent } from '../../core/protocol.js'; import type { ToolDefinition } from '../../core/tools.js'; +import type { AnalyticsRecorder } from './analytics.js'; import type { ProviderSession, RealtimeProvider } from './provider.js'; import { createRealtimeSession, resolveSessionConfig } from './session.js'; import { createToolRegistry } from './tools.js'; @@ -84,6 +85,30 @@ class FakeProvider implements RealtimeProvider { } } +class FakeAnalyticsRecorder implements AnalyticsRecorder { + sessionStartedCalls: Parameters[0][] = []; + sessionEndedCalls: Parameters[0][] = []; + messageLoggedCalls: Parameters[0][] = []; + toolCallLoggedCalls: Parameters[0][] = []; + eventLoggedCalls: Parameters[0][] = []; + + sessionStarted(p: Parameters[0]) { + this.sessionStartedCalls.push(p); + } + sessionEnded(p: Parameters[0]) { + this.sessionEndedCalls.push(p); + } + messageLogged(p: Parameters[0]) { + this.messageLoggedCalls.push(p); + } + toolCallLogged(p: Parameters[0]) { + this.toolCallLoggedCalls.push(p); + } + eventLogged(p: Parameters[0]) { + this.eventLoggedCalls.push(p); + } +} + function makeSocket() { const socket = new FakeBrowserSocket(); return { socket, asWs: socket as unknown as WebSocket }; @@ -112,20 +137,25 @@ describe('resolveSessionConfig', () => { describe('RealtimeSession', () => { it('starts a provider session on session.start and replies with session.ready', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(socket.sent).toContainEqual(expect.objectContaining({ type: 'session.ready' })) ); + expect(recorder.sessionStartedCalls).toContainEqual( + expect.objectContaining({ model: 'test-model', voice: undefined }) + ); }); it('rejects content events sent before session.start', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'input_text.send', text: 'too early' }); await vi.waitFor(() => @@ -134,12 +164,16 @@ describe('RealtimeSession', () => { ) ); expect(provider.sessions).toHaveLength(0); + expect(recorder.eventLoggedCalls).toContainEqual( + expect.objectContaining({ sessionId: null, type: 'error' }) + ); }); - it('forwards events to the provider session once active', async () => { + it('forwards events to the provider session once active and logs typed user text', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); @@ -148,12 +182,16 @@ describe('RealtimeSession', () => { await vi.waitFor(() => expect(provider.sessions[0].sent).toContainEqual({ type: 'input_text.send', text: 'hello' }) ); + expect(recorder.messageLoggedCalls).toContainEqual( + expect.objectContaining({ role: 'user', text: 'hello' }) + ); }); it('relays provider server events back to the browser socket', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); @@ -172,10 +210,42 @@ describe('RealtimeSession', () => { ); }); + it('assembles assistant transcript.delta chunks into one logged message on transcript.done', async () => { + const { session, socket, recorder } = await startActiveSession(DEFAULTS); + + session.emitToOrchestrator({ type: 'transcript.delta', role: 'assistant', text: 'Hel' }); + session.emitToOrchestrator({ type: 'transcript.delta', role: 'assistant', text: 'lo' }); + session.emitToOrchestrator({ type: 'transcript.done', role: 'assistant' }); + + await vi.waitFor(() => + expect(socket.sent).toContainEqual({ type: 'transcript.done', role: 'assistant' }) + ); + expect(recorder.messageLoggedCalls).toEqual([ + expect.objectContaining({ role: 'assistant', text: 'Hello' }) + ]); + }); + + it('logs a user voice transcript.delta immediately, since no transcript.done follows it today', async () => { + const { session, recorder } = await startActiveSession(DEFAULTS); + + session.emitToOrchestrator({ + type: 'transcript.delta', + role: 'user', + text: 'full transcript' + }); + + await vi.waitFor(() => + expect(recorder.messageLoggedCalls).toContainEqual( + expect.objectContaining({ role: 'user', text: 'full transcript' }) + ) + ); + }); + it('closes the provider session when the browser disconnects', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); @@ -183,12 +253,16 @@ describe('RealtimeSession', () => { socket.emit('close'); await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); expect(provider.sessions[0].closeReason).toBe('client_disconnected'); + expect(recorder.sessionEndedCalls).toContainEqual( + expect.objectContaining({ reason: 'client_disconnected' }) + ); }); it('closes the provider session on session.end and closes the browser socket', async () => { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); @@ -196,12 +270,16 @@ describe('RealtimeSession', () => { socket.receive({ type: 'session.end' }); await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); expect(socket.closeCalls).toBe(1); + expect(recorder.sessionEndedCalls).toContainEqual( + expect.objectContaining({ reason: 'client_requested' }) + ); }); it('does not leak an upstream session if the browser disconnects mid-handshake', async () => { const provider = new FakeProvider({ deferred: true }); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, DEFAULTS); + createRealtimeSession(asWs, provider, DEFAULTS, recorder); socket.receive({ type: 'session.start' }); socket.emit('close'); // disconnect while the provider handshake is still pending @@ -209,6 +287,21 @@ describe('RealtimeSession', () => { await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); + // No sessions row was ever created (sessionId never got set), so there's + // nothing to record an end time against. + expect(recorder.sessionEndedCalls).toEqual([]); + }); + + it('closes the browser side and records an end time when the provider disconnects first', async () => { + const { session, socket, recorder } = await startActiveSession(DEFAULTS); + + session.emitToOrchestrator({ type: 'session.closed', reason: 'upstream_disconnected' }); + + await vi.waitFor(() => expect(socket.closeCalls).toBe(1)); + expect(socket.sent).toContainEqual({ type: 'session.closed', reason: 'upstream_disconnected' }); + expect(recorder.sessionEndedCalls).toContainEqual( + expect.objectContaining({ reason: 'upstream_disconnected' }) + ); }); }); @@ -233,11 +326,12 @@ const CLIENT_TOOL: ToolDefinition = { async function startActiveSession(defaults: typeof DEFAULTS) { const provider = new FakeProvider(); + const recorder = new FakeAnalyticsRecorder(); const { socket, asWs } = makeSocket(); - createRealtimeSession(asWs, provider, defaults); + createRealtimeSession(asWs, provider, defaults, recorder); socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); - return { provider, session: provider.sessions[0], socket }; + return { provider, session: provider.sessions[0], socket, recorder }; } describe('RealtimeSession tool calls', () => { @@ -249,7 +343,7 @@ describe('RealtimeSession tool calls', () => { const toolRegistry = createToolRegistry([FAST_TOOL], { fast_tool: async () => ({ done: true }) }); - const { session } = await startActiveSession({ ...DEFAULTS, toolRegistry }); + const { session, recorder } = await startActiveSession({ ...DEFAULTS, toolRegistry }); session.emitToOrchestrator({ type: 'tool.call', @@ -266,6 +360,15 @@ describe('RealtimeSession tool calls', () => { }) ); await vi.waitFor(() => expect(session.sent).toContainEqual({ type: 'response.trigger' })); + expect(recorder.toolCallLoggedCalls).toContainEqual( + expect.objectContaining({ + callId: 'call-1', + name: 'fast_tool', + arguments: '{}', + output: JSON.stringify({ done: true }), + durationMs: expect.any(Number) + }) + ); }); it('only triggers response.trigger once every parallel tool call has resolved', async () => { diff --git a/src/lib/server/core/session.ts b/src/lib/server/core/session.ts index d4ec62d..9230898 100644 --- a/src/lib/server/core/session.ts +++ b/src/lib/server/core/session.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'; import WebSocket from 'ws'; import type { ClientEvent, ServerEvent, SessionStartConfig } from '../../core/protocol.js'; import { formatToolError } from '../../core/tools.js'; +import type { AnalyticsRecorder } from './analytics.js'; import type { ProviderSession, RealtimeProvider, ResolvedSessionConfig } from './provider.js'; import type { ToolRegistry } from './tools.js'; @@ -48,11 +49,21 @@ export class RealtimeSession { private outstandingToolCalls = 0; /** Client-executed tool calls forwarded to the browser, awaiting `tool.result`. */ private readonly pendingClientToolCalls = new Map(); + /** Set once `start()` reaches 'active'; null before then and stays that way if the session never gets that far. */ + private sessionId: string | null = null; + /** Populated at the top of handleToolCall (before any branching) so even the unknown-tool path is covered; consumed in resolveToolCall. */ + private readonly toolCallMeta = new Map< + string, + { name: string; arguments: string; startedAt: number } + >(); + /** Assistant voice replies stream as many transcript.delta chunks then one text-less transcript.done — see docs/ANALYTICS.md. */ + private assistantTranscriptBuffer = ''; constructor( private readonly browserSocket: WebSocket, private readonly provider: RealtimeProvider, - private readonly defaults: SessionDefaults + private readonly defaults: SessionDefaults, + private readonly recorder: AnalyticsRecorder ) { this.browserSocket.on('message', (data, isBinary) => { void this.handleBrowserMessage(data, isBinary); @@ -111,6 +122,16 @@ export class RealtimeSession { this.resolveToolCall(event.callId, event.output); return; } + if (event.type === 'input_text.send') { + // Typed text is always already a complete string — unlike the voice + // path, no buffering is needed. + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'user', + text: event.text, + loggedAt: Date.now() + }); + } this.providerSession.send(event); } @@ -149,9 +170,19 @@ export class RealtimeSession { } this.providerSession = providerSession; - this.providerSession.onEvent((event) => this.handleProviderEvent(event)); + const sessionId = randomUUID(); + this.sessionId = sessionId; this.state = 'active'; - this.sendToBrowser({ type: 'session.ready', sessionId: randomUUID() }); + this.recorder.sessionStarted({ + sessionId, + model: config.model, + voice: config.voice, + startedAt: Date.now() + }); + // Recorded above before wiring onEvent, so there's no window where a + // provider event could reach handleProviderEvent before sessionId is set. + this.providerSession.onEvent((event) => this.handleProviderEvent(event)); + this.sendToBrowser({ type: 'session.ready', sessionId }); } private handleProviderEvent(event: ServerEvent): void { @@ -159,6 +190,46 @@ export class RealtimeSession { void this.handleToolCall(event); return; } + if (event.type === 'transcript.delta') { + if (event.role === 'assistant') { + this.assistantTranscriptBuffer += event.text; + } else { + // User voice transcripts arrive as a single already-complete chunk + // (conversation.item.input_audio_transcription.completed in + // providers/openai.ts) with no corresponding + // transcript.done{role:'user'} ever emitted today — log + // immediately rather than buffering. Rests on unverified live + // behavior; see docs/ANALYTICS.md. + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'user', + text: event.text, + loggedAt: Date.now() + }); + } + this.sendToBrowser(event); + return; + } + if (event.type === 'transcript.done' && event.role === 'assistant') { + const text = this.assistantTranscriptBuffer; + this.assistantTranscriptBuffer = ''; + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'assistant', + text, + loggedAt: Date.now() + }); + this.sendToBrowser(event); + return; + } + if (event.type === 'session.closed') { + // Forward first, then close: close() moves the browser socket out of + // OPEN, and sendToBrowser no-ops once that happens, so closing first + // would silently swallow this notification. + this.sendToBrowser(event); + this.close('upstream_disconnected'); + return; + } this.sendToBrowser(event); } @@ -170,6 +241,11 @@ export class RealtimeSession { */ private async handleToolCall(event: Extract): Promise { this.outstandingToolCalls++; + this.toolCallMeta.set(event.callId, { + name: event.name, + arguments: event.arguments, + startedAt: Date.now() + }); const definition = this.defaults.toolRegistry.definitions.find((d) => d.name === event.name); if (!definition) { this.resolveToolCall(event.callId, formatToolError(`Unknown tool "${event.name}".`)); @@ -213,6 +289,23 @@ export class RealtimeSession { */ private resolveToolCall(callId: string, output: string): void { this.sendToProvider({ type: 'tool.result', callId, output }); + const meta = this.toolCallMeta.get(callId); + if (meta) { + // Should always be present given handleToolCall sets it for every + // path (including the unknown-tool branch) before this can run — + // guarded defensively rather than asserted, so a future refactor + // that breaks the invariant fails soft here instead of crashing. + this.toolCallMeta.delete(callId); + this.recorder.toolCallLogged({ + sessionId: this.sessionId!, + callId, + name: meta.name, + arguments: meta.arguments, + output, + durationMs: Date.now() - meta.startedAt, + loggedAt: Date.now() + }); + } this.outstandingToolCalls--; if (this.outstandingToolCalls === 0) { this.sendToProvider({ type: 'response.trigger' }); @@ -230,9 +323,26 @@ export class RealtimeSession { this.browserSocket.close(); } this.state = 'closed'; + if (this.sessionId) { + // Guarded: sessionId is still null if the browser disconnected + // mid-handshake, before start() ever reached 'active' — no sessions + // row exists yet to update in that case. + this.recorder.sessionEnded({ sessionId: this.sessionId, endedAt: Date.now(), reason }); + } } private sendToBrowser(event: ServerEvent): void { + if (event.type === 'error' || event.type === 'session.closed') { + // Logged before the delivery guard below: the event "happened" from + // the session's point of view even if the browser socket is already + // dead by the time we try to deliver it. + this.recorder.eventLogged({ + sessionId: this.sessionId, + type: event.type, + detail: JSON.stringify(event), + loggedAt: Date.now() + }); + } if (this.browserSocket.readyState !== WebSocket.OPEN) return; this.browserSocket.send(JSON.stringify(event)); } @@ -249,7 +359,8 @@ export class RealtimeSession { export function createRealtimeSession( browserSocket: WebSocket, provider: RealtimeProvider, - defaults: SessionDefaults + defaults: SessionDefaults, + recorder: AnalyticsRecorder ): RealtimeSession { - return new RealtimeSession(browserSocket, provider, defaults); + return new RealtimeSession(browserSocket, provider, defaults, recorder); } diff --git a/src/lib/server/core/ws-handler.ts b/src/lib/server/core/ws-handler.ts index 2abac01..ecec483 100644 --- a/src/lib/server/core/ws-handler.ts +++ b/src/lib/server/core/ws-handler.ts @@ -2,6 +2,7 @@ import type { IncomingMessage } from 'node:http'; import type { Duplex } from 'node:stream'; import { WebSocketServer } from 'ws'; import { REALTIME_WS_PATH } from '../../core/protocol.js'; +import type { AnalyticsRecorder } from './analytics.js'; import type { RealtimeProvider } from './provider.js'; import { createRealtimeSession, type SessionDefaults } from './session.js'; import { verifySessionToken } from './tokens.js'; @@ -9,6 +10,7 @@ import { verifySessionToken } from './tokens.js'; export interface RealtimeUpgradeDeps { provider: RealtimeProvider; defaults: SessionDefaults; + recorder: AnalyticsRecorder; } // `noServer: true` + `handleUpgrade` below is what lets this share an @@ -42,6 +44,6 @@ export function handleRealtimeUpgrade( } wss.handleUpgrade(req, socket, head, (browserSocket) => { - createRealtimeSession(browserSocket, deps.provider, deps.defaults); + createRealtimeSession(browserSocket, deps.provider, deps.defaults, deps.recorder); }); }