fix(goLive): gateway crash on screenshare stop — unhandledRejection during teardown

Test 00:32 confirmed the video pipeline WORKS (1410 frames @ 1280x720 sent,
ready=true, camera off) but the gateway crashed at stream stop:
unhandledRejection → graceful shutdown → systemd restart (bot offline).

Root cause candidates (both were fire-and-forget promises without .catch):
- BaseMediaConnection.setProtocols().then(...) — rejects when the PC is
  closed while setProtocols is in flight (stream teardown)
- void webRtcConn.createOffer().then(...) — rejects when the PC closes
  while the offer is still gathering

Fixes:
- .catch on both promise chains (log + continue; teardown is expected)
- unhandledRejection handler now treats transient stream errors (EPIPE,
  ERR_STREAM_DESTROYED, ERR_STREAM_WRITE_AFTER_END, ECONNRESET) like
  uncaughtException already does — warn + continue instead of shutting
  down the whole gateway. Non-transient rejections still log + shutdown
  (with String(reason) so the detail actually shows).

services/discord-gateway/src/app/bootstrap.ts

@@ -279,12 +279,51 @@ export async function initializeDiscordGateway() {
   });
 
   process.on("uncaughtException", (err) => {
+    const code =
+      typeof (err as NodeJS.ErrnoException).code === "string"
+        ? (err as NodeJS.ErrnoException).code
+        : "";
+    // Transient stream-teardown errors (voice stop/disconnect races, child
+    // process stdin closed while we still write) are NOT fatal — crashing the
+    // gateway on EPIPE takes the whole bot offline mid-music. Log + continue.
+    if (
+      code === "EPIPE" ||
+      code === "ERR_STREAM_DESTROYED" ||
+      code === "ERR_STREAM_WRITE_AFTER_END" ||
+      code === "ECONNRESET"
+    ) {
+      logger.warn(
+        { error: err },
+        "Uncaught transient stream error — continuing",
+      );
+      return;
+    }
     logger.error({ error: err }, "Uncaught exception");
     gracefulShutdown("uncaughtException");
   });
 
   process.on("unhandledRejection", (reason, promise) => {
-    logger.error({ reason, promise }, "Unhandled rejection");
+    const err =
+      reason instanceof Error ? reason : new Error(String(reason ?? "unknown"));
+    const code = (err as NodeJS.ErrnoException).code ?? "";
+    // Same transient-teardown policy as uncaughtException: a rejection that
+    // fires while a stream is being torn down (EPIPE after ffmpeg stdin
+    // closes, write-after-destroy, socket reset) must NOT take the whole
+    // gateway offline. Log detail + continue. Everything else still shuts
+    // down so real bugs surface.
+    if (
+      code === "EPIPE" ||
+      code === "ERR_STREAM_DESTROYED" ||
+      code === "ERR_STREAM_WRITE_AFTER_END" ||
+      code === "ECONNRESET"
+    ) {
+      logger.warn(
+        { error: err },
+        "Unhandled rejection transient stream error — continuing",
+      );
+      return;
+    }
+    logger.error({ error: err, reason: String(reason) }, "Unhandled rejection");
     gracefulShutdown("unhandledRejection");
   });
 

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

@@ -262,6 +262,9 @@ a=ice-lite
       [audioSection, videoSection, videoRtpMap].join("\n"),
       "answer",
     );
+    console.log(
+      `[goLive:${this.constructor.name}] SELECT_PROTOCOL_ACK processed — remote answer set (${[audioSection, videoSection].join("\n").length}B)`,
+    );
     this.emit("select_protocol_ack");
   }
 
@@ -330,7 +333,15 @@ a=ice-lite
       if (seq) this._sequenceNumber = seq;
       if (op === VoiceOpCodes.READY) {
         this.handleReady(d);
-        this.setProtocols().then(() => this.ready?.(this._webRtcWrapper));
+        this.setProtocols()
+          .then(() => this.ready?.(this._webRtcWrapper))
+          .catch((err: unknown) => {
+            // PC can be closed while setProtocols is in flight (stream
+            // teardown) — don't let that become an unhandledRejection.
+            console.log(
+              `[goLive:${this.constructor.name}] setProtocols rejected during teardown: ${err instanceof Error ? err.message : String(err)}`,
+            );
+          });
         this.setVideoAttributes(false);
       } else if (op >= 4000) {
         console.error(`${this.constructor.name} connection error`, d);
@@ -519,10 +530,14 @@ a=ice-lite
     const reconnect = () => {
       const webRtcConn = this._webRtcWrapper.initWebRtc();
       webRtcConn.onStateChange((state) => {
+        console.log(`[goLive:${this.constructor.name}] pc state => ${state}`);
         if (state === "closed" && !this._closed) reconnect();
       });
       this._webRtcWrapper.onLocalDescription = (sdp) => {
         const rtc_connection_id = randomUUID();
+        console.log(
+          `[goLive:${this.constructor.name}] sending SELECT_PROTOCOL (offer ${sdp.length}B, rtc_connection_id=${rtc_connection_id.slice(0, 8)})`,
+        );
         this.sendOpcode(VoiceOpCodes.SELECT_PROTOCOL, {
           protocol: "webrtc",
           codecs: Object.values(CodecPayloadType),
@@ -532,9 +547,18 @@ a=ice-lite
         });
       };
       // createOffer (binding resolves full SDP incl. candidates after gathering)
-      void webRtcConn.createOffer().then((sdp) => {
-        this._webRtcWrapper.onLocalDescription?.(sdp);
-      });
+      void webRtcConn
+        .createOffer()
+        .then((sdp) => {
+          this._webRtcWrapper.onLocalDescription?.(sdp);
+        })
+        .catch((err: unknown) => {
+          // PC closed while offer is gathering (stream teardown / reconnect) —
+          // swallow, the reconnect loop will start a fresh offer.
+          console.log(
+            `[goLive:${this.constructor.name}] createOffer rejected: ${err instanceof Error ? err.message : String(err)}`,
+          );
+        });
     };
     reconnect();
     return new Promise((resolve) => {

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

@@ -13,9 +13,7 @@
  */
 
 import { spawn } from "node:child_process";
-import { randomUUID } from "node:crypto";
-import { createWriteStream, existsSync, readdirSync } from "node:fs";
-import { tmpdir } from "node:os";
+import { existsSync, readdirSync } from "node:fs";
 import { join } from "node:path";
 import { PassThrough } from "node:stream";
 
@@ -140,9 +138,10 @@ export async function probeStreams(
 
 /**
  * Demux input (URL string or readable stream) into video frames on a
- * PassThrough. Uses ffmpeg -f h264 -c copy for video-only AnnexB output.
- * Returns stream info + the video pipe. Audio is not extracted (GoLive
- * screen share sends silence / uses Discord's mixed audio).
+ * PassThrough. Streams input DIRECTLY into ffmpeg (no spool-to-file — the
+ * live NUT/H264 source never ends, so spooling deadlocks). ffmpeg emits
+ * AnnexB H264 on stdout; NAL units are split into frames on the fly.
+ * Video metadata is parsed from ffmpeg stderr during init.
  */
 export async function demux(
   input: string | PassThrough,
@@ -152,109 +151,23 @@ export async function demux(
   audio: DemuxedStream | undefined;
   close: () => void;
 }> {
-  const _label = randomUUID();
   const vPipe = new PassThrough({ objectMode: true, highWaterMark: 128 });
   const aPipe = new PassThrough({ objectMode: true, highWaterMark: 128 });
 
-  // For stream input, spool to a temp file first so ffprobe can inspect it
-  // (ffprobe needs a seekable file; pipes can't be re-read). The stream is
-  // fully consumed before ffmpeg starts — acceptable for screen-share
-  // sources which are already fully buffered by yt-dlp in practice.
-  let spoolPath: string | null = null;
-  const cleanupSpool = () => {
-    if (spoolPath) {
-      import("node:fs").then(({ unlink }) => unlink(spoolPath!, () => {}));
-      spoolPath = null;
-    }
-  };
-
-  let effectiveInput: string;
-  if (typeof input === "string") {
-    effectiveInput = input;
-  } else {
-    spoolPath = join(tmpdir(), `golive-demux-${_label}.h264`);
-    const ws = createWriteStream(spoolPath);
-    await new Promise<void>((resolve, reject) => {
-      input.pipe(ws);
-      input.on("error", reject);
-      ws.on("finish", resolve);
-      ws.on("error", reject);
-    });
-    effectiveInput = spoolPath;
-  }
-
-  // Probe for codec + dimensions
-  let streams: Array<Record<string, unknown>> = [];
-  try {
-    streams = await probeStreams(effectiveInput);
-  } catch (_e) {
-    // probe failed (e.g. raw h264 without container) — infer h264 default
-    streams = [];
-  }
-
-  const v = streams.find((s) => s.codec_type === "video");
-  const a = streams.find((s) => s.codec_type === "audio");
-  let vInfo: DemuxedStream | undefined;
-  let aInfo: DemuxedStream | undefined;
-
-  if (v) {
-    const codecName = (v.codec_name as string) ?? "h264";
-    const rFrame = (v.r_frame_rate as string) ?? "0/1";
-    const [num, den] = rFrame.split("/").map((n) => Number(n));
-    vInfo = {
-      codec:
-        AVCodecID[
-          (codecName.toUpperCase() as keyof typeof AVCodecID) ??
-            "AV_CODEC_ID_H264"
-        ] ?? AVCodecID.AV_CODEC_ID_H264,
-      codecName,
-      width: (v.width as number) ?? 0,
-      height: (v.height as number) ?? 0,
-      framerate_num: num ?? 0,
-      framerate_den: den ?? 1,
-      sample_rate: 0,
-      stream: vPipe,
-    };
-  } else {
-    // Probe failed (e.g. raw AnnexB h264 input) — still emit frames on the
-    // video pipe; playStream infers dimensions from the first frame.
-    vInfo = {
-      codec: AVCodecID.AV_CODEC_ID_H264,
-      codecName: "h264",
-      width: 0,
-      height: 0,
-      framerate_num: 0,
-      framerate_den: 1,
-      sample_rate: 0,
-      stream: vPipe,
-    };
-  }
-
-  if (a) {
-    const codecName = (a.codec_name as string) ?? "opus";
-    aInfo = {
-      codec:
-        AVCodecID[
-          (codecName.toUpperCase() as keyof typeof AVCodecID) ??
-            "AV_CODEC_ID_OPUS"
-        ],
-      codecName,
-      width: 0,
-      height: 0,
-      framerate_num: 0,
-      framerate_den: 0,
-      sample_rate: Number(a.sample_rate) ?? 0,
-      stream: aPipe,
-    };
-  }
-
-  // Spawn ffmpeg — extract raw video (AnnexB for H264) to stdout
+  const isStream = typeof input !== "string";
   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",
-    "error",
+    "info",
+    // Input format hint: prepareStream always emits raw AnnexB H264 on
+    // pipe:0. Raw H264 has NO magic header, so ffmpeg's auto-detection
+    // fails with "Invalid data found when processing input" whenever the
+    // first bytes arrive late/buffered. Pin the demuxer input format.
+    ...(isStream ? ["-f", "h264"] : []),
     "-i",
-    effectiveInput,
+    isStream ? "pipe:0" : input,
     "-c:v",
     "copy",
     "-an", // no audio in this minimal demuxer
@@ -262,8 +175,113 @@ export async function demux(
     "h264",
     "pipe:1",
   ];
+  const proc = spawn(FFMPEG, args, {
+    stdio: isStream ? ["pipe", "pipe", "pipe"] : ["ignore", "pipe", "pipe"],
+  });
+  console.log(
+    `[goLive:Demuxer] spawn ffmpeg pid=${proc.pid} input=${isStream ? "stream" : input} args=${args.join(" ")}`,
+  );
 
-  const proc = spawn(FFMPEG, args, { stdio: ["ignore", "pipe", "pipe"] });
+  // Pipe live input straight into ffmpeg stdin — never await stream end.
+  if (isStream && proc.stdin) {
+    input.pipe(proc.stdin);
+    input.on("error", () => proc.stdin?.destroy());
+  }
+
+  // Set true once stderr metadata has been parsed (see handler below).
+  let parsedMeta = false;
+
+  // Parse stream metadata from ffmpeg stderr as it arrives (first chunk has
+  // the init lines). Fall back to H264 defaults if parsing fails.
+  let vInfo: DemuxedStream = {
+    codec: AVCodecID.AV_CODEC_ID_H264,
+    codecName: "h264",
+    width: 0,
+    height: 0,
+    framerate_num: 0,
+    framerate_den: 1,
+    sample_rate: 0,
+    stream: vPipe,
+  };
+  let aInfo: DemuxedStream | undefined;
+  let stderrBuf = "";
+  if (proc.stderr) {
+    proc.stderr.on("data", (d: Buffer) => {
+      const text = d.toString();
+      stderrBuf = (stderrBuf + text).slice(-16384);
+      // Surface actionable lines: ffmpeg errors + stream init lines
+      if (/error|invalid|no such|failed|cannot|not found|unable/i.test(text)) {
+        console.log(
+          `[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 }> = [];
+      // biome-ignore lint/suspicious/noAssignInExpressions: regex loop idiom
+      while ((m = streamRe.exec(stderrBuf)) !== null) {
+        found.push({ kind: m[2], codecRaw: m[3] });
+      }
+      const v = found.find((s) => s.kind === "Video");
+      const a = found.find((s) => s.kind === "Audio");
+      if (!v && !a) return;
+      parsedMeta = true;
+      if (v) {
+        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);
+        vInfo = {
+          codec:
+            AVCodecID[
+              (codecName.toUpperCase() as keyof typeof AVCodecID) ??
+                "AV_CODEC_ID_H264"
+            ] ?? AVCodecID.AV_CODEC_ID_H264,
+          codecName,
+          width: dim ? Number(dim[1]) : 0,
+          height: dim ? Number(dim[2]) : 0,
+          framerate_num: fps ? Math.round(Number(fps[1]) * 1000) : 0,
+          framerate_den: fps ? 1000 : 1,
+          sample_rate: 0,
+          stream: vPipe,
+        };
+      }
+      if (a) {
+        const codecName = a.codecRaw.split(" ")[0].toLowerCase();
+        const sr = /(\d+) Hz/.exec(stderrBuf);
+        aInfo = {
+          codec:
+            AVCodecID[
+              (codecName.toUpperCase() as keyof typeof AVCodecID) ??
+                "AV_CODEC_ID_OPUS"
+            ] ?? AVCodecID.AV_CODEC_ID_OPUS,
+          codecName,
+          width: 0,
+          height: 0,
+          framerate_num: 0,
+          framerate_den: 0,
+          sample_rate: sr ? Number(sr[1]) : 0,
+          stream: aPipe,
+        };
+      }
+    });
+  }
+
+  // Wait (briefly) for ffmpeg to print its stream init lines on stderr so
+  // vInfo carries real dimensions/fps. The lines arrive with the first chunk
+  // — a short timeout covers slow starts; callers fall back to sensible
+  // defaults when width/height are 0 anyway.
+  await Promise.race([
+    new Promise<void>((resolve) => {
+      const check = setInterval(() => {
+        if (parsedMeta) {
+          clearInterval(check);
+          resolve();
+        }
+      }, 25);
+    }),
+    new Promise<void>((resolve) => setTimeout(resolve, 1500)),
+  ]);
 
   // Scan stdout for NAL units. Each NAL unit (between start codes) is one frame
   // payload. We emit them individually; the packetizer chain handles FU-A.
@@ -281,6 +299,11 @@ export async function demux(
       free: () => {},
     });
     frameCount++;
+    if (frameCount === 1 || frameCount % 30 === 0) {
+      console.log(
+        `[goLive:Demuxer] frames=${frameCount} last=${nal.length}B key=${isKeyFrame}`,
+      );
+    }
   };
 
   if (proc.stdout) {
@@ -356,11 +379,6 @@ export async function demux(
     });
   }
 
-  if (proc.stderr) {
-    proc.stderr.on("data", () => {
-      /* errors swallowed */
-    });
-  }
   proc.on("close", () => {
     vPipe.end();
     aPipe.end();
@@ -370,7 +388,6 @@ export async function demux(
     proc.kill("SIGTERM");
     vPipe.end();
     aPipe.end();
-    cleanupSpool();
   };
 
   return { video: vInfo, audio: aInfo, close };

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

@@ -48,7 +48,19 @@ export class Streamer {
     this._client = client;
     // listen for gateway dispatch events
     this.client.on("raw", (packet) => {
-      this._gatewayEmitter.emit(packet.t, packet.d);
+      const t = packet.t as string;
+      if (
+        t === "STREAM_CREATE" ||
+        t === "STREAM_SERVER_UPDATE" ||
+        t === "VOICE_STATE_UPDATE" ||
+        t === "VOICE_SERVER_UPDATE"
+      ) {
+        console.log(
+          `[goLive:Streamer] raw dispatch ${t}`,
+          JSON.stringify(packet.d).slice(0, 220),
+        );
+      }
+      this._gatewayEmitter.emit(t, packet.d);
     });
   }
 
@@ -65,6 +77,11 @@ export class Streamer {
   }
 
   sendOpcode(code: number, data: unknown): void {
+    // Direct instrumentation — bypasses the bootstrap debug filter (which
+    // drops messages without [VOICE / [ffmpeg / error / stream).
+    console.log(
+      `[goLive:Streamer] sendOpcode op=${code} d=${JSON.stringify(data)}`,
+    );
     this.client.ws.broadcast({ op: code, d: data });
   }
 
@@ -141,7 +158,6 @@ export class Streamer {
         );
         return;
       }
-      this.signalStream();
       const {
         guildId: clientGuildId,
         channelId: clientChannelId,
@@ -155,40 +171,85 @@ export class Streamer {
         clientUserId,
         clientChannelId,
         (conn) => {
+          clearTimeout(streamTimeout);
+          clearInterval(retryInterval);
           resolve(conn);
         },
       );
       this.voiceConnection.streamConnection = streamConn;
-      this._gatewayEmitter.on(
-        "STREAM_CREATE",
-        (d: { stream_key: string; rtc_server_id: string }) => {
-          const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
-          if (
-            clientGuildId !== guildId ||
-            clientChannelId !== channelId ||
-            clientUserId !== userId
-          ) {
-            return;
-          }
-          streamConn.serverId = d.rtc_server_id;
-          streamConn.streamKey = d.stream_key;
-          streamConn.setSession(session_id);
-        },
-      );
-      this._gatewayEmitter.on(
-        "STREAM_SERVER_UPDATE",
-        (d: { stream_key: string; endpoint: string; token: string }) => {
-          const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
-          if (
-            clientGuildId !== guildId ||
-            clientChannelId !== channelId ||
-            clientUserId !== userId
-          ) {
-            return;
-          }
-          streamConn.setTokens(d.endpoint, d.token);
-        },
+
+      // Attach listeners BEFORE the first signal so a fast dispatch can't
+      // be lost between signalStream() and listener registration.
+      const onStreamCreate = (d: {
+        stream_key: string;
+        rtc_server_id: string;
+      }) => {
+        const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
+        if (
+          clientGuildId !== guildId ||
+          clientChannelId !== channelId ||
+          clientUserId !== userId
+        ) {
+          return;
+        }
+        streamConn.serverId = d.rtc_server_id;
+        streamConn.streamKey = d.stream_key;
+        streamConn.setSession(session_id);
+      };
+      const onStreamServerUpdate = (d: {
+        stream_key: string;
+        endpoint: string;
+        token: string;
+      }) => {
+        const { channelId, guildId, userId } = parseStreamKey(d.stream_key);
+        if (
+          clientGuildId !== guildId ||
+          clientChannelId !== channelId ||
+          clientUserId !== userId
+        ) {
+          return;
+        }
+        streamConn.setTokens(d.endpoint, d.token);
+      };
+      this._gatewayEmitter.on("STREAM_CREATE", onStreamCreate);
+      this._gatewayEmitter.on("STREAM_SERVER_UPDATE", onStreamServerUpdate);
+
+      const cleanup = () => {
+        clearTimeout(streamTimeout);
+        clearInterval(retryInterval);
+        this._gatewayEmitter.removeListener("STREAM_CREATE", onStreamCreate);
+        this._gatewayEmitter.removeListener(
+          "STREAM_SERVER_UPDATE",
+          onStreamServerUpdate,
+        );
+      };
+      const streamTimeout = setTimeout(() => {
+        cleanup();
+        reject(
+          new Error(
+            "Timed out waiting for STREAM_CREATE/STREAM_SERVER_UPDATE from Discord (stream handshake) — voice media session may not be active",
+          ),
+        );
+      }, 12_000);
+
+      // Discord sometimes drops the STREAM_CREATE request silently (upstream
+      // issue #217/#219) — resend a few times instead of giving up after one.
+      let attempt = 0;
+      const retryInterval = setInterval(() => {
+        attempt += 1;
+        if (attempt >= 4) {
+          clearInterval(retryInterval);
+          return;
+        }
+        console.log(
+          `[goLive:Streamer] createStream: retrying STREAM_CREATE (attempt ${attempt + 1}/4)`,
+        );
+        this.signalStream();
+      }, 3_000);
+      console.log(
+        `[goLive:Streamer] createStream: sending STREAM_CREATE (attempt 1/4)`,
       );
+      this.signalStream();
     });
   }
 
@@ -241,6 +302,16 @@ export class Streamer {
       channelId: channel_id,
       botId: user_id,
     } = this.voiceConnection;
+    // Un-deafen before requesting the stream (mimic real client). Do NOT
+    // set self_video: true — that flips on the bot's camera in Discord
+    // (visible to everyone); screen share should not enable the camera.
+    this.sendOpcode(GatewayOpCodes.VOICE_STATE_UPDATE, {
+      guild_id,
+      channel_id,
+      self_mute: false,
+      self_deaf: false,
+      self_video: false,
+    });
     this.sendOpcode(GatewayOpCodes.STREAM_CREATE, {
       type,
       guild_id,

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

@@ -68,6 +68,7 @@ export class WebRtcConnWrapper {
   private _audioTrack: NativeTrack | null = null;
   private _videoTrack: NativeTrack | null = null;
   private _videoCodec: WebRtcVideoCodec | null = null;
+  private _videoFrameLog = 0;
   /** Assigned by BaseMediaConnection to send the gathered SDP to Discord. */
   onLocalDescription: ((sdp: string) => void) | null = null;
 
@@ -114,7 +115,15 @@ export class WebRtcConnWrapper {
   }
 
   sendVideoFrame(frame: Buffer, frametime: number): void {
-    if (!this.ready || !this._videoTrack) return;
+    if (!this.ready || !this._videoTrack) {
+      if (this._videoFrameLog === 0) {
+        console.log(
+          `[goLive:WebRtc] sendVideoFrame DROPPED ready=${this.ready} track=${this._videoTrack !== null}`,
+        );
+        this._videoFrameLog++;
+      }
+      return;
+    }
     const clockRate = CodecPayloadType[this._videoCodec ?? "H264"].clockRate;
     if (this._videoCodec === "H264") {
       let spsRewritten = false;
@@ -157,6 +166,12 @@ export class WebRtcConnWrapper {
     }
     this._videoTrack.sendFrame(frame);
     this._videoTrack.addTimestamp(Math.round((frametime * clockRate) / 1000));
+    this._videoFrameLog++;
+    if (this._videoFrameLog === 1 || this._videoFrameLog % 30 === 0) {
+      console.log(
+        `[goLive:WebRtc] sendVideoFrame #${this._videoFrameLog} bytes=${frame.length} ready=${this.ready}`,
+      );
+    }
   }
 
   setPacketizer(videoCodec: string): void {

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

@@ -262,15 +262,22 @@ export async function playStream(
   options: PlayStreamOptions = {},
 ): Promise<void> {
   const conn = await streamer.createStream();
+  console.log("[goLive:playStream] createStream resolved");
 
   const { video, close: demuxClose } = await demux(prepared.output, {
     format: options.format ?? "nut",
   });
+  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}`,
+  );
 
   if (!video) throw new Error("No video stream in media");
 
   conn.setPacketizer(video.codecName);
   conn.mediaConnection.setSpeaking(true);
+  console.log(
+    `[goLive:playStream] setPacketizer(${video.codecName}) + setSpeaking done`,
+  );
 
   const w =
     typeof options.width === "function"

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

@@ -79,6 +79,22 @@ export function transcodeToHighQualityOgg(
   );
 
   input.pipe(proc.stdin);
+  // ffmpeg teardown closes stdin while the upstream source may still write —
+  // swallow EPIPE / destroyed-stream errors so they don't crash the gateway.
+  proc.stdin.on("error", (err: NodeJS.ErrnoException) => {
+    if (
+      err.code === "EPIPE" ||
+      err.code === "ERR_STREAM_DESTROYED" ||
+      err.code === "ERR_STREAM_WRITE_AFTER_END"
+    ) {
+      logger.debug(
+        { code: err.code },
+        "Transcode stdin closed during teardown",
+      );
+    } else {
+      logger.error({ error: err.message }, "Transcode stdin error");
+    }
+  });
   activeProcesses.add(proc);
 
   const cleanup = () => {
@@ -248,6 +264,20 @@ export function resolveMediaUrl(
     // `--print` headers to stderr — pipe stdout immediately so the child
     // never blocks on a full pipe while we wait for the headers on stderr.
     const mediaStream = new PassThrough();
+    // Teardown (player stop / ffmpeg exit) destroys this stream while
+    // yt-dlp may still push bytes — without a listener an EPIPE /
+    // ERR_STREAM_DESTROYED surfaces as an uncaughtException.
+    mediaStream.on("error", (err: NodeJS.ErrnoException) => {
+      if (
+        err.code === "EPIPE" ||
+        err.code === "ERR_STREAM_DESTROYED" ||
+        err.code === "ERR_STREAM_WRITE_AFTER_END"
+      ) {
+        logger.debug({ code: err.code }, "Media stream closed during teardown");
+      } else {
+        logger.error({ error: err.message }, "Media stream error");
+      }
+    });
     proc.stdout.pipe(mediaStream);
 
     let stderrBuf = "";

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

@@ -105,7 +105,11 @@ export class ScreenShareController {
         frameRate: 30,
         bitrateVideo: 2500,
         bitrateVideoMax: 4000,
-        includeAudio: true,
+        // Video-only GoLive: the -f h264 output muxer cannot carry audio
+        // ("h264 muxer does not support any stream of type audio" → header
+        // write fails → empty stdout → demux 'Invalid data' → black tile).
+        // Audio is not delivered by the GoLive demux path anyway.
+        includeAudio: false,
         videoCodec: normalizeVideoCodec("H264"),
       });
       const { command } = prepared;
@@ -145,6 +149,9 @@ export class ScreenShareController {
       };
       const done = playStream(prepared, this.streamer, {
         type: "go-live",
+        width: 1280,
+        height: 720,
+        frameRate: 30,
       })
         .catch((err: unknown) => {
           // Never let a stream failure become an unhandledRejection — that

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

@@ -56,6 +56,24 @@ export class VoiceTransmitter {
       // Create PCM input stream
       this.pcmStream = new PassThrough();
       this.pcmStream.setMaxListeners(32); // drain listeners accumulate during backpressure
+      // Voice teardown (stop / disconnect / ffmpeg exit) destroys this stream
+      // while Redis PCM messages may still be in flight. Without a listener,
+      // EPIPE / ERR_STREAM_DESTROYED / ERR_STREAM_WRITE_AFTER_END surface as
+      // an uncaughtException and crash the whole gateway.
+      this.pcmStream.on("error", (err: NodeJS.ErrnoException) => {
+        if (
+          err.code === "EPIPE" ||
+          err.code === "ERR_STREAM_DESTROYED" ||
+          err.code === "ERR_STREAM_WRITE_AFTER_END"
+        ) {
+          logger.debug(
+            { code: err.code },
+            "PCM stream closed during voice teardown — ignoring",
+          );
+        } else {
+          logger.error({ error: err.message }, "PCM stream error");
+        }
+      });
 
       // Spawn FFmpeg to encode 24kHz mono PCM → OggOpus
       // Input: 24kHz mono s16le (raw PCM)
@@ -146,7 +164,12 @@ export class VoiceTransmitter {
       );
 
       this.redisSub.on("message", (channel, message) => {
-        if (channel !== this.TRANSMIT_CHANNEL || !this.pcmStream) return;
+        if (
+          !this.isActive ||
+          channel !== this.TRANSMIT_CHANNEL ||
+          !this.pcmStream
+        )
+          return;
 
         try {
           const data = JSON.parse(message);
@@ -161,11 +184,21 @@ export class VoiceTransmitter {
                 this.draining = false;
                 // Re-acquire stream reference (could have been replaced by restart)
                 const currentStream = this.pcmStream;
-                if (!currentStream) return;
+                if (!currentStream || !this.isActive) return;
                 // Flush queued chunks
                 while (this.backpressureQueue.length > 0) {
                   const queued = this.backpressureQueue.shift()!;
-                  if (!currentStream.write(queued)) break;
+                  try {
+                    if (!currentStream.write(queued)) break;
+                  } catch (err) {
+                    logger.debug(
+                      {
+                        error: err instanceof Error ? err.message : String(err),
+                      },
+                      "PCM flush write failed during teardown — ignoring",
+                    );
+                    break;
+                  }
                 }
               });
             }

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

@@ -0,0 +1,63 @@
+// Regression test: demux must emit frames from a LIVE stream that never
+// ends (the NUT/H264 merge output during playback). The old implementation
+// spooled the whole stream to a file first → deadlocked forever → 0 frames.
+// Run: npx tsx tests/golive-demux-live-e2e.ts [ffmpeg-path]
+import { spawn } from "node:child_process";
+import { PassThrough } from "node:stream";
+import { demux } from "../src/goLive/Demuxer.js";
+
+const FFMPEG = process.argv[2] ?? "ffmpeg";
+
+// 1) Generate a 2s H264 test clip to a temp file
+const clip = "/tmp/golive-live-test.h264";
+await new Promise<void>((resolve, reject) => {
+  const p = spawn(
+    FFMPEG,
+    [
+      "-hide_banner", "-loglevel", "error",
+      "-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"] },
+  );
+  let err = "";
+  p.stderr?.on("data", (d: Buffer) => (err += d.toString()));
+  p.on("close", (code) => (code === 0 ? resolve() : reject(new Error(err))));
+});
+
+// 2) Feed the clip through a PassThrough but DON'T end it (live semantics),
+//    with a small pause after the first chunk so demux has time to emit.
+const input = new PassThrough();
+const demuxPromise = demux(input, { format: "h264" });
+const { video, close } = await demuxPromise;
+
+let frames = 0;
+let bytes = 0;
+video.stream.on("data", (frame: { data: Buffer }) => {
+  frames++;
+  bytes += frame.data.length;
+});
+
+const fs = await import("node:fs");
+const buf = fs.readFileSync(clip);
+const chunkSize = 16384;
+for (let i = 0; i < buf.length; i += chunkSize) {
+  input.write(buf.subarray(i, i + chunkSize));
+  if (i === 0) await new Promise((r) => setTimeout(r, 1500));
+}
+// Stream still open — if the old spool logic was here we'd never emit.
+await new Promise((r) => setTimeout(r, 500));
+
+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}, bytes: ${bytes}`);
+if (frames === 0) {
+  console.error("FAIL: no frames emitted while input still open (deadlock)");
+  close();
+  process.exit(1);
+}
+input.end();
+await new Promise((r) => setTimeout(r, 300));
+close();
+console.log("PASS: live stream demux works");
+process.exit(0);