Build a local Realtime voice agent with read-only tool calls

import express from "express";

const app = express();
// This demo accepts only the local page, including protection against DNS rebinding.
app.use((req, res, next) => {
  const hosts = ["localhost:3000", "127.0.0.1:3000"];
  if (!hosts.includes(req.get("host"))) return res.status(403).send("Local host required");
  if (req.method === "POST" && req.get("origin") !== `http://${req.get("host")}`) {
    return res.status(403).send("Same-origin local page required");
  }
  next();
});
app.use(express.text({ type: "application/sdp", limit: "64kb" }));

app.get("/", (_req, res) => res.sendFile("index.html", { root: process.cwd() }));
app.get("/controller.mjs", (_req, res) => res.sendFile("controller.mjs", { root: process.cwd() }));

app.post("/session", async (req, res) => {
  if (!process.env.OPENAI_API_KEY) return res.status(500).send("Server API key missing");
  if (typeof req.body !== "string" || !req.body.startsWith("v=")) {
    return res.status(400).send("Expected an SDP offer");
  }

  const form = new FormData();
  form.set("sdp", req.body);
  form.set("session", JSON.stringify({
    type: "realtime",
    model: "gpt-realtime-2.1",
    output_modalities: ["audio"],
    audio: { output: { voice: "marin" } },
    instructions: "Answer questions about the fictional demo library only using its read-only tools. State clearly that the data is synthetic software test data, not real venue information. For a lookup that may take noticeable time, give one short spoken progress preamble when useful, then call the tools. A preamble is never a result. Do not invent missing information. If a tool result is an error, say that lookup failed. Respond concisely in speech.",
    tools: [
      { type: "function", name: "get_demo_hours", description: "Read synthetic demo-library opening hours; no external account or network.", parameters: { type: "object", properties: {}, additionalProperties: false } },
      { type: "function", name: "get_demo_address", description: "Read synthetic demo-library address; no external account or network.", parameters: { type: "object", properties: {}, additionalProperties: false } }
    ],
    tool_choice: "auto"
  }));

  try {
    const upstream = await fetch("https://api.openai.com/v1/realtime/calls", {
      method: "POST",
      headers: { Authorization: `Bearer ${process.env.OPENAI_API_KEY}` },
      body: form,
      signal: AbortSignal.timeout(25000)
    });
    const answer = await upstream.text();
    if (!upstream.ok) {
      console.error("Realtime session rejected", upstream.status);
      return res.status(502).send(`OpenAI session request failed (${upstream.status})`);
    }
    res.type("application/sdp").send(answer);
  } catch (error) {
    console.error("Realtime session network failure", error.name);
    res.status(502).send("Could not reach the Realtime API");
  }
});

app.listen(3000, "127.0.0.1", () => console.log("Open http://localhost:3000"));

<!doctype html>
<html lang="en">
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Local Realtime voice and read-only tool check</title>
<h1>Local voice and tool check</h1>
<p>The library data in this demo is synthetic software test data.</p>
<button id="start">Start voice</button>
<button id="stop" disabled>Stop and release microphone</button>
<p id="status" role="status">Idle</p>
<audio id="reply" autoplay controls></audio>
<pre id="events" aria-label="Event status"></pre>
<script type="module">
import { ToolController } from "/controller.mjs";

const start = document.querySelector("#start");
const stop = document.querySelector("#stop");
const status = document.querySelector("#status");
const events = document.querySelector("#events");
const reply = document.querySelector("#reply");
let current;

function note(line) {
  events.textContent = `${line}\n${events.textContent}`.slice(0, 3000);
}

// These values are synthetic test fixtures, never real business records.
async function runDemoTool(call, signal) {
  if (signal.aborted) throw Error("interrupted");
  let args;
  try { args = JSON.parse(call.arguments); } catch { return { ok: false, error: "Invalid arguments" }; }
  if (!args || Array.isArray(args) || typeof args !== "object" || Object.keys(args).length) {
    return { ok: false, error: "Expected an empty object" };
  }
  if (call.name === "get_demo_hours") return { ok: true, value: "Synthetic demo library hours: Monday-Friday 09:00-17:00." };
  if (call.name === "get_demo_address") return { ok: true, value: "Synthetic demo library address: 1 Example Lane. This is not a real venue." };
  return { ok: false, error: "Tool is not allowed" };
}

function closeSession(session = current, message = "Stopped") {
  if (!session || session.closed) return;
  session.closed = true;
  clearTimeout(session.setupTimer);
  clearTimeout(session.sessionTimer);
  session.controller?.close();
  session.abort.abort();
  session.mic?.getTracks().forEach(track => track.stop());
  if (session.pc) session.pc.onconnectionstatechange = null;
  session.dc?.close();
  session.pc?.close();
  reply.pause();
  if (current === session) {
    current = null;
    reply.srcObject = null;
    start.disabled = false;
    stop.disabled = true;
    status.textContent = message;
    note(message);
  }
}

start.onclick = async () => {
  const session = { abort: new AbortController(), closed: false };
  current = session;
  start.disabled = true;
  stop.disabled = false;
  status.textContent = "Requesting microphone";
  events.textContent = "";
  session.setupTimer = setTimeout(() => closeSession(session, "Setup timed out; microphone stopped"), 30000);
  try {
    const mic = await navigator.mediaDevices.getUserMedia({ audio: true });
    if (session.closed) { mic.getTracks().forEach(track => track.stop()); return; }
    session.mic = mic;
    const pc = session.pc = new RTCPeerConnection();
    pc.ontrack = event => {
      if (session.closed) return;
      reply.srcObject = event.streams[0];
      reply.play().catch(() => note("Playback blocked; press Play on the audio control"));
    };
    pc.onconnectionstatechange = () => {
      if (session.closed) return;
      note(`Peer: ${pc.connectionState}`);
      if (pc.connectionState === "connected") {
        clearTimeout(session.setupTimer);
        clearTimeout(session.sessionTimer);
        session.sessionTimer = setTimeout(() => closeSession(session, "Local session limit reached; microphone stopped"), 55 * 60000);
        status.textContent = "Connected; ask about the synthetic demo library";
      }
      if (["failed", "closed", "disconnected"].includes(pc.connectionState)) closeSession(session, "Connection ended; microphone stopped");
    };
    pc.addTrack(mic.getAudioTracks()[0], mic);
    const dc = session.dc = pc.createDataChannel("oai-events");
    const send = event => {
      if (session.closed || dc.readyState !== "open") throw Error("Event channel is not open");
      dc.send(JSON.stringify(event));
    };
    session.controller = new ToolController({
      send, runTool: runDemoTool,
      status: note,
      fatal: message => closeSession(session, `${message}; microphone stopped`)
    });
    dc.onmessage = message => {
      if (session.closed) return;
      let event;
      try { event = JSON.parse(message.data); } catch { closeSession(session, "Invalid event; microphone stopped"); return; }
      if (event.type === "session.created") note("Realtime session created");
      if (event.type === "output_audio_buffer.cleared") note("Server cleared unplayed speech after interruption");
      session.controller.handle(event).catch(() => closeSession(session, "Controller failed; microphone stopped"));
    };
    dc.onclose = () => { if (!session.closed) closeSession(session, "Event channel closed; microphone stopped"); };
    dc.onerror = () => closeSession(session, "Event channel failed; microphone stopped");
    const offer = await pc.createOffer();
    await pc.setLocalDescription(offer);
    if (session.closed) return;
    const response = await fetch("/session", {
      method: "POST", headers: { "Content-Type": "application/sdp" },
      body: offer.sdp, signal: session.abort.signal
    });
    if (!response.ok) throw Error(`Session request failed (${response.status})`);
    const answer = await response.text();
    if (session.closed) return;
    await pc.setRemoteDescription({ type: "answer", sdp: answer });
    if (!session.closed) status.textContent = "Signaling complete; waiting for connection";
  } catch (error) {
    if (!session.closed) closeSession(session, `Start failed: ${error.message}`);
  }
};
stop.onclick = () => closeSession();
window.addEventListener("pagehide", () => closeSession());
</script>
</html>

// The same controller runs in the browser and in the synthetic local checks.
export class ToolController {
  constructor({ send, runTool, status, fatal, setTimer = setTimeout, clearTimer = clearTimeout }) {
    Object.assign(this, { send, runTool, status, fatal, setTimer, clearTimer });
    this.epoch = 0;
    this.rounds = 0;
    this.counter = 0;
    this.responses = new Map();
    this.pending = null;
    this.closed = false;
  }

  abortPending() {
    if (!this.pending) return;
    this.pending.abort.abort();
    if (this.pending.timer) this.clearTimer(this.pending.timer);
    this.pending = null;
  }

  close() {
    this.closed = true;
    this.epoch++;
    this.abortPending();
    this.responses.clear();
  }

  fail(message) {
    if (this.closed) return;
    this.status(message);
    this.close();
    this.fatal(message);
  }

  interrupt() {
    if (this.closed) return;
    this.epoch++;
    this.rounds = 0;
    this.abortPending();
    this.status("Speech detected; interrupted result discarded");
  }

  async handle(event) {
    if (this.closed) return;
    if (event.type === "error") {
      this.fail(`Realtime error: ${event.error?.code || event.error?.message || "unknown"}`);
      return;
    }
    if (event.type === "input_audio_buffer.speech_started") { this.interrupt(); return; }
    if (event.type === "response.created") {
      if (event.response?.id) this.responses.set(event.response.id, this.epoch);
      return;
    }
    if (event.type === "conversation.item.done") {
      const p = this.pending;
      const id = event.item?.id;
      if (!p || p.epoch !== this.epoch || !p.acks.has(id)) return;
      p.acks.delete(id);
      if (p.acks.size === 0) {
        if (p.timer) this.clearTimer(p.timer);
        this.pending = null;
        if (!p.abort.signal.aborted && !this.closed && p.epoch === this.epoch) {
          this.status("All tool outputs acknowledged; resuming response");
          this.send({ type: "response.create" });
        }
      }
      return;
    }
    if (event.type !== "response.done") return;
    const response = event.response;
    const id = response?.id;
    const responseEpoch = this.responses.get(id);
    this.responses.delete(id);
    if (responseEpoch === undefined || responseEpoch !== this.epoch) return;
    this.status(`Response ${id}: ${response.status || "unknown"}`);
    if (response.status !== "completed") return; // Never execute partial/cancelled calls.
    const calls = (response.output || []).filter(item => item.type === "function_call");
    if (!calls.length) return;
    if (this.pending) { this.fail("Overlapping tool batches in one turn"); return; }
    if (calls.length > 8 || ++this.rounds > 3) { this.fail("Tool call or round bound reached"); return; }
    const ids = new Set();
    for (const call of calls) {
      if (call.status !== "completed" || !call.call_id || ids.has(call.call_id)) {
        this.fail("Incomplete or repeated function call ID"); return;
      }
      ids.add(call.call_id);
    }
    const p = { epoch: this.epoch, abort: new AbortController(), acks: new Set(), timer: null };
    this.pending = p;
    this.status(`Executing ${calls.length} read-only local call(s) from ${id}`);
    const outputs = await Promise.all(calls.map(async call => {
      try { return { call, result: await this.runTool(call, p.abort.signal) }; }
      catch { return { call, result: { ok: false, error: "Local lookup failed" } }; }
    }));
    if (this.closed || this.pending !== p || p.epoch !== this.epoch || p.abort.signal.aborted) return;
    try {
      for (const { call, result } of outputs) {
        const itemId = `tool_output_${++this.counter}`;
        p.acks.add(itemId);
        this.send({
          type: "conversation.item.create",
          event_id: `tool_event_${this.counter}`,
          item: { id: itemId, type: "function_call_output", call_id: call.call_id, output: JSON.stringify(result) }
        });
        this.status(`Sent result for ${call.call_id}`);
      }
      p.timer = this.setTimer(() => this.fail("Tool output acknowledgement timed out"), 8000);
    } catch { this.fail("Could not send tool outputs"); }
  }
}

import assert from "node:assert/strict";
import { ToolController } from "./controller.mjs";

const deferred = () => { let resolve, reject; const promise = new Promise((a, b) => { resolve = a; reject = b; }); return { promise, resolve, reject }; };
const tick = () => new Promise(resolve => setImmediate(resolve));
const call = (id, name = "get_demo_hours") => ({ type: "function_call", status: "completed", call_id: id, name, arguments: "{}" });
const created = id => ({ type: "response.created", response: { id } });
const done = (id, status, output = []) => ({ type: "response.done", response: { id, status, output } });

function harness(runTool = async x => ({ ok: true, value: x.name })) {
  const sent = [], statuses = [], failures = [];
  const timers = new Map(); let nextTimer = 0;
  const controller = new ToolController({
    send: e => sent.push(e), runTool,
    status: s => statuses.push(s), fatal: e => failures.push(e),
    setTimer: callback => { const id = ++nextTimer; timers.set(id, callback); return id; },
    clearTimer: id => timers.delete(id)
  });
  return { controller, sent, statuses, failures, timers };
}

// Mixed audio/message output, two asynchronous calls resolved in reverse order.
{
  const a = deferred(), b = deferred();
  const h = harness(x => x.call_id === "call_A" ? a.promise : b.promise);
  await h.controller.handle(created("resp_1"));
  const work = h.controller.handle(done("resp_1", "completed", [
    { type: "message", status: "completed", content: [{ type: "output_audio", transcript: "Checking." }] },
    call("call_A"), call("call_B", "get_demo_address")
  ]));
  b.resolve({ ok: true, value: "address" }); await tick();
  assert.equal(h.sent.length, 0, "first finished lookup must not send or resume early");
  a.resolve({ ok: true, value: "hours" }); await work;
  assert.deepEqual(h.sent.map(x => x.item?.call_id), ["call_A", "call_B"]);
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 0);
  await h.controller.handle({ type: "conversation.item.done", item: { id: h.sent[1].item.id } });
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 0);
  await h.controller.handle({ type: "conversation.item.done", item: { id: h.sent[0].item.id } });
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 1, "one resume after both item acknowledgements");
}

// One local lookup error does not erase the other result.
{
  const h = harness(async x => { if (x.call_id === "bad") throw Error("synthetic failure"); return { ok: true, value: "fixture" }; });
  await h.controller.handle(created("resp_2"));
  await h.controller.handle(done("resp_2", "completed", [call("bad"), call("good")]));
  assert.equal(JSON.parse(h.sent[0].item.output).ok, false);
  assert.equal(JSON.parse(h.sent[1].item.output).ok, true);
  for (const e of h.sent.slice()) await h.controller.handle({ type: "conversation.item.done", item: { id: e.item.id } });
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 1);
}

// Partial/cancelled responses never dispatch a call.
for (const status of ["failed", "incomplete", "cancelled"]) {
  const h = harness();
  await h.controller.handle(created(`resp_${status}`));
  await h.controller.handle(done(`resp_${status}`, status, [call("ignored")]));
  assert.equal(h.sent.length, 0);
}

// Distinct rounds are bounded, and missing/duplicate IDs are rejected.
{
  const h = harness();
  for (let i = 1; i <= 3; i++) {
    await h.controller.handle(created(`round_${i}`));
    await h.controller.handle(done(`round_${i}`, "completed", [call(`call_${i}`)]));
    const last = h.sent.at(-1);
    await h.controller.handle({ type: "conversation.item.done", item: { id: last.item.id } });
  }
  await h.controller.handle(created("round_4"));
  await h.controller.handle(done("round_4", "completed", [call("too_many")]));
  assert.match(h.failures[0], /bound/);
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 3);
}
{
  const h = harness();
  await h.controller.handle(created("dup"));
  await h.controller.handle(done("dup", "completed", [call("same"), call("same")]));
  assert.match(h.failures[0], /repeated/);
}

// User speech or Stop invalidates in-flight lookups and ignores their late results.
for (const mode of ["speech", "stop"]) {
  const d = deferred(), h = harness(() => d.promise);
  await h.controller.handle(created(`resp_${mode}`));
  const work = h.controller.handle(done(`resp_${mode}`, "completed", [call("late")]));
  if (mode === "speech") await h.controller.handle({ type: "input_audio_buffer.speech_started" });
  else h.controller.close();
  d.resolve({ ok: true, value: "late fixture" }); await work;
  assert.equal(h.sent.length, 0);
}

// A late response completion or acknowledgement from the interrupted turn cannot resume it.
{
  const h = harness();
  await h.controller.handle(created("old"));
  await h.controller.handle({ type: "input_audio_buffer.speech_started" });
  await h.controller.handle(done("old", "completed", [call("late_call")]));
  assert.equal(h.sent.length, 0);
  await h.controller.handle(created("new"));
  await h.controller.handle(done("new", "completed", [call("fresh_call")]));
  const output = h.sent[0];
  await h.controller.handle({ type: "input_audio_buffer.speech_started" });
  await h.controller.handle({ type: "conversation.item.done", item: { id: output.item.id } });
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 0);
}

// Missing server acknowledgement closes instead of speculatively resuming.
{
  const h = harness();
  await h.controller.handle(created("no_ack"));
  await h.controller.handle(done("no_ack", "completed", [call("one")]));
  assert.equal(h.timers.size, 1);
  [...h.timers.values()][0]();
  assert.match(h.failures[0], /timed out/);
  assert.equal(h.sent.filter(x => x.type === "response.create").length, 0);
}

console.log("Synthetic controller checks passed: ordering, single resume, error isolation, bounds and interruption.");

import assert from "node:assert/strict";
import fs from "node:fs";
import vm from "node:vm";
import { ToolController } from "./controller.mjs";

const html = fs.readFileSync(new URL("./index.html", import.meta.url), "utf8");
const script = html.match(/<script type="module">([\s\S]*?)<\/script>/)?.[1]
  .replace('import { ToolController } from "/controller.mjs";', "");
assert.ok(script, "inline browser module found");

function fakePage(mediaPromise) {
  const nodes = Object.fromEntries(["#start", "#stop", "#status", "#events", "#reply"].map(k => [k, {
    disabled: k === "#stop", textContent: "", srcObject: null,
    pause() { this.paused = true; }, play() { return Promise.resolve(); }
  }]));
  let fetches = 0, peer;
  class FakePeer {
    constructor() { peer = this; this.connectionState = "new"; }
    addTrack() {}
    createDataChannel() { return this.dc = { readyState: "open", close() { this.closed = true; }, send() {} }; }
    async createOffer() { return { sdp: "v=0" }; }
    async setLocalDescription() {}
    async setRemoteDescription() {}
    close() { this.closed = true; }
  }
  vm.runInNewContext(script, {
    ToolController, document: { querySelector: k => nodes[k] },
    navigator: { mediaDevices: { getUserMedia: () => mediaPromise } },
    RTCPeerConnection: FakePeer,
    fetch: async () => { fetches++; return { ok: true, text: async () => "answer" }; },
    window: { addEventListener() {} },
    AbortController, setTimeout, clearTimeout, console
  });
  return { nodes, peer: () => peer, fetches: () => fetches };
}

// Stop while permission is pending; any late track is released without a session request.
{
  let allowMic;
  const page = fakePage(new Promise(resolve => { allowMic = resolve; }));
  const starting = page.nodes["#start"].onclick();
  page.nodes["#stop"].onclick();
  const track = { stopped: false, stop() { this.stopped = true; } };
  allowMic({ getTracks: () => [track], getAudioTracks: () => [track] });
  await starting;
  assert.equal(track.stopped, true);
  assert.equal(page.fetches(), 0);
  assert.equal(page.nodes["#start"].disabled, false);
}

// Stop after signaling: microphone, data channel, peer and playback all close.
{
  const track = { stopped: false, stop() { this.stopped = true; } };
  const page = fakePage(Promise.resolve({ getTracks: () => [track], getAudioTracks: () => [track] }));
  await page.nodes["#start"].onclick();
  page.nodes["#stop"].onclick();
  assert.equal(page.fetches(), 1);
  assert.equal(track.stopped, true);
  assert.equal(page.peer().closed, true);
  assert.equal(page.peer().dc.closed, true);
  assert.equal(page.nodes["#reply"].paused, true);
  assert.equal(page.nodes["#reply"].srcObject, null);
}

console.log("Synthetic browser lifecycle checks passed: late microphone and full Stop cleanup.");

npm init -y
npm install express
node server.mjs

node --check server.mjs
node --check controller.mjs
node protocol-check.mjs
node browser-check.mjs