Files
vocero-client/vocero-frontend/src/realtime/controller.ts
T

630 lines
19 KiB
TypeScript

import { z } from "zod";
import { createAudioContext } from "../audio/engine";
import {
errorMessage,
getConfig,
getProfile,
getReports,
getTools,
post,
reasoningSchema,
searchSchema,
} from "../lib/api";
import { buildInstructions, profileContext, ragContext, speechInstructions, WEB_SEARCH } from "../lib/context";
import {
defaultConfig,
type ConnectionStatus,
type DynamicTool,
type Profile,
type Reports,
type ServerConfig,
type ToolCall,
type Transcript,
} from "../lib/contracts";
import type { Message } from "../lib/messages";
import {
callerFromLocation,
connectionTarget,
GATE_OFF,
loadSettings,
saveSettings,
type Caller,
type Settings,
} from "../lib/settings";
import { RealtimeClient, SessionError } from "./client";
export const LIVE_STATES = new Set<ConnectionStatus>(["listening", "user-speaking", "processing", "ai-speaking"]);
export const ACTIVE_STATES = new Set<ConnectionStatus>([...LIVE_STATES, "connecting", "queued", "your-turn"]);
export interface AppState {
loading: boolean;
configReady: boolean;
configError: string;
contextLoading: boolean;
config: ServerConfig;
settings: Settings;
profile: Profile | null;
reports: Reports | null;
tools: DynamicTool[];
toolsLoading: boolean;
toolsError: string;
status: ConnectionStatus;
error: string;
notice: string;
messages: Message[];
muted: boolean;
rag: string;
queuePosition: number;
joinDeadline: number;
unlock: boolean;
wakeWordEnabled: boolean;
}
/** Observable application state; React owns rendering, this object owns session lifetime. */
export class VoiceController {
readonly caller: Caller;
private state: AppState;
private readonly listeners = new Set<() => void>();
private readonly levelListeners = new Set<() => void>();
private level = 0;
private client: RealtimeClient | null = null;
private subscriptions: (() => void)[] = [];
private requests = new AbortController();
private initialization: AbortController | null = null;
private reasoningRequest: AbortController | null = null;
private reasoningTurn = 0;
private externalSpeaking = false;
private idleTimer: ReturnType<typeof setInterval> | undefined;
private lastActivity = Date.now();
private lastMeter = 0;
private sessionNumber = 0;
private reasoned = new Set<string>();
private autoStarted = false;
constructor(caller = callerFromLocation()) {
this.caller = caller;
this.state = {
loading: true,
configReady: false,
configError: "",
contextLoading: false,
config: defaultConfig,
settings: loadSettings(),
profile: null,
reports: null,
tools: [],
toolsLoading: false,
toolsError: "",
status: "idle",
error: "",
notice: "",
messages: [],
muted: false,
rag: "",
queuePosition: 0,
joinDeadline: 0,
unlock: false,
wakeWordEnabled: false,
};
}
getSnapshot = () => this.state;
getLevel = () => this.level;
subscribeLevel = (listener: () => void) => {
this.levelListeners.add(listener);
return () => {
this.levelListeners.delete(listener);
};
};
private updateLevel(level: number) {
if (this.level === level) {
return;
}
this.level = level;
this.levelListeners.forEach((listener) => {
listener();
});
}
subscribe = (listener: () => void) => {
this.listeners.add(listener);
return () => {
this.listeners.delete(listener);
};
};
private update(patch: Partial<AppState>) {
this.state = { ...this.state, ...patch };
this.listeners.forEach((listener) => {
listener();
});
}
async initialize(parent?: AbortSignal) {
this.initialization?.abort();
const initialization = new AbortController();
this.initialization = initialization;
const { signal } = initialization;
const cancel = () => {
initialization.abort();
};
if (parent?.aborted) {
cancel();
return;
}
parent?.addEventListener("abort", cancel, { once: true });
this.update({ loading: true, configReady: false, configError: "", contextLoading: !!this.caller.phoneNumber });
const configuration = getConfig(signal);
const profile = getProfile(this.caller.phoneNumber, signal)
.then((profile) => {
if (!signal.aborted) {
this.update({ profile, contextLoading: false });
this.pushSettings();
}
})
.catch(() => undefined);
const reports = getReports(this.caller.phoneNumber, signal)
.then((reports) => {
if (!signal.aborted) {
this.update({ reports });
}
})
.catch(() => undefined);
const configure = async () => {
try {
const config = await configuration;
if (signal.aborted) {
return;
}
this.update({ config, configReady: true, loading: false });
if (config.apipilot) {
await profile;
if (!signal.aborted) {
await this.refreshTools(signal, !!this.state.profile);
}
}
} catch (error) {
if (!signal.aborted) {
this.update({ loading: false, configError: `No se pudo cargar la configuración. ${errorMessage(error)}` });
}
}
};
try {
await Promise.all([configure(), profile, reports]);
} finally {
parent?.removeEventListener("abort", cancel);
}
}
dispose() {
this.initialization?.abort();
void this.stop();
}
async refreshTools(signal: AbortSignal, refresh = false) {
this.update({ toolsLoading: true, toolsError: "" });
try {
const tools = await getTools(signal, refresh);
if (!signal.aborted) {
this.update({ tools, toolsLoading: false });
this.pushSettings();
}
} catch (error) {
if (!signal.aborted) {
this.update({ toolsError: errorMessage(error), toolsLoading: false });
}
}
}
private instructions() {
return buildInstructions(
this.state.config,
this.state.settings,
this.state.profile,
this.state.rag,
this.state.tools,
);
}
private tools() {
return this.state.settings.webSearch && (this.state.config.search || this.state.settings.searchKey)
? [WEB_SEARCH]
: [];
}
private gate() {
return { enabled: this.state.settings.noiseGate > GATE_OFF, thresholdDb: this.state.settings.noiseGate };
}
private pushSettings() {
this.client?.updateSession(
speechInstructions(this.instructions(), this.state.config),
this.state.settings.voice,
this.tools(),
);
}
changeSettings(patch: Partial<Settings>) {
const changed = Object.fromEntries(
Object.entries(patch).filter(([key, value]) => this.state.settings[key as keyof Settings] !== value),
) as Partial<Settings>;
if (!Object.keys(changed).length) {
return;
}
const settings = { ...this.state.settings, ...changed };
this.update({ settings });
saveSettings(changed);
if (changed.noiseGate !== undefined) {
this.client?.setGate(this.gate());
}
if (
changed.voice !== undefined ||
changed.instructions !== undefined ||
changed.webSearch !== undefined ||
changed.searchKey !== undefined
) {
this.pushSettings();
}
}
setRag(rag: string) {
this.update({ rag: rag.trim() });
this.pushSettings();
}
dismissError() {
this.update({ error: "" });
}
dismissUnlock() {
this.update({ unlock: false });
}
setWakeWordEnabled(wakeWordEnabled: boolean) {
this.update({ wakeWordEnabled });
}
offerAutoStart() {
if (this.autoStarted || this.client || !this.state.configReady || this.state.contextLoading) {
return;
}
this.autoStarted = true;
const mode = this.state.config.startSessionMode.trim().toUpperCase();
if (mode === "TALK") {
this.update({ unlock: true });
} else if (mode === "TALKFIRST") {
void this.start(true);
}
}
async start(automatic = false, preparedContext?: AudioContext, greeting?: string) {
if (this.client || !this.state.configReady || this.state.contextLoading) {
if (preparedContext) {
void preparedContext.close().catch(() => undefined);
}
return;
}
this.update({
unlock: false,
error: "",
notice: "",
status: "connecting",
muted: false,
joinDeadline: 0,
});
let context: AudioContext | undefined;
try {
if (!navigator.mediaDevices?.getUserMedia || typeof AudioContext === "undefined") {
throw new Error("El micrófono necesita HTTPS o localhost y un navegador compatible.");
}
const target = connectionTarget(this.state.config, this.state.settings, this.caller);
context = preparedContext ?? createAudioContext();
if (automatic && context.state !== "running") {
await context.close();
this.update({ status: "idle", unlock: true });
return;
}
this.requests = new AbortController();
this.reasoned.clear();
this.sessionNumber++;
const client = new RealtimeClient({
...target,
context,
voice: this.state.settings.voice,
instructions: speechInstructions(this.instructions(), this.state.config),
tools: this.tools(),
gate: this.gate(),
greeting:
greeting ??
(["TALK", "TALKFIRST"].includes(this.state.config.startSessionMode.toUpperCase())
? "hola vocero"
: undefined),
});
this.client = client;
this.subscriptions = [
client.on("status", (status) => {
if (client !== this.client) {
return;
}
if (status === "closed") {
void this.stop(this.state.error ? "error" : "closed");
return;
}
if (status === "listening" || status === "user-speaking" || status === "ai-speaking") {
this.lastActivity = Date.now();
}
if (status === "user-speaking") {
this.cancelReasoning();
window.speechSynthesis?.cancel();
this.update({
messages: this.state.messages.map((message) =>
message.role === "assistant" && message.partial
? { ...message, partial: false, interrupted: true }
: message,
),
});
}
if (
!["listening", "processing", "ai-speaking"].includes(status) ||
(!this.reasoningRequest && !this.externalSpeaking)
) {
this.update({ status });
}
}),
client.on("transcript", (transcript) => {
this.lastActivity = Date.now();
this.transcript(transcript);
if (transcript.role === "user" && !transcript.partial && !this.reasoned.has(transcript.id)) {
this.reasoned.add(transcript.id);
void this.reason(transcript, client);
}
}),
client.on("finished", (response) => {
if (response.transcript) {
this.transcript(
{ id: response.responseId, role: "assistant", text: response.transcript, partial: false },
response.status !== "completed",
);
}
}),
client.on("level", (level) => {
if (!this.state.muted && level > 0.0025) {
this.lastActivity = Date.now();
}
if (Date.now() - this.lastMeter > 150) {
this.lastMeter = Date.now();
this.updateLevel(level);
}
}),
client.on("queue", ({ position }) => {
this.update({ queuePosition: position });
}),
client.on("readyToJoin", ({ expiresSec }) => {
this.update({ joinDeadline: Date.now() + expiresSec * 1000 });
}),
client.on("grant", () => {
this.update({ joinDeadline: 0 });
}),
client.on("error", (error) => {
this.reportError(error);
}),
client.on("serverError", (error) => {
this.update({ error: error.message });
}),
client.on("tool", (call) => {
void this.runTool(call, client);
}),
];
this.lastActivity = Date.now();
this.idleTimer = setInterval(() => {
if (
this.state.status === "listening" &&
!this.reasoningRequest &&
!this.externalSpeaking &&
Date.now() - this.lastActivity >= 30000
) {
void this.stop();
this.update({ notice: "Sesión finalizada por inactividad. Toca para volver a conversar." });
}
}, 1000);
try {
await client.connect();
} catch (error) {
// A closed event may already have cleared the client; explicit cancellation is silent.
if (!(error instanceof SessionError && error.code === "aborted") && (!this.client || this.client === client)) {
this.reportError(error);
if (automatic) {
this.update({ unlock: true });
}
}
}
} catch (error) {
if (context) {
void context.close().catch(() => undefined);
}
this.reportError(error);
}
}
private reportError(error: unknown) {
this.update({
error: errorMessage(error),
status: "error",
});
}
private transcript(transcript: Transcript, interrupted = false) {
if (!transcript.text.trim()) {
return;
}
const id = `${String(this.sessionNumber)}:${transcript.role}:${transcript.id}`;
const existing = this.state.messages.find((message) => message.id === id);
const message: Message = {
...existing,
id,
role: transcript.role,
text: transcript.text,
partial: transcript.partial,
timestamp: existing?.timestamp ?? new Date().toISOString(),
interrupted: interrupted || existing?.interrupted,
};
this.update({
messages: existing
? this.state.messages.map((item) => (item.id === id ? message : item))
: [...this.state.messages, message],
});
}
private cancelReasoning() {
this.reasoningTurn++;
this.reasoningRequest?.abort();
this.reasoningRequest = null;
this.externalSpeaking = false;
}
private async reason(transcript: Transcript, client: RealtimeClient) {
const { config, profile, rag } = this.state;
if (config.reasoningType === "PROXY" || !transcript.text.trim()) {
return;
}
this.cancelReasoning();
const turn = this.reasoningTurn;
const request = new AbortController();
this.reasoningRequest = request;
this.update({ status: "processing" });
const current = () => this.client === client && turn === this.reasoningTurn && !request.signal.aborted;
window.speechSynthesis?.cancel();
try {
const result = await post(
"/api/reasoning",
reasoningSchema,
{
transcript: transcript.text.trim(),
instructions: this.instructions(),
language: "es-ES",
user_context: [profileContext(profile), ragContext(rag)].filter(Boolean).join("\n\n"),
phoneNumber: this.caller.phoneNumber || null,
episode_id: this.caller.episodeId || null,
},
request.signal,
);
if (!current()) {
return;
}
const answer = result.answer.trim();
if (!answer) {
throw new Error("El servicio de razonamiento no devolvió una respuesta.");
}
client.renderExternalAnswer(answer);
this.transcript({ role: "assistant", id: `reasoning:${transcript.id}`, text: answer, partial: false });
if ("speechSynthesis" in window) {
this.externalSpeaking = true;
const utterance = new SpeechSynthesisUtterance(answer);
utterance.lang = "es-ES";
const voice = window.speechSynthesis.getVoices().find((voice) => voice.lang.startsWith("es"));
if (voice) {
utterance.voice = voice;
}
utterance.onstart = () => {
if (current()) {
this.update({ status: "ai-speaking" });
}
};
utterance.onend = utterance.onerror = () => {
if (current()) {
this.externalSpeaking = false;
this.lastActivity = Date.now();
this.update({ status: "listening" });
}
};
window.speechSynthesis.cancel();
window.speechSynthesis.speak(utterance);
}
} catch (error) {
if (current()) {
this.externalSpeaking = false;
this.update({ error: errorMessage(error) });
}
} finally {
if (turn === this.reasoningTurn) {
this.reasoningRequest = null;
if (!this.externalSpeaking && current()) {
this.lastActivity = Date.now();
this.update({ status: "listening" });
}
}
}
}
private async runTool(call: ToolCall, client: RealtimeClient) {
const id = `${String(this.sessionNumber)}:tool:${call.callId}`;
let args = call.arguments;
let output: string;
try {
const parsed: unknown = JSON.parse(call.arguments || "{}");
args = JSON.stringify(parsed, null, 2);
if (call.name !== "web_search") {
throw new Error(`Herramienta desconocida: ${call.name}`);
}
const query = z.object({ query: z.string().trim().min(1) }).parse(parsed).query;
const result = await post(
"/api/search",
searchSchema,
{
query,
...(!this.state.config.search && this.state.settings.searchKey ? { key: this.state.settings.searchKey } : {}),
},
this.requests.signal,
);
output = [
`Google search result from ${new Date().toISOString().slice(0, 10)}:`,
result.answer ? `Answer: ${result.answer}` : "",
...result.results.map((item) => `- ${item.title}: ${item.snippet} (${item.url})`),
]
.filter(Boolean)
.join("\n");
if (!result.answer && !result.results.length) {
output += "\nNo results found.";
}
} catch (error) {
output = `Tool error: ${errorMessage(error)}`;
}
if (client !== this.client) {
return;
}
this.update({
messages: [
...this.state.messages,
{
id,
role: "tool",
text: output,
toolName: call.name,
toolArguments: args,
partial: false,
timestamp: new Date().toISOString(),
},
],
});
client.sendToolOutput(call.callId, output);
}
join() {
this.client?.join();
this.update({ joinDeadline: 0 });
}
toggleMuted() {
const muted = !this.state.muted;
this.client?.setMuted(muted);
this.update({ muted });
}
async restart() {
// Resume before awaiting teardown so browsers retain the click's audio permission.
const context = createAudioContext();
await this.stop();
await this.start(false, context);
}
async stop(status: ConnectionStatus = "closed") {
const client = this.client;
this.client = null;
this.subscriptions.forEach((unsubscribe) => {
unsubscribe();
});
this.subscriptions = [];
clearInterval(this.idleTimer);
this.requests.abort();
this.cancelReasoning();
window.speechSynthesis?.cancel();
this.updateLevel(0);
this.update({
status,
wakeWordEnabled: false,
muted: false,
joinDeadline: 0,
unlock: false,
messages: this.state.messages.map((message) =>
message.partial ? { ...message, partial: false, interrupted: message.role === "assistant" } : message,
),
});
await client?.close();
}
}