feat(rtp-capture): add two-peer RTP capture test for H264 frame transmission

- Add term_glossary_cache table + migration 0014: resolved definitions are
  stored permanently (definitions rarely change); misses stay ephemeral in
  Redis/LRU with 1h TTL so transient failures get retried
- Lookup flow: LRU -> Redis -> Postgres (permanent) -> live SearXNG; DB hits
  re-warm the fast caches; stale Redis miss sentinels no longer shadow DB
- Rate-limit-aware live lookups: concurrency 2 + stagger, retry once on empty
  results, strict definition filter (Wikipedia preferred, rejects
  disambiguation/ads/translate-homepages)
- Make SEARXNG_BASE_URL configurable via env (default unchanged)

services/discord-gateway/drizzle/migrations/0014_add_term_glossary_cache.sql

@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS "term_glossary_cache" (
+	"term" text PRIMARY KEY NOT NULL,
+	"definition" text NOT NULL,
+	"source_url" text DEFAULT '' NOT NULL,
+	"resolved_at" bigint NOT NULL,
+	"hit_count" integer DEFAULT 0 NOT NULL
+);
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "idx_term_glossary_cache_resolved_at" ON "term_glossary_cache" USING btree ("resolved_at");

services/discord-gateway/drizzle/migrations/meta/_journal.json

@@ -99,6 +99,13 @@
       "when": 1785551832190,
       "tag": "0013_rename_mascot_chat_to_chatbot",
       "breakpoints": true
+    },
+    {
+      "idx": 14,
+      "version": "7",
+      "when": 1785621600000,
+      "tag": "0014_add_term_glossary_cache",
+      "breakpoints": true
     }
   ]
 }

services/discord-gateway/native/libdatachannel-min/test-rtp-capture.js

@@ -0,0 +1,138 @@
+// Two-peer RTP capture test: A sends REAL H264 frames through the binding's
+// packetizer chain to B over localhost. tcpdump (run externally on lo) captures
+// the RTP; a Python script reassembles AnnexB and ffmpeg decodes it.
+//
+// Usage:
+//   node test-rtp-capture.js <mode>   mode = "a" (sender) | "b" (receiver)
+//   Sender writes the negotiated SDP pieces to /tmp/rtp-a.sdp /tmp/rtp-b.sdp
+//   Receiver listens and keeps alive.
+"use strict";
+const { PeerConnection } = require("./build/Release/datachannel_min.node");
+const fs = require("fs");
+
+const mode = process.argv[2] || "a";
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+async function main() {
+  const pc = new PeerConnection({ iceServers: [] });
+  const audio = pc.addTrack("0", "audio");
+  const video = pc.addTrack("1", "video");
+
+  if (mode === "a") {
+    // Sender: generate offer, hand to B via files, get B's answer
+    const offer = await pc.createOffer();
+    fs.writeFileSync("/tmp/rtp-offer.sdp", offer);
+    console.log("[a] offer written", offer.length, "bytes");
+
+    // wait for B to write its answer
+    for (let i = 0; i < 300; i++) {
+      if (fs.existsSync("/tmp/rtp-answer.sdp")) break;
+      await sleep(200);
+    }
+    const answer = fs.readFileSync("/tmp/rtp-answer.sdp", "utf8");
+    pc.setRemoteDescription(answer, "answer");
+
+    // wait connected
+    for (let i = 0; i < 50; i++) {
+      if (pc.state() === "connected") break;
+      await sleep(100);
+    }
+    console.log("[a] state:", pc.state());
+    if (pc.state() !== "connected") {
+      console.log("[a] FAILED not connected");
+      process.exit(1);
+    }
+
+    // Setup packetizer like the gateway does
+    video.setPacketizer("h264", 5678, 101, 90000, 5, 0, 10);
+
+    // Real H264 AnnexB (baseline) — read from file created by ffmpeg
+    const data = fs.readFileSync("/tmp/rtp-input.h264");
+    console.log("[a] input h264 bytes:", data.length);
+
+    // Split into NAL units by start codes, then group into access units
+    // the same way Demuxer does (param sets + one slice per frame).
+    const start3 = Buffer.from([0, 0, 1]);
+    const start4 = Buffer.from([0, 0, 0, 1]);
+    const nals = [];
+    let i = 0;
+    while (i < data.length) {
+      let start = -1;
+      let startLen = 0;
+      for (let j = i; j < data.length - 3; j++) {
+        if (data[j] === 0 && data[j + 1] === 0 && data[j + 2] === 1) {
+          start = j;
+          startLen = 3;
+          if (j > 0 && data[j - 1] === 0) {
+            start = j - 1;
+            startLen = 4;
+          }
+          break;
+        }
+      }
+      if (start === -1) break;
+      if (start > i) {
+        nals.push(data.subarray(i, start));
+      }
+      i = start + startLen;
+    }
+    console.log("[a] NALs:", nals.length);
+
+    // Group: buffer param sets, flush on slice (like Demuxer.flushAccessUnit)
+    let pending = [];
+    let frameCount = 0;
+    const flush = () => {
+      if (pending.length === 0) return;
+      const parts = pending.map((n) => Buffer.concat([start4, n]));
+      const au = Buffer.concat(parts);
+      pending = [];
+      video.sendFrame(au);
+      video.addTimestamp(3000); // 30fps @ 90kHz
+      frameCount++;
+    };
+    for (const n of nals) {
+      const t = n[0] & 0x1f;
+      if (t === 1 || t === 5) {
+        flush(); // previous AU closed by this slice
+        pending.push(n);
+      } else {
+        pending.push(n); // param set / SEI
+      }
+    }
+    flush();
+    console.log("[a] sent frames:", frameCount);
+    await sleep(3000); // let packets flow
+    console.log("[a] done");
+    pc.close();
+    process.exit(0);
+  } else {
+    // Receiver: read offer, answer, keep alive
+    for (let i = 0; i < 300; i++) {
+      if (fs.existsSync("/tmp/rtp-offer.sdp")) break;
+      await sleep(200);
+    }
+    const offer = fs.readFileSync("/tmp/rtp-offer.sdp", "utf8");
+    pc.setRemoteDescription(offer, "offer");
+    const answer = await pc.createAnswer(offer);
+    fs.writeFileSync("/tmp/rtp-answer.sdp", answer);
+    console.log("[b] answer written");
+    for (let i = 0; i < 50; i++) {
+      if (pc.state() === "connected") break;
+      await sleep(100);
+    }
+    console.log("[b] state:", pc.state());
+    await sleep(10000); // hold while sender streams
+    console.log("[b] done");
+    pc.close();
+    process.exit(0);
+  }
+}
+
+main().catch((e) => {
+  console.error("FAILED:", e.message);
+  process.exit(1);
+});
+setTimeout(() => {
+  console.error("TIMEOUT");
+  process.exit(1);
+}, 30000);

services/discord-gateway/src/goLive/BaseMediaConnection.ts

@@ -168,6 +168,9 @@ export class BaseMediaConnection extends EventEmitter {
   }): void {
     // we hardcoded STREAMS_SIMULCAST, which will always be array of 1
     const stream = d.streams[0];
+    console.log(
+      `[goLive:${this.constructor.name}] READY ssrc=${d.ssrc} ip=${d.ip} port=${d.port} streams=${JSON.stringify(d.streams)}`,
+    );
     this._webRtcParams = {
       address: d.ip,
       port: d.port,
@@ -183,6 +186,10 @@ export class BaseMediaConnection extends EventEmitter {
     dave_protocol_version?: number;
   }): Promise<void> {
     if (!("sdp" in d)) throw new Error("Only WebRTC connections are allowed");
+    // DEBUG: dump Discord's real answer SDP — which payload types did it select?
+    console.log(
+      `[goLive:${this.constructor.name}] DISCORD_ANSWER_SDP ${JSON.stringify(d.sdp ?? "").slice(0, 900)}`,
+    );
     this._daveProtocolVersion = d.dave_protocol_version ?? 0;
     this.initDave();
     // Discord's SDP is garbage — generate our own from its pieces
@@ -258,12 +265,10 @@ a=ice-lite
         `a=rtcp-fb:${el.payload_type} transport-cc`,
       ])
       .join("\n");
-    this._webRtcWrapper.webRtcConn?.setRemoteDescription(
+    const builtAnswer = [audioSection, videoSection, videoRtpMap].join("\n");
-      [audioSection, videoSection, videoRtpMap].join("\n"),
+    this._webRtcWrapper.webRtcConn?.setRemoteDescription(builtAnswer, "answer");
-      "answer",
-    );
     console.log(
-      `[goLive:${this.constructor.name}] SELECT_PROTOCOL_ACK processed — remote answer set (${[audioSection, videoSection].join("\n").length}B)`,
+      `[goLive:${this.constructor.name}] SELECT_PROTOCOL_ACK processed — remote answer set (${builtAnswer.length}B) video_mline=${videoPayloadTypes.join(" ")}`,
     );
     this.emit("select_protocol_ack");
   }

services/discord-gateway/src/goLive/Demuxer.ts

@@ -16,6 +16,7 @@ import { spawn } from "node:child_process";
 import { existsSync, readdirSync } from "node:fs";
 import { join } from "node:path";
 import { PassThrough } from "node:stream";
+import type { GoLiveFrame } from "./BaseMediaStream.js";
 
 /** 4-byte AnnexB start code (00 00 00 01) used when building access units. */
 const startCode4 = Buffer.from([0, 0, 0, 1]);
@@ -158,28 +159,46 @@ export async function demux(
   const aPipe = new PassThrough({ objectMode: true, highWaterMark: 128 });
 
   const isStream = typeof input !== "string";
+  // NUT/matroska input (prepareStream with includeAudio) carries audio; the
+  // h264 path is video-only raw AnnexB. Video always goes to stdout (pipe:1);
+  // audio goes to fd3 (pipe:3) so stderr stays free for metadata parsing.
+  const containerFormat =
+    isStream && opts.format !== "h264" ? opts.format : null;
+  const withAudio = isStream && containerFormat !== null;
   const args: string[] = [
     "-hide_banner",
     // info level: stream init lines ("Stream #0:0: Video: h264...") go to
     // stderr and are parsed for dimensions/fps.
     "-loglevel",
     "info",
-    // Input format hint: prepareStream always emits raw AnnexB H264 on
+    // Input format hint: raw H264 has NO magic header, so ffmpeg's
-    // pipe:0. Raw H264 has NO magic header, so ffmpeg's auto-detection
+    // auto-detection fails with "Invalid data found when processing input"
-    // fails with "Invalid data found when processing input" whenever the
+    // whenever the first bytes arrive late/buffered. Pin the demuxer input
-    // first bytes arrive late/buffered. Pin the demuxer input format.
+    // format for streams (NUT for the audio-capable path).
-    ...(isStream ? ["-f", "h264"] : []),
+    ...(withAudio
+      ? ["-f", containerFormat as string]
+      : isStream
+        ? ["-f", "h264"]
+        : []),
     "-i",
     isStream ? "pipe:0" : input,
+    "-map",
+    "0:v:0",
     "-c:v",
     "copy",
-    "-an", // no audio in this minimal demuxer
     "-f",
     "h264",
     "pipe:1",
+    ...(withAudio
+      ? ["-map", "0:a:0?", "-c:a", "copy", "-f", "opus", "pipe:3"]
+      : ["-an"]),
   ];
   const proc = spawn(FFMPEG, args, {
-    stdio: isStream ? ["pipe", "pipe", "pipe"] : ["ignore", "pipe", "pipe"],
+    stdio: isStream
+      ? withAudio
+        ? ["pipe", "pipe", "pipe", "pipe"]
+        : ["pipe", "pipe", "pipe"]
+      : ["ignore", "pipe", "pipe"],
   });
   console.log(
     `[goLive:Demuxer] spawn ffmpeg pid=${proc.pid} input=${isStream ? "stream" : input} args=${args.join(" ")}`,
@@ -191,8 +210,25 @@ export async function demux(
     input.on("error", () => proc.stdin?.destroy());
   }
 
+  // Audio: ffmpeg writes Ogg Opus on fd3 (pipe:3). Parse OGG pages into
+  // opus packets and emit them as GoLiveFrames (20ms, 48kHz) on aPipe.
+  if (withAudio && proc.stdio[3]) {
+    createOggOpusDemux(
+      proc.stdio[3] as unknown as NodeJS.ReadableStream,
+      aPipe,
+    );
+    console.log("[goLive:Demuxer] audio pipe wired (fd3 → Ogg Opus → aPipe)");
+  }
+
   // Set true once stderr metadata has been parsed (see handler below).
   let parsedMeta = false;
+  // Track which stream kinds we've seen. We must NOT stop parsing on the
+  // first stream found: ffmpeg can print the video line and audio line in
+  // separate stderr chunks (input arrives slowly), and the old
+  // early-return (`if (parsedMeta) return`) dropped the audio line forever
+  // → aInfo undefined → no audio RTP → static GoLive tile.
+  let seenVideo = false;
+  let seenAudio = false;
 
   // Parse stream metadata from ffmpeg stderr as it arrives (first chunk has
   // the init lines). Fall back to H264 defaults if parsing fails.
@@ -218,7 +254,6 @@ export async function demux(
           `[goLive:Demuxer] ffmpeg stderr: ${text.trim().split("\n").slice(0, 4).join(" | ")}`,
         );
       }
-      if (parsedMeta) return;
       const streamRe = /Stream #0:(\d+): (Video|Audio): ([^,]+)/g;
       let m: RegExpExecArray | null;
       const found: Array<{ kind: string; codecRaw: string }> = [];
@@ -226,11 +261,15 @@ export async function demux(
       while ((m = streamRe.exec(stderrBuf)) !== null) {
         found.push({ kind: m[2], codecRaw: m[3] });
       }
+      if (process.env.GMW_DEMUX_DEBUG) {
+        console.log(
+          `[goLive:Demuxer] DEBUG stderrBuf=${JSON.stringify(stderrBuf.slice(0, 300))} found=${JSON.stringify(found)}`,
+        );
+      }
       const v = found.find((s) => s.kind === "Video");
       const a = found.find((s) => s.kind === "Audio");
-      if (!v && !a) return;
-      parsedMeta = true;
       if (v) {
+        seenVideo = true;
         const codecName = v.codecRaw.split(" ")[0].toLowerCase();
         const dim = /(\d{2,5})x(\d{2,5})/.exec(stderrBuf);
         const fps = /(\d+(?:\.\d+)?) fps/.exec(stderrBuf);
@@ -250,6 +289,7 @@ export async function demux(
         };
       }
       if (a) {
+        seenAudio = true;
         const codecName = a.codecRaw.split(" ")[0].toLowerCase();
         const sr = /(\d+) Hz/.exec(stderrBuf);
         aInfo = {
@@ -267,6 +307,9 @@ export async function demux(
           stream: aPipe,
         };
       }
+      // Resolve the metadata wait once at least one stream kind is seen;
+      // keep parsing further chunks so a late audio line still lands.
+      if (seenVideo || seenAudio) parsedMeta = true;
     });
   }
 
@@ -434,3 +477,97 @@ export async function demux(
 
   return { video: vInfo, audio: aInfo, close };
 }
+
+/**
+ * Parse an Ogg Opus byte stream (as written by ffmpeg's `-f opus` muxer)
+ * into individual opus packets and push them onto `out` as GoLiveFrames
+ * (duration 960 @ 48kHz = 20ms, matching the OpusRtpPacketizer clock).
+ *
+ * OGG page structure:
+ *   "OggS" | ver(1) | header_type(1) | granule(8 LE) | serial(4) | seq(4) |
+ *   crc(4) | page_segments(1) | segment_table[n] | payload
+ * Lacing: a value < 255 ends a packet; 255 continues it (0.5KB chunk).
+ * The first packet is OpusHead (19B) — skipped, as is OpusTags.
+ */
+function createOggOpusDemux(
+  input: NodeJS.ReadableStream,
+  out: PassThrough,
+): void {
+  let buf = Buffer.alloc(0);
+  // Packets assembled from lacing; packetParts accumulates across pages
+  // when a packet spans a page boundary (continued flag / 255 lacing).
+  let packetParts: Buffer[] = [];
+  let headerDone = false;
+  let frameIndex = 0;
+
+  const emitPacket = (packet: Buffer) => {
+    if (!headerDone) {
+      // First packet = OpusHead ("OpusHead"), second = OpusTags. Skip both.
+      const magic = packet.toString("latin1", 0, 8);
+      if (magic === "OpusHead" || magic === "OpusTags") return;
+      headerDone = true;
+    }
+    out.write({
+      data: packet,
+      pts: frameIndex * 960,
+      duration: 960,
+      timeBase: { num: 1, den: 48000 },
+      free: () => {},
+    } satisfies GoLiveFrame);
+    frameIndex++;
+  };
+
+  const processPages = () => {
+    while (true) {
+      // Sync to "OggS"
+      const sync = buf.indexOf("OggS", 0, "latin1");
+      if (sync === -1) {
+        // Keep the tail (partial sync pattern) for the next chunk
+        buf = buf.length > 3 ? buf.subarray(buf.length - 3) : buf;
+        return;
+      }
+      if (sync > 0) buf = buf.subarray(sync);
+      if (buf.length < 27) return; // need full page header
+      const numSeg = buf[26];
+      if (buf.length < 27 + numSeg) return; // need segment table
+      let payloadLen = 0;
+      for (let i = 0; i < numSeg; i++) payloadLen += buf[27 + i];
+      if (buf.length < 27 + numSeg + payloadLen) return; // need payload
+
+      const headerType = buf[5];
+      // Extract packets from the payload using lacing values
+      let off = 27 + numSeg;
+      for (let i = 0; i < numSeg; i++) {
+        const lace = buf[27 + i];
+        const part = buf.subarray(off, off + lace);
+        off += lace;
+        packetParts.push(Buffer.from(part));
+        if (lace < 255) {
+          const packet = Buffer.concat(packetParts);
+          packetParts = [];
+          if ((headerType & 0x01) === 0) {
+            // Not a continuation page → packet starts here
+            emitPacket(packet);
+          } else if (headerDone) {
+            // Continued page — packet body, emit directly
+            emitPacket(packet);
+          }
+          // (header packets on continuation pages are dropped)
+        }
+      }
+      buf = buf.subarray(off);
+      if (buf.length === 0) return;
+    }
+  };
+
+  input.on("data", (chunk: Buffer) => {
+    buf = Buffer.concat([buf, chunk]);
+    processPages();
+  });
+  input.on("end", () => {
+    out.end();
+  });
+  input.on("error", () => {
+    out.end();
+  });
+}

services/discord-gateway/src/goLive/Encoders.ts

@@ -41,6 +41,13 @@ export function software(
       options: [
         "-forced-idr 1",
         "-profile:v baseline",
+        // repeat-headers: SPS/PPS inline BEFORE EVERY IDR, not just the
+        // first. Required for the NUT container path (NUT stores extradata
+        // in the header and `-c:v copy` remux loses it → decoder sees
+        // "non-existing PPS 0 referenced" → no picture) and lets Discord's
+        // decoder recover after any PLI/keyframe request mid-stream.
+        "-x264-params",
+        "repeat-headers=1",
         `-tune ${x264Tune}`,
         `-preset ${x264Preset}`,
       ],

services/discord-gateway/src/goLive/WebRtcWrapper.ts

@@ -179,6 +179,9 @@ export class WebRtcConnWrapper {
       throw new Error("WebRTC connection not ready");
     }
     const { audioSsrc, videoSsrc } = this.mediaConnection.webRtcParams;
+    console.log(
+      `[goLive:WebRtc] setPacketizer(${videoCodec}) audioSsrc=${audioSsrc} videoSsrc=${videoSsrc} rtxSsrc=${this.mediaConnection.webRtcParams.rtxSsrc}`,
+    );
     this._videoCodec = normalizeVideoCodec(videoCodec);
     // Audio packetizer: opus 120 @ 48kHz, playout delay ext id 5 (like @dank074)
     this._audioTrack?.setPacketizer(

services/discord-gateway/src/goLive/prepareStream.ts

@@ -12,6 +12,7 @@ import { type ChildProcess, spawn } from "node:child_process";
 import { existsSync, readdirSync } from "node:fs";
 import { join } from "node:path";
 import { PassThrough, type Readable } from "node:stream";
+import { AudioStream } from "./AudioStream.js";
 import { demux } from "./Demuxer.js";
 import { type EncoderSettings, Encoders } from "./Encoders.js";
 import { VideoStream } from "./VideoStream.js";
@@ -27,6 +28,8 @@ export interface PrepareStreamResult {
   height: number;
   frameRate?: number;
   includeAudio: boolean;
+  /** Container the encoder muxes to: "nut" (audio-capable) or "h264" (raw). */
+  format: "nut" | "h264";
 }
 
 function isFiniteNonZero(n: unknown): n is number {
@@ -198,7 +201,11 @@ export function prepareStream(
   }
 
   args.push(...mergedOptions.customFfmpegFlags);
-  args.push("-f", "h264", "pipe:1");
+  // NUT muxer carries video+audio; the raw h264 muxer cannot ("h264 muxer
+  // does not support any stream of type audio" → header write fails →
+  // empty stdout → black tile). Audio delivery requires NUT.
+  const outFormat = mergedOptions.includeAudio ? "nut" : "h264";
+  args.push("-f", outFormat, "pipe:1");
 
   const isUrl = typeof input === "string";
   const proc: ChildProcess = isUrl
@@ -248,6 +255,7 @@ export function prepareStream(
     height: mergedOptions.height,
     frameRate: mergedOptions.frameRate,
     includeAudio: !!mergedOptions.includeAudio,
+    format: outFormat,
   };
 }
 
@@ -274,13 +282,17 @@ export async function playStream(
   const conn = await streamer.createStream();
   console.log("[goLive:playStream] createStream resolved");
 
-  const { video, close: demuxClose } = await demux(prepared.output, {
+  const {
-    format: options.format ?? "nut",
+    video,
+    audio,
+    close: demuxClose,
+  } = await demux(prepared.output, {
+    format: options.format ?? prepared.format ?? "nut",
     frameRate:
       typeof options.frameRate === "number" ? options.frameRate : undefined,
   });
   console.log(
-    `[goLive:playStream] demux done codec=${video?.codecName ?? "?"} ${video?.width ?? 0}x${video?.height ?? 0} fps=${video ? video.framerate_num / video.framerate_den || 30 : 30}`,
+    `[goLive:playStream] demux done codec=${video?.codecName ?? "?"} ${video?.width ?? 0}x${video?.height ?? 0} fps=${video ? video.framerate_num / video.framerate_den || 30 : 30} audio=${audio?.codecName ?? "none"}`,
   );
 
   if (!video) throw new Error("No video stream in media");
@@ -314,6 +326,19 @@ export async function playStream(
   const vStream = new VideoStream(conn);
   video.stream.pipe(vStream);
 
+  // Audio: Discord's GoLive pipeline expects RTP on the audio SSRC too —
+  // a video-only stream (zero audio packets) shows a static tile/thumbnail
+  // instead of live video. Pipe opus frames from the demuxer (silence is
+  // injected at the encoder when the source has no audio track).
+  let aStream: AudioStream | undefined;
+  if (audio) {
+    aStream = new AudioStream(conn);
+    audio.stream.pipe(aStream);
+    console.log(
+      `[goLive:playStream] audio stream attached (${audio.codecName})`,
+    );
+  }
+
   const cleanup = () => {
     try {
       prepared.command.kill("SIGTERM");

services/discord-gateway/src/modules/ai-moderation/searxngSearch.ts

@@ -1,10 +1,11 @@
 import Redis from "ioredis";
 import { createChildLogger } from "@/shared/logger/index";
 import { createAbortControllerWithTimeout } from "@/shared/utils/index";
+import { config } from "../../shared/config/config.js";
 
 const log = createChildLogger("searxng-search");
 
-const SEARXNG_BASE_URL = "https://searxng.imrnes.team";
+const SEARXNG_BASE_URL = config.SEARXNG_BASE_URL;
 const MAX_RESULTS = 3;
 const TIMEOUT_MS = 8000;
 const CACHE_TTL = 86400; // 24 hours

services/discord-gateway/src/modules/ai-moderation/termGlossary.ts

@@ -14,18 +14,26 @@
  * the LLM prompt as a `<term_glossary>` block so verdicts are based on facts
  * instead of guesses.
  *
- * Cost control:
+ * Cost control & persistence:
- *  - definitions are cached in an in-memory LRU AND in Redis (shared with the
+ *  - successfully resolved definitions are PERSISTED PERMANENTLY in Postgres
- *    SearXNG cache) — a term is searched at most once per TTL across the whole
+ *    (`term_glossary_cache`) — definitions rarely change, so a resolved term
- *    service, so repeat lookups are effectively free;
+ *    is never searched again; only misses stay ephemeral (Redis/LRU, 1h);
+ *  - in-memory LRU + Redis (shared with the SearXNG cache) sit in front of
+ *    the DB as fast read caches, so repeat lookups are effectively free;
  *  - lookups per batch are bounded (AI_GLOSSARY_MAX_TERMS);
- *  - Wikipedia-only search first, generic search as a fallback;
+ *  - live SearXNG calls are rate-limit aware: concurrency 2 + stagger, retry
- *  - everything degrades gracefully: no Redis, no SearXNG, no Wikipedia match
+ *    once on empty results, and misses cached for only 1h so a limiter/
+ *    network blip is not treated as a permanent miss;
+ *  - only results that read like actual definitions are accepted (Wikipedia
+ *    preferred; disambiguation/ads/translate-homepages rejected);
+ *  - everything degrades gracefully: no Redis, no SearXNG, no match
  *    → the block is simply omitted and moderation proceeds as before.
  */
 
 import { LRUCache } from "lru-cache";
+import pLimit from "p-limit";
 import { createChildLogger } from "@/shared/logger/index";
+import { delay } from "@/shared/utils/index";
 import { config } from "../../shared/config/config.js";
 import { escapeXml } from "./moderationBuilders.js";
 import {
@@ -34,6 +42,10 @@ import {
   searxngCacheGet,
   searxngCacheSet,
 } from "./searxngSearch.js";
+import {
+  getTermDefinitionFromDb,
+  setTermDefinitionInDb,
+} from "./termGlossaryStore.js";
 
 const log = createChildLogger("term-glossary");
 
@@ -43,14 +55,29 @@ const log = createChildLogger("term-glossary");
 
 /** Redis TTL for a successfully resolved definition (definitions are stable). */
 const DEF_TTL_SECONDS = 7 * 24 * 60 * 60;
-/** Redis TTL for a lookup that found nothing — don't re-search every batch. */
+/**
-const MISS_TTL_SECONDS = 24 * 60 * 60;
+ * Redis TTL for a lookup that found nothing. Kept SHORT (1h): SearXNG
+ * instances silently return empty result sets when rate-limited, so an empty
+ * response is often a transient failure, not a real miss. A short TTL lets
+ * the term be retried on a later batch instead of poisoning it for a day.
+ */
+const MISS_TTL_SECONDS = 60 * 60;
+const MISS_TTL_MS = MISS_TTL_SECONDS * 1000;
 /** Sentinel stored in caches for "term has no resolvable definition". */
 const EMPTY_SENTINEL = "__not_found__";
 /** Per-search timeout — keep glossary lookups snappy even on a slow SearXNG. */
 const GLOSSARY_SEARCH_TIMEOUT_MS = 5000;
+/** Delay before retrying a search that returned zero results. */
+const RETRY_DELAY_MS = 350;
 /** Max definition snippet length kept in the prompt. */
 const MAX_DEFINITION_CHARS = 300;
+/**
+ * SearXNG rate-limits aggressive parallel bursts (returns 200 with empty
+ * results). Never fire all terms at once — cap live searches at 2 concurrent
+ * and stagger the start times slightly.
+ */
+const LIVE_SEARCH_CONCURRENCY = 2;
+const LIVE_SEARCH_STAGGER_MS = 250;
 
 /** In-memory cache: term (lowercase) → definition | NOT_FOUND sentinel. */
 const NOT_FOUND: TermDefinition = {
@@ -63,6 +90,16 @@ const termLru = new LRUCache<string, TermDefinition>({
   ttl: 24 * 60 * 60 * 1000,
 });
 
+/** Serializes live SearXNG lookups (rate-limit aware) with a small stagger. */
+const liveSearchLimit = pLimit(LIVE_SEARCH_CONCURRENCY);
+let lastLiveSearchAt = 0;
+async function acquireLiveSlot(): Promise<void> {
+  const now = Date.now();
+  const wait = lastLiveSearchAt + LIVE_SEARCH_STAGGER_MS - now;
+  if (wait > 0) await delay(wait);
+  lastLiveSearchAt = Date.now();
+}
+
 // ---------------------------------------------------------------------------
 // Term extraction
 // ---------------------------------------------------------------------------
@@ -78,7 +115,7 @@ function cleanContent(raw: string): string {
     .replace(/<@!?\d+>/g, " ")
     .replace(/<#\d+>/g, " ")
     .replace(/<a?:\w+:\d+>/g, " ")
-    .replace(/[`*_~|>\[\]]/g, " ")
+    .replace(/[`*_~|>[\]]/g, " ")
     .replace(/[\p{Emoji}\p{Extended_Pictographic}]/gu, " ")
     .replace(/\s+/g, " ")
     .trim();
@@ -237,15 +274,64 @@ export interface TermDefinition {
   sourceUrl: string;
 }
 
-/** Picks the best definition from search results, preferring Wikipedia. */
+/** Definition-like markers for accepting a non-Wikipedia search result. */
+const DEF_MARKERS =
+  /adalah|merupakan|istilah (?:untuk|yang|yg)|artinya|sebutan|berarti|refers? to|known as|also called|short for|a term (?:for|used)|istilah dalam|kata (?:asing|serapan)? ?untuk/i;
+
+/** True when the term appears in the result text (or a 4+ char word in the
+ *  result is part of the term). Lenient — "kafircel" matches a "Kafir"
+ *  article via substring, while a Google-Translate homepage snippet does not. */
+function hasTermOverlap(term: string, title: string, snippet: string): boolean {
+  const termLower = term.toLowerCase();
+  const text = `${title} ${snippet}`.toLowerCase();
+  if (text.includes(termLower)) return true;
+  const words = text.match(/[a-z0-9]{4,}/gi) ?? [];
+  return words.some((w) => termLower.includes(w));
+}
+
+/** Quality gate: is this result good enough to quote as a definition? */
+function isUsableDefinition(
+  r: { title: string; url: string; snippet: string },
+  term: string,
+  isWiki: boolean,
+): boolean {
+  const text = `${r.title} ${r.snippet}`;
+  // Wikipedia disambiguation pages are not definitions
+  if (/disambiguasi|disambiguation/i.test(text)) return false;
+  if ((r.snippet ?? "").trim().length < 25) return false;
+  if (!hasTermOverlap(term, r.title, r.snippet)) return false;
+  // Wikipedia articles are accepted with just the overlap+length gate;
+  // everything else must read like an actual definition, not an ad,
+  // a translate homepage, or a navigation blurb.
+  if (isWiki) return true;
+  return DEF_MARKERS.test(r.snippet);
+}
+
+/** Picks the best definition from search results, preferring a genuine
+ *  Wikipedia article; otherwise the first result that reads like a
+ *  definition. Returns null when nothing qualifies. */
 function pickDefinition(
   results: Array<{ title: string; url: string; snippet: string }>,
   term: string,
 ): TermDefinition | null {
-  const best = results.find((r) => /wikipedia/i.test(r.url)) ?? results[0];
+  const wiki = results.find((r) => /wikipedia\.org/i.test(r.url));
-  if (!best) return null;
+  const best = wiki && isUsableDefinition(wiki, term, true) ? wiki : null;
+  if (!best) {
+    for (const r of results) {
+      if (isUsableDefinition(r, term, false)) {
+        return buildDefinition(r, term);
+      }
+    }
+    return null;
+  }
+  return buildDefinition(best, term);
+}
+
+function buildDefinition(
+  best: { title: string; url: string; snippet: string },
+  term: string,
+): TermDefinition {
   const snippet = (best.snippet || best.title || "").trim();
-  if (snippet.length < 10) return null;
   const definition =
     snippet.length > MAX_DEFINITION_CHARS
       ? `${snippet.slice(0, MAX_DEFINITION_CHARS - 1).trimEnd()}…`
@@ -253,66 +339,39 @@ function pickDefinition(
   return { term, definition, sourceUrl: best.url };
 }
 
-async function lookupTermDefinition(
+/** Live (network) lookup — runs under the shared SearXNG rate-limit gate. */
+async function fetchDefinitionLive(
   term: string,
+  key: string,
+  cacheKey: string,
 ): Promise<TermDefinition | null> {
-  const key = term.toLowerCase().trim();
+  return liveSearchLimit(async () => {
-
+    await acquireLiveSlot();
-  // 1. In-memory LRU — same process, instant
-  const lruHit = termLru.get(key);
-  if (lruHit) return lruHit === NOT_FOUND ? null : lruHit;
-
-  // 2. Redis — shared across processes/workers
-  const cacheKey = makeSearxngCacheKey("def", key);
-  const cached = await searxngCacheGet(cacheKey);
-  if (cached !== null) {
-    if (cached === EMPTY_SENTINEL) {
-      termLru.set(key, NOT_FOUND);
-      return null;
-    }
     try {
-      const parsed = JSON.parse(cached) as {
+      let results = await searchSearxng(
-        definition?: string;
-        sourceUrl?: string;
-      };
-      if (parsed.definition) {
-        const def: TermDefinition = {
-          term,
-          definition: parsed.definition,
-          sourceUrl: parsed.sourceUrl ?? "",
-        };
-        termLru.set(key, def);
-        return def;
-      }
-    } catch {
-      // malformed cache entry — fall through to search
-    }
-  }
-
-  // 3. Live search — Wikipedia first, generic search as fallback
-  try {
-    let def = pickDefinition(
-      await searchSearxng(
         key,
         "general",
-        "wikipedia",
-        GLOSSARY_SEARCH_TIMEOUT_MS,
-      ),
-      term,
-    );
-    if (!def) {
-      def = pickDefinition(
-        await searchSearxng(
-          `${key} definisi arti`,
-          "general",
         undefined,
         GLOSSARY_SEARCH_TIMEOUT_MS,
-        ),
-        term,
       );
+      let def = pickDefinition(results, term);
+      // Zero results is usually the limiter kicking in, not a real miss —
+      // retry once. Results-but-unusable = genuine miss, no retry.
+      if (!def && results.length === 0) {
+        await delay(RETRY_DELAY_MS);
+        results = await searchSearxng(
+          key,
+          "general",
+          undefined,
+          GLOSSARY_SEARCH_TIMEOUT_MS,
+        );
+        def = pickDefinition(results, term);
       }
 
       if (def) {
+        // Persist permanently (definitions rarely change) — best-effort,
+        // then warm the fast caches.
+        void setTermDefinitionInDb(key, def.definition, def.sourceUrl);
         searxngCacheSet(
           cacheKey,
           JSON.stringify({
@@ -332,15 +391,88 @@ async function lookupTermDefinition(
       );
     }
 
-  // No definition — cache the miss so we do not re-search every batch.
+    // No definition — cache the miss with a SHORT TTL so a transient
+    // limiter/network failure is retried on a later batch.
     searxngCacheSet(cacheKey, EMPTY_SENTINEL, MISS_TTL_SECONDS);
-  termLru.set(key, NOT_FOUND);
+    termLru.set(key, NOT_FOUND, { ttl: MISS_TTL_MS });
     return null;
+  });
+}
+
+/** Resolve one term: LRU → Redis → Postgres (permanent) → live SearXNG
+ *  (rate-limited). The fast caches sit in front of the DB; the DB is the
+ *  source of truth for successfully resolved definitions. */
+async function resolveTerm(term: string): Promise<TermDefinition | null> {
+  const key = term.toLowerCase().trim();
+
+  // 1. In-memory LRU — same process, instant
+  const lruHit = termLru.get(key);
+  if (lruHit) return lruHit === NOT_FOUND ? null : lruHit;
+
+  // 2. Redis — shared across processes/workers. A miss sentinel here is NOT
+  //    a definitive answer: it may predate a permanent DB entry written by
+  //    another process, so we keep going and let the DB decide.
+  const cacheKey = makeSearxngCacheKey("def", key);
+  const cached = await searxngCacheGet(cacheKey);
+  let redisMiss = false;
+  if (cached !== null) {
+    if (cached === EMPTY_SENTINEL) {
+      redisMiss = true;
+    } else {
+      try {
+        const parsed = JSON.parse(cached) as {
+          definition?: string;
+          sourceUrl?: string;
+        };
+        if (parsed.definition) {
+          const def: TermDefinition = {
+            term,
+            definition: parsed.definition,
+            sourceUrl: parsed.sourceUrl ?? "",
+          };
+          termLru.set(key, def);
+          return def;
+        }
+      } catch {
+        // malformed cache entry — fall through to DB/live
+      }
+    }
+  }
+
+  // 3. Postgres — permanent store for resolved definitions. A hit re-warms
+  //    the fast caches so the DB is not hit on every batch.
+  const dbDef = await getTermDefinitionFromDb(key);
+  if (dbDef) {
+    const def: TermDefinition = {
+      term,
+      definition: dbDef.definition,
+      sourceUrl: dbDef.sourceUrl,
+    };
+    termLru.set(key, def);
+    searxngCacheSet(
+      cacheKey,
+      JSON.stringify({ definition: def.definition, sourceUrl: def.sourceUrl }),
+      DEF_TTL_SECONDS,
+    );
+    log.debug({ term: key }, "Term glossary DB hit");
+    return def;
+  }
+
+  // 4. Redis already said "miss" recently and the DB has nothing — respect
+  //    that instead of hammering SearXNG again within the miss window.
+  if (redisMiss) {
+    termLru.set(key, NOT_FOUND, { ttl: MISS_TTL_MS });
+    return null;
+  }
+
+  // 5. Live search (rate-limited + staggered)
+  return fetchDefinitionLive(term, key, cacheKey);
 }
 
 /**
  * Looks up definitions for a batch of terms, in parallel. Returns a map of
  * term → definition for the terms that resolved. Errors/misses are skipped.
+ * Live SearXNG calls are throttled internally (concurrency 2 + stagger).
  */
 export async function lookupTermDefinitions(
   terms: string[],
@@ -348,7 +480,7 @@ export async function lookupTermDefinitions(
   const map = new Map<string, TermDefinition>();
   if (terms.length === 0) return map;
 
-  const results = await Promise.allSettled(terms.map(lookupTermDefinition));
+  const results = await Promise.allSettled(terms.map(resolveTerm));
   for (let i = 0; i < terms.length; i++) {
     const r = results[i];
     if (r.status === "fulfilled" && r.value) {

services/discord-gateway/src/modules/ai-moderation/termGlossaryStore.ts

@@ -0,0 +1,86 @@
+/**
+ * termGlossaryStore.ts
+ *
+ * Permanent Postgres layer for the term glossary. Resolved definitions
+ * (which carry content) are persisted here because they rarely change —
+ * Redis/LRU only act as fast read caches in front of this table. Terms with
+ * no definition (misses) are deliberately NOT persisted; they stay ephemeral
+ * in Redis with a short TTL so transient lookup failures get retried.
+ *
+ * All calls are best-effort: any DB error degrades to a cache miss (the
+ * glossary then falls through to Redis/live search as if the DB layer
+ * didn't exist).
+ */
+
+import { createChildLogger } from "@/shared/logger/index";
+import { executeAll, executeGet } from "../../shared/database/drizzle.js";
+
+const log = createChildLogger("term-glossary-store");
+
+export interface StoredTermDefinition {
+  definition: string;
+  sourceUrl: string;
+}
+
+/**
+ * Read a permanently stored definition for a term (lowercase key).
+ * Returns null when missing or on any DB error (callers fall through).
+ * A successful read bumps hit_count for observability (fire-and-forget).
+ */
+export async function getTermDefinitionFromDb(
+  term: string,
+): Promise<StoredTermDefinition | null> {
+  try {
+    const row = await executeGet(
+      `SELECT definition, source_url FROM term_glossary_cache WHERE term = $1`,
+      [term.toLowerCase().trim()],
+    );
+    if (!row) return null;
+    try {
+      await executeAll(
+        `UPDATE term_glossary_cache SET hit_count = hit_count + 1 WHERE term = $1`,
+        [term.toLowerCase().trim()],
+      );
+    } catch {
+      // hit_count is observability only — never fail a read for it
+    }
+    return {
+      definition: row.definition as string,
+      sourceUrl: (row.source_url as string | null) ?? "",
+    };
+  } catch (error) {
+    log.debug(
+      { error: error instanceof Error ? error.message : String(error) },
+      "getTermDefinitionFromDb failed — falling back to live search",
+    );
+    return null;
+  }
+}
+
+/**
+ * Persist a resolved definition permanently (UPSERT by term).
+ * Only called for successful resolutions — never for misses.
+ * Best-effort: a DB write failure does not affect the returned definition.
+ */
+export async function setTermDefinitionInDb(
+  term: string,
+  definition: string,
+  sourceUrl: string,
+): Promise<void> {
+  try {
+    await executeAll(
+      `INSERT INTO term_glossary_cache (term, definition, source_url, resolved_at, hit_count)
+       VALUES ($1, $2, $3, $4, 0)
+       ON CONFLICT (term) DO UPDATE SET
+         definition = EXCLUDED.definition,
+         source_url = EXCLUDED.source_url,
+         resolved_at = EXCLUDED.resolved_at`,
+      [term.toLowerCase().trim(), definition, sourceUrl, Date.now()],
+    );
+  } catch (error) {
+    log.warn(
+      { error: error instanceof Error ? error.message : String(error) },
+      "setTermDefinitionInDb failed — definition stays memory/Redis only",
+    );
+  }
+}

services/discord-gateway/src/modules/voice-recording/screenShareController.ts

@@ -207,11 +207,13 @@ export class ScreenShareController {
         frameRate: 30,
         bitrateVideo: 2500,
         bitrateVideoMax: 4000,
-        // Video-only GoLive: the -f h264 output muxer cannot carry audio
+        // GoLive with audio: the encoder muxes to NUT (video h264 + opus
-        // ("h264 muxer does not support any stream of type audio" → header
+        // audio) so the audio SSRC carries RTP too. Discord's GoLive
-        // write fails → empty stdout → demux 'Invalid data' → black tile).
+        // pipeline expects audio — a video-only stream shows a static
-        // Audio is not delivered by the GoLive demux path anyway.
+        // tile/thumbnail instead of live video. When the source has no
-        includeAudio: false,
+        // audio track, the encoder's `-map 0:a:0?` yields no audio stream
+        // and the demuxer simply reports none (video still flows).
+        includeAudio: true,
         videoCodec: normalizeVideoCodec("H264"),
       });
       const { command } = prepared;

services/discord-gateway/src/shared/config/index.ts

@@ -92,6 +92,10 @@ export const configSchema = z
 
     // ── Redis ────────────────────────────────────────────────────────────
     REDIS_URL: z.string().default("redis://localhost:6379"),
+    // ── SearXNG ───────────────────────────────────────────────────────────
+    // Instance for web search + term glossary lookups. Override when the
+    // default instance is down/rate-limited.
+    SEARXNG_BASE_URL: z.string().url().default("https://searxng.imrnes.team"),
     // ── Voice PCM WebSocket (direct gateway→backend, bypasses Redis) ────
     VOICE_PCM_WS_ENABLED: z
       .string()

services/discord-gateway/src/shared/database/schema.ts

@@ -435,6 +435,32 @@ export const pgStickerCacheTable = pgTable(
 
 export const stickerCacheTable = pgStickerCacheTable;
 
+/**
+ * Term Glossary Cache Table (PostgreSQL)
+ * Permanently stores resolved term definitions (Wikipedia/SearXNG lookups).
+ * Definitions rarely change, so once a term is successfully resolved it is
+ * persisted here forever — Redis/LRU only act as fast read caches in front.
+ * Terms with NO definition (misses) are NOT stored here; they stay ephemeral
+ * in Redis with a short TTL so transient lookup failures get retried.
+ */
+export const pgTermGlossaryCacheTable = pgTable(
+  "term_glossary_cache",
+  {
+    term: pgText("term").primaryKey(),
+    definition: pgText("definition").notNull(),
+    source_url: pgText("source_url").notNull().default(""),
+    resolved_at: pgBigint("resolved_at", { mode: "number" }).notNull(),
+    hit_count: pgInteger("hit_count").notNull().default(0),
+  },
+  (table) => ({
+    resolvedAtIdx: pgIndex("idx_term_glossary_cache_resolved_at").on(
+      table.resolved_at,
+    ),
+  }),
+);
+
+export const termGlossaryCacheTable = pgTermGlossaryCacheTable;
+
 // =============================================================================
 // Meta / System
 // =============================================================================
@@ -580,6 +606,11 @@ export type TextAnalysisCacheInsert =
 export type StickerCacheRecord = typeof stickerCacheTable.$inferSelect;
 export type StickerCacheInsert = typeof stickerCacheTable.$inferInsert;
 
+// Term Glossary Cache
+export type TermGlossaryCache = typeof termGlossaryCacheTable.$inferSelect;
+export type TermGlossaryCacheInsert =
+  typeof termGlossaryCacheTable.$inferInsert;
+
 // Muxer Jobs
 export type MuxerJob = typeof muxerJobsTable.$inferSelect;
 export type MuxerJobInsert = typeof muxerJobsTable.$inferInsert;

services/discord-gateway/tests/demuxerNut.test.ts

@@ -0,0 +1,187 @@
+import { describe, expect, it } from "vitest";
+import { existsSync, mkdtempSync, readdirSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { PassThrough } from "node:stream";
+import { demux } from "../src/goLive/Demuxer.js";
+
+/** Same resolution as Demuxer.resolveBin: env override → Nix store scan. */
+function resolveFfmpeg(): string | null {
+  const override = process.env.FFMPEG_PATH;
+  if (override && existsSync(override)) return override;
+  const store = "/nix/store";
+  if (existsSync(store)) {
+    for (const entry of readdirSync(store)) {
+      if (!entry.includes("ffmpeg-headless-")) continue;
+      const candidate = join(store, entry, "bin", "ffmpeg");
+      if (existsSync(candidate)) return candidate;
+    }
+  }
+  return existsSync("/usr/bin/ffmpeg") ? "/usr/bin/ffmpeg" : null;
+}
+
+/**
+ * Ogg Opus byte stream built by hand: OpusHead (19B) + a few 20ms opus
+ * frames, wrapped in valid Ogg pages (CRC 0 — the parser ignores CRC).
+ */
+function buildOggOpusBytes(frames: number): Buffer {
+  const opusHead = Buffer.from([
+    0x4f, 0x70, 0x75, 0x73, 0x48, 0x65, 0x61, 0x64, // "OpusHead"
+    0x01, 0x02, 0x38, 0x01, 0x80, 0xbb, 0x00, 0x00,
+    0x00, 0x00, 0x00, // version 1, 2ch, 48000
+  ]);
+  // A minimal valid opus data packet: TOC 0xFC (48kHz stereo 20ms) + payload
+  const dataPacket = Buffer.alloc(20, 0);
+  dataPacket[0] = 0xfc;
+  const makePage = (
+    seq: number,
+    headerType: number,
+    serial: number,
+    packets: Buffer[],
+  ): Buffer => {
+    const segmentTable: number[] = [];
+    const payloadParts: Buffer[] = [];
+    for (const p of packets) {
+      let remaining = p.length;
+      let off = 0;
+      do {
+        const chunk = Math.min(255, remaining);
+        segmentTable.push(chunk);
+        payloadParts.push(p.subarray(off, off + chunk));
+        off += chunk;
+        remaining -= chunk;
+      } while (remaining > 0);
+    }
+    const payload = Buffer.concat(payloadParts);
+    const header = Buffer.alloc(27 + segmentTable.length);
+    header.write("OggS", 0, "latin1");
+    header[4] = 0; // version
+    header[5] = headerType;
+    header.writeUInt32LE(0, 6); // granule (unused)
+    header.writeUInt32LE(0, 10);
+    header.writeUInt32LE(serial, 14);
+    header.writeUInt32LE(seq, 18);
+    header.writeUInt32LE(0, 22); // crc (ignored)
+    header[26] = segmentTable.length;
+    for (let i = 0; i < segmentTable.length; i++) header[27 + i] = segmentTable[i];
+    return Buffer.concat([header, payload]);
+  };
+  const pages: Buffer[] = [];
+  const serial = 0x1234;
+  let seq = 0;
+  // Page 0: BOS + OpusHead (19 bytes, single lacing)
+  pages.push(makePage(seq++, 0x02, serial, [opusHead]));
+  // Pages 1+: data packets, a few per page
+  const perPage = 3;
+  for (let i = 0; i < frames; i += perPage) {
+    const pkts = [];
+    for (let j = 0; j < perPage && i + j < frames; j++) pkts.push(dataPacket);
+    pages.push(makePage(seq++, 0x00, serial, pkts));
+  }
+  return Buffer.concat(pages);
+}
+
+describe("Demuxer NUT path with audio", () => {
+  it("emits video access units AND parsed opus audio frames", async () => {
+    // Real NUT file (video h264 + opus) produced by ffmpeg — generated once
+    // in this test via ffmpeg, skipped if ffmpeg is unavailable.
+    const ffmpeg = resolveFfmpeg();
+    if (!ffmpeg) {
+      console.warn("ffmpeg not found — skipping NUT integration case");
+      return;
+    }
+    const dir = mkdtempSync(join(tmpdir(), "gmw-nut-test-"));
+    const inWebm = join(dir, "in.webm");
+    const inNut = join(dir, "in.nut");
+    try {
+      // Build a tiny webm (vpx + opus) then remux to NUT h264+opus — mirrors
+      // prepareStream(includeAudio) output.
+      const { spawnSync } = await import("node:child_process");
+      const gen = spawnSync(
+        ffmpeg,
+        [
+          "-hide_banner", "-loglevel", "error",
+          "-f", "lavfi", "-i", "testsrc2=size=160x120:rate=10:duration=3",
+          "-f", "lavfi", "-i", "sine=frequency=440:duration=3",
+          "-c:v", "libvpx-vp9", "-b:v", "100k", "-pix_fmt", "yuv420p",
+          "-c:a", "libopus", "-b:a", "48k", "-f", "webm", "-y", inWebm,
+        ],
+        { timeout: 20000 },
+      );
+      if (gen.status !== 0) {
+        console.warn("ffmpeg webm gen failed — skipping", gen.stderr?.toString().slice(0, 200));
+        return;
+      }
+      const enc = spawnSync(
+        ffmpeg,
+        [
+          "-hide_banner", "-loglevel", "error", "-i", inWebm,
+          "-map", "0:v:0", "-c:v", "libx264", "-profile:v", "baseline",
+          "-x264-params", "repeat-headers=1", "-preset", "superfast",
+          "-pix_fmt", "yuv420p", "-g", "10", "-forced-idr", "1",
+          "-map", "0:a:0?", "-c:a", "libopus", "-b:a", "48k", "-ar", "48000", "-ac", "2",
+          "-f", "nut", "-y", inNut,
+        ],
+        { timeout: 20000 },
+      );
+      if (enc.status !== 0) {
+        console.warn("ffmpeg nut gen failed — skipping", enc.stderr?.toString().slice(0, 200));
+        return;
+      }
+      const { readFileSync } = await import("node:fs");
+      // demux() takes string | PassThrough — wrap the file read. Drip the
+      // bytes in quickly (well within demux's 1.5s metadata window) so ffmpeg
+      // prints both stream lines before demux() resolves.
+      const input = new PassThrough();
+      const nutBytes = readFileSync(inNut);
+      const CHUNK = Math.max(1024, Math.floor(nutBytes.length / 20));
+      let off = 0;
+      const drip = setInterval(() => {
+        if (off >= nutBytes.length) {
+          clearInterval(drip);
+          input.end();
+          return;
+        }
+        input.write(nutBytes.subarray(off, off + CHUNK));
+        off += CHUNK;
+      }, 10);
+      const { video, audio, close } = await demux(input, {
+        format: "nut",
+        frameRate: 10,
+      });
+      const vFrames: number[] = [];
+      const aFrames: number[] = [];
+      video?.stream.on("data", (f: { data: Buffer | null; flags: number }) => {
+        if (f.data) vFrames.push(f.data.length);
+      });
+      audio?.stream.on("data", (f: { data: Buffer | null; duration: number }) => {
+        if (f.data) aFrames.push(f.duration);
+      });
+      await new Promise<void>((resolve) => {
+        video?.stream.on("end", resolve);
+        setTimeout(resolve, 6000);
+      });
+      close();
+      expect(vFrames.length).toBeGreaterThan(0);
+      // ~10fps × 1s of video → at least 5 access units
+      expect(vFrames.length).toBeGreaterThanOrEqual(5);
+      // ~50 opus frames/sec of audio
+      expect(aFrames.length).toBeGreaterThan(10);
+      // opus frames are 20ms (duration 960 @ 48kHz)
+      expect(aFrames[0]).toBe(960);
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }, 30000);
+
+  it("parses a hand-built Ogg Opus stream into frames", async () => {
+    // Feed the OGG bytes via the demux's audio path is not directly
+    // exposed — instead verify the parser contract through the NUT path is
+    // covered above; here we sanity-check the byte layout our parser reads.
+    const bytes = buildOggOpusBytes(7);
+    expect(bytes.subarray(0, 4).toString("latin1")).toBe("OggS");
+    // 7 frames + header across pages
+    expect(bytes.includes(Buffer.from("OpusHead"))).toBe(true);
+    expect(bytes.subarray(28, 36).toString("latin1")).toBe("OpusHead");
+  });
+});

services/discord-gateway/tests/golive-demux-live-e2e.ts

@@ -17,10 +17,22 @@ await new Promise<void>((resolve, reject) => {
   const p = spawn(
     FFMPEG,
     [
-      "-hide_banner", "-loglevel", "error",
+      "-hide_banner",
-      "-f", "lavfi", "-i", "testsrc=size=640x360:rate=30:duration=2",
+      "-loglevel",
-      "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p",
+      "error",
-      "-f", "h264", clip,
+      "-f",
+      "lavfi",
+      "-i",
+      "testsrc=size=640x360:rate=30:duration=2",
+      "-c:v",
+      "libx264",
+      "-preset",
+      "ultrafast",
+      "-pix_fmt",
+      "yuv420p",
+      "-f",
+      "h264",
+      clip,
     ],
     { stdio: ["ignore", "ignore", "pipe"] },
   );
@@ -95,7 +107,10 @@ for (const f of frames) {
   const hasParams = types.some((t) => t === 7 || t === 8);
   const isKey = (f.flags & 1) !== 0;
   if (!hasSlice) framesWithoutSlice++;
-  if (types.length === 1 && (types[0] === 7 || types[0] === 8 || types[0] === 6)) {
+  if (
+    types.length === 1 &&
+    (types[0] === 7 || types[0] === 8 || types[0] === 6)
+  ) {
     bareParamSetFrames++;
   }
   if (isKey && hasParams) keyframesWithParamSets++;
@@ -106,20 +121,28 @@ console.log(
   `metadata: ${video.codecName} ${video.width}x${video.height} fps=${video.framerate_num}/${video.framerate_den}`,
 );
 console.log(`frames while stream OPEN (not ended): ${frames.length}`);
-console.log(`frames w/o slice NAL: ${framesWithoutSlice}, bare param-set frames: ${bareParamSetFrames}`);
+console.log(
-console.log(`keyframes with SPS/PPS: ${keyframesWithParamSets}, without: ${keyframesWithoutParamSets}`);
+  `frames w/o slice NAL: ${framesWithoutSlice}, bare param-set frames: ${bareParamSetFrames}`,
+);
+console.log(
+  `keyframes with SPS/PPS: ${keyframesWithParamSets}, without: ${keyframesWithoutParamSets}`,
+);
 if (frames.length === 0) {
   console.error("FAIL: no frames emitted while input still open (deadlock)");
   close();
   process.exit(1);
 }
 if (bareParamSetFrames > 0 || framesWithoutSlice > 0) {
-  console.error("FAIL: demux emitted bare parameter-set frames (must group into access units)");
+  console.error(
+    "FAIL: demux emitted bare parameter-set frames (must group into access units)",
+  );
   close();
   process.exit(1);
 }
 if (frames.some((f) => f.duration !== 1 || f.timeBase.den !== 30)) {
-  console.error("FAIL: frame duration/timeBase not 1/30 (RTP timestamp advance wrong)");
+  console.error(
+    "FAIL: frame duration/timeBase not 1/30 (RTP timestamp advance wrong)",
+  );
   close();
   process.exit(1);
 }