Add first-frame audio sniff and session-end summary diagnostic logs (#84)

* feat: add first-frame audio sniff and session-end summary diagnostic logs

* review: sniff per tag, fix head-bytes label, add diagnostic-logging tests

* review: quote JSON log fields, reset sniff on reattach, nits

* diag: log per-transcription text length and preview at DEBUG

* worker: forward LOG_LEVEL to container (derives debug from DEBUG=true)

* review: shallow-copy inputFormat, gate debug log, flip DEBUG default to false

Author: rpurdel

src/OutgoingConnection.ts

@@ -72,6 +72,11 @@ export class OutgoingConnection {
 		this.initializeBackend();
 	}
 
+	getInputFormat(): AudioFormat {
+		// Return a shallow copy so callers can't mutate the decoder's format state.
+		return { ...this.inputAudioFormat };
+	}
+
 	updateInputFormat(inputFormat: AudioFormat): void {
 		// Validate synchronously so callers get an immediate error rather than
 		// an async failure deep in reinitializeDecoder -> createAudioDecoder.
@@ -179,11 +184,13 @@ export class OutgoingConnection {
 	private setupBackendHandlers(backend: TranscriptionBackend): void {
 		backend.onInterimTranscription = (message) => {
 			getInstruments().transcriptionsReceivedTotal.add(1, { provider: this.options.provider || 'unknown', is_interim: 'true' });
+			this.logTranscriptionSummary(message, true);
 			this.onInterimTranscription?.(message);
 		};
 
 		backend.onCompleteTranscription = (message) => {
 			getInstruments().transcriptionsReceivedTotal.add(1, { provider: this.options.provider || 'unknown', is_interim: 'false' });
+			this.logTranscriptionSummary(message, false);
 			this.clearIdleCommitTimeout();
 			this.onCompleteTranscription?.(message);
 		};
@@ -202,6 +209,21 @@ export class OutgoingConnection {
 		};
 	}
 
+	private logTranscriptionSummary(message: TranscriptionMessage, isInterim: boolean): void {
+		// DEBUG-only: summarises each transcription arriving from the backend so we
+		// can tell apart empty end-of-utterance finals from real speech without
+		// logging full PII.  Controlled by LOG_LEVEL=debug.
+		// Early-exit when debug isn't enabled so we don't allocate join/slice strings
+		// on every transcription in production.
+		if (!logger.isLevelEnabled('debug')) return;
+		const segments = message.transcript ?? [];
+		const text = segments.map((s) => s.text ?? '').join(' ').trim();
+		const preview = text.length > 40 ? text.slice(0, 40) + '…' : text;
+		logger.debug(
+			`Backend ${isInterim ? 'interim' : 'final'} tag=${this.localTag} lang=${message.language ?? 'n/a'} segments=${segments.length} textLen=${text.length} preview=${JSON.stringify(preview)}`,
+		);
+	}
+
 	private async reinitializeDecoder(): Promise<void> {
 		if (!this.backend) {
 			throw new Error('Cannot initialize decoder without a backend');

src/transcriberproxy.ts

@@ -42,6 +42,10 @@ export class TranscriberProxy extends EventEmitter {
 	private sessionId?: string;
 	private dispatcherConnection?: DispatcherConnection;
 	private createdAt: number;
+	private audioPacketCount = 0;
+	private interimTranscriptionCount = 0;
+	private finalTranscriptionCount = 0;
+	private firstFrameLoggedTags = new Set<string>();
 
 	constructor(ws: WebSocket, options: TranscriberProxyOptions) {
 		super({ captureRejections: true });
@@ -158,9 +162,11 @@ export class TranscriberProxy extends EventEmitter {
 		const newConnection = new OutgoingConnection(tag, mediaFormat, this.options);
 
 		newConnection.onInterimTranscription = (message) => {
+			this.interimTranscriptionCount++;
 			this.emit('interim_transcription', message);
 		};
 		newConnection.onCompleteTranscription = (message) => {
+			this.finalTranscriptionCount++;
 			// Dump transcript if enabled
 			if (this.transcriptDumpStream) {
 				try {
@@ -273,6 +279,24 @@ export class TranscriberProxy extends EventEmitter {
 				logger.warn(`Received media event for tag "${tag}" with no prior start event; creating connection with encoding "${encoding}"`);
 				connection = this.createConnection(tag, mediaFormat);
 			}
+			const payloadB64 = parsedMessage.media?.payload;
+			const hasAudio = typeof payloadB64 === 'string' && payloadB64.length > 0;
+			if (hasAudio) {
+				this.audioPacketCount++;
+				if (!this.firstFrameLoggedTags.has(tag)) {
+					// 64 base64 chars decode to at most 48 bytes; we only emit the first 16.
+					const head = Buffer.from(payloadB64.slice(0, 64), 'base64');
+					const headByteCount = Math.min(16, head.length);
+					const headHex = head.subarray(0, headByteCount).toString('hex');
+					const mediaSnapshot = { ...parsedMessage.media, payload: `<b64:${payloadB64.length} chars, first ${headByteCount} decoded bytes=${headHex}>` };
+					// JSON-valued fields are quoted so that downstream logfmt-style parsers
+					// don't misinterpret spaces inside the JSON payload (e.g. inside `tag`).
+					logger.info(
+						`First client frame sniff: sessionId=${this.sessionId} tag=${tag} provider=${this.options.provider ?? 'default'} urlEncoding=${this.options.encoding ?? 'opus'} startFormat='${JSON.stringify(connection.getInputFormat())}' media='${JSON.stringify(mediaSnapshot)}'`,
+					);
+					this.firstFrameLoggedTags.add(tag);
+				}
+			}
 			connection.handleMediaEvent(parsedMessage);
 		}
 	}
@@ -348,6 +372,11 @@ export class TranscriberProxy extends EventEmitter {
 		// Re-setup listeners on new WebSocket
 		this.setupWebSocketListeners();
 
+		// Treat a reattach as a new connection for diagnostic purposes: the client
+		// may negotiate a different audio format on reconnect, so fire the
+		// first-frame sniff again on the first real audio packet per tag.
+		this.firstFrameLoggedTags.clear();
+
 		// Reset chunk tracking on all connections so frames from the new client
 		// aren't discarded as "reordered" (chunk numbers restart from 0)
 		this.outgoingConnections.forEach((connection, tag) => {
@@ -360,6 +389,9 @@ export class TranscriberProxy extends EventEmitter {
 	}
 
 	close(): void {
+		logger.info(
+			`Session ended: sessionId=${this.sessionId} provider=${this.options.provider ?? 'default'} audioPackets=${this.audioPacketCount} interims=${this.interimTranscriptionCount} finals=${this.finalTranscriptionCount} durationSec=${this.getSessionDurationSec().toFixed(1)}`,
+		);
 		this.outgoingConnections.forEach((connection) => {
 			connection.close();
 		});

test/unit/OutgoingConnection.test.ts

@@ -23,6 +23,7 @@ vi.mock('../../src/logger', () => ({
 		error: vi.fn(),
 		warn: vi.fn(),
 		debug: vi.fn(),
+		isLevelEnabled: vi.fn(() => true),
 	},
 }));
 

test/unit/TranscriberProxy.test.ts

@@ -46,12 +46,14 @@ vi.mock('../../src/config', () => ({
 // to `this` in the implementation do not persist on the created instance. Passing the
 // implementation directly to vi.fn() avoids this issue.
 vi.mock('../../src/OutgoingConnection', () => ({
-	OutgoingConnection: vi.fn(function (this: any, tag: string) {
+	OutgoingConnection: vi.fn(function (this: any, tag: string, inputFormat: unknown) {
 		this.localTag = tag;
 		this.participantId = tag.split('-')[0]; // Extract participant ID from tag like "participant1-ssrc123"
 		this.handleMediaEvent = vi.fn();
 		this.addTranscriptContext = vi.fn();
 		this.updateInputFormat = vi.fn();
+		this.getInputFormat = vi.fn(() => inputFormat ?? { encoding: 'opus' });
+		this.resetChunkTracking = vi.fn();
 		this.close = vi.fn();
 		this.onInterimTranscription = undefined;
 		this.onCompleteTranscription = undefined;
@@ -578,4 +580,115 @@ describe('TranscriberProxy', () => {
 			expect(conn.handleMediaEvent).toHaveBeenCalledWith(mediaEvent);
 		});
 	});
+
+	describe('diagnostic logging', () => {
+		// 'T2dnUw==' is base64 for 'OggS' — the Ogg page capture pattern.
+		const OGG_PAYLOAD = 'T2dnUw==';
+
+		it('logs the first client frame sniff exactly once per tag', () => {
+			const proxy = new TranscriberProxy(mockWebSocket, options);
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag1', mediaFormat: { encoding: 'opus' } } });
+			vi.mocked(logger.info).mockClear();
+
+			const media1 = { event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } };
+			proxy.handleMediaEvent(media1);
+			proxy.handleMediaEvent(media1);
+			proxy.handleMediaEvent(media1);
+
+			const sniffCalls = vi.mocked(logger.info).mock.calls.filter(([msg]) => typeof msg === 'string' && msg.startsWith('First client frame sniff:'));
+			expect(sniffCalls).toHaveLength(1);
+			const msg = sniffCalls[0][0] as string;
+			expect(msg).toContain('tag=tag1');
+			expect(msg).toContain('urlEncoding=opus');
+			expect(msg).toContain(`startFormat='{"encoding":"opus"}'`);
+			expect(msg).toContain('4f676753'); // 'OggS' in hex
+			expect(msg).toContain(`<b64:${OGG_PAYLOAD.length} chars, first 4 decoded bytes=4f676753>`);
+		});
+
+		it('logs the first client frame sniff once per participant tag', () => {
+			const proxy = new TranscriberProxy(mockWebSocket, options);
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag1', mediaFormat: { encoding: 'opus' } } });
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag2', mediaFormat: { encoding: 'opus' } } });
+			vi.mocked(logger.info).mockClear();
+
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag2', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 1, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag2', payload: OGG_PAYLOAD, chunk: 1, timestamp: 0 } });
+
+			const sniffCalls = vi.mocked(logger.info).mock.calls.filter(([msg]) => typeof msg === 'string' && msg.startsWith('First client frame sniff:'));
+			expect(sniffCalls).toHaveLength(2);
+			expect(sniffCalls[0][0]).toContain('tag=tag1');
+			expect(sniffCalls[1][0]).toContain('tag=tag2');
+		});
+
+		it('does not sniff or count empty-payload frames, and retries the sniff on the next real frame', () => {
+			const proxy = new TranscriberProxy(mockWebSocket, options);
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag1', mediaFormat: { encoding: 'opus' } } });
+			vi.mocked(logger.info).mockClear();
+
+			// Missing payload → should not log, should not flip the flag
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', chunk: 0, timestamp: 0 } });
+			// Empty-string payload → same
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: '', chunk: 1, timestamp: 0 } });
+
+			let sniffCalls = vi.mocked(logger.info).mock.calls.filter(([msg]) => typeof msg === 'string' && msg.startsWith('First client frame sniff:'));
+			expect(sniffCalls).toHaveLength(0);
+
+			// Real audio frame → sniff now fires (not short-circuited by prior empty frames)
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 2, timestamp: 0 } });
+
+			sniffCalls = vi.mocked(logger.info).mock.calls.filter(([msg]) => typeof msg === 'string' && msg.startsWith('First client frame sniff:'));
+			expect(sniffCalls).toHaveLength(1);
+
+			// Session-end summary reflects that only the real frame was counted as audio
+			vi.mocked(logger.info).mockClear();
+			proxy.close();
+			const endCall = vi.mocked(logger.info).mock.calls.find(([msg]) => typeof msg === 'string' && msg.startsWith('Session ended:'));
+			expect(endCall?.[0]).toContain('audioPackets=1');
+		});
+
+		it('fires the first-frame sniff again after a WebSocket reattach', () => {
+			const proxy = new TranscriberProxy(mockWebSocket, options);
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag1', mediaFormat: { encoding: 'opus' } } });
+
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 1, timestamp: 0 } });
+
+			vi.mocked(logger.info).mockClear();
+			proxy.reattachWebSocket({ addEventListener: vi.fn(), send: vi.fn(), close: vi.fn() } as any);
+
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } });
+			const sniffCalls = vi.mocked(logger.info).mock.calls.filter(([msg]) => typeof msg === 'string' && msg.startsWith('First client frame sniff:'));
+			expect(sniffCalls).toHaveLength(1);
+		});
+
+		it('emits a session-end summary with audioPackets, interims, finals, and provider', () => {
+			const proxy = new TranscriberProxy(mockWebSocket, { ...options, provider: 'deepgram' });
+			proxy.handleStartEvent({ event: 'start', start: { tag: 'tag1', mediaFormat: { encoding: 'opus' } } });
+			const conn = vi.mocked(OutgoingConnection).mock.instances[0] as any;
+
+			// 3 audio packets
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 0, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 1, timestamp: 0 } });
+			proxy.handleMediaEvent({ event: 'media', media: { tag: 'tag1', payload: OGG_PAYLOAD, chunk: 2, timestamp: 0 } });
+
+			// 2 interims + 1 final via the connection callbacks
+			conn.onInterimTranscription({ transcript: [], is_interim: true, message_id: 'a', type: 'transcription-result', event: 'transcription-result', participant: { id: 'tag1' }, timestamp: 0 });
+			conn.onInterimTranscription({ transcript: [], is_interim: true, message_id: 'b', type: 'transcription-result', event: 'transcription-result', participant: { id: 'tag1' }, timestamp: 0 });
+			conn.onCompleteTranscription({ transcript: [{ text: 'hi' }], is_interim: false, message_id: 'c', type: 'transcription-result', event: 'transcription-result', participant: { id: 'tag1' }, timestamp: 0 });
+
+			vi.mocked(logger.info).mockClear();
+			proxy.close();
+
+			const endCall = vi.mocked(logger.info).mock.calls.find(([msg]) => typeof msg === 'string' && msg.startsWith('Session ended:'));
+			expect(endCall).toBeDefined();
+			const endMsg = endCall![0] as string;
+			expect(endMsg).toContain('provider=deepgram');
+			expect(endMsg).toContain('audioPackets=3');
+			expect(endMsg).toContain('interims=2');
+			expect(endMsg).toContain('finals=1');
+			expect(endMsg).toMatch(/durationSec=\d+\.\d/);
+		});
+	});
 });

worker/env.d.ts

@@ -34,6 +34,7 @@ export interface Env {
 	PROVIDERS_PRIORITY?: string;
 	FORCE_COMMIT_TIMEOUT?: string;
 	DEBUG?: string;
+	LOG_LEVEL?: string;
 	ROUTING_MODE?: string;
 	CONTAINER_POOL_SIZE?: string;
 	MAX_CONNECTIONS_PER_CONTAINER?: string;

worker/index.ts

@@ -56,7 +56,10 @@ function buildContainerEnvVars(env: Env): Record<string, string> {
 		ENABLE_OPENAI_CUSTOM_PROVIDER: env.ENABLE_OPENAI_CUSTOM_PROVIDER || 'false',
 		OPENAI_CUSTOM_REQUIRE_WSS: env.OPENAI_CUSTOM_REQUIRE_WSS || 'true',
 		FORCE_COMMIT_TIMEOUT: env.FORCE_COMMIT_TIMEOUT || '2',
-		DEBUG: env.DEBUG || 'true',
+		DEBUG: env.DEBUG || 'false',
+		// If DEBUG=true is explicitly set on the Worker env, default LOG_LEVEL to
+		// 'debug' unless an operator overrides it (e.g. LOG_LEVEL=info).
+		LOG_LEVEL: env.LOG_LEVEL || (env.DEBUG === 'true' ? 'debug' : 'info'),
 		ROUTING_MODE: env.ROUTING_MODE || 'session',
 		CONTAINER_POOL_SIZE: env.CONTAINER_POOL_SIZE || '5',
 		MAX_CONNECTIONS_PER_CONTAINER: env.MAX_CONNECTIONS_PER_CONTAINER || '10',