AIvoices/server-cloudflare/models/openai.ts

import { DurableObject } from "cloudflare:workers";
import type { Env } from "../src/types";
import { createOpusPacketizer } from "../src/opus";
import { getFirstMessagePrompt, getSystemPrompt } from "../src/prompt";

const AUDIO_OUTPUT_SAMPLE_RATE = 24_000;
const INPUT_SILENCE_DURATION_MS = 1000;
const INPUT_LEVEL_THRESHOLD = 180;

interface OpenAIChatMessage {
  role: "system" | "user" | "assistant";
  content: string;
}

interface SessionState {
  history: OpenAIChatMessage[];
}

function createAuthMessage() {
  return {
    type: "auth",
    volume_control: 100,
    pitch_factor: 1,
    is_ota: false,
    is_reset: false,
  };
}

function createServerMessage(msg: string, extra: Record<string, unknown> = {}) {
  return JSON.stringify({
    type: "server",
    msg,
    ...extra,
  });
}

function errorMessage(error: unknown): string {
  if (error instanceof Error) {
    return `${error.name}: ${error.message}`;
  }
  return String(error);
}

async function transcribePcm(env: Env, audio: Uint8Array): Promise<string> {
  const response = await env.AI.run("@cf/openai/whisper", {
    audio: [...audio],
  }) as { text?: string };

  return response.text?.trim() || "";
}

async function generateOpenAIReply(
  env: Env,
  transcript: string | null,
  history: OpenAIChatMessage[],
): Promise<string> {
  if (!env.OPENAI_API_KEY?.trim()) {
    throw new Error("OPENAI_API_KEY is missing");
  }

  const messages: OpenAIChatMessage[] = [
    { role: "system", content: getSystemPrompt(env) },
    ...history,
  ];

  if (transcript && transcript.trim().length > 0) {
    messages.push({ role: "user", content: transcript });
  } else {
    messages.push({ role: "user", content: getFirstMessagePrompt(env) });
  }

  const response = await fetch("https://api.openai.com/v1/chat/completions", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${env.OPENAI_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      model: env.ELATO_OPENAI_MODEL || "gpt-4.1-mini",
      messages,
      temperature: 0.7,
    }),
  });

  if (!response.ok) {
    throw new Error(`OpenAI request failed: ${response.status} ${await response.text()}`);
  }

  const data = (await response.json()) as {
    choices?: Array<{ message?: { content?: string } }>;
  };

  return (
    data.choices?.[0]?.message?.content?.trim() ||
    "I heard you, but I do not have a response yet."
  );
}

async function synthesizeSpeech(env: Env, text: string): Promise<Response> {
  if (!env.AI) {
    throw new Error("Cloudflare AI binding is missing");
  }

  return env.AI.run(
    "@cf/deepgram/aura-1",
    {
      text,
      speaker: "asteria",
      encoding: "linear16",
      container: "none",
      sample_rate: AUDIO_OUTPUT_SAMPLE_RATE,
    },
    {
      returnRawResponse: true,
    },
  ) as Promise<Response>;
}

export class ElatoOpenAiVoiceAgent extends DurableObject<Env> {
  private audioBuffer = new Uint8Array(0);
  private isGenerating = false;
  private opusPromise: Promise<Awaited<ReturnType<typeof createOpusPacketizer>>> | null = null;
  private hasStartedConversation = false;
  private sawUserSpeech = false;
  private vadTimer: ReturnType<typeof setTimeout> | null = null;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
  }

  private appendAudio(chunk: Uint8Array) {
    const next = new Uint8Array(this.audioBuffer.length + chunk.length);
    next.set(this.audioBuffer, 0);
    next.set(chunk, this.audioBuffer.length);
    this.audioBuffer = next;
  }

  private calculateAudioLevel(audio: Uint8Array): number {
    if (audio.byteLength < 2) {
      return 0;
    }

    const samples = new Int16Array(
      audio.buffer,
      audio.byteOffset,
      Math.floor(audio.byteLength / 2),
    );

    let sum = 0;
    for (let i = 0; i < samples.length; i += 1) {
      sum += Math.abs(samples[i] ?? 0);
    }

    return Math.round(sum / samples.length);
  }

  private async loadSessionState(): Promise<SessionState> {
    const stored = await this.ctx.storage.get<SessionState>("session_state");
    return stored || { history: [] };
  }

  private async saveSessionState(state: SessionState) {
    await this.ctx.storage.put("session_state", state);
  }

  private resetBufferedAudio() {
    this.audioBuffer = new Uint8Array(0);
  }

  private clearVadTimer() {
    if (this.vadTimer) {
      clearTimeout(this.vadTimer);
      this.vadTimer = null;
    }
  }

  private getOpusPacketizer(websocket: WebSocket) {
    if (!this.opusPromise) {
      this.opusPromise = createOpusPacketizer((packet) => websocket.send(packet));
    }
    return this.opusPromise;
  }

  private failStartup(websocket: WebSocket, stage: string, error: unknown) {
    console.error(`[cloudflare][startup:${stage}] ${errorMessage(error)}`);
    websocket.send(createServerMessage("RESPONSE.ERROR"));
    websocket.close(1011, "startup_failed");
  }

  private scheduleAutoCommit(websocket: WebSocket) {
    if (this.isGenerating || !this.sawUserSpeech) {
      return;
    }

    this.clearVadTimer();
    this.vadTimer = setTimeout(() => {
      void this.ctx.blockConcurrencyWhile(async () => {
        if (this.isGenerating || !this.sawUserSpeech || this.audioBuffer.byteLength === 0) {
          return;
        }

        console.log(
          `[cloudflare][vad] silence detected, auto-committing turn (${this.audioBuffer.byteLength} bytes buffered)`,
        );
        this.isGenerating = true;
        this.sawUserSpeech = false;

        try {
          await this.handleTurn(websocket);
        } catch (error) {
          console.error(`[cloudflare][turn] ${errorMessage(error)}`);
          websocket.send(createServerMessage("RESPONSE.ERROR"));
        } finally {
          this.isGenerating = false;
        }
      });
    }, INPUT_SILENCE_DURATION_MS);
  }

  private async streamAssistantReply(websocket: WebSocket, reply: string) {
    const opus = await this.getOpusPacketizer(websocket);
    opus.reset();
    websocket.send(createServerMessage("RESPONSE.CREATED"));

    const ttsResponse = await synthesizeSpeech(this.env, reply);
    if (!ttsResponse.ok || !ttsResponse.body) {
      console.error(
        `[cloudflare][tts] request failed: ${ttsResponse.status} ${ttsResponse.statusText}`,
      );
      websocket.send(createServerMessage("RESPONSE.ERROR"));
      return;
    }

    const reader = ttsResponse.body.getReader();
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        if (value) {
          opus.push(value);
        }
      }
      opus.flush(true);
      websocket.send(createServerMessage("RESPONSE.COMPLETE", { volume_control: 100 }));
      console.log(`[cloudflare][tts] streamed reply successfully (${reply.length} chars)`);
    } finally {
      reader.releaseLock();
    }
  }

  private async handleTurn(
    websocket: WebSocket,
  ) {
    const pcm = this.audioBuffer;
    this.resetBufferedAudio();

    if (pcm.byteLength === 0) {
      return;
    }

    websocket.send(createServerMessage("AUDIO.COMMITTED"));
    this.clearVadTimer();

    const transcript = await transcribePcm(this.env, pcm);
    if (!transcript) {
      console.error("[cloudflare][stt] empty transcript");
      websocket.send(createServerMessage("RESPONSE.ERROR"));
      return;
    }
    console.log(`[cloudflare][stt] transcript: ${transcript}`);
    /* Add user transcript DB call here */

    const session = await this.loadSessionState();
    const reply = await generateOpenAIReply(this.env, transcript, session.history);
    console.log(`[cloudflare][llm] generated reply (${reply.length} chars)`);
    session.history.push(
      { role: "user", content: transcript },
      { role: "assistant", content: reply },
    );
    await this.saveSessionState(session);
    /* Add AI transcript DB call here */
    await this.streamAssistantReply(websocket, reply);
  }

  private async startInitialTurn(websocket: WebSocket) {
    if (this.hasStartedConversation || this.isGenerating) {
      return;
    }

    this.hasStartedConversation = true;
    this.isGenerating = true;

    try {
      const session = await this.loadSessionState();
      const reply = await generateOpenAIReply(this.env, null, session.history);
      console.log(`[cloudflare][llm] initial reply (${reply.length} chars)`);
      session.history.push({ role: "assistant", content: reply });
      await this.saveSessionState(session);
      /* Add AI transcript DB call here */
      await this.streamAssistantReply(websocket, reply);
    } catch (error) {
      this.failStartup(websocket, "initial_turn", error);
    } finally {
      this.isGenerating = false;
    }
  }

  async fetch(request: Request): Promise<Response> {
    if (request.headers.get("Upgrade") !== "websocket") {
      return new Response("Expected websocket", { status: 426 });
    }

    const pair = new WebSocketPair();
    const [client, server] = Object.values(pair);
    server.accept();

    server.send(JSON.stringify(createAuthMessage()));
    void this.startInitialTurn(server);

    server.addEventListener("message", (event) => {
      void this.ctx.blockConcurrencyWhile(async () => {
        if (typeof event.data !== "string") {
          const chunk = new Uint8Array(event.data as ArrayBuffer);
          this.appendAudio(chunk);

          if (!this.isGenerating) {
            const level = this.calculateAudioLevel(chunk);
            if (level >= INPUT_LEVEL_THRESHOLD) {
              if (!this.sawUserSpeech) {
                console.log(`[cloudflare][vad] speech started (level=${level}, threshold=${INPUT_LEVEL_THRESHOLD})`);
              }
              this.sawUserSpeech = true;
              this.scheduleAutoCommit(server);
            } else if (this.sawUserSpeech) {
              this.scheduleAutoCommit(server);
            }
          }
          return;
        }

        const message = JSON.parse(event.data) as {
          type?: string;
          msg?: string;
        };

        if (message.type !== "instruction") {
          return;
        }

        if (message.msg === "end_of_speech") {
          if (this.isGenerating) {
            return;
          }
          this.isGenerating = true;
          this.sawUserSpeech = false;
          try {
            await this.handleTurn(server);
          } catch (error) {
            console.error(`[cloudflare][turn] ${errorMessage(error)}`);
            server.send(createServerMessage("RESPONSE.ERROR"));
          } finally {
            this.isGenerating = false;
          }
          return;
        }

        if (message.msg === "INTERRUPT") {
          this.isGenerating = false;
          this.sawUserSpeech = false;
          this.clearVadTimer();
          this.resetBufferedAudio();
          server.send(createServerMessage("RESPONSE.COMPLETE", { volume_control: 100 }));
          return;
        }

        if (message.msg === "END_SESSION") {
          server.send(createServerMessage("SESSION.END"));
          server.close(1000, "Session ended");
        }
      });
    });

    server.addEventListener("close", () => {
      this.isGenerating = false;
      this.sawUserSpeech = false;
      this.clearVadTimer();
      this.resetBufferedAudio();
      if (this.opusPromise) {
        void this.opusPromise.then((opus) => opus.close()).catch(() => {});
        this.opusPromise = null;
      }
    });

    return new Response(null, { status: 101, webSocket: client });
  }
}