diff --git a/docs/ANALYTICS.md b/docs/ANALYTICS.md index 231ff7c..187d2c8 100644 --- a/docs/ANALYTICS.md +++ b/docs/ANALYTICS.md @@ -8,7 +8,13 @@ Server-side recording of session volume, conversation content, tool usage, and f ## 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. +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 keeps analytics writes compatible with the session's existing synchronous lifecycle methods. + +## Failure isolation + +Analytics is best-effort telemetry, not part of session correctness. Every `AnalyticsRecorder` call in `session.ts` runs through one non-throwing boundary: a full disk, locked database, or other recorder failure is reported through the server logger under the `analytics` category, then normal provider traffic, browser delivery, and teardown continue. Failure reporting is itself guarded so a broken logger cannot turn a telemetry failure back into a session failure. + +This boundary covers writes after the recorder has been constructed. Recorder construction still happens eagerly at process startup; an invalid/unwritable database path is a configuration or boot failure rather than a mid-session degradation case. ## The transcript-buffering gotcha diff --git a/src/lib/server/core/session.spec.ts b/src/lib/server/core/session.spec.ts index 39ed7ad..7aad37f 100644 --- a/src/lib/server/core/session.spec.ts +++ b/src/lib/server/core/session.spec.ts @@ -112,6 +112,38 @@ class FakeAnalyticsRecorder implements AnalyticsRecorder { } } +type AnalyticsMethod = + 'sessionStarted' | 'sessionEnded' | 'messageLogged' | 'toolCallLogged' | 'eventLogged'; + +class ThrowingAnalyticsRecorder implements AnalyticsRecorder { + readonly calls: AnalyticsMethod[] = []; + + private fail(method: AnalyticsMethod): never { + this.calls.push(method); + throw new Error(`Simulated ${method} storage failure.`); + } + + sessionStarted(): void { + this.fail('sessionStarted'); + } + + sessionEnded(): void { + this.fail('sessionEnded'); + } + + messageLogged(): void { + this.fail('messageLogged'); + } + + toolCallLogged(): void { + this.fail('toolCallLogged'); + } + + eventLogged(): void { + this.fail('eventLogged'); + } +} + function makeSocket() { const socket = new FakeBrowserSocket(); return { socket, asWs: socket as unknown as WebSocket }; @@ -252,12 +284,103 @@ describe('RealtimeSession', () => { recorder.close(); }); - it('contains unexpected async message failures and leaves other sessions usable', async () => { + it('degrades gracefully when every analytics write fails', async () => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + const provider = new FakeProvider(); + const recorder = new ThrowingAnalyticsRecorder(); + const { socket, asWs } = makeSocket(); + const orchestrator = createRealtimeSession( + asWs, + provider, + DEFAULTS, + recorder, + TEST_TOOL_TIMEOUT_MS + ); + + 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' })) + ); + + socket.receive({ type: 'input_text.send' }); + await vi.waitFor(() => + expect(socket.sent).toContainEqual( + expect.objectContaining({ type: 'error', code: 'invalid_client_event' }) + ) + ); + + socket.receive({ type: 'input_text.send', text: 'still works' }); + await vi.waitFor(() => + expect(provider.sessions[0].sent).toContainEqual({ + type: 'input_text.send', + text: 'still works' + }) + ); + + provider.sessions[0].emitToOrchestrator({ + type: 'transcript.delta', + role: 'user', + text: 'voice input' + }); + provider.sessions[0].emitToOrchestrator({ + type: 'transcript.delta', + role: 'assistant', + text: 'voice output' + }); + provider.sessions[0].emitToOrchestrator({ type: 'transcript.done', role: 'assistant' }); + expect(socket.sent).toContainEqual({ + type: 'transcript.delta', + role: 'user', + text: 'voice input' + }); + expect(socket.sent).toContainEqual({ type: 'transcript.done', role: 'assistant' }); + + provider.sessions[0].emitToOrchestrator({ + type: 'tool.call', + callId: 'call-1', + name: 'nonexistent_tool', + arguments: '{}' + }); + await vi.waitFor(() => + expect(provider.sessions[0].sent).toContainEqual({ type: 'response.trigger' }) + ); + + provider.sessions[0].emitToOrchestrator({ + type: 'session.closed', + reason: 'upstream_disconnected' + }); + expect(socket.sent).toContainEqual({ + type: 'session.closed', + reason: 'upstream_disconnected' + }); + expect(orchestrator.getState()).toBe('closed'); + expect(provider.sessions[0].closed).toBe(true); + expect(socket.closeCalls).toBe(1); + expect(socket.sent).not.toContainEqual( + expect.objectContaining({ type: 'error', code: 'internal_error' }) + ); + expect(recorder.calls).toEqual([ + 'sessionStarted', + 'eventLogged', + 'messageLogged', + 'messageLogged', + 'messageLogged', + 'toolCallLogged', + 'eventLogged', + 'sessionEnded' + ]); + expect(errorSpy).toHaveBeenCalledTimes(recorder.calls.length); + expect(errorSpy).toHaveBeenCalledWith( + '[analytics] Failed to record session start.', + expect.objectContaining({ message: expect.stringContaining('storage failure') }) + ); + errorSpy.mockRestore(); + }); + + it('contains non-analytics async message failures and leaves other sessions usable', async () => { const provider = new FakeProvider(); const recorder = new FakeAnalyticsRecorder(); - recorder.messageLogged = () => { - throw new Error('analytics unavailable'); - }; const affected = makeSocket(); const other = makeSocket(); createRealtimeSession(affected.asWs, provider, DEFAULTS, recorder, TEST_TOOL_TIMEOUT_MS); @@ -267,8 +390,11 @@ describe('RealtimeSession', () => { await vi.waitFor(() => expect(provider.sessions).toHaveLength(1)); other.socket.receive({ type: 'session.start' }); await vi.waitFor(() => expect(provider.sessions).toHaveLength(2)); - affected.socket.receive({ type: 'input_text.send', text: 'hello' }); + provider.sessions[0].send = () => { + throw new Error('provider send failed'); + }; + affected.socket.receive({ type: 'response.cancel' }); await vi.waitFor(() => expect(affected.socket.sent).toContainEqual({ type: 'error', diff --git a/src/lib/server/core/session.ts b/src/lib/server/core/session.ts index d0236b4..84c0944 100644 --- a/src/lib/server/core/session.ts +++ b/src/lib/server/core/session.ts @@ -164,11 +164,13 @@ export class RealtimeSession { // Typed text is always already a complete string — unlike the voice // path, no buffering is needed. this.logger.info('messages', 'User message', { role: 'user', text: event.text }); - this.recorder.messageLogged({ - sessionId: this.sessionId!, - role: 'user', - text: event.text, - loggedAt: Date.now() + this.recordAnalytics('message', () => { + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'user', + text: event.text, + loggedAt: Date.now() + }); }); } this.providerSession.send(event); @@ -259,11 +261,13 @@ export class RealtimeSession { role: 'system', text: config.instructions }); - this.recorder.sessionStarted({ - sessionId, - model: config.model, - voice: config.voice, - startedAt: Date.now() + this.recordAnalytics('session start', () => { + 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. @@ -287,11 +291,13 @@ export class RealtimeSession { // immediately rather than buffering. Rests on unverified live // behavior; see docs/ANALYTICS.md. this.logger.info('messages', 'User message', { role: 'user', text: event.text }); - this.recorder.messageLogged({ - sessionId: this.sessionId!, - role: 'user', - text: event.text, - loggedAt: Date.now() + this.recordAnalytics('message', () => { + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'user', + text: event.text, + loggedAt: Date.now() + }); }); } this.sendToBrowser(event); @@ -301,11 +307,13 @@ export class RealtimeSession { const text = this.assistantTranscriptBuffer; this.assistantTranscriptBuffer = ''; this.logger.info('messages', 'Assistant message', { role: 'assistant', text }); - this.recorder.messageLogged({ - sessionId: this.sessionId!, - role: 'assistant', - text, - loggedAt: Date.now() + this.recordAnalytics('message', () => { + this.recorder.messageLogged({ + sessionId: this.sessionId!, + role: 'assistant', + text, + loggedAt: Date.now() + }); }); this.sendToBrowser(event); return; @@ -391,14 +399,16 @@ export class RealtimeSession { callId, durationMs: Date.now() - meta.startedAt }); - this.recorder.toolCallLogged({ - sessionId: this.sessionId!, - callId, - name: meta.name, - arguments: meta.arguments, - output, - durationMs: Date.now() - meta.startedAt, - loggedAt: Date.now() + this.recordAnalytics('tool call', () => { + this.recorder.toolCallLogged({ + sessionId: this.sessionId!, + callId, + name: meta.name, + arguments: meta.arguments, + output, + durationMs: Date.now() - meta.startedAt, + loggedAt: Date.now() + }); }); } this.outstandingToolCalls--; @@ -423,7 +433,13 @@ export class RealtimeSession { // 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 }); + this.recordAnalytics('session end', () => { + this.recorder.sessionEnded({ + sessionId: this.sessionId!, + endedAt: Date.now(), + reason + }); + }); } } @@ -432,17 +448,42 @@ export class RealtimeSession { // 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() + this.recordAnalytics('event', () => { + 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)); } + /** + * Analytics is best-effort telemetry, never part of session correctness. + * A full disk, locked database, or broken recorder must not interrupt + * provider traffic, browser delivery, or teardown. + */ + private recordAnalytics(operation: string, write: () => void): void { + try { + write(); + } catch (error) { + let message = 'Unknown error'; + try { + message = error instanceof Error ? error.message : String(error); + } catch { + // Even a hostile thrown value with a broken toString stays contained. + } + try { + this.logger.error('analytics', `Failed to record ${operation}.`, { message }); + } catch { + // Telemetry failure reporting is best-effort too. + } + } + } + /** * Verifies a `debug.subscribe` request and replies with the result. In * dev, any request succeeds (frictionless — see docs/LOGGING.md); in