talk-controller.js

/**
 * TalkController — orchestrates the live voice loop on /avatars/:id.
 *
 *   user mic ─▶ Web Speech API STT ─▶ /api/chat (SSE)
 *                                          │
 *                                          ▼
 *                          /api/tts/eleven (cloned voice)
 *                          /api/tts/edge   (fallback)
 *                                          │
 *                                          ▼
 *                                 audio element + analyser
 *                                          │
 *                                          ▼
 *                           LipsyncDriver ▶ AvatarMouthTarget
 *
 * Every piece is real:
 *   - Web Speech API mic capture (browser-native, no key)
 *   - /api/chat streams from Anthropic / OpenRouter / etc (existing)
 *   - /api/tts/eleven is the existing R2-cached ElevenLabs proxy
 *   - /api/tts/edge is the existing Microsoft Edge Neural TTS fallback
 *   - Voice ID is read from /api/agents/:agent_id/voice when the avatar is
 *     bound to an agent with a cloned voice; otherwise we use the Edge path
 *
 * The controller takes ownership of an AvatarMouthTarget — it doesn't own the
 * scene that drives the visuals. Tear down by calling stop().
 */

import { LipsyncDriver, tapAudioElement } from './lipsync-driver.js';
import { MicCapture } from './mic-capture.js';
import { log } from '../shared/log.js';

// Riva interim recognition cadence. While holding to talk we fire a recognition
// pass over the audio-so-far on this interval so partial words surface live;
// MAX_INTERIMS caps the round-trips per utterance so a long hold can't drain the
// metered ASR budget. The release always runs one authoritative final pass.
const INTERIM_INTERVAL_MS = 1400;
const MAX_INTERIMS = 4;

const EDGE_VOICES_BY_GENDER = {
	female: 'en-US-AriaNeural',
	male: 'en-US-GuyNeural',
	neutral: 'en-US-AriaNeural',
};

const ELEVEN_DEFAULT_VOICE = 'EXAVITQu4vr4xnSDxMaL'; // Bella — ElevenLabs default voice

export class TalkController {
	/**
	 * @param {object} opts
	 * @param {object} opts.avatar       Avatar record (must include id; optionally agent_id, source_meta)
	 * @param {() => string} [opts.systemPromptFn]  Optional system prompt builder
	 * @param {(msg: { role: 'user'|'assistant', content: string }) => void} [opts.onMessage]
	 *        Hook so the host UI can append a transcript line.
	 * @param {(state: 'idle'|'listening'|'transcribing'|'thinking'|'speaking') => void} [opts.onStateChange]
	 * @param {(partial: string) => void} [opts.onInterim]  Live (non-final) transcript while listening.
	 * @param {(err: Error) => void} [opts.onError]
	 * @param {{ attach: Function, setMouthShape: Function }} opts.mouthTarget
	 */
	constructor({ avatar, systemPromptFn, onMessage, onStateChange, onInterim, onError, mouthTarget, commandInterceptor }) {
		if (!avatar?.id) throw new Error('TalkController: avatar.id required');
		if (!mouthTarget) throw new Error('TalkController: mouthTarget required');
		this.avatar = avatar;
		this.systemPromptFn = systemPromptFn || (() => '');
		this.onMessage = onMessage || (() => {});
		this.onStateChange = onStateChange || (() => {});
		this.onInterim = onInterim || (() => {});
		this.onError = onError || ((e) => log.warn('[talk]', e?.message));
		this.mouthTarget = mouthTarget;
		// Optional async hook: gets first crack at a final transcript. If it returns
		// true, the utterance was handled out-of-band (e.g. a wallet command) and the
		// normal chat round-trip is skipped. Used by the Conversational Wallet.
		this.commandInterceptor = commandInterceptor || null;

		this._state = 'idle';
		this._history = [];
		this._recognizer = null;
		this._audioCtx = null;
		this._currentAudioEl = null;
		this._currentTap = null;
		this._driver = null;
		this._voicePromise = null; // resolves to { provider, voiceId } | null

		// Speech-to-text routing. 'riva' = server-side NVIDIA Riva (cross-browser),
		// 'browser' = window.SpeechRecognition (Chrome/Edge/Safari), 'none' = text
		// only. Resolved once by prepare(); language tracks the avatar's locale.
		this._sttMode = null;
		this._sttModePromise = null;
		this._listenMode = null; // mode of the in-flight turn
		this._mic = null;
		this._interimTimer = null;
		this._interimBusy = false;
		this._interimCount = 0;
		this.language = 'en-US';
	}

	get state() {
		return this._state;
	}

	/** Resolved STT path: 'riva' | 'browser' | 'none' (null until prepare()). */
	get sttMode() {
		return this._sttMode;
	}

	/** Live mic level (0..1) while the Riva lane is capturing; 0 otherwise. */
	get micLevel() {
		return this._mic ? this._mic.getLevel() : 0;
	}

	/**
	 * Decide the STT path once, before the first turn. Probes /api/asr for the
	 * free NVIDIA Riva lane (works in every browser, including Firefox); falls
	 * back to the browser's own SpeechRecognition, then to text-only. Safe to
	 * call repeatedly — the probe runs at most once. Returns the resolved mode.
	 */
	async prepare() {
		if (this._sttMode) return this._sttMode;
		if (!this._sttModePromise) {
			this._sttModePromise = (async () => {
				const hasBrowserSR = !!(window.SpeechRecognition || window.webkitSpeechRecognition);
				const canCapture = MicCapture.isSupported();
				if (canCapture) {
					try {
						const r = await fetch('/api/asr', { headers: { accept: 'application/json' } });
						if (r.ok) {
							const j = await r.json();
							if (j?.configured) return (this._sttMode = 'riva');
						}
					} catch {
						// Probe failure is not fatal — fall back to whatever the browser offers.
					}
				}
				return (this._sttMode = hasBrowserSR ? 'browser' : 'none');
			})();
		}
		return this._sttModePromise;
	}

	/**
	 * Begin a single push-to-talk turn. Returns immediately. The recognized speech
	 * triggers the chat call when stopListening() lands a final transcript (Riva)
	 * or on the recognizer's `end` event (browser). Routes by the mode resolved in
	 * prepare(), with a synchronous fallback if prepare() hasn't settled yet.
	 */
	startListening() {
		if (this._state !== 'idle') return false;

		const mode = this._sttMode || this._fallbackSttMode();
		if (mode === 'riva') {
			this._listenMode = 'riva';
			this._startRivaListening();
			return true;
		}
		if (mode === 'browser') {
			this._listenMode = 'browser';
			return this._startBrowserListening();
		}
		this._listenMode = null;
		this.onError(coded('Voice input isn’t available here — type your message instead.', 'stt-unavailable'));
		return false;
	}

	/** Stop an in-flight recognition. State transitions to idle (or transcribing). */
	stopListening() {
		if (this._listenMode === 'riva') {
			this._stopRivaListening().catch((err) => {
				this._setState('idle');
				this.onError(err);
			});
			return;
		}
		if (this._recognizer) {
			try {
				this._recognizer.stop();
			} catch {}
		}
	}

	// Best mode to use before prepare() has resolved. Prefer the browser's own
	// recognizer (zero setup, instant) when present; otherwise attempt Riva if the
	// environment can capture audio at all.
	_fallbackSttMode() {
		if (window.SpeechRecognition || window.webkitSpeechRecognition) return 'browser';
		return MicCapture.isSupported() ? 'riva' : 'none';
	}

	// ── Browser SpeechRecognition path (Chrome/Edge/Safari) ────────────────
	_startBrowserListening() {
		const RecCls = window.SpeechRecognition || window.webkitSpeechRecognition;
		if (!RecCls) {
			this.onError(coded('Your browser does not support speech input. Try Chrome, Edge, or Safari.', 'stt-unavailable'));
			return false;
		}

		const rec = new RecCls();
		rec.lang = this.language;
		rec.continuous = false;
		rec.interimResults = true;
		rec.maxAlternatives = 1;
		this._recognizer = rec;

		let finalText = '';
		rec.onresult = (e) => {
			let interim = '';
			for (let i = e.resultIndex; i < e.results.length; i++) {
				const res = e.results[i];
				if (res.isFinal) finalText += res[0].transcript;
				else interim += res[0].transcript;
			}
			if (interim) this.onInterim(interim);
		};
		rec.onerror = (e) => {
			const kind = e.error || 'unknown';
			// 'no-speech'/'aborted' are benign end-of-hold outcomes, not failures.
			if (kind !== 'no-speech' && kind !== 'aborted') {
				this.onError(coded(`Speech recognition error: ${kind}`, kind === 'not-allowed' ? 'permission-denied' : 'stt-failed'));
			}
		};
		rec.onend = () => {
			this._recognizer = null;
			this.onInterim('');
			const transcript = finalText.trim();
			if (!transcript) {
				this._setState('idle');
				return;
			}
			this._handleTranscript(transcript).catch((err) => this.onError(err));
		};

		try {
			rec.start();
			this._setState('listening');
			return true;
		} catch (err) {
			this.onError(coded(`Could not start mic: ${err.message}`, 'capture-failed'));
			this._setState('idle');
			return false;
		}
	}

	// ── NVIDIA Riva path (cross-browser, server-side) ──────────────────────
	_startRivaListening() {
		const mic = new MicCapture();
		this._mic = mic;
		this._setState('listening'); // optimistic — the getUserMedia prompt is showing
		mic.start().then(
			() => {
				// stopListening() may have run during the permission prompt.
				if (this._mic !== mic) {
					mic.dispose();
					return;
				}
				this._interimCount = 0;
				this._interimBusy = false;
				this._interimTimer = setInterval(() => this._fireInterim(), INTERIM_INTERVAL_MS);
			},
			(err) => {
				if (this._mic === mic) this._mic = null;
				mic.dispose();
				this._setState('idle');
				this.onError(err); // .code: permission-denied | no-mic | unsupported | capture-failed
			},
		);
	}

	async _stopRivaListening() {
		this._clearInterim();
		const mic = this._mic;
		if (!mic) return;
		this._mic = null;

		// Nothing captured (released before the mic opened) — quietly reset.
		if (!mic.capturing) {
			mic.dispose();
			this._setState('idle');
			return;
		}

		this._setState('transcribing');
		let wav = null;
		try {
			wav = await mic.stop();
		} catch {
			// fall through — treated as no audio below
		} finally {
			mic.dispose();
		}
		this.onInterim('');

		if (!wav) {
			this._setState('idle');
			return;
		}

		let transcript = '';
		try {
			transcript = (await this._recognize(wav)).trim();
		} catch (err) {
			this._setState('idle');
			this.onError(err);
			return;
		}

		if (!transcript) {
			this._setState('idle');
			this.onError(coded('No speech detected — hold the button and speak, or type your message.', 'no-speech'));
			return;
		}

		this._handleTranscript(transcript).catch((err) => this.onError(err));
	}

	// Fire one interim recognition over the audio captured so far so partial words
	// surface live. Strictly best-effort: a failed or rate-limited interim is
	// swallowed; the authoritative transcript comes from the final pass on release.
	async _fireInterim() {
		if (!this._mic || this._interimBusy || this._interimCount >= MAX_INTERIMS) return;
		const snapshot = this._mic.snapshotWav();
		if (!snapshot) return;
		this._interimBusy = true;
		this._interimCount += 1;
		try {
			const text = await this._recognize(snapshot);
			if (this._mic && text) this.onInterim(text); // still listening
		} catch {
			// interim is optional — ignore
		} finally {
			this._interimBusy = false;
		}
	}

	_clearInterim() {
		if (this._interimTimer) {
			clearInterval(this._interimTimer);
			this._interimTimer = null;
		}
		this._interimBusy = false;
	}

	// POST a WAV clip to the free Riva ASR lane and return the transcript text.
	async _recognize(wavBlob) {
		const r = await fetch(`/api/asr?language=${encodeURIComponent(this.language)}`, {
			method: 'POST',
			headers: { 'content-type': 'audio/wav' },
			credentials: 'include',
			body: wavBlob,
		});
		if (!r.ok) {
			const j = await r.json().catch(() => ({}));
			const err = coded(
				j.error_description || j.error || `Speech recognition failed (${r.status})`,
				j.error || (r.status === 429 ? 'rate_limited' : 'asr_failed'),
			);
			err.status = r.status;
			throw err;
		}
		const j = await r.json();
		return j.text || '';
	}

	/**
	 * Force a turn from text (e.g. typed message). Same downstream path as
	 * speech input — chat → TTS → lipsync.
	 */
	async say(text) {
		const trimmed = String(text || '').trim();
		if (!trimmed) return;
		await this._handleTranscript(trimmed);
	}

	/** Stop everything immediately and detach. Idempotent. */
	stop() {
		this._clearInterim();
		if (this._mic) {
			this._mic.dispose();
			this._mic = null;
		}
		this.stopListening();
		this._stopPlayback();
		this._driver?.dispose();
		this._driver = null;
		this._setState('idle');
	}

	/**
	 * Invalidate the cached voice lookup so the next turn re-checks the agent
	 * for a (possibly newly cloned) voice_id. Call after the user finishes a
	 * voice-clone flow inside the overlay.
	 */
	refreshVoice() {
		this._voicePromise = null;
	}

	/** Recent conversation turns (for grounding an out-of-band command parse). */
	get history() {
		return this._history.slice();
	}

	/**
	 * Voice a line through the avatar WITHOUT a chat round-trip — same TTS + lipsync
	 * path as a reply. Used by the Conversational Wallet to speak read-backs,
	 * clarifying questions, and confirmations in character.
	 */
	async speakText(text) {
		const trimmed = String(text || '').trim();
		if (!trimmed) return;
		try {
			await this._speak(trimmed);
		} catch (err) {
			this.onError(err);
		}
	}

	// ── pipeline ─────────────────────────────────────────────────────────

	async _handleTranscript(transcript) {
		this.onMessage({ role: 'user', content: transcript });
		this._history.push({ role: 'user', content: transcript });

		// Wallet commands (and any other registered interceptor) get first crack at
		// the utterance. A handled command never reaches the chat model — the
		// interceptor owns the read-back, confirm, and execution path.
		if (this.commandInterceptor) {
			this._setState('thinking');
			let handled = false;
			try {
				handled = await this.commandInterceptor(transcript);
			} catch (err) {
				this.onError(err);
			}
			if (handled) {
				this._setState('idle');
				return;
			}
		}

		this._setState('thinking');

		let replyText = '';
		try {
			replyText = await this._streamChat(transcript);
		} catch (err) {
			this.onError(err);
			this._setState('idle');
			return;
		}

		if (!replyText) {
			this._setState('idle');
			return;
		}

		this._history.push({ role: 'assistant', content: replyText });
		this.onMessage({ role: 'assistant', content: replyText });

		try {
			await this._speak(replyText);
		} catch (err) {
			this.onError(err);
		}
	}

	async _streamChat(message) {
		const isUuid =
			typeof this.avatar.id === 'string' &&
			/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(this.avatar.id);

		const r = await fetch('/api/chat', {
			method: 'POST',
			headers: { 'content-type': 'application/json' },
			credentials: 'include',
			body: JSON.stringify({
				message,
				system_prompt: this.systemPromptFn(),
				history: this._history.slice(-10, -1),
				...(isUuid ? { agentId: this.avatar.id } : {}),
				...(this.avatar.agent_id ? { agentId: this.avatar.agent_id } : {}),
			}),
		});
		if (!r.ok) {
			const j = await r.json().catch(() => ({}));
			throw new Error(j.error_description || j.error || `Chat failed (${r.status})`);
		}
		const reader = r.body.getReader();
		const decoder = new TextDecoder();
		let buf = '';
		let acc = '';
		while (true) {
			const { value, done } = await reader.read();
			if (done) break;
			buf += decoder.decode(value, { stream: true });
			const blocks = buf.split('\n\n');
			buf = blocks.pop() || '';
			for (const block of blocks) {
				const dataLine = block.split('\n').find((l) => l.startsWith('data:'));
				if (!dataLine) continue;
				const payload = dataLine.slice(5).trim();
				if (!payload) continue;
				let evt;
				try {
					evt = JSON.parse(payload);
				} catch {
					continue;
				}
				if (evt.type === 'chunk' && evt.text) acc += evt.text;
				else if (evt.type === 'error') throw new Error(evt.message || evt.error || 'Stream error');
			}
		}
		return acc.trim();
	}

	async _resolveVoice() {
		if (this._voicePromise) return this._voicePromise;
		const agentId = this.avatar.agent_id;
		if (!agentId) {
			this._voicePromise = Promise.resolve(null);
			return this._voicePromise;
		}
		this._voicePromise = (async () => {
			try {
				const r = await fetch(`/api/agents/${encodeURIComponent(agentId)}/voice`, {
					credentials: 'include',
				});
				if (!r.ok) return null;
				const j = await r.json();
				if (j.voice_provider === 'elevenlabs' && j.voice_id) {
					return { provider: 'elevenlabs', voiceId: j.voice_id };
				}
				return null;
			} catch {
				return null;
			}
		})();
		return this._voicePromise;
	}

	async _speak(text) {
		// Stop any in-flight playback first so consecutive turns don't overlap.
		this._stopPlayback();

		const voice = await this._resolveVoice();
		const blob = voice
			? await this._fetchTtsEleven(text, voice.voiceId)
			: await this._fetchTtsEdge(text);

		const url = URL.createObjectURL(blob);
		const audio = new Audio();
		audio.crossOrigin = 'anonymous';
		audio.src = url;
		this._currentAudioEl = audio;

		// Build the audio graph so the analyser can read what's about to play.
		// MediaElementSource can only be created once per element — we tear it
		// down on `ended` to free the slot for the next turn.
		if (!this._audioCtx) {
			this._audioCtx = new (window.AudioContext || window.webkitAudioContext)();
		}
		if (this._audioCtx.state === 'suspended') {
			await this._audioCtx.resume().catch(() => {});
		}
		this._currentTap = tapAudioElement(audio, this._audioCtx);

		// Drive the lipsync.
		this._driver?.dispose();
		this._driver = new LipsyncDriver({
			analyser: this._currentTap.analyser,
			target: this.mouthTarget,
		});

		this._setState('speaking');

		const cleanup = () => {
			URL.revokeObjectURL(url);
			this._driver?.stop();
			this._currentTap?.disconnect();
			this._currentTap = null;
			this._currentAudioEl = null;
			this._setState('idle');
		};
		audio.onended = cleanup;
		audio.onerror = () => {
			cleanup();
			this.onError(new Error('Audio playback failed'));
		};

		this._driver.start();
		try {
			await audio.play();
		} catch (err) {
			cleanup();
			throw err;
		}
	}

	_stopPlayback() {
		if (this._currentAudioEl) {
			try {
				this._currentAudioEl.pause();
			} catch {}
			this._currentAudioEl = null;
		}
		if (this._currentTap) {
			this._currentTap.disconnect();
			this._currentTap = null;
		}
		this._driver?.stop();
	}

	async _fetchTtsEleven(text, voiceId) {
		const r = await fetch('/api/tts/eleven', {
			method: 'POST',
			headers: { 'content-type': 'application/json' },
			credentials: 'include',
			body: JSON.stringify({
				voiceId: voiceId || ELEVEN_DEFAULT_VOICE,
				text: text.slice(0, 500),
			}),
		});
		if (!r.ok) {
			// Fall back to Edge so the talk loop still completes if ElevenLabs
			// is rate-limited or down.
			log.warn('[talk] eleven TTS failed, falling back to edge');
			return this._fetchTtsEdge(text);
		}
		return r.blob();
	}

	async _fetchTtsEdge(text) {
		const gender =
			this.avatar?.source_meta?.gender ||
			this.avatar?.source_meta?.bodyType ||
			'neutral';
		const voice = EDGE_VOICES_BY_GENDER[gender] || EDGE_VOICES_BY_GENDER.neutral;
		const r = await fetch('/api/tts/edge', {
			method: 'POST',
			headers: { 'content-type': 'application/json' },
			credentials: 'include',
			body: JSON.stringify({ voice, text: text.slice(0, 1500) }),
		});
		if (!r.ok) throw new Error(`TTS failed (${r.status})`);
		return r.blob();
	}

	_setState(state) {
		if (this._state === state) return;
		this._state = state;
		this.onStateChange(state);
	}
}

// Error carrying a machine-readable `.code` so the host UI can route each
// failure precisely (mic denial → text input, rate limit → "try later", etc).
function coded(message, code) {
	const err = new Error(message);
	err.code = code;
	return err;
}