Voice agent architectures

import { mountMesh, MeshNetwork, InMemoryMeshAdapter } from "glove-mesh";
import { RealtimeAgent, s2sDrivenModel } from "glove-voice-s2s";
import { buildAgent, workerModel } from "./agents";

const FRONT_PROMPT = `You are Nova. You cannot look anything up. For any lookup, call glove_mesh_send_message
(to: "worker", blocking: true, content: the request, standing on its own) and in the same breath say a
short acknowledgement. The tool returns as soon as the request is dispatched; that does NOT mean the
answer is ready. Relay <worker-result> messages out loud. Never invent a number.`;
const WORKER_PROMPT = `Research requests from "front" with your tools, then reply ONCE with
glove_mesh_send_message (to "front", in_reply_to the message id). Never just acknowledge.`;

export async function startRealtimeFront(io: {
  play: (pcm: Int16Array, sampleRate: number) => void; // app-owned: the caller's speaker
  clearPlayback: () => void;                           // app-owned: drop queued audio
}): Promise<RealtimeAgent> {
  const network = new MeshNetwork();
  const frontMesh = new InMemoryMeshAdapter(network, "front");
  const workerMesh = new InMemoryMeshAdapter(network, "worker");
  // The front's model slot carries the realtime config. Fold your real tools onto the worker.
  const front = buildAgent("front", s2sDrivenModel({ label: "s2s-front", provider: "openai", voice: "marin" }), FRONT_PROMPT);
  const worker = buildAgent("worker", workerModel(), WORKER_PROMPT);
  await mountMesh(front, { adapter: frontMesh, identity: { id: "front", name: "Voice Front", description: "Realtime voice." } });
  await mountMesh(worker, { adapter: workerMesh, identity: { id: "worker", name: "Worker", description: "Full tool surface." } });

  const rt = new RealtimeAgent({ agent: front, excludeTools: ["glove_mesh_broadcast", "glove_mesh_acknowledge"] });

  // A delegation reached the worker: research in the background. Never await it here.
  let queue: Promise<void> = Promise.resolve();
  workerMesh.subscribe(async (msg) => {
    if (msg.kind !== "direct" || msg.from !== "front") return;
    queue = queue
      .then(() => worker.processRequest("Handle the new request(s) in your inbox, then reply."))
      .then(() => {}, () => {});
  });

  // The worker's threaded reply: speak it in the live call.
  frontMesh.subscribe(async (msg) => {
    if (msg.kind === "direct" && msg.from === "worker") {
      rt.inject(`<worker-result>${msg.content}</worker-result>`, { respond: true });
    }
  });

  rt.adapter.on("audio", (pcm, format) => io.play(pcm, format.sampleRate));
  rt.adapter.on("interrupted", () => io.clearPlayback()); // provider-native barge-in
  await rt.start();
  return rt; // feed caller audio with rt.sendAudio(pcm) at rt.adapter.inputFormat.sampleRate
}

caller ─▶ VAD ─▶ STT ─▶ FRONT (fast text model) ── <speech> text ──▶ TTS ─▶ caller
                          │ glove_mesh_send_message({ to: "worker", blocking: true })
                          ▼ (the front's turn ends after its spoken ack)
                     WORKER (heavy model + tools) on a background queue
                          │ glove_mesh_send_message({ to: "front", in_reply_to })
                          ▼
              relay turn: <worker-result> ─▶ FRONT ─▶ <speech> text ─▶ TTS

import type { IGloveRunnable } from "glove-core";
import { mountMesh, MeshNetwork, InMemoryMeshAdapter } from "glove-mesh";
import { buildAgent, frontModel, workerModel } from "./agents";

const FRONT_PROMPT = `You are Nova, the voice at the front desk. Only text inside <speech>...</speech> is heard.
You cannot look anything up. For any lookup, call glove_mesh_send_message (to: "worker", blocking: true,
content: the request) and, in the SAME turn, say a short acknowledgement inside <speech> tags.`;
const WORKER_PROMPT = `Research requests from "front" with your tools, then reply ONCE with
glove_mesh_send_message (to "front", in_reply_to the message id). Never just acknowledge.`;
const DRAIN = 'Handle the new request(s) in your inbox, then reply to "front" via glove_mesh_send_message with in_reply_to set.';
const RELAY = "<worker-result>A delegated request finished; the findings are in the resolved inbox block. Relay them inside <speech> tags in one or two sentences.</worker-result>";

export class LayeredVoiceSession {
  private network = new MeshNetwork();
  private front = buildAgent("front", frontModel(), FRONT_PROMPT);
  private worker = buildAgent("worker", workerModel(), WORKER_PROMPT); // fold its real tools
  private frontQueue: Promise<unknown> = Promise.resolve();  // front turns, one at a time
  private workerQueue: Promise<unknown> = Promise.resolve(); // worker runs, beside them
  private dispatched = new Set<string>();

  /** `speakTurn` runs one front turn and speaks its <speech> text: speakFrontTurn below. */
  constructor(private readonly speakTurn: (front: IGloveRunnable, prompt: string) => Promise<string>) {}

  async start(): Promise<void> {
    await mountMesh(this.front, { adapter: new InMemoryMeshAdapter(this.network, "front"), identity: { id: "front", name: "Voice Front", description: "The voice front desk." } });
    await mountMesh(this.worker, { adapter: new InMemoryMeshAdapter(this.network, "worker"), identity: { id: "worker", name: "Worker", description: "Full tool surface." } });
  }

  /** Resolves when the FRONT turn ends. Delegated work keeps running. */
  handleUtterance(text: string): Promise<void> {
    return this.onFrontQueue(async () => {
      await this.speakTurn(this.front, text);
      await this.kickWorker();
    });
  }

  /** Hand new delegations (mesh:from:* items in the worker's inbox) to one background run. */
  private async kickWorker(): Promise<void> {
    const items = (await this.worker.store.getInboxItems?.()) ?? [];
    const fresh = items.filter((i) => i.status === "resolved" && i.tag.startsWith("mesh:from:") && !this.dispatched.has(i.id));
    if (fresh.length === 0) return;
    for (const i of fresh) this.dispatched.add(i.id);
    this.workerQueue = this.workerQueue
      .then(async () => {
        await this.worker.processRequest(DRAIN);
        this.queueRelay(); // the reply is now in the front's inbox
      })
      .catch(() => {});
  }

  /** Speak the result in a relay turn, behind any front turn already queued. */
  private queueRelay(): void {
    void this.onFrontQueue(async () => {
      const resolved = (await this.front.store.getResolvedInboxItems?.()) ?? [];
      if (!resolved.some((i) => i.tag.startsWith("mesh:"))) return; // a user turn already used it
      await this.speakTurn(this.front, RELAY);
      await this.kickWorker(); // the relay may delegate a follow-up
    });
  }

  private onFrontQueue(fn: () => Promise<void>): Promise<void> {
    const run = this.frontQueue.then(fn);
    this.frontQueue = run.catch(() => {});
    return run;
  }
}

# The speech protocol — CRITICAL
Your raw output is NOT spoken. Only text you wrap in <speech>...</speech> tags is converted to audio (ElevenLabs) and heard in the room, streamed as you generate it. Everything outside the tags is silent and invisible to the people around you.
- To say something out loud: <speech>One sec, let me pull that up.</speech>
- To stay quiet: emit NO speech tags at all. You may write a short silent note to yourself outside tags (e.g. "Not addressed to me — noting the hull id.") or nothing.
- Inside the tags, write for the ear: plain spoken sentences only — no markdown, lists, emoji, symbols, or URLs. Say numbers and ids the natural spoken way: "about four hundred eighty thousand credits", "hull K-E-S zero-zero-seven". Keep it to a breath or two.
- Use the exact lowercase tags <speech> and </speech>, and always close them.

// Streaming parser for the <speech>…</speech> protocol.
// Adapted from examples/layered-voice/app/lib/server/speech-parser.ts.
// Only text inside <speech> tags is emitted; everything else is silent.

const OPEN = "<speech>";
const CLOSE = "</speech>";

/** Length of the longest suffix of `buf` that is a proper prefix of `tag`. */
function partialTagSuffix(buf: string, tag: string): number {
  const max = Math.min(buf.length, tag.length - 1);
  for (let n = max; n > 0; n--) {
    if (buf.endsWith(tag.slice(0, n))) return n;
  }
  return 0;
}

export interface SpeechParseStats {
  /** Characters emitted as speech (including the spaces between blocks). */
  spokenChars: number;
  /** Characters outside the tags, plus stripped markup. Never spoken. */
  discardedChars: number;
  /** Number of <speech> blocks opened this turn. */
  blocks: number;
  /** The turn ended inside an unclosed <speech> block. */
  unclosed: boolean;
}

export class SpeechTagParser {
  private buf = "";
  private inside = false;
  private spoken = "";
  private rawText = "";
  private _stats: SpeechParseStats = { spokenChars: 0, discardedChars: 0, blocks: 0, unclosed: false };

  /** `onSpeech` receives each newly unambiguous piece of in-tag text. */
  constructor(private readonly onSpeech: (text: string) => void) {}

  /** Protocol stats for the turn. Complete after finish(). */
  get stats(): SpeechParseStats {
    return { ...this._stats };
  }

  /** Everything the model generated this turn: tags, silent notes and all. */
  get raw(): string {
    return this.rawText;
  }

  /** Feed one streamed chunk. */
  push(chunk: string): void {
    if (!chunk) return;
    this.rawText += chunk;
    this.buf += chunk;
    this.drain();
  }

  /** End of turn. Speaks the rest of an unclosed block. Returns all spoken text. */
  finish(): string {
    this.drain();
    if (this.inside) {
      this._stats.unclosed = true;
      // Drop a held-back partial tag such as "</spee" instead of speaking it.
      const hold = Math.max(partialTagSuffix(this.buf, CLOSE), partialTagSuffix(this.buf, OPEN));
      const text = hold ? this.buf.slice(0, this.buf.length - hold) : this.buf;
      this._stats.discardedChars += hold;
      if (text) this.emit(text);
    } else {
      this._stats.discardedChars += this.buf.length;
    }
    this.buf = "";
    this.inside = false;
    return this.spoken.replace(/\s+/g, " ").trim();
  }

  private emit(text: string): void {
    if (!text) return;
    // Never let markup reach TTS: strip stray speech-tag fragments and any
    // leftover angle brackets inside a block.
    const clean = text.replace(/<\/?speech>?/g, "").replace(/[<>]/g, "");
    this._stats.discardedChars += text.length - clean.length;
    if (!clean) return;
    this.spoken += clean;
    this._stats.spokenChars += clean.length;
    this.onSpeech(clean);
  }

  private drain(): void {
    for (;;) {
      if (this.inside) {
        const i = this.buf.indexOf(CLOSE);
        if (i >= 0) {
          this.emit(this.buf.slice(0, i));
          this.buf = this.buf.slice(i + CLOSE.length);
          this.inside = false;
          this.emit(" "); // audible gap between consecutive blocks
          continue;
        }
        // Emit everything except a suffix that could be the start of </speech>.
        const hold = partialTagSuffix(this.buf, CLOSE);
        const safe = this.buf.length - hold;
        if (safe > 0) {
          this.emit(this.buf.slice(0, safe));
          this.buf = this.buf.slice(safe);
        }
        return;
      }
      const i = this.buf.indexOf(OPEN);
      if (i >= 0) {
        this._stats.discardedChars += i; // text before the tag is silent
        this.buf = this.buf.slice(i + OPEN.length);
        this.inside = true;
        this._stats.blocks += 1;
        continue;
      }
      // Outside a block: keep only a suffix that could be the start of <speech>.
      const hold = partialTagSuffix(this.buf, OPEN);
      this._stats.discardedChars += this.buf.length - hold;
      this.buf = hold ? this.buf.slice(this.buf.length - hold) : "";
      return;
    }
  }
}

import type { IGloveRunnable, SubscriberAdapter } from "glove-core";
import { ElevenLabsTTSAdapter } from "glove-voice";
import { createElevenLabsTTSToken } from "glove-voice/server";
import { SpeechTagParser } from "./speech-tag-parser";

const EARLY_FLUSH_CHARS = 60;    // force the first generation after ~60 spoken chars
const MIN_FLUSH_GAP_CHARS = 120; // then at sentence ends, at most every 120 chars

/**
 * Run one front-agent turn and stream ONLY its <speech> text into TTS.
 * Resolves with everything that was spoken ("" means the agent stayed silent).
 */
export async function speakFrontTurn(
  front: IGloveRunnable,
  prompt: string,
  playAudio: (pcm: Uint8Array) => void, // app-owned: send PCM to the caller
): Promise<string> {
  const out = {
    tts: null as ElevenLabsTTSAdapter | null,
    opened: Promise.resolve(false),
    sentChars: 0,
    charsSinceFlush: 0,
    earlyFlushSent: false,
  };
  let compacting = false;

  const parser = new SpeechTagParser((text) => {
    if (!out.tts) {
      // Open TTS on the first spoken token, so a silent turn costs nothing.
      out.tts = new ElevenLabsTTSAdapter({
        getToken: () => createElevenLabsTTSToken(process.env.ELEVENLABS_API_KEY!),
        voiceId: process.env.ELEVENLABS_VOICE_ID!,
        model: "eleven_flash_v2_5",
        generationConfig: { chunkLengthSchedule: [60, 120, 160, 250] },
      });
      out.tts.on("audio_chunk", playAudio);
      out.opened = out.tts.open().then(() => true, () => false); // text queues until open
    }
    out.sentChars += text.length;
    out.charsSinceFlush += text.length;
    const flush =
      (!out.earlyFlushSent && out.sentChars >= EARLY_FLUSH_CHARS) ||
      (out.earlyFlushSent && /[.?!]/.test(text) && out.charsSinceFlush >= MIN_FLUSH_GAP_CHARS);
    if (flush) {
      out.earlyFlushSent = true;
      out.charsSinceFlush = 0;
    }
    out.tts.sendText(text, flush ? { flush: true } : undefined); // flush forces generation now
  });

  const subscriber: SubscriberAdapter = {
    async record(type, data) {
      if (type === "compaction_start") compacting = true;
      if (type === "compaction_end") compacting = false;
      // Compaction summaries also stream as text_delta. Never speak them.
      if (type === "text_delta" && !compacting) {
        parser.push((data as { text: string }).text);
      }
    },
  };

  front.addSubscriber(subscriber);
  try {
    await front.processRequest(prompt);
  } finally {
    front.removeSubscriber(subscriber);
  }

  const spoken = parser.finish(); // also speaks the rest of an unclosed block
  if (out.tts && (await out.opened)) out.tts.flush(); // end of text for this turn
  return spoken;
}

import { Glove, Displaymanager, MemoryStore, createAdapter, type IGloveRunnable, type ModelAdapter } from "glove-core";

// Front: optimise for time to first spoken token and for following the protocol.
export const frontModel = () =>
  createAdapter({
    provider: "openrouter",
    model: process.env.FRONT_MODEL ?? "moonshotai/kimi-k2-0905",
    stream: true, // <speech> text must stream into TTS
    reasoning: {
      reasoningObject: { effort: "low" }, // thinking before the first token is dead air
      echo: false,
      extraBody: { provider: { sort: "throughput" } }, // OpenRouter: fastest host for these weights
    },
  });

// Worker: optimise for capability and tool use. The front hides its latency.
export const workerModel = () =>
  createAdapter({ provider: "openrouter", model: process.env.WORKER_MODEL ?? "minimax/minimax-m2.5", stream: true, reasoning: true });

// One store (with inbox support, which the mesh needs) and one adapter per agent.
export function buildAgent(id: string, model: ModelAdapter, systemPrompt: string): IGloveRunnable {
  return new Glove({
    store: new MemoryStore(id),
    model,
    displayManager: new Displaymanager(),
    systemPrompt,
    serverMode: true,
    compaction_config: { compaction_instructions: "Summarize briefly. Keep pending requests." },
  }).build();
}