-
@@ -194,6 +201,7 @@
:statsVisible="showStatsForNerds"
:isHandRaised="isHandRaised"
:isReactionPickerOpen="isReactionPickerOpen"
+ :isCaptionsEnabled="captionStore.isCaptionsEnabled"
@update:isReactionPickerOpen="isReactionPickerOpen = $event"
:meetingId="meetingId"
:meetingTitle="meetingTitle"
@@ -211,6 +219,7 @@
@toggle-screen-share="mediaControls.toggleScreenShare()"
@toggle-fullscreen="toggleFullscreen"
@toggle-raise-hand="raiseHand.toggleRaiseHand()"
+ @toggle-captions="toggleCaptions"
@report-problem="handleReportProblem"
@toggle-stats="toggleStatsForNerds"
@end-call="sfuConnection.endCall()"
@@ -257,6 +266,7 @@ import { Badge, Button, createResource, frappeRequest, toast } from "frappe-ui";
import { computed, h, onMounted, onUnmounted, provide, ref, watch } from "vue";
import { useRoute, useRouter } from "vue-router";
+import CaptionOverlay from "../components/CaptionOverlay.vue";
import ChatPanel from "../components/ChatPanel.vue";
import JoinRequestNotifications from "../components/JoinRequestNotifications.vue";
import LobbyOverlay from "../components/LobbyOverlay.vue";
@@ -272,6 +282,11 @@ import PeoplePanel from "../components/PeoplePanel.vue";
import RejectionOverlay from "../components/RejectionOverlay.vue";
import StatsForNerdsOverlay from "../components/StatsForNerdsOverlay.vue";
import { useBackgroundEffects } from "../composables/useBackgroundEffects";
+import { useCaptionStore } from "../composables/useCaptionStore";
+import {
+ restoreCaptionSubscription,
+ useCaptions,
+} from "../composables/useCaptions";
import { useChat } from "../composables/useChat";
import { useChatStore } from "../composables/useChatStore";
import { useConnectionState } from "../composables/useConnectionState";
@@ -352,6 +367,8 @@ const lobbyStore = useLobbyStore();
const reactionStore = useReactionStore();
const raiseHandStore = useRaiseHandStore();
const gridLayout = useGridLayout(mediaState);
+const captionStore = useCaptionStore();
+let captionRestoreGeneration = 0;
// --- Lobby notification tracking ---
const notifiedLobbyUsers = ref(new Set
());
@@ -552,6 +569,18 @@ const sfuConnection = useSFUConnection({
onActiveSpeakerChanged: (participantIds: string[]) => {
participantStore.activeSpeakerIds = participantIds;
},
+ onRoomRejoined: (sfuClient) => {
+ const generation = ++captionRestoreGeneration;
+ void restoreCaptionSubscription(
+ sfuClient,
+ captionStore.isCaptionsEnabled,
+ ).then((restored) => {
+ if (generation === captionRestoreGeneration && !restored) {
+ captionStore.setCaptionsEnabled(false);
+ }
+ });
+ },
+ onE2EERequired: () => captions.disableCaptionsForE2EE(),
onRecordingState: recording.syncState,
onRecordingEnabled: recording.setGlobalEnabled,
});
@@ -647,6 +676,14 @@ const raiseHand = useRaiseHand({
sfuClient: sfuConnection.sfuClient,
});
+const captions = useCaptions({
+ sfuClient: sfuConnection.sfuClient,
+});
+const toggleCaptions = async () => {
+ captionRestoreGeneration++;
+ await captions.toggleCaptions();
+};
+
// --- Lobby ---
const lobby = useLobby({
lobbyStore,
@@ -1028,6 +1065,7 @@ onMounted(async () => {
lobbyStore.$reset();
reactionStore.$reset();
raiseHandStore.$reset();
+ captionStore.$reset();
gridLayout.resetGridLayout();
currentUser.resetCurrentUser();
e2eeState.reset();
diff --git a/frontend/src/apps/meet/utils/sfu/ParticipantConnection.ts b/frontend/src/apps/meet/utils/sfu/ParticipantConnection.ts
index 1b5c67985d..527d048fe0 100644
--- a/frontend/src/apps/meet/utils/sfu/ParticipantConnection.ts
+++ b/frontend/src/apps/meet/utils/sfu/ParticipantConnection.ts
@@ -127,6 +127,7 @@ export interface SFUEventHandlers {
detail?: string,
) => void;
onRecoveryExhausted?: () => void;
+ onRoomRejoined?: () => void;
onLifecycleStateChange?: (state: ParticipantConnectionState) => void;
onInitialPublicationError?: (error: unknown) => void;
}
@@ -724,6 +725,7 @@ export class ParticipantConnection {
this.getCurrentRejoinMediaState(),
);
if (generation !== this.lifecycleGeneration) return;
+ this.eventHandlers.onRoomRejoined?.();
if (!(await this.waitForE2EEContextIfRequired(signal))) {
throw new Error(
"E2EE context is not ready after signaling reconnect",
diff --git a/frontend/src/apps/meet/utils/sfu/__tests__/ParticipantConnectionEvents.test.ts b/frontend/src/apps/meet/utils/sfu/__tests__/ParticipantConnectionEvents.test.ts
index 23c1187233..a57f059197 100644
--- a/frontend/src/apps/meet/utils/sfu/__tests__/ParticipantConnectionEvents.test.ts
+++ b/frontend/src/apps/meet/utils/sfu/__tests__/ParticipantConnectionEvents.test.ts
@@ -212,11 +212,13 @@ describe("ParticipantConnection", () => {
it("rejoins the room and rebuilds media after signaling reconnect", async () => {
const { manager, mediaManager, sfuClient, transportManager, recoveryManager } =
createManager();
- manager.initialize("meeting-1", { user_id: "me" });
+ const onRoomRejoined = vi.fn();
+ manager.initialize("meeting-1", { user_id: "me" }, { onRoomRejoined });
await manager.joinRoom(
{ name: "Me", userId: "me" },
{ audio_enabled: true, video_enabled: true },
);
+ expect(onRoomRejoined).not.toHaveBeenCalled();
await manager.rejoinAfterSignalingReconnect();
@@ -232,6 +234,31 @@ describe("ParticipantConnection", () => {
expect(transportManager.createReceiveTransport).toHaveBeenCalledTimes(1);
expect(mediaManager.rebuildSendSide).toHaveBeenCalledTimes(1);
expect(recoveryManager.setupTransportEventHandlers).toHaveBeenCalledTimes(1);
+ expect(onRoomRejoined).toHaveBeenCalledTimes(1);
+ });
+
+ it("notifies room rejoin only after the join request succeeds", async () => {
+ const { manager, sfuClient } = createManager();
+ const onRoomRejoined = vi.fn();
+ let finishJoin: () => void = () => {};
+ manager.initialize("meeting-1", { user_id: "me" }, { onRoomRejoined });
+ await manager.joinRoom(
+ { name: "Me", userId: "me" },
+ { audio_enabled: true, video_enabled: true },
+ );
+ sfuClient.joinRoom.mockReturnValueOnce(
+ new Promise((resolve) => {
+ finishJoin = resolve;
+ }),
+ );
+
+ const rejoin = manager.rejoinAfterSignalingReconnect();
+ await vi.waitFor(() => expect(sfuClient.joinRoom).toHaveBeenCalledTimes(2));
+ expect(onRoomRejoined).not.toHaveBeenCalled();
+
+ finishJoin();
+ await rejoin;
+ expect(onRoomRejoined).toHaveBeenCalledTimes(1);
});
it("uses current live tracks for rejoin media state", async () => {
diff --git a/knip.json b/knip.json
index f8ff1a6180..3e8999f6a1 100644
--- a/knip.json
+++ b/knip.json
@@ -27,10 +27,10 @@
]
},
"suite/meet/sfu-server": {
- "entry": ["src/server.ts", "src/**/*.test.ts"],
+ "entry": ["src/**/*.test.ts"],
"project": ["src/**/*.ts", "!src/server.ts", "!src/**/*.test.ts"],
"vitest": false,
- "ignoreBinaries": ["vitest"],
+ "ignoreBinaries": ["ffmpeg"],
"ignoreIssues": {
"src/types/index.ts": ["types"]
}
diff --git a/suite/meet/sfu-server/.env.example b/suite/meet/sfu-server/.env.example
index 4bece472ab..e9c442e0ce 100644
--- a/suite/meet/sfu-server/.env.example
+++ b/suite/meet/sfu-server/.env.example
@@ -29,6 +29,24 @@ E2EE_ROSTER_PERSISTENCE_DIR=
# TURN_USERNAME=username
# TURN_PASSWORD=password
+# STT / Captions
+STT_SERVER_URL=http://localhost:8000
+# NEMOTRON_MODEL=nvidia/nemotron-3.5-asr-streaming-0.6b
+# NEMOTRON_LANGUAGE=en-US # pin language for stability; use auto only for multilingual rooms
+# NEMOTRON_ATT_CONTEXT_SIZE=56,3
+# NEMOTRON_FINAL_SILENCE_MS=600
+# Local diagnosis only; captures participant audio.
+# STT_CAPTURE_DIR=./data/stt-captures
+# STT_PRE_ROLL_MS=300
+# STT_SILENCE_MS=500
+# STT_MIN_SPEECH_MS=600
+# STT_MIN_TAIL_MS=200
+# STT_SHORT_UTTERANCE_SILENCE_MS=700
+# STT_VAD_THRESHOLD=0.012
+
+# Hugging Face token (optional, avoids rate-limit warnings when downloading models)
+# HF_TOKEN=hf_...
+
# Legacy settings (for backward compatibility)
API_KEY=your-api-key-here
SECRET_KEY=your-secret-key-here
diff --git a/suite/meet/sfu-server/Dockerfile b/suite/meet/sfu-server/Dockerfile
index 319cae66ca..75e58dfe28 100644
--- a/suite/meet/sfu-server/Dockerfile
+++ b/suite/meet/sfu-server/Dockerfile
@@ -26,7 +26,7 @@ WORKDIR /app/sfu-server
ENV NODE_ENV=production \
PORT=3000
-RUN apt-get update && apt-get install -y --no-install-recommends tini ca-certificates curl \
+RUN apt-get update && apt-get install -y --no-install-recommends tini ca-certificates curl ffmpeg \
&& rm -rf /var/lib/apt/lists/*
COPY --from=build /app /app
diff --git a/suite/meet/sfu-server/README.md b/suite/meet/sfu-server/README.md
index b5486884df..75783738f4 100644
--- a/suite/meet/sfu-server/README.md
+++ b/suite/meet/sfu-server/README.md
@@ -2,6 +2,36 @@
Mediasoup-based Selective Forwarding Unit (SFU) for Frappe Meet.
+## Speech-to-Text (Captions)
+
+Real-time captions are powered by an on-premise NVIDIA Nemotron ASR backend.
+
+### Local Development
+
+Set `STT_SERVER_URL` to a running STT service that implements `/health` and the OpenAI Realtime transcription endpoint at `/v1/realtime`.
+
+### Docker Compose
+
+Set `STT_SERVER_URL` to an externally managed STT backend. The SFU deployment does not start an STT sidecar.
+
+### Environment Variables
+
+| Variable | Description | Default |
+|---|---|---|
+| `STT_SERVER_URL` | SFU URL for the STT service | — |
+| `STT_CAPTURE_DIR` | Optional local directory for diagnostic utterance WAV/JSON captures | — |
+| `NEMOTRON_MODEL` | Hugging Face model ID | `nvidia/nemotron-3.5-asr-streaming-0.6b` |
+| `NEMOTRON_LANGUAGE` | Locale prompt such as `en-US`, or `auto` for multilingual rooms | `en-US` |
+| `NEMOTRON_ATT_CONTEXT_SIZE` | NeMo streaming attention context, `left,right` | `56,3` |
+| `NEMOTRON_FINAL_SILENCE_MS` | Silence padding appended before final decode | `600` |
+| `STT_SILENCE_MS` | Silence duration before finalizing an utterance | `500` |
+| `STT_MIN_SPEECH_MS` | Minimum speech duration before normal silence final | `600` |
+| `STT_MIN_TAIL_MS` | Minimum speech duration for short utterance final | `200` |
+| `STT_SHORT_UTTERANCE_SILENCE_MS` | Silence duration before finalizing short utterances | `700` |
+| `STT_VAD_THRESHOLD` | Speech detection sensitivity (0.0–1.0) | `0.012` |
+| `STT_PRE_ROLL_MS` | Audio retained before speech detection to avoid clipped words | `300` |
+| `HF_TOKEN` | Hugging Face token (optional, avoids rate limits) | — |
+
## Development Setup
From the Suite app directory, install the SFU dependencies and create a local environment file:
diff --git a/suite/meet/sfu-server/deploy/.env.example b/suite/meet/sfu-server/deploy/.env.example
index e5675cf717..4cd6d9534d 100644
--- a/suite/meet/sfu-server/deploy/.env.example
+++ b/suite/meet/sfu-server/deploy/.env.example
@@ -52,6 +52,27 @@ MEDIASOUP_NUM_WORKERS=
# Worker log verbosity: debug | warn | error | none
MEDIASOUP_WORKER_LOGLEVEL=warn
+# === STT / Captions ==========================================================
+# URL of an external STT backend that supports /health and /v1/realtime.
+STT_SERVER_URL=http://127.0.0.1:8000
+
+# Nemotron ASR settings for the standalone stt-server/server.py process.
+NEMOTRON_MODEL=nvidia/nemotron-3.5-asr-streaming-0.6b
+NEMOTRON_LANGUAGE=en-US
+NEMOTRON_ATT_CONTEXT_SIZE=56,3
+NEMOTRON_FINAL_SILENCE_MS=600
+STT_SILENCE_MS=500
+STT_MIN_SPEECH_MS=600
+STT_MIN_TAIL_MS=200
+STT_SHORT_UTTERANCE_SILENCE_MS=700
+
+# Speech detection sensitivity (0.0–1.0). Lower = more sensitive.
+STT_VAD_THRESHOLD=0.012
+
+# Hugging Face token (optional). Setting this avoids rate-limit warnings
+# when downloading models. Get one at https://huggingface.co/settings/tokens
+# HF_TOKEN=hf_...
+
# === Logging ==================================================================
SFU_LOG_LEVEL=info
diff --git a/suite/meet/sfu-server/deploy/docker-compose.yml b/suite/meet/sfu-server/deploy/docker-compose.yml
index 2ab8159e5a..89d25f6215 100644
--- a/suite/meet/sfu-server/deploy/docker-compose.yml
+++ b/suite/meet/sfu-server/deploy/docker-compose.yml
@@ -52,6 +52,8 @@ services:
- MEDIASOUP_NUM_WORKERS=${MEDIASOUP_NUM_WORKERS:-}
- MEDIASOUP_WORKER_LOGLEVEL=${MEDIASOUP_WORKER_LOGLEVEL:-warn}
- SFU_LOG_LEVEL=${SFU_LOG_LEVEL:-info}
+ - STT_SERVER_URL=${STT_SERVER_URL:-}
+ - STT_VAD_THRESHOLD=${STT_VAD_THRESHOLD:-0.012}
- SENTRY_DSN=${SENTRY_DSN:-}
- SENTRY_ENVIRONMENT=${SENTRY_ENVIRONMENT:-production}
- SENTRY_RELEASE=${SENTRY_RELEASE:-}
diff --git a/suite/meet/sfu-server/package.json b/suite/meet/sfu-server/package.json
index 2534b66ec6..9c8f2fc90e 100644
--- a/suite/meet/sfu-server/package.json
+++ b/suite/meet/sfu-server/package.json
@@ -35,6 +35,7 @@
"prom-client": "^15.1.3",
"socket.io": "^4.7.5",
"socket.io-client": "^4.7.5",
+ "ws": "^8.21.0",
"yargs": "^18.0.0"
},
"devDependencies": {
@@ -43,6 +44,7 @@
"@types/express": "^4.17.21",
"@types/jsonwebtoken": "^9.0.6",
"@types/node": "^20.11.24",
+ "@types/ws": "^8.18.1",
"@vitest/coverage-v8": "4.1.9",
"nodemon": "^3.0.1",
"ts-node": "^10.9.2",
diff --git a/suite/meet/sfu-server/src/config.test.ts b/suite/meet/sfu-server/src/config.test.ts
index 5430a07900..67687d0975 100644
--- a/suite/meet/sfu-server/src/config.test.ts
+++ b/suite/meet/sfu-server/src/config.test.ts
@@ -34,6 +34,11 @@ describe('loadConfig', () => {
allowPlainTransport: false,
bypassRateLimits: false,
});
+ expect(config.stt).toEqual({
+ serverUrl: undefined,
+ allowMockFallback: false,
+ captureDirectory: undefined,
+ });
expect(Object.isFrozen(config)).toBe(true);
expect(Object.isFrozen(config.mediasoup.worker)).toBe(true);
});
@@ -61,6 +66,21 @@ describe('loadConfig', () => {
expect(config.runtime.bypassRateLimits).toBe(true);
});
+ it('loads optional STT diagnostics configuration', () => {
+ const config = loadConfig(
+ validEnv({
+ STT_SERVER_URL: 'https://stt.example.test',
+ STT_CAPTURE_DIR: './data/stt-captures',
+ }),
+ system,
+ );
+ expect(config.stt).toEqual({
+ serverUrl: 'https://stt.example.test',
+ allowMockFallback: false,
+ captureDirectory: './data/stt-captures',
+ });
+ });
+
it('rejects partial numbers and invalid enum values', () => {
expect(() =>
loadConfig(
diff --git a/suite/meet/sfu-server/src/config.ts b/suite/meet/sfu-server/src/config.ts
index e3822fad13..d6f445040e 100644
--- a/suite/meet/sfu-server/src/config.ts
+++ b/suite/meet/sfu-server/src/config.ts
@@ -34,6 +34,11 @@ export interface SFUConfig {
metrics: {
token?: string;
};
+ stt: {
+ serverUrl?: string;
+ allowMockFallback: boolean;
+ captureDirectory?: string;
+ };
logging: {
level: SFULogLevel;
};
@@ -340,6 +345,11 @@ export function loadConfig(
bypassRateLimits: mode === 'development' || ci || githubActions,
},
metrics: { token: optional(env, 'METRICS_TOKEN') },
+ stt: {
+ serverUrl: optional(env, 'STT_SERVER_URL'),
+ allowMockFallback: mode === 'development',
+ captureDirectory: optional(env, 'STT_CAPTURE_DIR'),
+ },
logging: { level: logLevel },
sentry: {
dsn: sentryDsn,
diff --git a/suite/meet/sfu-server/src/mediasoup/MediasoupManager.ts b/suite/meet/sfu-server/src/mediasoup/MediasoupManager.ts
index 44d8e8ff32..92969af8f7 100644
--- a/suite/meet/sfu-server/src/mediasoup/MediasoupManager.ts
+++ b/suite/meet/sfu-server/src/mediasoup/MediasoupManager.ts
@@ -1,3 +1,4 @@
+import type { SttManager } from '../stt/SttManager';
import type {
CloseProducerResult,
Consumer,
@@ -36,6 +37,7 @@ export class MediasoupManager {
private transportManager = new TransportManager();
private producerManager = new ProducerManager();
consumerManager = new ConsumerManager();
+ private sttManager: SttManager | null = null;
private networkQualityListeners: Array<
(
@@ -100,16 +102,41 @@ export class MediasoupManager {
this.producerManager.on(
'producer_closed',
- (roomId: string, peerId: string, kind: 'audio' | 'video') => {
+ (
+ roomId: string,
+ peerId: string,
+ kind: 'audio' | 'video',
+ producerId: string,
+ ) => {
const peerState = this.peerScores.get(peerId);
if (peerState) {
delete peerState[kind];
this.evaluateAndEmitNetworkQuality(roomId, peerId);
}
+ if (this.sttManager && kind === 'audio') {
+ this.sttManager
+ .stopTranscription(roomId, peerId, producerId)
+ .catch((error) => {
+ loggers.mediasoupManager.warn(
+ 'STT stop error on producer close: %s',
+ (error as Error).message,
+ );
+ });
+ }
},
);
}
+ setSttManager(sttManager: SttManager): void {
+ this.sttManager = sttManager;
+ this.sttManager.setGetRouter((roomId) =>
+ this.roomManager.getRouter(roomId),
+ );
+ this.sttManager.setRestartRoomTranscription((roomId) =>
+ this.startSttForExistingProducers(roomId, sttManager),
+ );
+ }
+
private evaluateAndEmitNetworkQuality(roomId: string, peerId: string) {
const peerState = this.peerScores.get(peerId);
if (!peerState) return;
@@ -230,6 +257,9 @@ export class MediasoupManager {
this.closingRooms.add(roomId);
const room = this.roomManager.getRoom(roomId);
try {
+ if (this.sttManager) {
+ await this.sttManager.stopRoom(roomId);
+ }
for (const peerId of [...(room?.peers.keys() ?? [])]) {
await this.removePeer(roomId, peerId);
}
@@ -416,6 +446,18 @@ export class MediasoupManager {
room.audioLevelObserver.addProducer({ producerId: result.id });
}
+ if (this.sttManager && kind === 'audio') {
+ const peerInfo = peer.info;
+ this.sttManager
+ .startTranscription(roomId, peerId, peerInfo?.name, producer)
+ .catch((error) => {
+ loggers.mediasoupManager.warn(
+ 'STT start error: %s',
+ (error as Error).message,
+ );
+ });
+ }
+
return result;
}
@@ -978,6 +1020,52 @@ export class MediasoupManager {
return room?.peers.has(peerId) || false;
}
+ async startSttForExistingProducers(
+ roomId: string,
+ sttManager: SttManager,
+ ): Promise {
+ const room = this.roomManager.getRoom(roomId);
+ if (!room) {
+ loggers.mediasoupManager.warn(
+ 'Cannot start STT: room %s not found',
+ roomId,
+ );
+ return;
+ }
+
+ const started: string[] = [];
+ for (const [peerId, peer] of room.peers) {
+ for (const producer of peer.producers.values()) {
+ if (producer.kind !== 'audio' || producer.closed) continue;
+
+ try {
+ await sttManager.startTranscription(
+ roomId,
+ peerId,
+ peer.info.name,
+ producer,
+ );
+ started.push(peerId);
+ } catch (error) {
+ loggers.mediasoupManager.warn(
+ 'Failed to start STT for existing producer %s in room %s: %s',
+ producer.id,
+ roomId,
+ (error as Error).message,
+ );
+ }
+ }
+ }
+
+ if (started.length > 0) {
+ loggers.mediasoupManager.info(
+ 'Started STT for %d existing audio producer(s) in room %s',
+ started.length,
+ roomId,
+ );
+ }
+ }
+
get rooms() {
return this.roomManager;
}
@@ -1011,6 +1099,15 @@ export class MediasoupManager {
};
loggers.mediasoupManager.info('Initial cleanup stats: %o', initialStats);
+ if (this.sttManager) {
+ const sttManager = this.sttManager;
+ await Promise.all(
+ this.roomManager
+ .getAllRooms()
+ .map((room) => sttManager.stopRoom(room.id)),
+ );
+ }
+
// Close all rooms (this will also close peers, transports, producers, consumers)
await this.roomManager.cleanup();
diff --git a/suite/meet/sfu-server/src/mediasoup/RoomManager.ts b/suite/meet/sfu-server/src/mediasoup/RoomManager.ts
index 80f4636467..e4f4e51bcd 100644
--- a/suite/meet/sfu-server/src/mediasoup/RoomManager.ts
+++ b/suite/meet/sfu-server/src/mediasoup/RoomManager.ts
@@ -95,6 +95,10 @@ export class RoomManager {
return this.rooms.get(roomId);
}
+ getAllRooms(): Room[] {
+ return Array.from(this.rooms.values());
+ }
+
getRouter(roomId: string): mediasoup.types.Router | undefined {
return this.routers.get(roomId);
}
diff --git a/suite/meet/sfu-server/src/server.ts b/suite/meet/sfu-server/src/server.ts
index a67aeaed2b..aaa73c083c 100644
--- a/suite/meet/sfu-server/src/server.ts
+++ b/suite/meet/sfu-server/src/server.ts
@@ -14,6 +14,7 @@ import { RecordingGrantManager } from './server/RecordingGrantManager';
import { RecordingGrantPersistenceFile } from './server/RecordingGrantPersistenceFile';
import { RouteManager } from './server/RouteManager';
import { SocketHandlerManager } from './server/SocketHandlerManager';
+import { SttManager } from './stt/SttManager';
import { Telemetry } from './telemetry/Telemetry';
import { configureLogging, loggers } from './utils/logger';
import { captureException, flushSentry, initSentry } from './utils/sentry';
@@ -28,6 +29,7 @@ export class SFUServer {
private authManager: AuthManager;
private routeManager: RouteManager;
private socketHandlerManager: SocketHandlerManager;
+ private sttManager: SttManager;
private config: SFUConfig['server'];
private telemetry: Telemetry;
private recordingGrantPersistence?: RecordingGrantPersistenceFile;
@@ -65,6 +67,12 @@ export class SFUServer {
this.mediasoup.onMediaScore((direction, media, score) =>
this.telemetry.mediaScore.observe({ direction, media }, score),
);
+ this.sttManager = new SttManager({
+ sttServerUrl: config.stt.serverUrl,
+ allowMockFallback: config.stt.allowMockFallback,
+ captureDirectory: config.stt.captureDirectory,
+ });
+ this.mediasoup.setSttManager(this.sttManager);
const recordingPersistencePath = config.persistence.recordingGrantFile;
this.recordingGrantPersistence = recordingPersistencePath
? new RecordingGrantPersistenceFile(recordingPersistencePath)
@@ -103,6 +111,7 @@ export class SFUServer {
config.runtime,
e2eeCoordinatorPersistence,
recordingGrantManager,
+ this.sttManager,
);
this.setupMiddleware();
diff --git a/suite/meet/sfu-server/src/server/RoomRegistry.ts b/suite/meet/sfu-server/src/server/RoomRegistry.ts
index 38407a2504..0d34d75916 100644
--- a/suite/meet/sfu-server/src/server/RoomRegistry.ts
+++ b/suite/meet/sfu-server/src/server/RoomRegistry.ts
@@ -288,6 +288,22 @@ export class RoomRegistry {
this.emitToScope(roomId, 'full', event, ...args);
}
+ emitToFullAccessSockets(
+ roomId: string,
+ socketIds: ReadonlySet,
+ event: Event,
+ ...args: Parameters
+ ): void {
+ const fullSocketIds = this.io.sockets.adapter.rooms.get(fullRoom(roomId));
+ if (!fullSocketIds) return;
+ for (const socketId of socketIds) {
+ if (!fullSocketIds.has(socketId)) continue;
+ const socket: ServerSocket | undefined =
+ this.io.sockets.sockets.get(socketId);
+ socket?.emit(event, ...args);
+ }
+ }
+
emitToPreviewParticipants(
roomId: string,
event: Event,
diff --git a/suite/meet/sfu-server/src/server/SocketHandlerManager.ts b/suite/meet/sfu-server/src/server/SocketHandlerManager.ts
index 7e4135fd0e..3bfbe1a1ae 100644
--- a/suite/meet/sfu-server/src/server/SocketHandlerManager.ts
+++ b/suite/meet/sfu-server/src/server/SocketHandlerManager.ts
@@ -1,6 +1,7 @@
import type { Server } from 'socket.io';
import type { SFUConfig } from '../config';
import type { MediasoupManager } from '../mediasoup/MediasoupManager';
+import type { SttManager } from '../stt/SttManager';
import type { Telemetry } from '../telemetry/Telemetry';
import type { ClientToServerEvents, ServerToClientEvents } from '../types';
import { loggers } from '../utils/logger';
@@ -25,6 +26,7 @@ import { registerReactionHandlers } from './handlers/ReactionHandlers';
import { registerRoomJoinHandlers } from './handlers/RoomJoinHandlers';
import { registerRoomQueryHandlers } from './handlers/RoomQueryHandlers';
import { registerScreenShareHandlers } from './handlers/ScreenShareHandlers';
+import { registerSttHandlers } from './handlers/SttHandlers';
import { registerWebRtcTransportHandlers } from './handlers/WebRtcTransportHandlers';
import type { RecordingGrantManager } from './RecordingGrantManager';
import { RoomLifecycleCoordinator } from './RoomLifecycleCoordinator';
@@ -53,6 +55,7 @@ export class SocketHandlerManager {
private readonly runtime: SFUConfig['runtime'],
coordinatorPersistence?: E2eeCoordinatorPersistence,
private readonly recordingGrantManager?: RecordingGrantManager,
+ sttManager?: SttManager,
) {
this.io = io;
this.mediasoup = mediasoup;
@@ -70,6 +73,9 @@ export class SocketHandlerManager {
this.runtime.bypassRateLimits,
);
this.e2eeEpochRelay.setRoster(roster);
+ sttManager?.setEmitToSubscribers((roomId, socketIds, event, data) => {
+ this.registry.emitToFullAccessSockets(roomId, socketIds, event, data);
+ });
this.roomLifecycle = new RoomLifecycleCoordinator(
this.registry,
this.e2eeEpochRelay,
@@ -84,6 +90,7 @@ export class SocketHandlerManager {
mediasoup,
authManager,
rateLimiter: this.rateLimiter,
+ sttManager,
e2eeEpochRelay: this.e2eeEpochRelay,
e2eeRoster: roster,
telemetry,
@@ -102,6 +109,7 @@ export class SocketHandlerManager {
registerHostControlHandlers(deps),
registerScreenShareHandlers(deps),
registerPollHandlers(deps),
+ registerSttHandlers(deps),
registerChatHandlers(deps),
registerReactionHandlers(deps),
registerRaiseHandHandlers(deps),
diff --git a/suite/meet/sfu-server/src/server/__tests__/RoomRegistry.test.ts b/suite/meet/sfu-server/src/server/__tests__/RoomRegistry.test.ts
index 60faeadd71..5cb94f2a87 100644
--- a/suite/meet/sfu-server/src/server/__tests__/RoomRegistry.test.ts
+++ b/suite/meet/sfu-server/src/server/__tests__/RoomRegistry.test.ts
@@ -252,6 +252,34 @@ describe('RoomRegistry', () => {
);
});
+ it('emitToFullAccessSockets reaches only selected full sockets', () => {
+ const setup = makeIo();
+ const registry = new RoomRegistry(setup.io);
+ const subscribed = makeSocket('subscribed');
+ const other = makeSocket('other');
+ const preview = makeSocket('preview');
+ addFullSocket(setup, 'r1', subscribed);
+ addFullSocket(setup, 'r1', other);
+ addPreviewSocket(setup, 'r1', preview);
+
+ registry.emitToFullAccessSockets(
+ 'r1',
+ new Set(['subscribed', 'preview']),
+ 'hello',
+ { x: 1 },
+ );
+
+ expect(
+ (subscribed as unknown as { _emitCalls: unknown[] })._emitCalls,
+ ).toEqual([{ event: 'hello', data: { x: 1 } }]);
+ expect(
+ (other as unknown as { _emitCalls: unknown[] })._emitCalls,
+ ).toEqual([]);
+ expect(
+ (preview as unknown as { _emitCalls: unknown[] })._emitCalls,
+ ).toEqual([]);
+ });
+
it('emitToScope is a no-op when the room has no sockets', () => {
const { io } = makeIo();
const registry = new RoomRegistry(io);
diff --git a/suite/meet/sfu-server/src/server/handlers/AuthHandlers.ts b/suite/meet/sfu-server/src/server/handlers/AuthHandlers.ts
index bdacde9e87..e8fbe722eb 100644
--- a/suite/meet/sfu-server/src/server/handlers/AuthHandlers.ts
+++ b/suite/meet/sfu-server/src/server/handlers/AuthHandlers.ts
@@ -25,6 +25,15 @@ export function registerAuthHandlers(deps: HandlerDeps) {
}
deps.authManager.updateSocketToken(socket, token);
+ if (socket.e2eeRequired && socket.roomId) {
+ const wasLastSubscriber = deps.sttManager?.removeSubscriber(
+ socket.roomId,
+ socket.id,
+ );
+ if (wasLastSubscriber) {
+ void deps.sttManager?.stopRoom(socket.roomId, true);
+ }
+ }
deps.telemetry.authEvents.inc({
stage: 'refresh',
reason: 'valid',
diff --git a/suite/meet/sfu-server/src/server/handlers/DisconnectHandlers.ts b/suite/meet/sfu-server/src/server/handlers/DisconnectHandlers.ts
index e05dc2b28f..dfd777dcd4 100644
--- a/suite/meet/sfu-server/src/server/handlers/DisconnectHandlers.ts
+++ b/suite/meet/sfu-server/src/server/handlers/DisconnectHandlers.ts
@@ -83,6 +83,13 @@ export function registerDisconnectHandlers(deps: HandlerDeps) {
roomId,
);
}
+ const wasLastSubscriber = deps.sttManager?.removeSubscriber(
+ roomId,
+ socket.id,
+ );
+ if (wasLastSubscriber) {
+ await deps.sttManager?.stopRoom(roomId, true);
+ }
deps.roomLifecycle.scheduleCleanupIfHumanEmpty(roomId);
}
} catch (error) {
diff --git a/suite/meet/sfu-server/src/server/handlers/Handler.ts b/suite/meet/sfu-server/src/server/handlers/Handler.ts
index a18c61299a..c508d8ca7e 100644
--- a/suite/meet/sfu-server/src/server/handlers/Handler.ts
+++ b/suite/meet/sfu-server/src/server/handlers/Handler.ts
@@ -1,6 +1,7 @@
import type { Server, Socket } from 'socket.io';
import type { SFUConfig } from '../../config';
import type { MediasoupManager } from '../../mediasoup/MediasoupManager';
+import type { SttManager } from '../../stt/SttManager';
import type { Telemetry } from '../../telemetry/Telemetry';
import type {
ClientToServerEvents,
@@ -28,6 +29,7 @@ export interface HandlerDeps {
mediasoup: MediasoupManager;
authManager: AuthManager;
rateLimiter: RateLimiter;
+ sttManager?: SttManager;
e2eeEpochRelay: E2EEEpochRelay;
e2eeRoster: E2eeRosterStore;
telemetry: Telemetry;
diff --git a/suite/meet/sfu-server/src/server/handlers/SttHandlers.test.ts b/suite/meet/sfu-server/src/server/handlers/SttHandlers.test.ts
new file mode 100644
index 0000000000..a29841f269
--- /dev/null
+++ b/suite/meet/sfu-server/src/server/handlers/SttHandlers.test.ts
@@ -0,0 +1,65 @@
+import { describe, expect, it, vi } from 'vitest';
+import { registerSttHandlers } from './SttHandlers';
+
+describe('registerSttHandlers', () => {
+ it('acknowledges subscriptions before starting existing producers', async () => {
+ let toggle:
+ | ((data: unknown, callback: (result: unknown) => void) => void)
+ | undefined;
+ const socket = {
+ id: 'socket-1',
+ roomId: 'room-1',
+ e2eeRequired: false,
+ on: (event: string, handler: typeof toggle) => {
+ if (event === 'stt:toggle') toggle = handler;
+ },
+ };
+ let finishStart: () => void = () => {};
+ const startSttForExistingProducers = vi.fn(
+ () =>
+ new Promise((resolve) => {
+ finishStart = resolve;
+ }),
+ );
+ registerSttHandlers({
+ authManager: { ensureFullAccess: vi.fn() },
+ sttManager: { addSubscriber: vi.fn(() => true) },
+ mediasoup: { startSttForExistingProducers },
+ } as never)(socket as never);
+ const callback = vi.fn();
+
+ toggle?.({ enabled: true }, callback);
+
+ expect(callback).toHaveBeenCalledWith({ success: true, enabled: true });
+ expect(startSttForExistingProducers).toHaveBeenCalledOnce();
+ finishStart();
+ });
+
+ it('rejects caption subscriptions when E2EE is required', () => {
+ let toggle:
+ | ((data: unknown, callback: (result: unknown) => void) => void)
+ | undefined;
+ const socket = {
+ id: 'socket-1',
+ roomId: 'room-1',
+ e2eeRequired: true,
+ on: (event: string, handler: typeof toggle) => {
+ if (event === 'stt:toggle') toggle = handler;
+ },
+ };
+ const addSubscriber = vi.fn();
+ registerSttHandlers({
+ authManager: { ensureFullAccess: vi.fn() },
+ sttManager: { addSubscriber },
+ } as never)(socket as never);
+ const callback = vi.fn();
+
+ toggle?.({ enabled: true }, callback);
+
+ expect(callback).toHaveBeenCalledWith({
+ success: false,
+ error: 'Captions are unavailable when E2EE is required',
+ });
+ expect(addSubscriber).not.toHaveBeenCalled();
+ });
+});
diff --git a/suite/meet/sfu-server/src/server/handlers/SttHandlers.ts b/suite/meet/sfu-server/src/server/handlers/SttHandlers.ts
new file mode 100644
index 0000000000..a940775185
--- /dev/null
+++ b/suite/meet/sfu-server/src/server/handlers/SttHandlers.ts
@@ -0,0 +1,69 @@
+import type { Socket } from 'socket.io';
+import { loggers } from '../../utils/logger';
+import type { HandlerDeps, TypedSocket } from './Handler';
+
+/** Registers per-socket caption subscription controls. */
+export function registerSttHandlers(deps: HandlerDeps) {
+ return (socket: Socket) => {
+ socket.on('stt:toggle', async (data, callback) => {
+ try {
+ deps.authManager.ensureFullAccess(socket);
+ if (!deps.sttManager) {
+ callback({ success: false, error: 'STT is not configured' });
+ return;
+ }
+
+ const typedSocket = socket as TypedSocket;
+ const roomId = typedSocket.roomId;
+ const enabled =
+ typeof data?.enabled === 'boolean' ? data.enabled : false;
+
+ if (!roomId) {
+ callback({ success: false, error: 'Not in a room' });
+ return;
+ }
+ if (enabled && typedSocket.e2eeRequired) {
+ callback({
+ success: false,
+ error: 'Captions are unavailable when E2EE is required',
+ });
+ return;
+ }
+
+ if (enabled) {
+ const wasFirst = deps.sttManager.addSubscriber(roomId, socket.id);
+ callback({ success: true, enabled });
+ if (wasFirst) {
+ void deps.mediasoup
+ .startSttForExistingProducers(roomId, deps.sttManager)
+ .catch((error) => {
+ loggers.socketHandler.warn(
+ 'Failed to start STT for room %s: %s',
+ roomId,
+ (error as Error).message,
+ );
+ });
+ }
+ } else {
+ const wasLast = deps.sttManager.removeSubscriber(roomId, socket.id);
+ callback({ success: true, enabled });
+ if (wasLast) {
+ void deps.sttManager.stopRoom(roomId, true).catch((error) => {
+ loggers.socketHandler.warn(
+ 'Failed to stop STT for room %s: %s',
+ roomId,
+ (error as Error).message,
+ );
+ });
+ }
+ }
+ } catch (error) {
+ loggers.socketHandler.warn(
+ 'stt:toggle failed: %s',
+ (error as Error).message,
+ );
+ callback({ success: false, error: (error as Error).message });
+ }
+ });
+ };
+}
diff --git a/suite/meet/sfu-server/src/stt/AudioIngester.test.ts b/suite/meet/sfu-server/src/stt/AudioIngester.test.ts
new file mode 100644
index 0000000000..4e748fc530
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/AudioIngester.test.ts
@@ -0,0 +1,122 @@
+import type { Producer, Router } from 'mediasoup/types';
+import { describe, expect, it, vi } from 'vitest';
+import { AudioIngester } from './AudioIngester';
+import { AudioPreRoll } from './AudioPreRoll';
+import type { ISttClient, ISttStream } from './SttClient';
+
+const FRAME_BYTES = 4800;
+
+function speechFrame(): Buffer {
+ const frame = Buffer.alloc(FRAME_BYTES);
+ for (let offset = 0; offset < frame.length; offset += 2) {
+ frame.writeInt16LE(16_000, offset);
+ }
+ return frame;
+}
+
+describe('AudioIngester', () => {
+ it('cleans up a transport created after stop begins', async () => {
+ const sttClient = {
+ isAvailable: () => true,
+ onAvailable: vi.fn(),
+ createStream: vi.fn(),
+ } satisfies ISttClient;
+ const ingester = new AudioIngester({
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producer: { id: 'producer-1' } as Producer,
+ router: {} as Router,
+ sttClient,
+ onUnexpectedStreamClose: vi.fn(),
+ onTranscript: vi.fn(),
+ });
+ let finishSetup: () => void = () => {};
+ const transport = { close: vi.fn() };
+ const internals = ingester as unknown as {
+ plainTransport: typeof transport | null;
+ setupPlainTransport(): Promise;
+ createConsumer(): Promise;
+ };
+ vi.spyOn(internals, 'setupPlainTransport').mockImplementation(
+ () =>
+ new Promise((resolve) => {
+ finishSetup = () => {
+ internals.plainTransport = transport;
+ resolve();
+ };
+ }),
+ );
+ const createConsumer = vi.spyOn(internals, 'createConsumer');
+
+ const start = ingester.start();
+ await ingester.stop();
+ finishSetup();
+ await start;
+
+ expect(transport.close).toHaveBeenCalledOnce();
+ expect(createConsumer).not.toHaveBeenCalled();
+ });
+
+ it('drains every complete queued VAD frame in one check', async () => {
+ const stream = {
+ sendAudio: vi.fn(),
+ markFinal: vi.fn(),
+ onUnexpectedClose: vi.fn(),
+ close: vi.fn<() => Promise>().mockResolvedValue(),
+ } satisfies ISttStream;
+ const sttClient = {
+ isAvailable: () => true,
+ onAvailable: vi.fn(),
+ createStream: vi.fn(),
+ } satisfies ISttClient;
+ const ingester = new AudioIngester({
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producer: { id: 'producer-1' } as Producer,
+ router: {} as Router,
+ sttClient,
+ onUnexpectedStreamClose: vi.fn(),
+ onTranscript: vi.fn(),
+ });
+ const silence = Buffer.alloc(FRAME_BYTES);
+ const speech1 = speechFrame();
+ const speechSilence = Buffer.alloc(FRAME_BYTES);
+ const speech2 = speechFrame();
+ const remainder = Buffer.alloc(FRAME_BYTES / 2);
+ const internals = ingester as unknown as {
+ vadQueue: Buffer[];
+ vadQueueBytes: number;
+ sttStream: ISttStream;
+ preRoll: AudioPreRoll;
+ speechCheckCount: number;
+ silenceCheckCount: number;
+ isInSpeech: boolean;
+ streamedBytes: number;
+ runVadCheck(): Promise;
+ shouldFlush(): boolean;
+ };
+ internals.vadQueue = [silence, speech1, speechSilence, speech2, remainder];
+ internals.vadQueueBytes = FRAME_BYTES * 4.5;
+ internals.sttStream = stream;
+ internals.preRoll = new AudioPreRoll(3);
+ const shouldFlush = vi
+ .spyOn(internals, 'shouldFlush')
+ .mockReturnValue(false);
+
+ await internals.runVadCheck();
+
+ expect(stream.sendAudio.mock.calls.map(([frame]) => frame)).toEqual([
+ silence,
+ speech1,
+ speechSilence,
+ speech2,
+ ]);
+ expect(internals.vadQueueBytes).toBe(FRAME_BYTES / 2);
+ expect(internals.vadQueue).toEqual([remainder]);
+ expect(internals.speechCheckCount).toBe(2);
+ expect(internals.silenceCheckCount).toBe(0);
+ expect(internals.isInSpeech).toBe(true);
+ expect(internals.streamedBytes).toBe(FRAME_BYTES * 4);
+ expect(shouldFlush).toHaveBeenCalledTimes(4);
+ });
+});
diff --git a/suite/meet/sfu-server/src/stt/AudioIngester.ts b/suite/meet/sfu-server/src/stt/AudioIngester.ts
new file mode 100644
index 0000000000..063b47ce88
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/AudioIngester.ts
@@ -0,0 +1,582 @@
+import { type ChildProcess, spawn } from 'node:child_process';
+import { randomUUID } from 'node:crypto';
+import dgram from 'node:dgram';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import type {
+ Consumer,
+ PlainTransport,
+ Producer,
+ Router,
+ RtpCapabilities,
+} from 'mediasoup/types';
+import { loggers } from '../utils/logger';
+import { AudioPreRoll } from './AudioPreRoll';
+import { updatePcmCaptureTranscript, writePcmCapture } from './PcmCapture';
+import type { ISttClient, ISttStream } from './SttClient';
+
+interface AudioIngesterOptions {
+ roomId: string;
+ participantId: string;
+ participantName?: string;
+ producer: Producer;
+ router: Router;
+ sttClient: ISttClient;
+ captureDirectory?: string;
+ /** Called before each flush; if false, audio is discarded (active-speaker-only mode) */
+ isActiveSpeaker?: () => boolean;
+ onUnexpectedStreamClose: () => void;
+ onTranscript: (text: string, isFinal: boolean, durationMs: number) => void;
+}
+
+// ── VAD / Streaming Config ───────────────────────────────────────────────────
+const SAMPLE_RATE = 24000;
+const BYTES_PER_SAMPLE = 2; // s16le
+const OUTPUT_CHANNELS = 1; // ASR input is mono; Meet still publishes stereo Opus.
+
+/** How often we check audio energy (ms) */
+const VAD_CHECK_MS = 100;
+/** Bytes of audio per VAD check */
+const BYTES_PER_CHECK = (SAMPLE_RATE * BYTES_PER_SAMPLE * VAD_CHECK_MS) / 1000;
+const PRE_ROLL_CHECKS = Math.max(
+ 0,
+ Math.ceil(
+ Number.parseInt(process.env.STT_PRE_ROLL_MS || '300', 10) / VAD_CHECK_MS,
+ ),
+);
+
+/** Consecutive silent checks before we flush (500 ms pause by default) */
+const SILENCE_CHECKS_TO_FLUSH = Math.max(
+ 1,
+ Math.ceil(
+ Number.parseInt(process.env.STT_SILENCE_MS || '500', 10) / VAD_CHECK_MS,
+ ),
+);
+/** Minimum speech before a normal silence final (600 ms by default). */
+const MIN_SPEECH_CHECKS = Math.max(
+ 1,
+ Math.ceil(
+ Number.parseInt(process.env.STT_MIN_SPEECH_MS || '600', 10) / VAD_CHECK_MS,
+ ),
+);
+/** Min speech for short utterance / tail-end catch-up flush (200 ms by default). */
+const MIN_TAIL_CHECKS = Math.max(
+ 1,
+ Math.ceil(
+ Number.parseInt(process.env.STT_MIN_TAIL_MS || '200', 10) / VAD_CHECK_MS,
+ ),
+);
+/** Silence before finalizing a short utterance (700 ms by default). */
+const SHORT_UTTERANCE_SILENCE_CHECKS = Math.max(
+ SILENCE_CHECKS_TO_FLUSH,
+ Math.ceil(
+ Number.parseInt(process.env.STT_SHORT_UTTERANCE_SILENCE_MS || '700', 10) /
+ VAD_CHECK_MS,
+ ),
+);
+/**
+ * Normalized RMS threshold for speech vs silence.
+ * 0.0 = absolute silence, 1.0 = full-scale square wave.
+ * 0.012 works well for typical mic input routed through Mediasoup.
+ */
+const SPEECH_RMS_THRESHOLD = Number.parseFloat(
+ process.env.STT_VAD_THRESHOLD || '0.012',
+);
+
+/** Captures one producer, decodes its audio, and streams VAD-delimited speech to STT. */
+export class AudioIngester {
+ private roomId: string;
+ private participantId: string;
+ private participantName?: string;
+ private producer: Producer;
+ private router: Router;
+ private sttClient: ISttClient;
+ private sttStream: ISttStream | null = null;
+ private captureDirectory?: string;
+ private captureFrames: Buffer[] = [];
+ private pendingCaptureMetadata: string[] = [];
+ private sessionId = randomUUID();
+ private isActiveSpeaker?: () => boolean;
+ private onUnexpectedStreamClose: () => void;
+ private onTranscript: (
+ text: string,
+ isFinal: boolean,
+ durationMs: number,
+ ) => void;
+
+ private plainTransport: PlainTransport | null = null;
+ private consumer: Consumer | null = null;
+ private ffmpeg: ChildProcess | null = null;
+ private ffmpegPort = 0;
+ private sdpPath = '';
+ private running = false;
+
+ // ── VAD state ──────────────────────────────────────────────────────────────
+ private vadQueue: Buffer[] = [];
+ private vadQueueBytes = 0;
+ private speechCheckCount = 0;
+ private silenceCheckCount = 0;
+ private isInSpeech = false;
+ private vadTimer: NodeJS.Timeout | null = null;
+ private streamedBytes = 0;
+ private preRoll = new AudioPreRoll(PRE_ROLL_CHECKS);
+
+ constructor(options: AudioIngesterOptions) {
+ this.roomId = options.roomId;
+ this.participantId = options.participantId;
+ this.participantName = options.participantName;
+ this.producer = options.producer;
+ this.router = options.router;
+ this.sttClient = options.sttClient;
+ this.captureDirectory = options.captureDirectory;
+ this.isActiveSpeaker = options.isActiveSpeaker;
+ this.onUnexpectedStreamClose = options.onUnexpectedStreamClose;
+ this.onTranscript = options.onTranscript;
+ }
+
+ async start(): Promise {
+ if (this.running) return;
+ this.running = true;
+
+ try {
+ await this.setupPlainTransport();
+ if (!this.running) {
+ await this.stop();
+ return;
+ }
+ await this.createConsumer();
+ if (!this.running) {
+ await this.stop();
+ return;
+ }
+ await this.startFfmpeg();
+ if (!this.running) {
+ await this.stop();
+ return;
+ }
+ await this.plainTransport!.connect({
+ ip: '127.0.0.1',
+ port: this.ffmpegPort,
+ });
+ if (!this.running) {
+ await this.stop();
+ return;
+ }
+ const stream = await this.sttClient.createStream(
+ {
+ sessionId: this.sessionId,
+ roomId: this.roomId,
+ participantId: this.participantId,
+ producerId: this.producer.id,
+ participantName: this.participantName,
+ sampleRate: SAMPLE_RATE,
+ language: process.env.NEMOTRON_LANGUAGE || 'en-US',
+ },
+ (event) => {
+ if (event.isFinal) this.recordCaptureTranscript(event.text);
+ this.onTranscript(event.text, event.isFinal, event.durationMs);
+ },
+ );
+ if (!this.running) {
+ await stream.close();
+ await this.stop();
+ return;
+ }
+ this.sttStream = stream;
+ stream.onUnexpectedClose(() => {
+ if (!this.running || this.sttStream !== stream) return;
+ this.onUnexpectedStreamClose();
+ });
+ if (!this.running || this.sttStream !== stream) return;
+ this.startVadLoop();
+
+ loggers.stt.info(
+ 'AudioIngester started for %s in room %s (producer %s, session %s, ffmpeg port %d, vadThreshold=%.4f)',
+ this.participantId,
+ this.roomId,
+ this.producer.id,
+ this.sessionId,
+ this.ffmpegPort,
+ SPEECH_RMS_THRESHOLD,
+ );
+ } catch (error) {
+ await this.stop();
+ loggers.stt.error(
+ 'Failed to start AudioIngester for %s: %s',
+ this.participantId,
+ (error as Error).message,
+ );
+ throw error;
+ }
+ }
+
+ async stop(): Promise {
+ this.running = false;
+
+ if (this.vadTimer) {
+ clearTimeout(this.vadTimer);
+ this.vadTimer = null;
+ }
+
+ if (this.streamedBytes > 0 && this.speechCheckCount >= MIN_TAIL_CHECKS) {
+ this.markFinal();
+ }
+
+ const stream = this.sttStream;
+ this.sttStream = null;
+ if (stream) await stream.close();
+
+ const consumer = this.consumer;
+ this.consumer = null;
+ if (consumer) {
+ try {
+ consumer.close();
+ } catch {
+ /* ignore */
+ }
+ }
+ const plainTransport = this.plainTransport;
+ this.plainTransport = null;
+ if (plainTransport) {
+ try {
+ plainTransport.close();
+ } catch {
+ /* ignore */
+ }
+ }
+ const ffmpeg = this.ffmpeg;
+ this.ffmpeg = null;
+ if (ffmpeg && !ffmpeg.killed) {
+ ffmpeg.kill('SIGTERM');
+ setTimeout(() => {
+ if (ffmpeg.exitCode === null && ffmpeg.signalCode === null) {
+ ffmpeg.kill('SIGKILL');
+ }
+ }, 1000);
+ }
+ const sdpPath = this.sdpPath;
+ this.sdpPath = '';
+ if (sdpPath) {
+ try {
+ fs.unlinkSync(sdpPath);
+ } catch {
+ /* ignore */
+ }
+ }
+
+ loggers.stt.info('AudioIngester stopped for %s', this.participantId);
+ }
+
+ // ── Mediasoup plumbing ─────────────────────────────────────────────────────
+
+ private async setupPlainTransport(): Promise {
+ this.plainTransport = await this.router.createPlainTransport({
+ listenInfo: { protocol: 'udp', ip: '127.0.0.1' },
+ rtcpMux: true,
+ comedia: false,
+ });
+ }
+
+ private async createConsumer(): Promise {
+ const rtpCapabilities: RtpCapabilities = {
+ codecs: [
+ {
+ mimeType: 'audio/opus',
+ kind: 'audio',
+ preferredPayloadType: 111,
+ clockRate: 48000,
+ channels: 2,
+ parameters: {},
+ rtcpFeedback: [],
+ },
+ ],
+ headerExtensions: [],
+ };
+
+ this.consumer = await this.plainTransport!.consume({
+ producerId: this.producer.id,
+ rtpCapabilities,
+ });
+ }
+
+ private async startFfmpeg(): Promise {
+ this.ffmpegPort = await this.findAvailablePort();
+ const payloadType =
+ this.consumer?.rtpParameters?.codecs?.[0]?.payloadType ?? 111;
+
+ this.sdpPath = path.join(
+ os.tmpdir(),
+ `stt_${this.roomId}_${this.participantId}_${Date.now()}.sdp`,
+ );
+ fs.writeFileSync(this.sdpPath, this.buildSdp(this.ffmpegPort, payloadType));
+
+ const args = [
+ '-protocol_whitelist',
+ 'file,crypto,udp,rtp',
+ '-i',
+ this.sdpPath,
+ '-f',
+ 's16le',
+ '-ar',
+ String(SAMPLE_RATE),
+ '-ac',
+ String(OUTPUT_CHANNELS),
+ 'pipe:1',
+ ];
+
+ this.ffmpeg = spawn('ffmpeg', args, {
+ stdio: ['ignore', 'pipe', 'pipe'],
+ });
+
+ this.ffmpeg.stdout!.on('data', (data: Buffer) => {
+ this.vadQueue.push(data);
+ this.vadQueueBytes += data.length;
+ });
+
+ this.ffmpeg.stderr!.on('data', (data: Buffer) => {
+ const msg = data.toString().trim();
+ if (msg && process.env.SFU_LOG_LEVEL === 'debug') {
+ loggers.stt.debug('ffmpeg: %s', msg.slice(0, 200));
+ }
+ });
+
+ this.ffmpeg.on('error', (error) => {
+ loggers.stt.error(
+ 'ffmpeg error for %s: %s',
+ this.participantId,
+ error.message,
+ );
+ });
+
+ this.ffmpeg.on('exit', (code) => {
+ if (code !== 0 && this.running) {
+ loggers.stt.warn(
+ 'ffmpeg exited with code %d for %s',
+ code,
+ this.participantId,
+ );
+ }
+ });
+ }
+
+ // ── VAD loop ───────────────────────────────────────────────────────────────
+
+ private startVadLoop(): void {
+ const run = () => {
+ if (!this.running) return;
+ this.runVadCheck()
+ .then(() => {
+ if (this.running) {
+ this.vadTimer = setTimeout(run, VAD_CHECK_MS);
+ }
+ })
+ .catch((error) => {
+ loggers.stt.error('VAD check error: %s', (error as Error).message);
+ if (this.running) {
+ this.vadTimer = setTimeout(run, VAD_CHECK_MS);
+ }
+ });
+ };
+ run();
+ }
+
+ private async runVadCheck(): Promise {
+ while (this.vadQueueBytes >= BYTES_PER_CHECK) {
+ const frame = this.dequeueBytes(BYTES_PER_CHECK);
+ const frameSumSq = this.calculateSumSq(frame);
+ const rms =
+ Math.sqrt(frameSumSq / (BYTES_PER_CHECK / BYTES_PER_SAMPLE)) / 32768;
+ const isSpeech = rms > SPEECH_RMS_THRESHOLD;
+
+ if (isSpeech) {
+ if (!this.isInSpeech) {
+ for (const preRollFrame of this.preRoll.drain()) {
+ this.sendFrame(preRollFrame);
+ }
+ }
+ this.silenceCheckCount = 0;
+ this.speechCheckCount++;
+ this.isInSpeech = true;
+ this.sendFrame(frame);
+ } else {
+ this.silenceCheckCount++;
+ if (this.isInSpeech) {
+ this.sendFrame(frame);
+ } else {
+ this.preRoll.remember(frame);
+ }
+ }
+
+ if (this.shouldFlush()) {
+ this.markFinal();
+ }
+ }
+ }
+
+ private shouldFlush(): boolean {
+ // Flush on silence after enough speech
+ if (
+ this.isInSpeech &&
+ this.silenceCheckCount >= SILENCE_CHECKS_TO_FLUSH &&
+ this.speechCheckCount >= MIN_SPEECH_CHECKS
+ ) {
+ return true;
+ }
+ // Extended silence: flush whatever audio we have, even short utterances.
+ // Catches trailing words that didn't reach MIN_SPEECH_CHECKS.
+ if (
+ this.isInSpeech &&
+ this.silenceCheckCount >= SHORT_UTTERANCE_SILENCE_CHECKS &&
+ this.speechCheckCount >= MIN_TAIL_CHECKS
+ ) {
+ return true;
+ }
+ return false;
+ }
+
+ private sendFrame(frame: Buffer): void {
+ if (this.isActiveSpeaker && !this.isActiveSpeaker()) {
+ loggers.stt.debug(
+ 'Speaker %s not active, discarding frame',
+ this.participantId,
+ );
+ return;
+ }
+ this.sttStream?.sendAudio(frame);
+ if (this.captureDirectory) this.captureFrames.push(Buffer.from(frame));
+ this.streamedBytes += frame.length;
+ }
+
+ private markFinal(): void {
+ if (this.streamedBytes < BYTES_PER_CHECK * MIN_TAIL_CHECKS) {
+ this.resetVadState();
+ return;
+ }
+ const durationMs =
+ (this.streamedBytes / BYTES_PER_SAMPLE / SAMPLE_RATE) * 1000;
+ loggers.stt.debug(
+ 'Marking final %d ms (%d checks) for %s session %s',
+ durationMs.toFixed(0),
+ this.speechCheckCount,
+ this.participantId,
+ this.sessionId,
+ );
+ this.writeCapture(durationMs);
+ this.sttStream?.markFinal(durationMs);
+ this.resetVadState();
+ }
+
+ private resetVadState(): void {
+ this.speechCheckCount = 0;
+ this.silenceCheckCount = 0;
+ this.isInSpeech = false;
+ this.streamedBytes = 0;
+ this.captureFrames = [];
+ this.preRoll.clear();
+ }
+
+ private writeCapture(durationMs: number): void {
+ if (!this.captureDirectory || this.captureFrames.length === 0) return;
+ try {
+ const metadataPath = writePcmCapture(
+ this.captureDirectory,
+ Buffer.concat(this.captureFrames),
+ {
+ sessionId: this.sessionId,
+ roomId: this.roomId,
+ participantId: this.participantId,
+ producerId: this.producer.id,
+ sampleRate: SAMPLE_RATE,
+ channels: OUTPUT_CHANNELS,
+ durationMs,
+ },
+ );
+ this.pendingCaptureMetadata.push(metadataPath);
+ loggers.stt.info('Captured STT utterance at %s', metadataPath);
+ } catch (error) {
+ loggers.stt.warn(
+ 'Failed to capture STT utterance: %s',
+ (error as Error).message,
+ );
+ }
+ }
+
+ private recordCaptureTranscript(transcript: string): void {
+ const metadataPath = this.pendingCaptureMetadata.shift();
+ if (!metadataPath) return;
+ try {
+ updatePcmCaptureTranscript(metadataPath, transcript);
+ } catch (error) {
+ loggers.stt.warn(
+ 'Failed to update STT capture transcript: %s',
+ (error as Error).message,
+ );
+ }
+ }
+
+ // ── Helpers ────────────────────────────────────────────────────────────────
+
+ private calculateSumSq(buffer: Buffer): number {
+ let sum = 0;
+ for (let i = 0; i < buffer.length; i += BYTES_PER_SAMPLE) {
+ const sample = buffer.readInt16LE(i);
+ sum += sample * sample;
+ }
+ return sum;
+ }
+
+ /**
+ * Read exactly `n` bytes from the front of the vad queue.
+ * Handles partial buffers by splitting/consuming from the head.
+ */
+ private dequeueBytes(n: number): Buffer {
+ const out = Buffer.alloc(n);
+ let written = 0;
+
+ while (written < n && this.vadQueue.length > 0) {
+ const head = this.vadQueue[0];
+ const remaining = n - written;
+
+ if (head.length <= remaining) {
+ head.copy(out, written);
+ written += head.length;
+ this.vadQueue.shift();
+ this.vadQueueBytes -= head.length;
+ } else {
+ head.copy(out, written, 0, remaining);
+ this.vadQueue[0] = head.subarray(remaining);
+ this.vadQueueBytes -= remaining;
+ written += remaining;
+ }
+ }
+
+ return out;
+ }
+
+ private buildSdp(port: number, payloadType: number): string {
+ return [
+ 'v=0',
+ 'o=- 0 0 IN IP4 127.0.0.1',
+ 's=STT',
+ 'c=IN IP4 127.0.0.1',
+ 't=0 0',
+ `m=audio ${port} RTP/AVP ${payloadType}`,
+ `a=rtpmap:${payloadType} opus/48000/2`,
+ '',
+ ].join('\n');
+ }
+
+ private findAvailablePort(): Promise {
+ return new Promise((resolve, reject) => {
+ const socket = dgram.createSocket('udp4');
+ socket.bind(0, '127.0.0.1', () => {
+ const address = socket.address();
+ socket.close(() => {
+ resolve(address.port);
+ });
+ });
+ socket.on('error', reject);
+ });
+ }
+}
diff --git a/suite/meet/sfu-server/src/stt/AudioPreRoll.test.ts b/suite/meet/sfu-server/src/stt/AudioPreRoll.test.ts
new file mode 100644
index 0000000000..69ac50e642
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/AudioPreRoll.test.ts
@@ -0,0 +1,18 @@
+import { describe, expect, it } from 'vitest';
+import { AudioPreRoll } from './AudioPreRoll';
+
+describe('AudioPreRoll', () => {
+ it('retains only the frames immediately before speech begins', () => {
+ const preRoll = new AudioPreRoll(3);
+ for (let value = 1; value <= 5; value++) {
+ preRoll.remember(Buffer.from([value]));
+ }
+
+ expect(preRoll.drain()).toEqual([
+ Buffer.from([3]),
+ Buffer.from([4]),
+ Buffer.from([5]),
+ ]);
+ expect(preRoll.drain()).toEqual([]);
+ });
+});
diff --git a/suite/meet/sfu-server/src/stt/AudioPreRoll.ts b/suite/meet/sfu-server/src/stt/AudioPreRoll.ts
new file mode 100644
index 0000000000..3799777d21
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/AudioPreRoll.ts
@@ -0,0 +1,21 @@
+export class AudioPreRoll {
+ private frames: Buffer[] = [];
+
+ constructor(private maxFrames: number) {}
+
+ remember(frame: Buffer): void {
+ if (this.maxFrames === 0) return;
+ this.frames.push(frame);
+ if (this.frames.length > this.maxFrames) this.frames.shift();
+ }
+
+ drain(): Buffer[] {
+ const frames = this.frames;
+ this.frames = [];
+ return frames;
+ }
+
+ clear(): void {
+ this.frames = [];
+ }
+}
diff --git a/suite/meet/sfu-server/src/stt/PcmCapture.test.ts b/suite/meet/sfu-server/src/stt/PcmCapture.test.ts
new file mode 100644
index 0000000000..34014c26bc
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/PcmCapture.test.ts
@@ -0,0 +1,57 @@
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { afterEach, describe, expect, it } from 'vitest';
+import {
+ encodePcm16Wav,
+ type PcmCaptureMetadata,
+ updatePcmCaptureTranscript,
+ writePcmCapture,
+} from './PcmCapture';
+
+describe('PCM capture', () => {
+ let directory: string | undefined;
+
+ afterEach(() => {
+ if (directory) fs.rmSync(directory, { recursive: true, force: true });
+ });
+
+ it('encodes exact PCM bytes in a 24 kHz mono WAV', () => {
+ const pcm = Buffer.from([0, 0, 1, 0, 255, 255]);
+ const wav = encodePcm16Wav(pcm, 24000);
+
+ expect(wav.toString('ascii', 0, 4)).toBe('RIFF');
+ expect(wav.toString('ascii', 8, 12)).toBe('WAVE');
+ expect(wav.readUInt16LE(22)).toBe(1);
+ expect(wav.readUInt32LE(24)).toBe(24000);
+ expect(wav.readUInt16LE(34)).toBe(16);
+ expect(wav.subarray(44)).toEqual(pcm);
+ });
+
+ it('writes metadata and records the final transcript', () => {
+ directory = fs.mkdtempSync(path.join(os.tmpdir(), 'stt-capture-'));
+ const metadataPath = writePcmCapture(directory, Buffer.alloc(4800), {
+ sessionId: 'session-1',
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producerId: 'producer-1',
+ sampleRate: 24000,
+ channels: 1,
+ durationMs: 100,
+ });
+ updatePcmCaptureTranscript(metadataPath, 'hello world');
+
+ const metadata = JSON.parse(
+ fs.readFileSync(metadataPath, 'utf8'),
+ ) as PcmCaptureMetadata;
+ expect(metadata).toMatchObject({
+ sessionId: 'session-1',
+ sampleRate: 24000,
+ durationMs: 100,
+ transcript: 'hello world',
+ });
+ expect(fs.readFileSync(path.join(directory, metadata.wavFile)).length).toBe(
+ 4844,
+ );
+ });
+});
diff --git a/suite/meet/sfu-server/src/stt/PcmCapture.ts b/suite/meet/sfu-server/src/stt/PcmCapture.ts
new file mode 100644
index 0000000000..9a8da04573
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/PcmCapture.ts
@@ -0,0 +1,71 @@
+import { randomUUID } from 'node:crypto';
+import fs from 'node:fs';
+import path from 'node:path';
+
+export interface PcmCaptureMetadata {
+ sessionId: string;
+ roomId: string;
+ participantId: string;
+ producerId: string;
+ sampleRate: number;
+ channels: number;
+ durationMs: number;
+ capturedAt: string;
+ wavFile: string;
+ transcript?: string;
+}
+
+export function encodePcm16Wav(
+ pcm: Buffer,
+ sampleRate: number,
+ channels = 1,
+): Buffer {
+ const header = Buffer.alloc(44);
+ header.write('RIFF', 0);
+ header.writeUInt32LE(36 + pcm.length, 4);
+ header.write('WAVE', 8);
+ header.write('fmt ', 12);
+ header.writeUInt32LE(16, 16);
+ header.writeUInt16LE(1, 20);
+ header.writeUInt16LE(channels, 22);
+ header.writeUInt32LE(sampleRate, 24);
+ header.writeUInt32LE(sampleRate * channels * 2, 28);
+ header.writeUInt16LE(channels * 2, 32);
+ header.writeUInt16LE(16, 34);
+ header.write('data', 36);
+ header.writeUInt32LE(pcm.length, 40);
+ return Buffer.concat([header, pcm]);
+}
+
+export function writePcmCapture(
+ directory: string,
+ pcm: Buffer,
+ metadata: Omit,
+): string {
+ fs.mkdirSync(directory, { recursive: true });
+ const captureId = `${Date.now()}-${metadata.sessionId}-${randomUUID()}`;
+ const wavFile = `${captureId}.wav`;
+ const metadataPath = path.join(directory, `${captureId}.json`);
+ fs.writeFileSync(
+ path.join(directory, wavFile),
+ encodePcm16Wav(pcm, metadata.sampleRate, metadata.channels),
+ );
+ fs.writeFileSync(
+ metadataPath,
+ `${JSON.stringify({ ...metadata, capturedAt: new Date().toISOString(), wavFile }, null, 2)}\n`,
+ );
+ return metadataPath;
+}
+
+export function updatePcmCaptureTranscript(
+ metadataPath: string,
+ transcript: string,
+): void {
+ const metadata = JSON.parse(
+ fs.readFileSync(metadataPath, 'utf8'),
+ ) as PcmCaptureMetadata;
+ fs.writeFileSync(
+ metadataPath,
+ `${JSON.stringify({ ...metadata, transcript }, null, 2)}\n`,
+ );
+}
diff --git a/suite/meet/sfu-server/src/stt/SttClient.test.ts b/suite/meet/sfu-server/src/stt/SttClient.test.ts
new file mode 100644
index 0000000000..b65b0eb468
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/SttClient.test.ts
@@ -0,0 +1,259 @@
+import { createServer, type Server } from 'node:http';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { type WebSocket, WebSocketServer } from 'ws';
+import { SttClient, type SttTranscriptEvent } from './SttClient';
+
+interface ClientEvent {
+ type?: string;
+ audio?: string;
+ session?: {
+ type?: string;
+ audio?: { input?: { format?: { type?: string; rate?: number } } };
+ };
+}
+
+describe('SttClient Realtime protocol', () => {
+ let server: Server | undefined;
+ let websocketServer: WebSocketServer | undefined;
+ let client: SttClient | undefined;
+
+ afterEach(async () => {
+ client?.destroy();
+ vi.restoreAllMocks();
+ await new Promise(
+ (resolve) => websocketServer?.close(() => resolve()) ?? resolve(),
+ );
+ await new Promise(
+ (resolve) => server?.close(() => resolve()) ?? resolve(),
+ );
+ });
+
+ it('configures a transcription session and maps committed item events to Meet transcripts', async () => {
+ server = createServer((_request, response) => {
+ response.writeHead(200, { 'Content-Type': 'application/json' });
+ response.end('{"status":"ok"}');
+ });
+ websocketServer = new WebSocketServer({ server, path: '/v1/realtime' });
+ await new Promise((resolve) =>
+ server!.listen(0, '127.0.0.1', resolve),
+ );
+ const address = server.address();
+ if (!address || typeof address === 'string')
+ throw new Error('Missing test server address');
+
+ const clientEvents: ClientEvent[] = [];
+ websocketServer.on('connection', (socket) => {
+ socket.send(
+ JSON.stringify({
+ type: 'session.created',
+ event_id: 'event-created',
+ session: { id: 'sess-1', type: 'transcription' },
+ }),
+ );
+ socket.on('message', (raw) => {
+ const event = JSON.parse(raw.toString()) as ClientEvent;
+ clientEvents.push(event);
+ if (event.type === 'session.update') {
+ socket.send(
+ JSON.stringify({
+ type: 'session.updated',
+ event_id: 'event-updated',
+ session: { id: 'sess-1', type: 'transcription' },
+ }),
+ );
+ }
+ if (event.type === 'input_audio_buffer.commit') {
+ socket.send(
+ JSON.stringify({
+ type: 'input_audio_buffer.committed',
+ event_id: 'event-committed',
+ item_id: 'item-1',
+ previous_item_id: null,
+ }),
+ );
+ socket.send(
+ JSON.stringify({
+ type: 'conversation.item.input_audio_transcription.delta',
+ event_id: 'event-delta',
+ item_id: 'item-1',
+ content_index: 0,
+ delta: 'hello',
+ }),
+ );
+ socket.send(
+ JSON.stringify({
+ type: 'conversation.item.input_audio_transcription.completed',
+ event_id: 'event-completed',
+ item_id: 'item-1',
+ content_index: 0,
+ transcript: 'hello world',
+ usage: { type: 'duration', seconds: 0.1 },
+ }),
+ );
+ }
+ });
+ });
+
+ client = new SttClient(`http://127.0.0.1:${address.port}`);
+ const transcripts: SttTranscriptEvent[] = [];
+ const stream = await client.createStream(
+ {
+ sessionId: 'meet-session-1',
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producerId: 'producer-1',
+ sampleRate: 24000,
+ language: 'en-US',
+ },
+ (event) => transcripts.push(event),
+ );
+ const unexpectedClose = vi.fn();
+ stream.onUnexpectedClose(unexpectedClose);
+
+ stream.sendAudio(Buffer.from([0, 0, 1, 0]));
+ stream.markFinal(100);
+ await stream.close();
+
+ const update = clientEvents.find(
+ (event) => event.type === 'session.update',
+ );
+ const append = clientEvents.find(
+ (event) => event.type === 'input_audio_buffer.append',
+ );
+ expect(update).toMatchObject({
+ session: {
+ type: 'transcription',
+ audio: { input: { format: { type: 'audio/pcm', rate: 24000 } } },
+ },
+ });
+ expect(append).toMatchObject({
+ audio: Buffer.from([0, 0, 1, 0]).toString('base64'),
+ });
+ expect(transcripts).toEqual([
+ { text: 'hello', isFinal: false, durationMs: 100, sequence: 1 },
+ { text: 'hello world', isFinal: true, durationMs: 100, sequence: 2 },
+ ]);
+ expect(unexpectedClose).not.toHaveBeenCalled();
+ });
+
+ it('reports a configured Realtime stream closing unexpectedly', async () => {
+ server = createServer((_request, response) => {
+ response.writeHead(200, { 'Content-Type': 'application/json' });
+ response.end('{"status":"ok"}');
+ });
+ websocketServer = new WebSocketServer({ server, path: '/v1/realtime' });
+ await new Promise((resolve) =>
+ server!.listen(0, '127.0.0.1', resolve),
+ );
+ const address = server.address();
+ if (!address || typeof address === 'string')
+ throw new Error('Missing test server address');
+
+ let serverSocket: WebSocket | undefined;
+ websocketServer.on('connection', (socket) => {
+ serverSocket = socket;
+ socket.send(JSON.stringify({ type: 'session.created' }));
+ socket.on('message', (raw) => {
+ const event = JSON.parse(raw.toString()) as ClientEvent;
+ if (event.type === 'session.update') {
+ socket.send(JSON.stringify({ type: 'session.updated' }));
+ }
+ });
+ });
+
+ client = new SttClient(`http://127.0.0.1:${address.port}`);
+ const stream = await client.createStream(
+ {
+ sessionId: 'meet-session-1',
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producerId: 'producer-1',
+ sampleRate: 24000,
+ },
+ vi.fn(),
+ );
+ const unexpectedClose = vi.fn();
+ stream.onUnexpectedClose(unexpectedClose);
+
+ serverSocket?.close(1011, 'backend failure');
+ await vi.waitFor(() => expect(unexpectedClose).toHaveBeenCalledTimes(1));
+ await stream.close();
+ expect(unexpectedClose).toHaveBeenCalledTimes(1);
+ });
+
+ it('delivers an unexpected close that occurs before listener registration', async () => {
+ server = createServer((_request, response) => {
+ response.writeHead(200, { 'Content-Type': 'application/json' });
+ response.end('{"status":"ok"}');
+ });
+ websocketServer = new WebSocketServer({ server, path: '/v1/realtime' });
+ await new Promise((resolve) =>
+ server!.listen(0, '127.0.0.1', resolve),
+ );
+ const address = server.address();
+ if (!address || typeof address === 'string')
+ throw new Error('Missing test server address');
+
+ websocketServer.on('connection', (socket) => {
+ socket.send(JSON.stringify({ type: 'session.created' }));
+ socket.on('message', (raw) => {
+ const event = JSON.parse(raw.toString()) as ClientEvent;
+ if (event.type === 'session.update') {
+ socket.send(JSON.stringify({ type: 'session.updated' }), () => {
+ socket.close(1011, 'backend failure');
+ });
+ }
+ });
+ });
+
+ client = new SttClient(`http://127.0.0.1:${address.port}`);
+ const stream = await client.createStream(
+ {
+ sessionId: 'meet-session-1',
+ roomId: 'room-1',
+ participantId: 'participant-1',
+ producerId: 'producer-1',
+ sampleRate: 24000,
+ },
+ vi.fn(),
+ );
+ await vi.waitFor(() => {
+ const internals = stream as unknown as { unexpectedlyClosed: boolean };
+ expect(internals.unexpectedlyClosed).toBe(true);
+ });
+ const unexpectedClose = vi.fn();
+
+ stream.onUnexpectedClose(unexpectedClose);
+
+ expect(unexpectedClose).toHaveBeenCalledTimes(1);
+ await stream.close();
+ });
+
+ it('notifies after each unhealthy-to-healthy recovery', async () => {
+ const fetchMock = vi
+ .spyOn(globalThis, 'fetch')
+ .mockResolvedValueOnce({ ok: false, status: 503 } as Response)
+ .mockResolvedValueOnce({ ok: true } as Response)
+ .mockResolvedValueOnce({ ok: false, status: 503 } as Response)
+ .mockResolvedValueOnce({ ok: true } as Response);
+ client = new SttClient('http://stt.example');
+ const recovered = vi.fn();
+ client.onAvailable(recovered);
+ const internals = client as unknown as {
+ checkHealth: () => void;
+ healthCheckInFlight: boolean;
+ };
+
+ await vi.waitFor(() => expect(internals.healthCheckInFlight).toBe(false));
+ internals.checkHealth();
+ await vi.waitFor(() => expect(recovered).toHaveBeenCalledTimes(1));
+ await vi.waitFor(() => expect(internals.healthCheckInFlight).toBe(false));
+ internals.checkHealth();
+ await vi.waitFor(() => expect(internals.healthCheckInFlight).toBe(false));
+ expect(client.isAvailable()).toBe(false);
+ internals.checkHealth();
+ await vi.waitFor(() => expect(recovered).toHaveBeenCalledTimes(2));
+
+ expect(fetchMock).toHaveBeenCalledTimes(4);
+ });
+});
diff --git a/suite/meet/sfu-server/src/stt/SttClient.ts b/suite/meet/sfu-server/src/stt/SttClient.ts
new file mode 100644
index 0000000000..ef179f9e95
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/SttClient.ts
@@ -0,0 +1,430 @@
+import WebSocket from 'ws';
+import { loggers } from '../utils/logger';
+
+export interface SttStreamMetadata {
+ sessionId: string;
+ roomId: string;
+ participantId: string;
+ producerId: string;
+ participantName?: string;
+ sampleRate: number;
+ language?: string;
+}
+
+export interface SttTranscriptEvent {
+ text: string;
+ isFinal: boolean;
+ durationMs: number;
+ sequence: number;
+}
+
+export interface ISttStream {
+ sendAudio(frame: Buffer): void;
+ markFinal(durationMs: number): void;
+ onUnexpectedClose(listener: () => void): void;
+ close(): Promise;
+}
+
+export interface ISttClient {
+ createStream(
+ metadata: SttStreamMetadata,
+ onTranscript: (event: SttTranscriptEvent) => void,
+ ): Promise;
+ isAvailable(): boolean;
+ onAvailable(listener: () => void): void;
+}
+
+interface RealtimeServerMessage {
+ type?: string;
+ item_id?: string;
+ delta?: string;
+ transcript?: string;
+ error?: { message?: string };
+}
+
+export class SttClient implements ISttClient {
+ private serverUrl: string;
+ private available = false;
+ private healthCheckInFlight = false;
+ private healthCheckTimer: NodeJS.Timeout | null = null;
+ private availableListeners = new Set<() => void>();
+ private readonly healthCheckIntervalMs = 10_000;
+
+ constructor(serverUrl: string) {
+ this.serverUrl = serverUrl.replace(/\/$/, '');
+ this.checkHealth();
+ this.startHealthCheckLoop();
+ }
+
+ private startHealthCheckLoop(): void {
+ this.healthCheckTimer = setInterval(() => {
+ this.checkHealth();
+ }, this.healthCheckIntervalMs);
+ }
+
+ private checkHealth(): void {
+ if (this.healthCheckInFlight) return;
+ this.healthCheckInFlight = true;
+ fetch(`${this.serverUrl}/health`)
+ .then((res) => {
+ if (res.ok) {
+ const recovered = !this.available;
+ this.available = true;
+ loggers.stt.info('STT server reachable at %s', this.serverUrl);
+ if (recovered) {
+ for (const listener of this.availableListeners) listener();
+ }
+ } else {
+ this.available = false;
+ loggers.stt.warn(
+ 'STT server health check failed (status %d)',
+ res.status,
+ );
+ }
+ })
+ .catch((err) => {
+ this.available = false;
+ loggers.stt.debug(
+ 'STT server unreachable at %s: %s',
+ this.serverUrl,
+ err.message,
+ );
+ })
+ .finally(() => {
+ this.healthCheckInFlight = false;
+ });
+ }
+
+ destroy(): void {
+ if (this.healthCheckTimer) clearInterval(this.healthCheckTimer);
+ this.healthCheckTimer = null;
+ }
+
+ isAvailable(): boolean {
+ return this.available;
+ }
+
+ onAvailable(listener: () => void): void {
+ this.availableListeners.add(listener);
+ }
+
+ async createStream(
+ metadata: SttStreamMetadata,
+ onTranscript: (event: SttTranscriptEvent) => void,
+ ): Promise {
+ const socket = new WebSocket(this.getStreamUrl());
+ const stream = new SttStream(socket, metadata, onTranscript);
+ try {
+ await stream.connect();
+ return stream;
+ } catch (error) {
+ this.available = false;
+ await stream.close();
+ throw error;
+ }
+ }
+
+ private getStreamUrl(): string {
+ const wsBase = this.serverUrl
+ .replace(/^http:/, 'ws:')
+ .replace(/^https:/, 'wss:');
+ return `${wsBase}/v1/realtime`;
+ }
+}
+
+class SttStream implements ISttStream {
+ private sequence = 0;
+ private bufferedBytes = 0;
+ private pendingCommits = 0;
+ private pendingDurations: number[] = [];
+ private durationByItem = new Map();
+ private textByItem = new Map();
+ private pendingWaiters = new Set<() => void>();
+ private readyResolve: (() => void) | null = null;
+ private readyReject: ((error: Error) => void) | null = null;
+ private ready = false;
+ private closeRequested = false;
+ private unexpectedlyClosed = false;
+ private unexpectedCloseDelivered = false;
+ private unexpectedCloseListener: (() => void) | null = null;
+
+ constructor(
+ private socket: WebSocket,
+ private metadata: SttStreamMetadata,
+ private onTranscript: (event: SttTranscriptEvent) => void,
+ ) {
+ this.socket.on('message', (data) => this.handleMessage(data.toString()));
+ this.socket.on('error', (error) => this.readyReject?.(error));
+ this.socket.on('close', (code, reason) => {
+ const wasReady = this.ready;
+ this.ready = false;
+ this.readyReject?.(
+ new Error(
+ `STT stream closed before setup (${code}: ${reason.toString()})`,
+ ),
+ );
+ this.resolvePendingWaiters();
+ if (wasReady && !this.closeRequested) {
+ this.unexpectedlyClosed = true;
+ this.deliverUnexpectedClose();
+ }
+ loggers.stt.debug(
+ 'STT stream closed for %s (code=%d, reason=%s)',
+ this.metadata.sessionId,
+ code,
+ reason.toString(),
+ );
+ });
+ }
+
+ connect(): Promise {
+ return new Promise((resolve, reject) => {
+ const timer = setTimeout(
+ () => reject(new Error('Timed out configuring STT Realtime session')),
+ 5000,
+ );
+ this.readyResolve = () => {
+ clearTimeout(timer);
+ this.ready = true;
+ resolve();
+ };
+ this.readyReject = (error) => {
+ clearTimeout(timer);
+ reject(error);
+ };
+ });
+ }
+
+ sendAudio(frame: Buffer): void {
+ if (!this.ready || this.socket.readyState !== WebSocket.OPEN) return;
+ this.bufferedBytes += frame.length;
+ this.sendEvent({
+ type: 'input_audio_buffer.append',
+ audio: frame.toString('base64'),
+ });
+ }
+
+ markFinal(durationMs: number): void {
+ if (
+ !this.ready ||
+ this.socket.readyState !== WebSocket.OPEN ||
+ this.bufferedBytes === 0
+ )
+ return;
+ this.pendingDurations.push(durationMs);
+ this.pendingCommits++;
+ this.bufferedBytes = 0;
+ this.sendEvent({ type: 'input_audio_buffer.commit' });
+ }
+
+ onUnexpectedClose(listener: () => void): void {
+ this.unexpectedCloseListener = listener;
+ this.deliverUnexpectedClose();
+ }
+
+ async close(): Promise {
+ this.closeRequested = true;
+ if (this.isSocketClosed()) return;
+ await this.waitForPendingCommits();
+ if (this.isSocketClosed()) return;
+ await new Promise((resolve) => {
+ this.socket.once('close', () => resolve());
+ this.socket.close();
+ });
+ }
+
+ private handleMessage(raw: string): void {
+ let message: RealtimeServerMessage;
+ try {
+ message = JSON.parse(raw) as RealtimeServerMessage;
+ } catch {
+ loggers.stt.warn('Dropping malformed STT Realtime message');
+ return;
+ }
+
+ if (message.type === 'session.created') {
+ this.sendEvent({
+ type: 'session.update',
+ session: {
+ type: 'transcription',
+ audio: {
+ input: {
+ format: { type: 'audio/pcm', rate: this.metadata.sampleRate },
+ transcription: {
+ model:
+ process.env.NEMOTRON_MODEL ||
+ 'nemotron-3.5-asr-streaming-0.6b',
+ language: this.metadata.language || 'en-US',
+ },
+ turn_detection: null,
+ },
+ },
+ },
+ });
+ return;
+ }
+ if (message.type === 'session.updated') {
+ this.readyResolve?.();
+ this.readyResolve = null;
+ this.readyReject = null;
+ return;
+ }
+ if (message.type === 'error') {
+ const error = new Error(message.error?.message || 'STT Realtime error');
+ if (!this.ready) this.readyReject?.(error);
+ else loggers.stt.warn('%s', error.message);
+ return;
+ }
+
+ const itemId = message.item_id;
+ if (!itemId) return;
+ if (message.type === 'input_audio_buffer.committed') {
+ this.durationByItem.set(itemId, this.pendingDurations.shift() || 0);
+ return;
+ }
+ if (message.type === 'conversation.item.input_audio_transcription.delta') {
+ const text = `${this.textByItem.get(itemId) || ''}${message.delta || ''}`;
+ this.textByItem.set(itemId, text);
+ if (text.trim())
+ this.emitTranscript(
+ text.trim(),
+ false,
+ this.durationByItem.get(itemId) || 0,
+ );
+ return;
+ }
+ if (
+ message.type === 'conversation.item.input_audio_transcription.completed'
+ ) {
+ this.emitTranscript(
+ (message.transcript || this.textByItem.get(itemId) || '').trim(),
+ true,
+ this.durationByItem.get(itemId) || 0,
+ );
+ this.finishItem(itemId);
+ return;
+ }
+ if (message.type === 'conversation.item.input_audio_transcription.failed') {
+ loggers.stt.warn(
+ 'STT transcription failed for item %s: %s',
+ itemId,
+ message.error?.message || 'unknown',
+ );
+ this.finishItem(itemId);
+ }
+ }
+
+ private emitTranscript(
+ text: string,
+ isFinal: boolean,
+ durationMs: number,
+ ): void {
+ if (!text && !isFinal) return;
+ this.sequence++;
+ this.onTranscript({ text, isFinal, durationMs, sequence: this.sequence });
+ }
+
+ private finishItem(itemId: string): void {
+ this.durationByItem.delete(itemId);
+ this.textByItem.delete(itemId);
+ this.pendingCommits = Math.max(0, this.pendingCommits - 1);
+ if (this.pendingCommits === 0) this.resolvePendingWaiters();
+ }
+
+ private sendEvent(event: object): void {
+ if (this.socket.readyState === WebSocket.OPEN)
+ this.socket.send(JSON.stringify(event));
+ }
+
+ private isSocketClosed(): boolean {
+ return this.socket.readyState === WebSocket.CLOSED;
+ }
+
+ private deliverUnexpectedClose(): void {
+ if (
+ !this.unexpectedlyClosed ||
+ this.unexpectedCloseDelivered ||
+ !this.unexpectedCloseListener
+ )
+ return;
+ this.unexpectedCloseDelivered = true;
+ this.unexpectedCloseListener();
+ }
+
+ private waitForPendingCommits(): Promise {
+ if (this.pendingCommits === 0) return Promise.resolve();
+ return new Promise((resolve) => {
+ const done = () => {
+ clearTimeout(timer);
+ this.pendingWaiters.delete(done);
+ resolve();
+ };
+ const timer = setTimeout(done, 15_000);
+ this.pendingWaiters.add(done);
+ });
+ }
+
+ private resolvePendingWaiters(): void {
+ for (const resolve of [...this.pendingWaiters]) resolve();
+ }
+}
+
+export class MockSttClient implements ISttClient {
+ private available = true;
+
+ isAvailable(): boolean {
+ return this.available;
+ }
+
+ onAvailable(_listener: () => void): void {}
+
+ async createStream(
+ metadata: SttStreamMetadata,
+ onTranscript: (event: SttTranscriptEvent) => void,
+ ): Promise {
+ return new MockSttStream(metadata, onTranscript);
+ }
+}
+
+class MockSttStream implements ISttStream {
+ private chunks: Buffer[] = [];
+ private bytes = 0;
+ private sequence = 0;
+
+ constructor(
+ private metadata: SttStreamMetadata,
+ private onTranscript: (event: SttTranscriptEvent) => void,
+ ) {}
+
+ sendAudio(frame: Buffer): void {
+ this.chunks.push(frame);
+ this.bytes += frame.length;
+ }
+
+ markFinal(durationMs: number): void {
+ if (this.bytes === 0) return;
+ this.sequence++;
+ const seconds = this.bytes / 2 / this.metadata.sampleRate;
+ loggers.stt.info(
+ '[MockSTT] Would transcribe %d bytes (~%ds audio) for session %s',
+ this.bytes,
+ seconds.toFixed(1),
+ this.metadata.sessionId,
+ );
+ this.onTranscript({
+ text: `[Mock #${this.sequence}: ~${seconds.toFixed(1)}s]`,
+ isFinal: true,
+ durationMs,
+ sequence: this.sequence,
+ });
+ this.chunks = [];
+ this.bytes = 0;
+ }
+
+ onUnexpectedClose(_listener: () => void): void {}
+
+ async close(): Promise {
+ this.chunks = [];
+ this.bytes = 0;
+ }
+}
diff --git a/suite/meet/sfu-server/src/stt/SttManager.test.ts b/suite/meet/sfu-server/src/stt/SttManager.test.ts
new file mode 100644
index 0000000000..2fa7dd8444
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/SttManager.test.ts
@@ -0,0 +1,146 @@
+import type { Producer, Router } from 'mediasoup/types';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { AudioIngester } from './AudioIngester';
+import type { ISttClient, ISttStream } from './SttClient';
+import { SttManager } from './SttManager';
+
+function createSttClient(available = false) {
+ let onAvailable: (() => void) | undefined;
+ const client: ISttClient = {
+ isAvailable: () => available,
+ onAvailable: (listener) => {
+ onAvailable = listener;
+ },
+ createStream: vi.fn<() => Promise>(),
+ };
+ return { client, recover: () => onAvailable?.() };
+}
+
+describe('SttManager', () => {
+ afterEach(() => {
+ vi.restoreAllMocks();
+ });
+
+ it('restarts subscribed rooms when the STT service recovers', async () => {
+ const sttClient = createSttClient();
+ const manager = new SttManager({ sttClient: sttClient.client });
+ const restartRoom = vi.fn<() => Promise>().mockResolvedValue();
+ manager.addSubscriber('room-1', 'socket-1');
+ manager.setRestartRoomTranscription(restartRoom);
+
+ sttClient.recover();
+ await vi.waitFor(() => expect(restartRoom).toHaveBeenCalledWith('room-1'));
+ });
+
+ it('passes only room subscribers to the transcript emitter', () => {
+ const sttClient = createSttClient();
+ const manager = new SttManager({ sttClient: sttClient.client });
+ const emit = vi.fn();
+ manager.addSubscriber('room-1', 'socket-1');
+ manager.setEmitToSubscribers(emit);
+
+ const internals = manager as unknown as {
+ handleTranscript: (
+ roomId: string,
+ participantId: string,
+ participantName: string,
+ text: string,
+ isFinal: boolean,
+ durationMs: number,
+ ) => void;
+ };
+ internals.handleTranscript(
+ 'room-1',
+ 'participant-1',
+ 'Alice',
+ 'Hello',
+ true,
+ 100,
+ );
+
+ expect(emit).toHaveBeenCalledWith(
+ 'room-1',
+ new Set(['socket-1']),
+ 'stt:segment',
+ expect.objectContaining({ roomId: 'room-1' }),
+ );
+ });
+
+ it('replaces only the ingester whose Realtime stream closed', async () => {
+ vi.spyOn(AudioIngester.prototype, 'start').mockResolvedValue();
+ const stop = vi.spyOn(AudioIngester.prototype, 'stop').mockResolvedValue();
+ const sttClient = createSttClient(true);
+ const manager = new SttManager({ sttClient: sttClient.client });
+ manager.setGetRouter(() => ({}) as Router);
+ manager.addSubscriber('room-1', 'socket-1');
+ const producerA = { id: 'producer-a', closed: false } as Producer;
+ const producerB = { id: 'producer-b', closed: false } as Producer;
+
+ await manager.startTranscription(
+ 'room-1',
+ 'participant-a',
+ 'Alice',
+ producerA,
+ );
+ await manager.startTranscription(
+ 'room-1',
+ 'participant-b',
+ 'Bob',
+ producerB,
+ );
+ const internals = manager as unknown as {
+ activeSessions: Map;
+ };
+ const failed = internals.activeSessions.get(
+ 'room-1:participant-a:producer-a',
+ )!;
+ const healthy = internals.activeSessions.get(
+ 'room-1:participant-b:producer-b',
+ )!;
+ const failedInternals = failed as unknown as {
+ onUnexpectedStreamClose: () => void;
+ };
+
+ failedInternals.onUnexpectedStreamClose();
+
+ await vi.waitFor(() => {
+ expect(
+ internals.activeSessions.get('room-1:participant-a:producer-a'),
+ ).not.toBe(failed);
+ });
+ expect(stop).toHaveBeenCalledOnce();
+ expect(stop.mock.contexts[0]).toBe(failed);
+ expect(
+ internals.activeSessions.get('room-1:participant-b:producer-b'),
+ ).toBe(healthy);
+ expect(internals.activeSessions).toHaveLength(2);
+ expect(AudioIngester.prototype.start).toHaveBeenCalledTimes(3);
+ });
+
+ it('blocks new sessions until overlapping room stops finish', async () => {
+ vi.spyOn(AudioIngester.prototype, 'start').mockResolvedValue();
+ let finishStop: () => void = () => {};
+ vi.spyOn(AudioIngester.prototype, 'stop').mockImplementation(
+ () =>
+ new Promise((resolve) => {
+ finishStop = resolve;
+ }),
+ );
+ const sttClient = createSttClient(true);
+ const manager = new SttManager({ sttClient: sttClient.client });
+ manager.setGetRouter(() => ({}) as Router);
+ manager.addSubscriber('room-1', 'socket-1');
+ await manager.startTranscription('room-1', 'participant-a', 'Alice', {
+ id: 'producer-a',
+ closed: false,
+ } as Producer);
+
+ const firstStop = manager.stopRoom('room-1');
+ const secondStop = manager.stopRoom('room-1');
+ expect(manager.addSubscriber('room-1', 'socket-2')).toBe(false);
+
+ finishStop();
+ await Promise.all([firstStop, secondStop]);
+ expect(manager.addSubscriber('room-1', 'socket-2')).toBe(true);
+ });
+});
diff --git a/suite/meet/sfu-server/src/stt/SttManager.ts b/suite/meet/sfu-server/src/stt/SttManager.ts
new file mode 100644
index 0000000000..7153ff35d8
--- /dev/null
+++ b/suite/meet/sfu-server/src/stt/SttManager.ts
@@ -0,0 +1,407 @@
+import type { Producer, Router } from 'mediasoup/types';
+import type { ServerToClientEvents, TranscriptSegment } from '../types';
+import { loggers } from '../utils/logger';
+import { AudioIngester } from './AudioIngester';
+import { type ISttClient, MockSttClient, SttClient } from './SttClient';
+
+interface SttManagerOptions {
+ /** URL of the STT server (e.g. http://127.0.0.1:8080) */
+ sttServerUrl?: string;
+ /** Use mock client in development when no STT server is configured */
+ allowMockFallback?: boolean;
+ captureDirectory?: string;
+ sttClient?: ISttClient;
+}
+
+type EmitSttToSubscribers = (
+ roomId: string,
+ socketIds: ReadonlySet,
+ event: 'stt:segment',
+ data: Parameters[0],
+) => void;
+
+export class SttManager {
+ private static readonly STREAM_RECOVERY_DELAYS_MS = [0, 1000, 5000, 10_000];
+ private sttClient: ISttClient;
+ private activeSessions = new Map();
+ private roomSubscribers = new Map>();
+ private roomActiveSpeakers = new Map>();
+ private sessionRecoveries = new Map();
+ private stoppingRooms = new Map();
+ private emitToSubscribers: EmitSttToSubscribers | undefined;
+ private getRouter: ((roomId: string) => Router | undefined) | undefined;
+ private restartRoomTranscription:
+ | ((roomId: string) => Promise)
+ | undefined;
+ private captureDirectory?: string;
+
+ constructor(options: SttManagerOptions) {
+ this.captureDirectory = options.captureDirectory;
+ if (this.captureDirectory) {
+ loggers.stt.warn(
+ 'STT diagnostic audio capture enabled at %s',
+ this.captureDirectory,
+ );
+ }
+ if (options.sttClient) {
+ this.sttClient = options.sttClient;
+ } else if (options.sttServerUrl) {
+ const url = options.sttServerUrl.trim();
+ loggers.stt.info('Using STT server: %s', url);
+ this.sttClient = new SttClient(url);
+ } else if (options.allowMockFallback) {
+ loggers.stt.warn('No STT server URL configured. Using mock client.');
+ this.sttClient = new MockSttClient();
+ } else {
+ loggers.stt.warn('STT disabled: no server URL and mock fallback is off.');
+ this.sttClient = new MockSttClient();
+ (this.sttClient as MockSttClient).isAvailable = () => false;
+ }
+ this.sttClient.onAvailable(() => this.restartSubscribedRooms());
+ }
+
+ setEmitToSubscribers(fn: EmitSttToSubscribers): void {
+ this.emitToSubscribers = fn;
+ }
+
+ setGetRouter(fn: (roomId: string) => Router | undefined): void {
+ this.getRouter = fn;
+ }
+
+ setRestartRoomTranscription(fn: (roomId: string) => Promise): void {
+ this.restartRoomTranscription = fn;
+ if (this.sttClient.isAvailable()) this.restartSubscribedRooms();
+ }
+
+ setActiveSpeakers(roomId: string, participantIds: string[]): void {
+ this.roomActiveSpeakers.set(roomId, new Set(participantIds));
+ }
+
+ isActiveSpeaker(roomId: string, participantId: string): boolean {
+ const speakers = this.roomActiveSpeakers.get(roomId);
+ if (!speakers) return true;
+ return speakers.has(participantId);
+ }
+
+ hasSubscribers(roomId: string): boolean {
+ return (this.roomSubscribers.get(roomId)?.size ?? 0) > 0;
+ }
+
+ getSubscribers(roomId: string): Set | undefined {
+ return this.roomSubscribers.get(roomId);
+ }
+
+ addSubscriber(roomId: string, socketId: string): boolean {
+ if ((this.stoppingRooms.get(roomId) ?? 0) > 0) return false;
+ if (!this.roomSubscribers.has(roomId)) {
+ this.roomSubscribers.set(roomId, new Set());
+ }
+ const set = this.roomSubscribers.get(roomId)!;
+ const wasFirst = set.size === 0;
+ set.add(socketId);
+ loggers.stt.info(
+ 'STT subscriber added for room %s (socket: %s, total: %d)',
+ roomId,
+ socketId,
+ set.size,
+ );
+ return wasFirst;
+ }
+
+ removeSubscriber(roomId: string, socketId: string): boolean {
+ const set = this.roomSubscribers.get(roomId);
+ if (!set) return false;
+ set.delete(socketId);
+ const isEmpty = set.size === 0;
+ if (isEmpty) {
+ this.roomSubscribers.delete(roomId);
+ }
+ loggers.stt.info(
+ 'STT subscriber removed for room %s (socket: %s, total: %d)',
+ roomId,
+ socketId,
+ set.size,
+ );
+ return isEmpty;
+ }
+
+ async startTranscription(
+ roomId: string,
+ participantId: string,
+ participantName: string | undefined,
+ producer: Producer,
+ ): Promise {
+ if ((this.stoppingRooms.get(roomId) ?? 0) > 0) return;
+ if (!this.hasSubscribers(roomId)) {
+ loggers.stt.debug('STT has no subscribers for room %s, skipping', roomId);
+ return;
+ }
+
+ const sessionKey = this.getSessionKey(roomId, participantId, producer.id);
+ if (this.activeSessions.has(sessionKey)) {
+ loggers.stt.debug('Transcription already active for %s', sessionKey);
+ return;
+ }
+
+ if (!this.sttClient.isAvailable()) {
+ loggers.stt.warn('STT server unavailable, cannot start transcription');
+ return;
+ }
+
+ const router = this.getRouter?.(roomId);
+ if (!router) {
+ loggers.stt.error('Router not found for room %s', roomId);
+ return;
+ }
+
+ const ingester = new AudioIngester({
+ roomId,
+ participantId,
+ participantName,
+ producer,
+ router,
+ sttClient: this.sttClient,
+ captureDirectory: this.captureDirectory,
+ isActiveSpeaker: () => this.isActiveSpeaker(roomId, participantId),
+ onUnexpectedStreamClose: () => {
+ void this.recoverIngester(
+ sessionKey,
+ ingester,
+ roomId,
+ participantId,
+ participantName,
+ producer,
+ ).catch((error) => {
+ loggers.stt.warn(
+ 'Failed to recover STT stream for %s: %s',
+ sessionKey,
+ (error as Error).message,
+ );
+ });
+ },
+ onTranscript: (text, isFinal, durationMs) => {
+ this.handleTranscript(
+ roomId,
+ participantId,
+ participantName,
+ text,
+ isFinal,
+ durationMs,
+ );
+ },
+ });
+
+ this.activeSessions.set(sessionKey, ingester);
+ try {
+ await ingester.start();
+ } catch (error) {
+ if (this.activeSessions.get(sessionKey) === ingester) {
+ this.activeSessions.delete(sessionKey);
+ }
+ throw error;
+ }
+ }
+
+ async stopTranscription(
+ roomId: string,
+ participantId: string,
+ producerId?: string,
+ ): Promise {
+ if (producerId) {
+ const sessionKey = this.getSessionKey(roomId, participantId, producerId);
+ this.sessionRecoveries.delete(sessionKey);
+ const ingester = this.activeSessions.get(sessionKey);
+ if (!ingester) return;
+
+ this.activeSessions.delete(sessionKey);
+ await ingester.stop();
+ return;
+ }
+
+ const stops: Promise[] = [];
+ for (const [key, ingester] of this.activeSessions) {
+ if (key.startsWith(`${roomId}:${participantId}:`)) {
+ this.sessionRecoveries.delete(key);
+ this.activeSessions.delete(key);
+ stops.push(ingester.stop());
+ }
+ }
+ await Promise.all(stops);
+ }
+
+ async stopRoom(roomId: string, restartIfSubscribed = false): Promise {
+ const subscribers = this.roomSubscribers.get(roomId);
+ if (!restartIfSubscribed) {
+ this.stoppingRooms.set(roomId, (this.stoppingRooms.get(roomId) ?? 0) + 1);
+ }
+ this.roomSubscribers.delete(roomId);
+ this.roomActiveSpeakers.delete(roomId);
+ try {
+ await this.stopRoomTranscriptions(roomId);
+ if (restartIfSubscribed && subscribers?.size) {
+ this.roomSubscribers.set(roomId, subscribers);
+ await this.restartRoomTranscription?.(roomId);
+ }
+ } finally {
+ if (!restartIfSubscribed) {
+ this.roomSubscribers.delete(roomId);
+ this.roomActiveSpeakers.delete(roomId);
+ const remainingStops = (this.stoppingRooms.get(roomId) ?? 1) - 1;
+ if (remainingStops > 0) this.stoppingRooms.set(roomId, remainingStops);
+ else this.stoppingRooms.delete(roomId);
+ }
+ }
+ }
+
+ private async stopRoomTranscriptions(roomId: string): Promise {
+ for (const sessionKey of this.sessionRecoveries.keys()) {
+ if (sessionKey.startsWith(`${roomId}:`)) {
+ this.sessionRecoveries.delete(sessionKey);
+ }
+ }
+ const stops: Promise[] = [];
+ for (const [key, ingester] of this.activeSessions) {
+ if (key.startsWith(`${roomId}:`)) {
+ this.activeSessions.delete(key);
+ stops.push(
+ ingester.stop().catch((error) => {
+ loggers.stt.error(
+ 'Error stopping ingester: %s',
+ (error as Error).message,
+ );
+ }),
+ );
+ }
+ }
+ await Promise.all(stops);
+ }
+
+ destroy(): void {
+ if (typeof (this.sttClient as SttClient).destroy === 'function') {
+ (this.sttClient as SttClient).destroy();
+ }
+ }
+
+ private handleTranscript(
+ roomId: string,
+ participantId: string,
+ participantName: string | undefined,
+ text: string,
+ isFinal: boolean,
+ durationMs: number,
+ ): void {
+ const now = Date.now();
+ const segment: TranscriptSegment = {
+ participantId,
+ participantName,
+ text,
+ isFinal,
+ timestamp: new Date(now).toISOString(),
+ segmentStart: now - durationMs,
+ segmentEnd: now,
+ };
+
+ const subscribers = this.roomSubscribers.get(roomId);
+ if (this.emitToSubscribers && subscribers?.size) {
+ this.emitToSubscribers(roomId, subscribers, 'stt:segment', {
+ roomId,
+ segment,
+ });
+ }
+ }
+
+ private restartSubscribedRooms(): void {
+ if (!this.restartRoomTranscription) return;
+ for (const roomId of this.roomSubscribers.keys()) {
+ this.restartSubscribedRoom(roomId).catch((error) => {
+ loggers.stt.warn(
+ 'Failed to restart STT for room %s: %s',
+ roomId,
+ (error as Error).message,
+ );
+ });
+ }
+ }
+
+ private async restartSubscribedRoom(roomId: string): Promise {
+ await this.stopRoomTranscriptions(roomId);
+ if (this.hasSubscribers(roomId)) {
+ await this.restartRoomTranscription?.(roomId);
+ }
+ }
+
+ private async recoverIngester(
+ sessionKey: string,
+ failedIngester: AudioIngester,
+ roomId: string,
+ participantId: string,
+ participantName: string | undefined,
+ producer: Producer,
+ ): Promise {
+ if (this.activeSessions.get(sessionKey) !== failedIngester) return;
+ const recovery = Symbol(sessionKey);
+ this.sessionRecoveries.set(sessionKey, recovery);
+
+ try {
+ await failedIngester.stop();
+ let currentIngester = failedIngester;
+ let attempt = 0;
+ while (this.sessionRecoveries.get(sessionKey) === recovery) {
+ const delayMs =
+ SttManager.STREAM_RECOVERY_DELAYS_MS[
+ Math.min(attempt, SttManager.STREAM_RECOVERY_DELAYS_MS.length - 1)
+ ];
+ attempt++;
+ if (this.sessionRecoveries.get(sessionKey) !== recovery) return;
+ if (this.activeSessions.get(sessionKey) !== currentIngester) return;
+ if (
+ !this.hasSubscribers(roomId) ||
+ producer.closed ||
+ !this.sttClient.isAvailable()
+ ) {
+ this.activeSessions.delete(sessionKey);
+ return;
+ }
+ if (delayMs > 0)
+ await new Promise((resolve) => setTimeout(resolve, delayMs));
+ if (this.sessionRecoveries.get(sessionKey) !== recovery) return;
+ if (this.activeSessions.get(sessionKey) !== currentIngester) return;
+
+ this.activeSessions.delete(sessionKey);
+ try {
+ await this.startTranscription(
+ roomId,
+ participantId,
+ participantName,
+ producer,
+ );
+ const replacement = this.activeSessions.get(sessionKey);
+ if (replacement && replacement !== currentIngester) return;
+ throw new Error('STT replacement did not start');
+ } catch (error) {
+ if (this.sessionRecoveries.get(sessionKey) !== recovery) return;
+ currentIngester =
+ this.activeSessions.get(sessionKey) ?? currentIngester;
+ this.activeSessions.set(sessionKey, currentIngester);
+ loggers.stt.warn(
+ 'STT stream recovery attempt failed for %s: %s',
+ sessionKey,
+ (error as Error).message,
+ );
+ }
+ }
+ } finally {
+ if (this.sessionRecoveries.get(sessionKey) === recovery) {
+ this.sessionRecoveries.delete(sessionKey);
+ }
+ }
+ }
+
+ private getSessionKey(
+ roomId: string,
+ participantId: string,
+ producerId: string,
+ ): string {
+ return `${roomId}:${participantId}:${producerId}`;
+ }
+}
diff --git a/suite/meet/sfu-server/src/types/index.ts b/suite/meet/sfu-server/src/types/index.ts
index acaf0bd000..6e7cad196b 100644
--- a/suite/meet/sfu-server/src/types/index.ts
+++ b/suite/meet/sfu-server/src/types/index.ts
@@ -55,6 +55,9 @@ import type {
ScreenShareStoppedEvent,
SFUErrorEvent,
SFUScope,
+ SttSegmentEvent,
+ SttToggleRequest,
+ TranscriptSegment,
UpdateTokenRequest,
UserData,
} from '../../../types';
@@ -111,6 +114,9 @@ export type {
ScreenShareStoppedEvent,
SFUErrorEvent,
SFUScope,
+ SttSegmentEvent,
+ SttToggleRequest,
+ TranscriptSegment,
UpdateTokenRequest,
UserData,
WebRtcServer,
@@ -145,6 +151,7 @@ export interface ServerToClientEvents {
hand_raised: (data: HandRaisedEvent) => void;
existing_raised_hands: (data: ExistingRaisedHandsEvent) => void;
network_quality_update: (data: NetworkQualityUpdateEvent) => void;
+ 'stt:segment': (data: SttSegmentEvent) => void;
'e2ee:epoch': (data: E2eeEpochEnvelope) => void;
}
@@ -270,6 +277,10 @@ export interface ClientToServerEvents {
callback: (response: SFUResponse) => void,
) => void;
leave_room: (data?: LeaveRoomRequest) => void;
+ 'stt:toggle': (
+ data: SttToggleRequest,
+ callback: (response: SFUResponse & { enabled?: boolean }) => void,
+ ) => void;
'e2ee:epoch': (data: E2eeEpochEnvelope) => void;
}
diff --git a/suite/meet/sfu-server/src/utils/logger.ts b/suite/meet/sfu-server/src/utils/logger.ts
index b22270f286..fe33297d3e 100644
--- a/suite/meet/sfu-server/src/utils/logger.ts
+++ b/suite/meet/sfu-server/src/utils/logger.ts
@@ -101,6 +101,7 @@ export const loggers = {
authManager: new Logger('AuthManager'),
server: new Logger('Server'),
config: new Logger('Config'),
+ stt: new Logger('STT'),
telemetry: new Logger('Telemetry'),
} as const;
diff --git a/suite/meet/sfu-server/stt-server/.dockerignore b/suite/meet/sfu-server/stt-server/.dockerignore
new file mode 100644
index 0000000000..3dfcdcf01d
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/.dockerignore
@@ -0,0 +1,5 @@
+__pycache__/
+*.py[cod]
+*.log
+.env
+.venv/
diff --git a/suite/meet/sfu-server/stt-server/Dockerfile b/suite/meet/sfu-server/stt-server/Dockerfile
new file mode 100644
index 0000000000..d21662641e
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/Dockerfile
@@ -0,0 +1,38 @@
+ARG BASE_IMAGE=nvcr.io/nvidia/pytorch:25.06-py3
+FROM ${BASE_IMAGE}
+
+LABEL org.opencontainers.image.title="Frappe Meet Nemotron STT" \
+ org.opencontainers.image.description="Nemotron 3.5 ASR runtime for Frappe Meet captions" \
+ org.opencontainers.image.source="https://github.com/frappe/suite"
+
+ENV PYTHONUNBUFFERED=1 \
+ PIP_NO_CACHE_DIR=1 \
+ HF_HOME=/models/huggingface \
+ NEMO_CACHE_DIR=/models/nemo \
+ TORCH_HOME=/models/torch \
+ STT_HOST=0.0.0.0 \
+ STT_PORT=8000
+
+WORKDIR /app
+
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends ffmpeg git \
+ && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt ./
+RUN python -m pip install --no-cache-dir -r requirements.txt
+
+COPY server.py protocol.py resampling.py ./
+
+RUN useradd --create-home --uid 10001 stt \
+ && mkdir -p /models/huggingface /models/nemo /models/torch \
+ && chown -R stt:stt /app /models
+
+USER stt
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=15m --retries=3 \
+ CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=4)"]
+
+CMD ["python", "server.py"]
diff --git a/suite/meet/sfu-server/stt-server/README.md b/suite/meet/sfu-server/stt-server/README.md
new file mode 100644
index 0000000000..a633059e1b
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/README.md
@@ -0,0 +1,51 @@
+# Nemotron STT Runtime
+
+GPU inference image for Frappe Meet captions using NVIDIA Nemotron 3.5 ASR through NeMo.
+
+## Runtime Contract
+
+- Container port: `8000`
+- Health check: `GET /health`
+- OpenAI transcription: `POST /v1/audio/transcriptions`
+- OpenAI Realtime transcription: `WS /v1/realtime`
+- Model listing: `GET /v1/models`
+- GPU: NVIDIA CUDA-compatible GPU
+
+The model is downloaded at startup. Mount `/models` to persist the Hugging Face, NeMo, and Torch caches across container replacements.
+
+## Configuration
+
+| Variable | Default |
+|---|---|
+| `STT_HOST` | `0.0.0.0` |
+| `STT_PORT` | `8000` |
+| `NEMOTRON_MODEL` | `nvidia/nemotron-3.5-asr-streaming-0.6b` |
+| `NEMOTRON_LANGUAGE` | `en-US` |
+| `NEMOTRON_ATT_CONTEXT_SIZE` | `56,3` |
+| `NEMOTRON_FINAL_SILENCE_MS` | `600` |
+| `STT_STREAM_QUEUE_FRAMES` | `400` |
+| `HF_TOKEN` | unset |
+
+## OpenAI-Compatible API
+
+```bash
+curl http://localhost:8000/v1/audio/transcriptions \
+ -F file=@audio.wav \
+ -F model=nemotron-3.5-asr-streaming-0.6b \
+ -F language=en-US
+```
+
+Set `stream=true` to receive `transcript.text.delta` and `transcript.text.done` Server-Sent Events.
+
+For live input, connect to `/v1/realtime`, send a transcription `session.update` configured for 24 kHz PCM16 mono, append base64 audio with `input_audio_buffer.append`, and finalize turns with `input_audio_buffer.commit`. The server emits OpenAI Realtime transcription delta and completed events. Authentication is expected to be enforced by the private deployment boundary.
+
+## Run
+
+```bash
+docker run --rm --gpus all \
+ -p 8000:8000 \
+ -v nemotron-models:/models \
+ ghcr.io/frappe/suite/nemotron-stt:
+```
+
+The PR workflow publishes same-repository pull requests as `pr-` and all feature branches as both their branch name and short commit SHA. Fork branches publish under the fork owner's GHCR namespace.
diff --git a/suite/meet/sfu-server/stt-server/protocol.py b/suite/meet/sfu-server/stt-server/protocol.py
new file mode 100644
index 0000000000..f543a2c61f
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/protocol.py
@@ -0,0 +1,93 @@
+import json
+import re
+import time
+import uuid
+
+MODEL_SAMPLE_RATE = 16000
+REALTIME_SAMPLE_RATE = 24000
+
+
+def clean_transcript(text: str) -> str:
+ text = re.sub(r"\s*<[a-z]{2,3}(?:-[a-z0-9]{2,8})?>\s*", " ", text, flags=re.IGNORECASE)
+ return re.sub(r"\s+", " ", text).strip()
+
+
+def event_id() -> str:
+ return f"event_{uuid.uuid4().hex}"
+
+
+def item_id() -> str:
+ return f"item_{uuid.uuid4().hex}"
+
+
+def realtime_session(session_id: str, model: str, language: str) -> dict:
+ return {
+ "id": session_id,
+ "object": "realtime.transcription_session",
+ "type": "transcription",
+ "expires_at": int(time.time()) + 3600,
+ "audio": {
+ "input": {
+ "format": {"type": "audio/pcm", "rate": REALTIME_SAMPLE_RATE},
+ "transcription": {"model": model, "language": language},
+ "turn_detection": None,
+ }
+ },
+ "include": [],
+ }
+
+
+def validate_session_update(
+ message: dict,
+ supported_models: set[str],
+ default_model: str,
+ default_language: str,
+) -> tuple[dict | None, str | None]:
+ session = message.get("session")
+ if message.get("type") != "session.update" or not isinstance(session, dict):
+ return None, "Expected a session.update event"
+ if session.get("type") != "transcription":
+ return None, "session.type must be transcription"
+ audio = session.get("audio") or {}
+ audio_input = audio.get("input") or {}
+ audio_format = audio_input.get("format")
+ if audio_format and (
+ audio_format.get("type") != "audio/pcm" or audio_format.get("rate") != REALTIME_SAMPLE_RATE
+ ):
+ return None, f"Only {REALTIME_SAMPLE_RATE} Hz audio/pcm is supported"
+ transcription = audio_input.get("transcription") or {}
+ model = transcription.get("model") or default_model
+ if model not in supported_models:
+ return None, f"Unsupported transcription model: {model}"
+ language = (
+ transcription.get("language")
+ or next(iter(transcription.get("languages") or []), None)
+ or default_language
+ )
+ return {"model": model, "language": language}, None
+
+
+def realtime_error(
+ message: str, client_event_id: str | None = None, code: str = "invalid_request_error"
+) -> dict:
+ return {
+ "event_id": event_id(),
+ "type": "error",
+ "error": {
+ "type": "invalid_request_error",
+ "code": code,
+ "message": message,
+ "param": None,
+ "event_id": client_event_id,
+ },
+ }
+
+
+def openai_sse_event(event: dict) -> str:
+ return f"data: {json.dumps(event)}\n\n"
+
+
+def transcript_delta(previous: str, current: str) -> str | None:
+ if current == previous:
+ return None
+ return current[len(previous) :] if current.startswith(previous) else None
diff --git a/suite/meet/sfu-server/stt-server/protocol_test.py b/suite/meet/sfu-server/stt-server/protocol_test.py
new file mode 100644
index 0000000000..01f9545f97
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/protocol_test.py
@@ -0,0 +1,83 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from protocol import (
+ REALTIME_SAMPLE_RATE,
+ clean_transcript,
+ openai_sse_event,
+ realtime_error,
+ realtime_session,
+ transcript_delta,
+ validate_session_update,
+)
+
+
+class ProtocolTest(unittest.TestCase):
+ def setUp(self):
+ self.update = {
+ "type": "session.update",
+ "session": {
+ "type": "transcription",
+ "audio": {
+ "input": {
+ "format": {"type": "audio/pcm", "rate": REALTIME_SAMPLE_RATE},
+ "transcription": {"model": "nemotron", "language": "en-US"},
+ "turn_detection": None,
+ }
+ },
+ },
+ }
+
+ def test_validates_realtime_transcription_session(self):
+ config, error = validate_session_update(self.update, {"nemotron"}, "nemotron", "en-US")
+ self.assertIsNone(error)
+ self.assertEqual(config, {"model": "nemotron", "language": "en-US"})
+
+ invalid = self.update | {
+ "session": self.update["session"]
+ | {
+ "audio": {
+ "input": self.update["session"]["audio"]["input"]
+ | {"format": {"type": "audio/pcm", "rate": 16000}}
+ }
+ }
+ }
+ _, error = validate_session_update(invalid, {"nemotron"}, "nemotron", "en-US")
+ self.assertEqual(error, "Only 24000 Hz audio/pcm is supported")
+
+ config, error = validate_session_update(
+ {"type": "session.update", "session": {"type": "transcription"}},
+ {"nemotron"},
+ "nemotron",
+ "en-US",
+ )
+ self.assertIsNone(error)
+ self.assertEqual(config, {"model": "nemotron", "language": "en-US"})
+
+ def test_builds_realtime_session_and_error_events(self):
+ session = realtime_session("sess_1", "nemotron", "en-US")
+ self.assertEqual(session["object"], "realtime.transcription_session")
+ self.assertEqual(session["audio"]["input"]["format"]["rate"], 24000)
+ self.assertEqual(session["audio"]["input"]["transcription"]["model"], "nemotron")
+ error = realtime_error("bad event", "client_event_1")
+ self.assertEqual(error["type"], "error")
+ self.assertEqual(error["error"]["event_id"], "client_event_1")
+
+ def test_cleans_language_tags_and_frames_openai_events(self):
+ self.assertEqual(clean_transcript(" Hello world "), "Hello world")
+ self.assertEqual(
+ openai_sse_event({"type": "transcript.text.delta", "delta": "Hello"}),
+ 'data: {"type": "transcript.text.delta", "delta": "Hello"}\n\n',
+ )
+
+ def test_transcript_delta_handles_growth_and_hypothesis_rewrites(self):
+ self.assertEqual(transcript_delta("Hello", "Hello world"), " world")
+ self.assertIsNone(transcript_delta("Hello word", "Hello world"))
+ self.assertIsNone(transcript_delta("Hello", "Hello"))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/suite/meet/sfu-server/stt-server/requirements.txt b/suite/meet/sfu-server/stt-server/requirements.txt
new file mode 100644
index 0000000000..11aced19e9
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/requirements.txt
@@ -0,0 +1,11 @@
+fastapi>=0.110.0
+uvicorn[standard]>=0.29.0
+python-multipart>=0.0.9
+setuptools<82
+torch>=2.4.0
+numpy>=1.24.0
+librosa>=0.10.0
+soundfile>=0.12.1
+soxr>=0.3.2
+# Nemotron 3.5 ASR requires prompt-aware RNNT classes that may not be in PyPI NeMo yet.
+nemo_toolkit[asr] @ git+https://github.com/NVIDIA/NeMo.git@fc6a475f3d8b8fc0c8743d3107c5e933ecd5b99d
diff --git a/suite/meet/sfu-server/stt-server/resampling.py b/suite/meet/sfu-server/stt-server/resampling.py
new file mode 100644
index 0000000000..16683d66a9
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/resampling.py
@@ -0,0 +1,15 @@
+import numpy as np
+import soxr
+
+
+class StreamingResampler:
+ """Stateful resampler that preserves filter history across audio chunks."""
+
+ def __init__(self, input_rate: int, output_rate: int):
+ self.stream = soxr.ResampleStream(input_rate, output_rate, 1, dtype="float32", quality="HQ")
+
+ def process(self, audio: np.ndarray) -> np.ndarray:
+ return self.stream.resample_chunk(np.asarray(audio, dtype=np.float32))
+
+ def flush(self) -> np.ndarray:
+ return self.stream.resample_chunk(np.empty(0, dtype=np.float32), last=True)
diff --git a/suite/meet/sfu-server/stt-server/resampling_test.py b/suite/meet/sfu-server/stt-server/resampling_test.py
new file mode 100644
index 0000000000..6d4cc9f536
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/resampling_test.py
@@ -0,0 +1,30 @@
+import sys
+import unittest
+from pathlib import Path
+
+import numpy as np
+import soxr
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from resampling import StreamingResampler
+
+
+class StreamingResamplerTest(unittest.TestCase):
+ def test_chunked_output_matches_continuous_resampling(self):
+ sample_count = int(2.35 * 24000)
+ time = np.arange(sample_count, dtype=np.float32) / 24000
+ audio = (0.4 * np.sin(2 * np.pi * (180 + 1200 * time) * time)).astype(np.float32)
+ expected = soxr.resample(audio, 24000, 16000, quality="HQ")
+
+ resampler = StreamingResampler(24000, 16000)
+ chunks = [resampler.process(audio[offset : offset + 2400]) for offset in range(0, len(audio), 2400)]
+ chunks.append(resampler.flush())
+ actual = np.concatenate(chunks)
+
+ self.assertEqual(actual.shape, expected.shape)
+ np.testing.assert_allclose(actual, expected, rtol=1e-5, atol=1e-5)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/suite/meet/sfu-server/stt-server/server.py b/suite/meet/sfu-server/stt-server/server.py
new file mode 100644
index 0000000000..7aef86f744
--- /dev/null
+++ b/suite/meet/sfu-server/stt-server/server.py
@@ -0,0 +1,795 @@
+#!/usr/bin/env python3
+"""Nemotron ASR service for OpenAI clients and session-scoped Meet streams."""
+
+import asyncio
+import base64
+import binascii
+import copy
+import json
+import os
+import subprocess
+import tempfile
+import time
+import uuid
+from collections.abc import AsyncGenerator, Callable
+from contextlib import asynccontextmanager
+from typing import Annotated, Any
+
+import nemo.collections.asr as nemo_asr
+import numpy as np
+import soundfile as sf
+import torch
+import uvicorn
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile, WebSocket, WebSocketDisconnect
+from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
+from nemo.collections.asr.parts.preprocessing.features import normalize_batch
+from nemo.collections.asr.parts.utils.streaming_utils import CacheAwareStreamingAudioBuffer
+from omegaconf import OmegaConf
+from protocol import (
+ MODEL_SAMPLE_RATE,
+ REALTIME_SAMPLE_RATE,
+ clean_transcript,
+ event_id,
+ item_id,
+ openai_sse_event,
+ realtime_error,
+ realtime_session,
+ transcript_delta,
+ validate_session_update,
+)
+from resampling import StreamingResampler
+
+NEMOTRON_MODEL = os.getenv("NEMOTRON_MODEL", "nvidia/nemotron-3.5-asr-streaming-0.6b")
+NEMOTRON_LANGUAGE = os.getenv("NEMOTRON_LANGUAGE", "en-US").strip() or "en-US"
+NEMOTRON_ATT_CONTEXT_SIZE = os.getenv("NEMOTRON_ATT_CONTEXT_SIZE", "56,3")
+NEMOTRON_FINAL_SILENCE_MS = int(os.getenv("NEMOTRON_FINAL_SILENCE_MS", "600"))
+STT_STREAM_QUEUE_FRAMES = max(1, int(os.getenv("STT_STREAM_QUEUE_FRAMES", "400")))
+
+MODEL_ID = NEMOTRON_MODEL.rsplit("/", 1)[-1]
+MEL_HOP_SAMPLES = 160
+
+model = None
+inference_semaphore: asyncio.Semaphore | None = None
+ready = False
+
+
+def parse_att_context_size() -> list[int]:
+ try:
+ parts = [int(part.strip()) for part in NEMOTRON_ATT_CONTEXT_SIZE.strip("[] ").split(",")]
+ except ValueError:
+ parts = [56, 3]
+ return parts if len(parts) == 2 else [56, 3]
+
+
+def _label(**parts) -> None:
+ print("[stt] " + " ".join(f"{key}={value}" for key, value in parts.items()))
+
+
+def pcm16le_to_float32(audio_bytes: bytes) -> np.ndarray:
+ audio_i16 = np.frombuffer(audio_bytes, dtype=np.int16)
+ return audio_i16.astype(np.float32) / 32768.0
+
+
+def model_device():
+ try:
+ return next(model.parameters()).device
+ except (AttributeError, StopIteration):
+ return None
+
+
+def move_to_model_device(value):
+ device = model_device()
+ if device is not None and torch.is_tensor(value):
+ return value.to(device)
+ return value
+
+
+def apply_language(language: str | None) -> str:
+ resolved = (language or NEMOTRON_LANGUAGE).strip() or NEMOTRON_LANGUAGE
+ model.set_inference_prompt(resolved)
+ return resolved
+
+
+def load_model() -> None:
+ global model
+ device = "cuda" if torch.cuda.is_available() else "cpu"
+ att_context_size = parse_att_context_size()
+ _label(event="model_loading", model=NEMOTRON_MODEL, device=device, language=NEMOTRON_LANGUAGE)
+ t0 = time.time()
+
+ model = nemo_asr.models.ASRModel.from_pretrained(NEMOTRON_MODEL).eval()
+ if device == "cuda":
+ model = model.to("cuda")
+ apply_language(NEMOTRON_LANGUAGE)
+ model.encoder.set_default_att_context_size(att_context_size)
+ _label(event="model_loaded", backend="nemo", context=att_context_size, elapsed=f"{time.time() - t0:.2f}s")
+
+
+class FinalDecoder:
+ """Proven utterance decoder retained as a fallback for incremental failures."""
+
+ def __init__(self):
+ self.buffer = CacheAwareStreamingAudioBuffer(model, online_normalization=False)
+ self.cfg = model.encoder.streaming_cfg
+ self.cache_last_channel, self.cache_last_time, self.cache_last_channel_len = (
+ model.encoder.get_initial_cache_state(batch_size=1)
+ )
+ self.cache_last_channel = move_to_model_device(self.cache_last_channel)
+ self.cache_last_time = move_to_model_device(self.cache_last_time)
+ self.cache_last_channel_len = move_to_model_device(self.cache_last_channel_len)
+ self.previous_hypotheses = None
+ self.step = 0
+ self.last_text = ""
+
+ def transcribe(self, audio: np.ndarray) -> str:
+ if audio.size == 0:
+ return ""
+ self.buffer.append_audio(audio, stream_id=-1)
+ for chunk, chunk_len in self.buffer:
+ with torch.inference_mode():
+ (
+ _,
+ _,
+ self.cache_last_channel,
+ self.cache_last_time,
+ self.cache_last_channel_len,
+ self.previous_hypotheses,
+ ) = model.conformer_stream_step(
+ processed_signal=move_to_model_device(chunk),
+ processed_signal_length=move_to_model_device(chunk_len),
+ cache_last_channel=self.cache_last_channel,
+ cache_last_time=self.cache_last_time,
+ cache_last_channel_len=self.cache_last_channel_len,
+ previous_hypotheses=self.previous_hypotheses,
+ drop_extra_pre_encoded=self.cfg.drop_extra_pre_encoded if self.step else 0,
+ keep_all_outputs=self.buffer.is_buffer_empty(),
+ return_transcription=True,
+ )
+ self.step += 1
+ if self.previous_hypotheses:
+ self.last_text = clean_transcript(self.previous_hypotheses[0].text)
+ return self.last_text
+
+
+def final_transcribe(audio: np.ndarray) -> str:
+ return FinalDecoder().transcribe(audio)
+
+
+def _streaming_value(value):
+ if isinstance(value, list | tuple):
+ return value[1] if len(value) > 1 else value[0]
+ return value
+
+
+class StreamingFeatureBuffer:
+ """Rolling normalized mel-feature window for cache-aware inference."""
+
+ def __init__(self):
+ cfg = copy.deepcopy(model._cfg)
+ OmegaConf.set_struct(cfg.preprocessor, False)
+ self.normalize_type = cfg.preprocessor.normalize
+ cfg.preprocessor.normalize = "None"
+ cfg.preprocessor.dither = 0.0
+ cfg.preprocessor.pad_to = 0
+
+ streaming_cfg = model.encoder.streaming_cfg
+ self.chunk_frames = int(_streaming_value(streaming_cfg.chunk_size))
+ self.precache_frames = int(_streaming_value(streaming_cfg.pre_encode_cache_size))
+ self.buffer_frames = self.precache_frames + self.chunk_frames
+ self.chunk_samples = self.chunk_frames * MEL_HOP_SAMPLES
+ self.look_back = 2 * MEL_HOP_SAMPLES
+ self.device = model_device()
+ self.raw_preprocessor = model.from_config_dict(cfg.preprocessor).to(self.device)
+ self.reset()
+
+ def reset(self) -> None:
+ self.sample_ring = torch.zeros(
+ self.chunk_samples + self.look_back,
+ dtype=torch.float32,
+ device=self.device,
+ )
+ silence = torch.zeros(
+ self.buffer_frames * MEL_HOP_SAMPLES + self.look_back,
+ dtype=torch.float32,
+ device=self.device,
+ )
+ zero_level = self._extract(silence)[:, :1]
+ self.feature_buffer = zero_level.repeat(1, self.buffer_frames).contiguous()
+
+ def _extract(self, samples: torch.Tensor) -> torch.Tensor:
+ signal = samples.unsqueeze(0)
+ length = torch.tensor([samples.shape[0]], device=self.device)
+ features, _ = self.raw_preprocessor(input_signal=signal, length=length)
+ return features.squeeze(0)
+
+ def update(self, chunk_audio: np.ndarray) -> None:
+ chunk = torch.from_numpy(np.ascontiguousarray(chunk_audio)).float().to(self.device)
+ self.sample_ring[: -self.chunk_samples] = self.sample_ring[self.chunk_samples :].clone()
+ self.sample_ring[-self.chunk_samples :] = chunk
+ chunk_features = self._extract(self.sample_ring)[:, -self.chunk_frames :]
+ if chunk_features.shape[1] < self.chunk_frames:
+ padding = self.feature_buffer[:, -1:].repeat(1, self.chunk_frames - chunk_features.shape[1])
+ chunk_features = torch.cat([padding, chunk_features], dim=1)
+ self.feature_buffer[:, : -self.chunk_frames] = self.feature_buffer[:, self.chunk_frames :].clone()
+ self.feature_buffer[:, -self.chunk_frames :] = chunk_features
+
+ def normalized_window(self):
+ features = self.feature_buffer.unsqueeze(0)
+ length = torch.tensor([self.buffer_frames], device=self.device)
+ normalized, _, _ = normalize_batch(
+ x=features,
+ seq_len=length,
+ normalize_type=self.normalize_type,
+ )
+ return normalized, length
+
+
+class IncrementalDecoder:
+ """Stateful decoder that owns one stream's encoder cache and hypothesis."""
+
+ def __init__(self):
+ streaming_cfg = model.encoder.streaming_cfg
+ self.chunk_samples = int(_streaming_value(streaming_cfg.chunk_size)) * MEL_HOP_SAMPLES
+ self.drop_extra_pre_encoded = streaming_cfg.drop_extra_pre_encoded
+ self.features = StreamingFeatureBuffer()
+ self.reset()
+
+ def reset(self) -> None:
+ self.cache_last_channel, self.cache_last_time, self.cache_last_channel_len = (
+ model.encoder.get_initial_cache_state(batch_size=1)
+ )
+ self.cache_last_channel = move_to_model_device(self.cache_last_channel)
+ self.cache_last_time = move_to_model_device(self.cache_last_time)
+ self.cache_last_channel_len = move_to_model_device(self.cache_last_channel_len)
+ self.previous_hypotheses = None
+ self.current_text = ""
+ self.audio_buffer = np.zeros(0, dtype=np.float32)
+ self.step = 0
+ self.features.reset()
+
+ def feed(self, audio: np.ndarray) -> str:
+ if audio.size:
+ self.audio_buffer = np.concatenate([self.audio_buffer, audio])
+ while len(self.audio_buffer) >= self.chunk_samples:
+ chunk = self.audio_buffer[: self.chunk_samples]
+ self.audio_buffer = self.audio_buffer[self.chunk_samples :]
+ self.current_text = self._process_chunk(chunk)
+ return self.current_text
+
+ def flush(self) -> str:
+ if self.audio_buffer.size:
+ padding = self.chunk_samples - len(self.audio_buffer)
+ chunk = np.pad(self.audio_buffer, (0, padding))
+ self.current_text = self._process_chunk(chunk, is_final=True)
+ final = self.current_text.strip()
+ self.reset()
+ return final
+
+ def _process_chunk(self, audio: np.ndarray, is_final: bool = False) -> str:
+ self.features.update(audio)
+ processed, processed_len = self.features.normalized_window()
+ with torch.inference_mode():
+ (
+ _,
+ _,
+ self.cache_last_channel,
+ self.cache_last_time,
+ self.cache_last_channel_len,
+ best_hypotheses,
+ ) = model.conformer_stream_step(
+ processed_signal=processed,
+ processed_signal_length=processed_len,
+ cache_last_channel=self.cache_last_channel,
+ cache_last_time=self.cache_last_time,
+ cache_last_channel_len=self.cache_last_channel_len,
+ keep_all_outputs=is_final,
+ previous_hypotheses=self.previous_hypotheses,
+ drop_extra_pre_encoded=self.drop_extra_pre_encoded if self.step else 0,
+ return_transcription=True,
+ )
+ self.step += 1
+ self.previous_hypotheses = best_hypotheses
+ if best_hypotheses:
+ return clean_transcript(best_hypotheses[0].text)
+ return self.current_text
+
+
+class RealtimeTranscriptionSession:
+ def __init__(self, language: str):
+ self.language = language or NEMOTRON_LANGUAGE
+ self.last_sent_text = ""
+ self.utterance_audio: list[np.ndarray] = []
+ self.input_sample_count = 0
+ self.resampler = StreamingResampler(REALTIME_SAMPLE_RATE, MODEL_SAMPLE_RATE)
+ self.incremental_decoder = IncrementalDecoder()
+ self.incremental_failed = False
+
+ def append_and_decode(self, audio_bytes: bytes) -> str:
+ audio = pcm16le_to_float32(audio_bytes)
+ self.input_sample_count += len(audio)
+ audio = self.resampler.process(audio)
+ if audio.size:
+ self.utterance_audio.append(audio)
+ if self.incremental_failed:
+ return ""
+ try:
+ return self.incremental_decoder.feed(audio)
+ except Exception as incremental_error:
+ self.incremental_failed = True
+ _label(event="incremental_fallback", error=str(incremental_error))
+ return ""
+
+ def finalize(self) -> str:
+ if not self.has_audio:
+ self.incremental_decoder.reset()
+ self.reset_utterance()
+ return ""
+ tail = self.resampler.flush()
+ if tail.size:
+ self.utterance_audio.append(tail)
+ audio = np.concatenate(self.utterance_audio)
+ if NEMOTRON_FINAL_SILENCE_MS > 0:
+ audio = np.pad(
+ audio,
+ (0, int(MODEL_SAMPLE_RATE * NEMOTRON_FINAL_SILENCE_MS / 1000)),
+ )
+ text = FinalDecoder().transcribe(audio)
+ self.incremental_decoder.reset()
+ self.reset_utterance()
+ return text
+
+ def audio_duration_seconds(self) -> float:
+ return self.input_sample_count / REALTIME_SAMPLE_RATE
+
+ @property
+ def has_audio(self) -> bool:
+ return self.input_sample_count > 0
+
+ def clear(self) -> None:
+ self.incremental_decoder.reset()
+ self.reset_utterance()
+
+ def reset_utterance(self) -> None:
+ self.last_sent_text = ""
+ self.utterance_audio = []
+ self.input_sample_count = 0
+ self.resampler = StreamingResampler(REALTIME_SAMPLE_RATE, MODEL_SAMPLE_RATE)
+ self.incremental_failed = False
+
+
+def _run_with_language(language: str, operation: Callable[..., Any], *args):
+ apply_language(language)
+ return operation(*args)
+
+
+async def run_inference(language: str, operation: Callable[..., Any], *args):
+ if inference_semaphore is None:
+ raise RuntimeError("Inference service is not initialized")
+ async with inference_semaphore:
+ return await asyncio.to_thread(_run_with_language, language, operation, *args)
+
+
+def direct_transcribe(audio: np.ndarray) -> str:
+ device = model_device()
+ audio_tensor = torch.from_numpy(np.ascontiguousarray(audio)).float().unsqueeze(0).to(device)
+ audio_len = torch.tensor([audio.shape[0]], dtype=torch.long, device=device)
+ with torch.inference_mode():
+ processed, processed_len = model.preprocessor(input_signal=audio_tensor, length=audio_len)
+ encoded, encoded_len = model.encoder(audio_signal=processed, length=processed_len)
+ hypotheses = model.decoding.rnnt_decoder_predictions_tensor(
+ encoded,
+ encoded_len,
+ return_hypotheses=False,
+ )
+ hypothesis = hypotheses[0]
+ return clean_transcript(hypothesis.text if hasattr(hypothesis, "text") else str(hypothesis))
+
+
+def load_uploaded_audio(audio_bytes: bytes, filename: str) -> np.ndarray:
+ suffix = os.path.splitext(filename)[1] or ".wav"
+ with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as audio_file:
+ audio_file.write(audio_bytes)
+ path = audio_file.name
+ try:
+ try:
+ audio, sample_rate = sf.read(path, dtype="float32")
+ except Exception:
+ decoded = subprocess.run(
+ [
+ "ffmpeg",
+ "-v",
+ "error",
+ "-i",
+ path,
+ "-f",
+ "f32le",
+ "-ac",
+ "1",
+ "-ar",
+ str(MODEL_SAMPLE_RATE),
+ "pipe:1",
+ ],
+ check=True,
+ capture_output=True,
+ )
+ return np.frombuffer(decoded.stdout, dtype=np.float32).copy()
+ finally:
+ os.unlink(path)
+
+ if audio.ndim > 1:
+ audio = audio.mean(axis=-1) if audio.shape[-1] <= audio.shape[0] else audio.mean(axis=0)
+ if sample_rate != MODEL_SAMPLE_RATE:
+ import librosa
+
+ audio = librosa.resample(
+ np.asarray(audio, dtype=np.float32),
+ orig_sr=sample_rate,
+ target_sr=MODEL_SAMPLE_RATE,
+ )
+ return np.asarray(audio, dtype=np.float32)
+
+
+def run_warmup() -> None:
+ t0 = time.time()
+ _run_with_language(NEMOTRON_LANGUAGE, direct_transcribe, np.zeros(MODEL_SAMPLE_RATE, dtype=np.float32))
+ _label(event="warmup", elapsed=f"{time.time() - t0:.2f}s")
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+ global inference_semaphore, ready
+ load_model()
+ inference_semaphore = asyncio.Semaphore(1)
+ try:
+ await asyncio.to_thread(run_warmup)
+ finally:
+ ready = True
+ _label(event="ready", max_concurrency=1)
+ yield
+ ready = False
+
+
+app = FastAPI(title="Nemotron STT Server", lifespan=lifespan)
+
+
+@app.get("/health")
+async def health():
+ if not ready:
+ return JSONResponse(
+ {"status": "loading", "backend": "nemo", "model": NEMOTRON_MODEL},
+ status_code=503,
+ )
+ return {"status": "ok", "backend": "nemo", "model": NEMOTRON_MODEL}
+
+
+@app.get("/v1/models")
+async def list_models():
+ return {
+ "object": "list",
+ "data": [
+ {
+ "id": MODEL_ID,
+ "object": "model",
+ "created": int(time.time()),
+ "owned_by": "nvidia",
+ }
+ ],
+ }
+
+
+@app.post("/v1/audio/transcriptions")
+async def transcribe_audio_file(
+ file: Annotated[UploadFile, File()],
+ model_name: Annotated[str, Form(alias="model")] = MODEL_ID,
+ response_format: Annotated[str, Form()] = "json",
+ stream: Annotated[bool, Form()] = False,
+ language: Annotated[str | None, Form()] = None,
+ temperature: Annotated[str | None, Form()] = None,
+ prompt: Annotated[str | None, Form()] = None,
+):
+ del temperature, prompt
+ if not ready or model is None:
+ raise HTTPException(status_code=503, detail="Model not loaded")
+ if model_name not in {MODEL_ID, NEMOTRON_MODEL}:
+ raise HTTPException(status_code=400, detail=f"Unsupported model: {model_name}")
+ if response_format not in {"json", "text", "verbose_json"}:
+ raise HTTPException(status_code=400, detail=f"Unsupported response_format: {response_format}")
+ audio_bytes = await file.read()
+ if not audio_bytes:
+ raise HTTPException(status_code=400, detail="Empty audio file")
+ try:
+ audio = await asyncio.to_thread(load_uploaded_audio, audio_bytes, file.filename or "audio.wav")
+ except Exception as error:
+ raise HTTPException(status_code=400, detail=f"Failed to process audio: {error}") from error
+
+ resolved_language = language or NEMOTRON_LANGUAGE
+ duration = len(audio) / MODEL_SAMPLE_RATE
+ if stream:
+
+ async def event_stream() -> AsyncGenerator[str]:
+ decoder = await run_inference(resolved_language, IncrementalDecoder)
+ previous = ""
+ for offset in range(0, len(audio), decoder.chunk_samples):
+ current = await run_inference(
+ resolved_language,
+ decoder.feed,
+ audio[offset : offset + decoder.chunk_samples],
+ )
+ if delta := transcript_delta(previous, current):
+ yield openai_sse_event({"type": "transcript.text.delta", "delta": delta})
+ previous = current
+ final = await run_inference(resolved_language, decoder.flush)
+ if delta := transcript_delta(previous, final):
+ yield openai_sse_event({"type": "transcript.text.delta", "delta": delta})
+ yield openai_sse_event({"type": "transcript.text.done", "text": final})
+ yield "data: [DONE]\n\n"
+
+ return StreamingResponse(
+ event_stream(),
+ media_type="text/event-stream",
+ headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+ )
+
+ text = await run_inference(resolved_language, final_transcribe, audio)
+ if response_format == "text":
+ return PlainTextResponse(text)
+ if response_format == "verbose_json":
+ return {
+ "text": text,
+ "task": "transcribe",
+ "language": resolved_language,
+ "duration": duration,
+ }
+ return {"text": text}
+
+
+@app.websocket("/v1/realtime")
+async def realtime_transcription(websocket: WebSocket):
+ await websocket.accept()
+ if not ready or model is None:
+ await websocket.send_json(realtime_error("Model not loaded", code="server_not_ready"))
+ await websocket.close(code=1013, reason="Model not loaded")
+ return
+
+ requested_model = websocket.query_params.get("model") or MODEL_ID
+ if requested_model not in {MODEL_ID, NEMOTRON_MODEL}:
+ await websocket.send_json(realtime_error(f"Unsupported transcription model: {requested_model}"))
+ await websocket.close(code=1008, reason="Unsupported model")
+ return
+
+ realtime_session_id = f"sess_{uuid.uuid4().hex}"
+ effective_session = realtime_session(realtime_session_id, requested_model, NEMOTRON_LANGUAGE)
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "session.created",
+ "session": effective_session,
+ }
+ )
+
+ transcription: RealtimeTranscriptionSession | None = None
+ current_item_id = item_id()
+ previous_item_id: str | None = None
+ try:
+ queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=STT_STREAM_QUEUE_FRAMES)
+ closed = asyncio.Event()
+ _label(event="realtime_start", session=realtime_session_id, model=requested_model)
+
+ async def reader() -> None:
+ try:
+ while not closed.is_set():
+ message = await websocket.receive()
+ if message["type"] == "websocket.disconnect":
+ break
+ if message.get("bytes") is not None:
+ await websocket.send_json(
+ realtime_error("Binary WebSocket messages are not supported")
+ )
+ elif message.get("text") is not None:
+ await queue.put(message["text"])
+ except WebSocketDisconnect:
+ pass
+ finally:
+ await queue.put(None)
+
+ async def worker() -> None:
+ nonlocal transcription, effective_session, current_item_id, previous_item_id
+ while not closed.is_set():
+ payload = await queue.get()
+ try:
+ if payload is None:
+ return
+ try:
+ client_event = json.loads(payload)
+ except json.JSONDecodeError:
+ await websocket.send_json(realtime_error("Invalid JSON event"))
+ continue
+ if not isinstance(client_event, dict):
+ await websocket.send_json(realtime_error("WebSocket events must be JSON objects"))
+ continue
+
+ client_event_id = client_event.get("event_id")
+ event_type = client_event.get("type")
+ if event_type == "session.update":
+ config, error = validate_session_update(
+ client_event,
+ {MODEL_ID, NEMOTRON_MODEL},
+ effective_session["audio"]["input"]["transcription"]["model"],
+ effective_session["audio"]["input"]["transcription"]["language"],
+ )
+ if error:
+ await websocket.send_json(realtime_error(error, client_event_id))
+ continue
+ language = config.get("language") or NEMOTRON_LANGUAGE
+ if transcription is None:
+ transcription = await run_inference(
+ language, RealtimeTranscriptionSession, language
+ )
+ elif transcription.has_audio:
+ await websocket.send_json(
+ realtime_error(
+ "Cannot update the session while audio is buffered", client_event_id
+ )
+ )
+ continue
+ else:
+ transcription.language = language
+ effective_session = realtime_session(realtime_session_id, config["model"], language)
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "session.updated",
+ "session": effective_session,
+ }
+ )
+ continue
+
+ if transcription is None:
+ await websocket.send_json(
+ realtime_error("Send a valid session.update before audio events", client_event_id)
+ )
+ continue
+
+ if event_type == "input_audio_buffer.append":
+ try:
+ encoded_audio = client_event.get("audio")
+ if not isinstance(encoded_audio, str):
+ raise ValueError("audio must be a base64 string")
+ audio_bytes = base64.b64decode(encoded_audio, validate=True)
+ if not audio_bytes or len(audio_bytes) % 2:
+ raise ValueError("audio must contain PCM16 samples")
+ if len(audio_bytes) > 15 * 1024 * 1024:
+ raise ValueError("audio event exceeds the 15 MiB limit")
+ except (binascii.Error, ValueError) as decode_error:
+ await websocket.send_json(realtime_error(str(decode_error), client_event_id))
+ continue
+ text = await run_inference(
+ transcription.language,
+ transcription.append_and_decode,
+ audio_bytes,
+ )
+ if delta := transcript_delta(transcription.last_sent_text, text):
+ transcription.last_sent_text = text
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "conversation.item.input_audio_transcription.delta",
+ "item_id": current_item_id,
+ "content_index": 0,
+ "delta": delta,
+ "logprobs": None,
+ }
+ )
+ continue
+
+ if event_type == "input_audio_buffer.clear":
+ await run_inference(transcription.language, transcription.clear)
+ current_item_id = item_id()
+ await websocket.send_json(
+ {"event_id": event_id(), "type": "input_audio_buffer.cleared"}
+ )
+ continue
+
+ if event_type == "input_audio_buffer.commit":
+ if not transcription.has_audio:
+ await websocket.send_json(
+ realtime_error("Cannot commit an empty audio buffer", client_event_id)
+ )
+ continue
+ committed_item_id = current_item_id
+ audio_seconds = transcription.audio_duration_seconds()
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "input_audio_buffer.committed",
+ "previous_item_id": previous_item_id,
+ "item_id": committed_item_id,
+ }
+ )
+ previous_text = transcription.last_sent_text
+ t0 = time.time()
+ try:
+ text = await run_inference(transcription.language, transcription.finalize)
+ except Exception as inference_error:
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "conversation.item.input_audio_transcription.failed",
+ "item_id": committed_item_id,
+ "content_index": 0,
+ "error": {
+ "type": "server_error",
+ "code": "transcription_failed",
+ "message": str(inference_error),
+ "param": None,
+ },
+ }
+ )
+ continue
+ _label(
+ event="realtime_final",
+ session=realtime_session_id,
+ audio_seconds=f"{audio_seconds:.1f}",
+ text_len=len(text),
+ elapsed=f"{time.time() - t0:.2f}s",
+ )
+ if delta := transcript_delta(previous_text, text):
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "conversation.item.input_audio_transcription.delta",
+ "item_id": committed_item_id,
+ "content_index": 0,
+ "delta": delta,
+ "logprobs": None,
+ }
+ )
+ await websocket.send_json(
+ {
+ "event_id": event_id(),
+ "type": "conversation.item.input_audio_transcription.completed",
+ "item_id": committed_item_id,
+ "content_index": 0,
+ "transcript": text,
+ "usage": {"type": "duration", "seconds": audio_seconds},
+ "logprobs": None,
+ }
+ )
+ previous_item_id = committed_item_id
+ current_item_id = item_id()
+ continue
+
+ await websocket.send_json(
+ realtime_error(f"Unsupported event type: {event_type}", client_event_id)
+ )
+ finally:
+ queue.task_done()
+
+ reader_task = asyncio.create_task(reader())
+ worker_task = asyncio.create_task(worker())
+ done, _ = await asyncio.wait(
+ [reader_task, worker_task],
+ return_when=asyncio.FIRST_COMPLETED,
+ )
+ if worker_task in done:
+ closed.set()
+ reader_task.cancel()
+ try:
+ await reader_task
+ except asyncio.CancelledError:
+ pass
+ else:
+ await worker_task
+ if reader_task.done() and not reader_task.cancelled():
+ reader_task.result()
+ worker_task.result()
+ except WebSocketDisconnect:
+ pass
+ finally:
+ _label(event="realtime_end", session=realtime_session_id)
+
+
+if __name__ == "__main__":
+ host = os.getenv("STT_HOST", "127.0.0.1")
+ port = int(os.getenv("STT_PORT", "8000"))
+ uvicorn.run(app, host=host, port=port)
diff --git a/suite/meet/sfu-server/yarn.lock b/suite/meet/sfu-server/yarn.lock
index b0234658c1..32616758ce 100644
--- a/suite/meet/sfu-server/yarn.lock
+++ b/suite/meet/sfu-server/yarn.lock
@@ -551,6 +551,13 @@
"@types/node" "*"
"@types/send" "<1"
+"@types/ws@^8.18.1":
+ version "8.18.1"
+ resolved "https://registry.yarnpkg.com/@types/ws/-/ws-8.18.1.tgz#48464e4bf2ddfd17db13d845467f6070ffea4aa9"
+ integrity sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==
+ dependencies:
+ "@types/node" "*"
+
"@types/ws@^8.5.12":
version "8.18.1"
resolved "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz"
@@ -2138,6 +2145,11 @@ wrap-ansi@^9.0.0:
string-width "^7.0.0"
strip-ansi "^7.1.0"
+ws@^8.21.0:
+ version "8.21.0"
+ resolved "https://registry.yarnpkg.com/ws/-/ws-8.21.0.tgz#012e413fc07429945121b0c153158c4343086951"
+ integrity sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==
+
ws@~8.18.3:
version "8.18.3"
resolved "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz"
diff --git a/suite/meet/types/index.ts b/suite/meet/types/index.ts
index 88f7f098e6..083a449085 100644
--- a/suite/meet/types/index.ts
+++ b/suite/meet/types/index.ts
@@ -305,3 +305,24 @@ export interface PresenceJoinResponse {
success: boolean;
error?: string;
}
+
+// ── STT / Transcription Types ──
+
+export interface TranscriptSegment {
+ participantId: string;
+ participantName?: string;
+ text: string;
+ isFinal: boolean;
+ timestamp: string; // ISO
+ segmentStart: number; // ms from meeting start
+ segmentEnd: number; // ms from meeting start
+}
+
+export interface SttSegmentEvent {
+ roomId: string;
+ segment: TranscriptSegment;
+}
+
+export interface SttToggleRequest {
+ enabled: boolean;
+}