Spaces:

taqarub
/

node-api

Sleeping

node-api / shared /realtime-voice-protocol.ts

update persistent PCM voice bridge

79f41f5 4 months ago

6.28 kB

export interface BufferedAudioFrame {
  audioSequence: number;
  audio: string;
}

/**
 * Bounded replay buffer. Frames remain here until the server acknowledges
 * them, which makes reconnect replay deterministic and idempotent.
 */
export class AudioReplayBuffer {
  private frames: BufferedAudioFrame[] = [];

  constructor(private readonly capacity: number) {
    if (!Number.isInteger(capacity) || capacity < 1) {
      throw new Error("AudioReplayBuffer capacity must be a positive integer");
    }
  }

  push(frame: BufferedAudioFrame): BufferedAudioFrame | null {
    if (
      this.frames.some(
        (existing) => existing.audioSequence === frame.audioSequence,
      )
    ) {
      return null;
    }

    this.frames.push(frame);
    this.frames.sort((a, b) => a.audioSequence - b.audioSequence);
    return this.frames.length > this.capacity ? this.frames.shift()! : null;
  }

  acknowledge(audioSequence: number): void {
    this.frames = this.frames.filter(
      (frame) => frame.audioSequence > audioSequence,
    );
  }

  unacknowledged(): BufferedAudioFrame[] {
    return [...this.frames];
  }

  firstSequence(): number | null {
    return this.frames[0]?.audioSequence ?? null;
  }

  clear(): void {
    this.frames = [];
  }

  get size(): number {
    return this.frames.length;
  }
}

/**
 * Turns potentially duplicated or reordered frames back into one contiguous
 * stream. WebSocket normally preserves order, but reconnect replay can overlap
 * with frames already accepted by the server.
 */
export class OrderedAudioFrames {
  private pending = new Map<number, BufferedAudioFrame>();

  constructor(
    private nextSequence: number,
    private readonly capacity = 600,
  ) {}

  receive(frame: BufferedAudioFrame): {
    duplicate: boolean;
    ready: BufferedAudioFrame[];
  } {
    if (
      frame.audioSequence < this.nextSequence ||
      this.pending.has(frame.audioSequence)
    ) {
      return { duplicate: true, ready: [] };
    }

    this.pending.set(frame.audioSequence, frame);
    if (this.pending.size > this.capacity) {
      const highest = Math.max(...this.pending.keys());
      this.pending.delete(highest);
    }

    const ready: BufferedAudioFrame[] = [];
    while (this.pending.has(this.nextSequence)) {
      ready.push(this.pending.get(this.nextSequence)!);
      this.pending.delete(this.nextSequence);
      this.nextSequence++;
    }
    return { duplicate: false, ready };
  }

  expectedSequence(): number {
    return this.nextSequence;
  }
}

export interface RealtimeTranscriptEvent {
  sessionId: string;
  utteranceId: string;
  utteranceIndex: number;
  revision: number;
  text: string;
}

export interface TranscriptUpdate {
  accepted: boolean;
  confirmedText: string;
  interimText: string;
  appendedFinals: string[];
}

/**
 * Applies recognizer revisions while keeping exactly one replaceable interim
 * hypothesis. Finals are buffered until every earlier utterance is present.
 */
export class TranscriptAccumulator {
  private latestRevision = new Map<string, number>();
  private finalized = new Set<string>();
  private pendingFinals = new Map<number, RealtimeTranscriptEvent>();
  private nextFinalIndex = 0;
  private activeInterim: RealtimeTranscriptEvent | null = null;
  private confirmedParts: string[] = [];
  private sessionId: string | null = null;

  reset(sessionId?: string): void {
    this.latestRevision.clear();
    this.finalized.clear();
    this.pendingFinals.clear();
    this.nextFinalIndex = 0;
    this.activeInterim = null;
    this.confirmedParts = [];
    this.sessionId = sessionId ?? null;
  }

  /** Start a replacement audio session without discarding confirmed text. */
  continueSession(sessionId?: string): void {
    const confirmedParts = [...this.confirmedParts];
    this.reset(sessionId);
    this.confirmedParts = confirmedParts;
  }

  applyInterim(event: RealtimeTranscriptEvent): TranscriptUpdate {
    if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
    if (this.finalized.has(event.utteranceId)) return this.snapshot(false);

    const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
    if (event.revision <= previousRevision) return this.snapshot(false);
    if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);

    this.latestRevision.set(event.utteranceId, event.revision);
    this.activeInterim = event;
    return this.snapshot(true);
  }

  applyFinal(event: RealtimeTranscriptEvent): TranscriptUpdate {
    if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
    if (this.finalized.has(event.utteranceId)) return this.snapshot(false);

    const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
    if (event.revision < previousRevision) return this.snapshot(false);
    if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);

    const pending = this.pendingFinals.get(event.utteranceIndex);
    if (pending && pending.revision >= event.revision) {
      return this.snapshot(false);
    }

    this.latestRevision.set(event.utteranceId, event.revision);
    this.pendingFinals.set(event.utteranceIndex, event);

    const appendedFinals: string[] = [];
    while (this.pendingFinals.has(this.nextFinalIndex)) {
      const next = this.pendingFinals.get(this.nextFinalIndex)!;
      this.pendingFinals.delete(this.nextFinalIndex);
      this.finalized.add(next.utteranceId);
      const text = next.text.replace(/\s+/g, " ").trim();
      if (text) {
        this.confirmedParts.push(text);
        appendedFinals.push(text);
      }
      if (this.activeInterim?.utteranceId === next.utteranceId) {
        this.activeInterim = null;
      }
      this.nextFinalIndex++;
    }

    return this.snapshot(true, appendedFinals);
  }

  private acceptSession(sessionId: string): boolean {
    if (this.sessionId === null) {
      this.sessionId = sessionId;
      return true;
    }
    return this.sessionId === sessionId;
  }

  private snapshot(
    accepted: boolean,
    appendedFinals: string[] = [],
  ): TranscriptUpdate {
    return {
      accepted,
      confirmedText: this.confirmedParts.join(" "),
      interimText: this.activeInterim?.text.trim() ?? "",
      appendedFinals,
    };
  }
}

	export interface BufferedAudioFrame {
	audioSequence: number;
	audio: string;
	}

	/**
	* Bounded replay buffer. Frames remain here until the server acknowledges
	* them, which makes reconnect replay deterministic and idempotent.
	*/
	export class AudioReplayBuffer {
	private frames: BufferedAudioFrame[] = [];

	constructor(private readonly capacity: number) {
	if (!Number.isInteger(capacity) \|\| capacity < 1) {
	throw new Error("AudioReplayBuffer capacity must be a positive integer");
	}
	}

	push(frame: BufferedAudioFrame): BufferedAudioFrame \| null {
	if (
	this.frames.some(
	(existing) => existing.audioSequence === frame.audioSequence,
	)
	) {
	return null;
	}

	this.frames.push(frame);
	this.frames.sort((a, b) => a.audioSequence - b.audioSequence);
	return this.frames.length > this.capacity ? this.frames.shift()! : null;
	}

	acknowledge(audioSequence: number): void {
	this.frames = this.frames.filter(
	(frame) => frame.audioSequence > audioSequence,
	);
	}

	unacknowledged(): BufferedAudioFrame[] {
	return [...this.frames];
	}

	firstSequence(): number \| null {
	return this.frames[0]?.audioSequence ?? null;
	}

	clear(): void {
	this.frames = [];
	}

	get size(): number {
	return this.frames.length;
	}
	}

	/**
	* Turns potentially duplicated or reordered frames back into one contiguous
	* stream. WebSocket normally preserves order, but reconnect replay can overlap
	* with frames already accepted by the server.
	*/
	export class OrderedAudioFrames {
	private pending = new Map<number, BufferedAudioFrame>();

	constructor(
	private nextSequence: number,
	private readonly capacity = 600,
	) {}

	receive(frame: BufferedAudioFrame): {
	duplicate: boolean;
	ready: BufferedAudioFrame[];
	} {
	if (
	frame.audioSequence < this.nextSequence \|\|
	this.pending.has(frame.audioSequence)
	) {
	return { duplicate: true, ready: [] };
	}

	this.pending.set(frame.audioSequence, frame);
	if (this.pending.size > this.capacity) {
	const highest = Math.max(...this.pending.keys());
	this.pending.delete(highest);
	}

	const ready: BufferedAudioFrame[] = [];
	while (this.pending.has(this.nextSequence)) {
	ready.push(this.pending.get(this.nextSequence)!);
	this.pending.delete(this.nextSequence);
	this.nextSequence++;
	}
	return { duplicate: false, ready };
	}

	expectedSequence(): number {
	return this.nextSequence;
	}
	}

	export interface RealtimeTranscriptEvent {
	sessionId: string;
	utteranceId: string;
	utteranceIndex: number;
	revision: number;
	text: string;
	}

	export interface TranscriptUpdate {
	accepted: boolean;
	confirmedText: string;
	interimText: string;
	appendedFinals: string[];
	}

	/**
	* Applies recognizer revisions while keeping exactly one replaceable interim
	* hypothesis. Finals are buffered until every earlier utterance is present.
	*/
	export class TranscriptAccumulator {
	private latestRevision = new Map<string, number>();
	private finalized = new Set<string>();
	private pendingFinals = new Map<number, RealtimeTranscriptEvent>();
	private nextFinalIndex = 0;
	private activeInterim: RealtimeTranscriptEvent \| null = null;
	private confirmedParts: string[] = [];
	private sessionId: string \| null = null;

	reset(sessionId?: string): void {
	this.latestRevision.clear();
	this.finalized.clear();
	this.pendingFinals.clear();
	this.nextFinalIndex = 0;
	this.activeInterim = null;
	this.confirmedParts = [];
	this.sessionId = sessionId ?? null;
	}

	/** Start a replacement audio session without discarding confirmed text. */
	continueSession(sessionId?: string): void {
	const confirmedParts = [...this.confirmedParts];
	this.reset(sessionId);
	this.confirmedParts = confirmedParts;
	}

	applyInterim(event: RealtimeTranscriptEvent): TranscriptUpdate {
	if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
	if (this.finalized.has(event.utteranceId)) return this.snapshot(false);

	const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
	if (event.revision <= previousRevision) return this.snapshot(false);
	if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);

	this.latestRevision.set(event.utteranceId, event.revision);
	this.activeInterim = event;
	return this.snapshot(true);
	}

	applyFinal(event: RealtimeTranscriptEvent): TranscriptUpdate {
	if (!this.acceptSession(event.sessionId)) return this.snapshot(false);
	if (this.finalized.has(event.utteranceId)) return this.snapshot(false);

	const previousRevision = this.latestRevision.get(event.utteranceId) ?? -1;
	if (event.revision < previousRevision) return this.snapshot(false);
	if (event.utteranceIndex < this.nextFinalIndex) return this.snapshot(false);

	const pending = this.pendingFinals.get(event.utteranceIndex);
	if (pending && pending.revision >= event.revision) {
	return this.snapshot(false);
	}

	this.latestRevision.set(event.utteranceId, event.revision);
	this.pendingFinals.set(event.utteranceIndex, event);

	const appendedFinals: string[] = [];
	while (this.pendingFinals.has(this.nextFinalIndex)) {
	const next = this.pendingFinals.get(this.nextFinalIndex)!;
	this.pendingFinals.delete(this.nextFinalIndex);
	this.finalized.add(next.utteranceId);
	const text = next.text.replace(/\s+/g, " ").trim();
	if (text) {
	this.confirmedParts.push(text);
	appendedFinals.push(text);
	}
	if (this.activeInterim?.utteranceId === next.utteranceId) {
	this.activeInterim = null;
	}
	this.nextFinalIndex++;
	}

	return this.snapshot(true, appendedFinals);
	}

	private acceptSession(sessionId: string): boolean {
	if (this.sessionId === null) {
	this.sessionId = sessionId;
	return true;
	}
	return this.sessionId === sessionId;
	}

	private snapshot(
	accepted: boolean,
	appendedFinals: string[] = [],
	): TranscriptUpdate {
	return {
	accepted,
	confirmedText: this.confirmedParts.join(" "),
	interimText: this.activeInterim?.text.trim() ?? "",
	appendedFinals,
	};
	}
	}