Browse Source

Add server-side analytics recording to a local SQLite database

Records session start/end, conversation messages, tool calls, and
errors via better-sqlite3, hooked into RealtimeSession's existing
choke points so no client-side changes are needed. Also fixes a
pre-existing gap where an upstream-initiated disconnect only forwarded
session.closed to the browser without ever calling close(), so no
session end time was ever recorded for that path.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
main
Jonathan Hinkel 2 months ago
parent
commit
d3a5b1ad49
  1. 3
      .gitignore
  2. 66
      docs/ANALYTICS.md
  3. 1
      docs/ARCHITECTURE.md
  4. 33
      package-lock.json
  5. 2
      package.json
  6. 6
      server.js
  7. 151
      src/lib/server/core/analytics.spec.ts
  8. 198
      src/lib/server/core/analytics.ts
  9. 10
      src/lib/server/core/dev-plugin.ts
  10. 125
      src/lib/server/core/session.spec.ts
  11. 121
      src/lib/server/core/session.ts
  12. 4
      src/lib/server/core/ws-handler.ts

3
.gitignore

@ -9,6 +9,9 @@ node_modules
/build /build
/build-server /build-server
# Analytics (local SQLite db — see docs/ANALYTICS.md)
/data
# OS # OS
.DS_Store .DS_Store
Thumbs.db Thumbs.db

66
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.

1
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) | | 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) | | 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. 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.

33
package-lock.json

@ -8,6 +8,7 @@
"name": "webavatar-base", "name": "webavatar-base",
"version": "0.0.1", "version": "0.0.1",
"dependencies": { "dependencies": {
"better-sqlite3": "^13.0.3",
"ws": "^8.21.3" "ws": "^8.21.3"
}, },
"devDependencies": { "devDependencies": {
@ -16,6 +17,7 @@
"@sveltejs/adapter-node": "^5.5.7", "@sveltejs/adapter-node": "^5.5.7",
"@sveltejs/kit": "^2.63.0", "@sveltejs/kit": "^2.63.0",
"@sveltejs/vite-plugin-svelte": "^7.1.2", "@sveltejs/vite-plugin-svelte": "^7.1.2",
"@types/better-sqlite3": "^9.6.0",
"@types/node": "^24", "@types/node": "^24",
"@types/ws": "^8.18.1", "@types/ws": "^8.18.1",
"@vitest/browser-playwright": "^4.1.8", "@vitest/browser-playwright": "^4.1.8",
@ -1279,6 +1281,16 @@
"svelte": "^3 || ^4 || ^5 || ^5.0.0-next.0" "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": { "node_modules/@types/chai": {
"version": "5.2.3", "version": "5.2.3",
"resolved": "https://registry.npmjs.org/@types/chai/-/chai-5.2.3.tgz", "resolved": "https://registry.npmjs.org/@types/chai/-/chai-5.2.3.tgz",
@ -1829,6 +1841,18 @@
"node": "18 || 20 || >=22" "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": { "node_modules/brace-expansion": {
"version": "5.0.9", "version": "5.0.9",
"resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.9.tgz", "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.9.tgz",
@ -2959,6 +2983,15 @@
"dev": true, "dev": true,
"license": "MIT" "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": { "node_modules/obug": {
"version": "2.1.4", "version": "2.1.4",
"resolved": "https://registry.npmjs.org/obug/-/obug-2.1.4.tgz", "resolved": "https://registry.npmjs.org/obug/-/obug-2.1.4.tgz",

2
package.json

@ -23,6 +23,7 @@
"@sveltejs/adapter-node": "^5.5.7", "@sveltejs/adapter-node": "^5.5.7",
"@sveltejs/kit": "^2.63.0", "@sveltejs/kit": "^2.63.0",
"@sveltejs/vite-plugin-svelte": "^7.1.2", "@sveltejs/vite-plugin-svelte": "^7.1.2",
"@types/better-sqlite3": "^9.6.0",
"@types/node": "^24", "@types/node": "^24",
"@types/ws": "^8.18.1", "@types/ws": "^8.18.1",
"@vitest/browser-playwright": "^4.1.8", "@vitest/browser-playwright": "^4.1.8",
@ -42,6 +43,7 @@
"vitest-browser-svelte": "^2.1.1" "vitest-browser-svelte": "^2.1.1"
}, },
"dependencies": { "dependencies": {
"better-sqlite3": "^13.0.3",
"ws": "^8.21.3" "ws": "^8.21.3"
} }
} }

6
server.js

@ -7,7 +7,9 @@
// provide OPENAI_API_KEY / REALTIME_TOKEN_SECRET via your deployment's real // provide OPENAI_API_KEY / REALTIME_TOKEN_SECRET via your deployment's real
// secret mechanism. // secret mechanism.
import { createServer } from 'node:http'; import { createServer } from 'node:http';
import { join } from 'node:path';
import { handler } from './build/handler.js'; 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 { DEFAULT_SESSION_DEFAULTS } from './build-server/server/core/defaults.js';
import { OpenAiRealtimeProvider } from './build-server/server/core/providers/openai.js'; import { OpenAiRealtimeProvider } from './build-server/server/core/providers/openai.js';
import { handleRealtimeUpgrade } from './build-server/server/core/ws-handler.js'; import { handleRealtimeUpgrade } from './build-server/server/core/ws-handler.js';
@ -19,6 +21,8 @@ if (!apiKey) {
} }
const provider = new OpenAiRealtimeProvider({ 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 // 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 // composition root is one of the two places (alongside vite.config.ts) that
// merges core defaults with the project's tool registry. // merges core defaults with the project's tool registry.
@ -26,7 +30,7 @@ const defaults = { ...DEFAULT_SESSION_DEFAULTS, toolRegistry };
const server = createServer(handler); const server = createServer(handler);
server.on('upgrade', (req, socket, head) => { 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; const port = process.env.PORT ? Number(process.env.PORT) : 3000;

151
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();
});
});
});

198
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<AnalyticsRecorder['sessionStarted']>[0]): void {
this.insertSession.run(sessionId, model, voice ?? null, startedAt);
}
sessionEnded({
sessionId,
endedAt,
reason
}: Parameters<AnalyticsRecorder['sessionEnded']>[0]): void {
this.updateSessionEnd.run(endedAt, reason, sessionId);
}
messageLogged({
sessionId,
role,
text,
loggedAt
}: Parameters<AnalyticsRecorder['messageLogged']>[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<AnalyticsRecorder['toolCallLogged']>[0]): void {
this.insertToolCall.run(sessionId, callId, name, args, output, durationMs, loggedAt);
}
eventLogged({
sessionId,
type,
detail,
loggedAt
}: Parameters<AnalyticsRecorder['eventLogged']>[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();
}
}

10
src/lib/server/core/dev-plugin.ts

@ -1,4 +1,6 @@
import { join } from 'node:path';
import { loadEnv, type Plugin } from 'vite'; import { loadEnv, type Plugin } from 'vite';
import { SqliteAnalyticsRecorder } from './analytics.js';
import { OpenAiRealtimeProvider } from './providers/openai.js'; import { OpenAiRealtimeProvider } from './providers/openai.js';
import type { SessionDefaults } from './session.js'; import type { SessionDefaults } from './session.js';
import { handleRealtimeUpgrade } from './ws-handler.js'; import { handleRealtimeUpgrade } from './ws-handler.js';
@ -43,9 +45,15 @@ export function realtimeDevWsPlugin(options: RealtimeDevWsPluginOptions): Plugin
); );
} }
const provider = new OpenAiRealtimeProvider({ apiKey: apiKey ?? '' }); 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) => { server.httpServer?.on('upgrade', (req, socket, head) => {
handleRealtimeUpgrade(req, socket, head, { provider, defaults: options.defaults }); handleRealtimeUpgrade(req, socket, head, {
provider,
defaults: options.defaults,
recorder
});
}); });
} }
}; };

125
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 WebSocket from 'ws';
import type { ClientEvent, ServerEvent } from '../../core/protocol.js'; import type { ClientEvent, ServerEvent } from '../../core/protocol.js';
import type { ToolDefinition } from '../../core/tools.js'; import type { ToolDefinition } from '../../core/tools.js';
import type { AnalyticsRecorder } from './analytics.js';
import type { ProviderSession, RealtimeProvider } from './provider.js'; import type { ProviderSession, RealtimeProvider } from './provider.js';
import { createRealtimeSession, resolveSessionConfig } from './session.js'; import { createRealtimeSession, resolveSessionConfig } from './session.js';
import { createToolRegistry } from './tools.js'; import { createToolRegistry } from './tools.js';
@ -84,6 +85,30 @@ class FakeProvider implements RealtimeProvider {
} }
} }
class FakeAnalyticsRecorder implements AnalyticsRecorder {
sessionStartedCalls: Parameters<AnalyticsRecorder['sessionStarted']>[0][] = [];
sessionEndedCalls: Parameters<AnalyticsRecorder['sessionEnded']>[0][] = [];
messageLoggedCalls: Parameters<AnalyticsRecorder['messageLogged']>[0][] = [];
toolCallLoggedCalls: Parameters<AnalyticsRecorder['toolCallLogged']>[0][] = [];
eventLoggedCalls: Parameters<AnalyticsRecorder['eventLogged']>[0][] = [];
sessionStarted(p: Parameters<AnalyticsRecorder['sessionStarted']>[0]) {
this.sessionStartedCalls.push(p);
}
sessionEnded(p: Parameters<AnalyticsRecorder['sessionEnded']>[0]) {
this.sessionEndedCalls.push(p);
}
messageLogged(p: Parameters<AnalyticsRecorder['messageLogged']>[0]) {
this.messageLoggedCalls.push(p);
}
toolCallLogged(p: Parameters<AnalyticsRecorder['toolCallLogged']>[0]) {
this.toolCallLoggedCalls.push(p);
}
eventLogged(p: Parameters<AnalyticsRecorder['eventLogged']>[0]) {
this.eventLoggedCalls.push(p);
}
}
function makeSocket() { function makeSocket() {
const socket = new FakeBrowserSocket(); const socket = new FakeBrowserSocket();
return { socket, asWs: socket as unknown as WebSocket }; return { socket, asWs: socket as unknown as WebSocket };
@ -112,20 +137,25 @@ describe('resolveSessionConfig', () => {
describe('RealtimeSession', () => { describe('RealtimeSession', () => {
it('starts a provider session on session.start and replies with session.ready', async () => { it('starts a provider session on session.start and replies with session.ready', async () => {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
await vi.waitFor(() => await vi.waitFor(() =>
expect(socket.sent).toContainEqual(expect.objectContaining({ type: 'session.ready' })) 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 () => { it('rejects content events sent before session.start', async () => {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'input_text.send', text: 'too early' }); socket.receive({ type: 'input_text.send', text: 'too early' });
await vi.waitFor(() => await vi.waitFor(() =>
@ -134,12 +164,16 @@ describe('RealtimeSession', () => {
) )
); );
expect(provider.sessions).toHaveLength(0); 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 provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
@ -148,12 +182,16 @@ describe('RealtimeSession', () => {
await vi.waitFor(() => await vi.waitFor(() =>
expect(provider.sessions[0].sent).toContainEqual({ type: 'input_text.send', text: 'hello' }) 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 () => { it('relays provider server events back to the browser socket', async () => {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); 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 () => { it('closes the provider session when the browser disconnects', async () => {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
@ -183,12 +253,16 @@ describe('RealtimeSession', () => {
socket.emit('close'); socket.emit('close');
await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true));
expect(provider.sessions[0].closeReason).toBe('client_disconnected'); 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 () => { it('closes the provider session on session.end and closes the browser socket', async () => {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); await vi.waitFor(() => expect(provider.sessions).toHaveLength(1));
@ -196,12 +270,16 @@ describe('RealtimeSession', () => {
socket.receive({ type: 'session.end' }); socket.receive({ type: 'session.end' });
await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true));
expect(socket.closeCalls).toBe(1); 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 () => { it('does not leak an upstream session if the browser disconnects mid-handshake', async () => {
const provider = new FakeProvider({ deferred: true }); const provider = new FakeProvider({ deferred: true });
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, DEFAULTS); createRealtimeSession(asWs, provider, DEFAULTS, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
socket.emit('close'); // disconnect while the provider handshake is still pending 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).toHaveLength(1));
await vi.waitFor(() => expect(provider.sessions[0].closed).toBe(true)); 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) { async function startActiveSession(defaults: typeof DEFAULTS) {
const provider = new FakeProvider(); const provider = new FakeProvider();
const recorder = new FakeAnalyticsRecorder();
const { socket, asWs } = makeSocket(); const { socket, asWs } = makeSocket();
createRealtimeSession(asWs, provider, defaults); createRealtimeSession(asWs, provider, defaults, recorder);
socket.receive({ type: 'session.start' }); socket.receive({ type: 'session.start' });
await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); 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', () => { describe('RealtimeSession tool calls', () => {
@ -249,7 +343,7 @@ describe('RealtimeSession tool calls', () => {
const toolRegistry = createToolRegistry([FAST_TOOL], { const toolRegistry = createToolRegistry([FAST_TOOL], {
fast_tool: async () => ({ done: true }) fast_tool: async () => ({ done: true })
}); });
const { session } = await startActiveSession({ ...DEFAULTS, toolRegistry }); const { session, recorder } = await startActiveSession({ ...DEFAULTS, toolRegistry });
session.emitToOrchestrator({ session.emitToOrchestrator({
type: 'tool.call', type: 'tool.call',
@ -266,6 +360,15 @@ describe('RealtimeSession tool calls', () => {
}) })
); );
await vi.waitFor(() => expect(session.sent).toContainEqual({ type: 'response.trigger' })); 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 () => { it('only triggers response.trigger once every parallel tool call has resolved', async () => {

121
src/lib/server/core/session.ts

@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto';
import WebSocket from 'ws'; import WebSocket from 'ws';
import type { ClientEvent, ServerEvent, SessionStartConfig } from '../../core/protocol.js'; import type { ClientEvent, ServerEvent, SessionStartConfig } from '../../core/protocol.js';
import { formatToolError } from '../../core/tools.js'; import { formatToolError } from '../../core/tools.js';
import type { AnalyticsRecorder } from './analytics.js';
import type { ProviderSession, RealtimeProvider, ResolvedSessionConfig } from './provider.js'; import type { ProviderSession, RealtimeProvider, ResolvedSessionConfig } from './provider.js';
import type { ToolRegistry } from './tools.js'; import type { ToolRegistry } from './tools.js';
@ -48,11 +49,21 @@ export class RealtimeSession {
private outstandingToolCalls = 0; private outstandingToolCalls = 0;
/** Client-executed tool calls forwarded to the browser, awaiting `tool.result`. */ /** Client-executed tool calls forwarded to the browser, awaiting `tool.result`. */
private readonly pendingClientToolCalls = new Map<string, NodeJS.Timeout>(); private readonly pendingClientToolCalls = new Map<string, NodeJS.Timeout>();
/** 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( constructor(
private readonly browserSocket: WebSocket, private readonly browserSocket: WebSocket,
private readonly provider: RealtimeProvider, private readonly provider: RealtimeProvider,
private readonly defaults: SessionDefaults private readonly defaults: SessionDefaults,
private readonly recorder: AnalyticsRecorder
) { ) {
this.browserSocket.on('message', (data, isBinary) => { this.browserSocket.on('message', (data, isBinary) => {
void this.handleBrowserMessage(data, isBinary); void this.handleBrowserMessage(data, isBinary);
@ -111,6 +122,16 @@ export class RealtimeSession {
this.resolveToolCall(event.callId, event.output); this.resolveToolCall(event.callId, event.output);
return; 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); this.providerSession.send(event);
} }
@ -149,9 +170,19 @@ export class RealtimeSession {
} }
this.providerSession = providerSession; this.providerSession = providerSession;
this.providerSession.onEvent((event) => this.handleProviderEvent(event)); const sessionId = randomUUID();
this.sessionId = sessionId;
this.state = 'active'; 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 { private handleProviderEvent(event: ServerEvent): void {
@ -159,6 +190,46 @@ export class RealtimeSession {
void this.handleToolCall(event); void this.handleToolCall(event);
return; 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); this.sendToBrowser(event);
} }
@ -170,6 +241,11 @@ export class RealtimeSession {
*/ */
private async handleToolCall(event: Extract<ServerEvent, { type: 'tool.call' }>): Promise<void> { private async handleToolCall(event: Extract<ServerEvent, { type: 'tool.call' }>): Promise<void> {
this.outstandingToolCalls++; 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); const definition = this.defaults.toolRegistry.definitions.find((d) => d.name === event.name);
if (!definition) { if (!definition) {
this.resolveToolCall(event.callId, formatToolError(`Unknown tool "${event.name}".`)); this.resolveToolCall(event.callId, formatToolError(`Unknown tool "${event.name}".`));
@ -213,6 +289,23 @@ export class RealtimeSession {
*/ */
private resolveToolCall(callId: string, output: string): void { private resolveToolCall(callId: string, output: string): void {
this.sendToProvider({ type: 'tool.result', callId, output }); 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--; this.outstandingToolCalls--;
if (this.outstandingToolCalls === 0) { if (this.outstandingToolCalls === 0) {
this.sendToProvider({ type: 'response.trigger' }); this.sendToProvider({ type: 'response.trigger' });
@ -230,9 +323,26 @@ export class RealtimeSession {
this.browserSocket.close(); this.browserSocket.close();
} }
this.state = 'closed'; 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 { 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; if (this.browserSocket.readyState !== WebSocket.OPEN) return;
this.browserSocket.send(JSON.stringify(event)); this.browserSocket.send(JSON.stringify(event));
} }
@ -249,7 +359,8 @@ export class RealtimeSession {
export function createRealtimeSession( export function createRealtimeSession(
browserSocket: WebSocket, browserSocket: WebSocket,
provider: RealtimeProvider, provider: RealtimeProvider,
defaults: SessionDefaults defaults: SessionDefaults,
recorder: AnalyticsRecorder
): RealtimeSession { ): RealtimeSession {
return new RealtimeSession(browserSocket, provider, defaults); return new RealtimeSession(browserSocket, provider, defaults, recorder);
} }

4
src/lib/server/core/ws-handler.ts

@ -2,6 +2,7 @@ import type { IncomingMessage } from 'node:http';
import type { Duplex } from 'node:stream'; import type { Duplex } from 'node:stream';
import { WebSocketServer } from 'ws'; import { WebSocketServer } from 'ws';
import { REALTIME_WS_PATH } from '../../core/protocol.js'; import { REALTIME_WS_PATH } from '../../core/protocol.js';
import type { AnalyticsRecorder } from './analytics.js';
import type { RealtimeProvider } from './provider.js'; import type { RealtimeProvider } from './provider.js';
import { createRealtimeSession, type SessionDefaults } from './session.js'; import { createRealtimeSession, type SessionDefaults } from './session.js';
import { verifySessionToken } from './tokens.js'; import { verifySessionToken } from './tokens.js';
@ -9,6 +10,7 @@ import { verifySessionToken } from './tokens.js';
export interface RealtimeUpgradeDeps { export interface RealtimeUpgradeDeps {
provider: RealtimeProvider; provider: RealtimeProvider;
defaults: SessionDefaults; defaults: SessionDefaults;
recorder: AnalyticsRecorder;
} }
// `noServer: true` + `handleUpgrade` below is what lets this share an // `noServer: true` + `handleUpgrade` below is what lets this share an
@ -42,6 +44,6 @@ export function handleRealtimeUpgrade(
} }
wss.handleUpgrade(req, socket, head, (browserSocket) => { wss.handleUpgrade(req, socket, head, (browserSocket) => {
createRealtimeSession(browserSocket, deps.provider, deps.defaults); createRealtimeSession(browserSocket, deps.provider, deps.defaults, deps.recorder);
}); });
} }

Loading…
Cancel
Save