make ICE recover on failure, resolve keyboard not responding to some users, handle blank screen caused by STATUS_STACK_BUFFER_OVERRUN with a watchdog implementation

start.sh

@@ -113,7 +113,10 @@ socat TCP-LISTEN:12345,reuseaddr,fork \
   EXEC:"pacat --playback --device=mic_sink --format=s16le --rate=48000 --channels=1 --latency-msec=20" &
 echo "socat mic listener started on TCP :12345"
 
-# ── Start streaming agent ─────────────────────────────────────────────────────
+# ── Start streaming agent (with auto-restart) ─────────────────────────────────
+# gst_agent exits with code 1 on ICE failure or other fatal errors.
+# Without a restart loop, the container sits dead until manually restarted.
+# 3 second delay before each restart to avoid a tight crash loop.
 while true; do
     python3 ~/streaming-agent/gst_agent.py
     echo "[gst_agent] exited with code $?, restarting in 3s..."

streaming-agent/gst_agent.py

@@ -20,6 +20,7 @@ import asyncio
 import json
 import logging
 import os
+import queue as _queue
 import subprocess
 import sys
 import threading
@@ -69,15 +70,63 @@ def cf_put(path, body):
         raise RuntimeError(f"CF API error: {data['errorDescription']}")
     return data
 
-# ── Input forwarding via xdotool ──────────────────────────────────────────────
-def xdotool(*args):
+# ── Input forwarding via a serialized xdotool worker ──────────────────────────
+# Previously every input event spawned a fresh `xdotool` process via Popen with
+# NO ordering guarantee. Under load (rapid mouse-move + key events) a keyup
+# could execute before its keydown, leaving a key logically stuck down in X11 —
+# that's the "keyboard stops responding until handoff/restart" bug.
+#
+# Fix: push all input commands onto a single queue processed by ONE worker
+# thread. Commands execute strictly in the order they were received, so a
+# keydown always runs before its matching keyup. Each command still runs as a
+# short-lived xdotool invocation, but serialized, so ordering is guaranteed.
+_xdotool_queue = _queue.Queue()
+
+def _xdotool_worker():
     env = {**os.environ, 'DISPLAY': DISPLAY}
-    subprocess.Popen(
-        ['xdotool'] + list(args),
-        env=env,
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    )
+    while True:
+        args = _xdotool_queue.get()
+
+        # Coalesce a run of consecutive mouse moves into one, but only by
+        # peeking non-destructively: we drain moves, and if we hit a non-move
+        # command we execute the accumulated move, then execute that command.
+        if args and args[0] == 'mousemove_relative':
+            sum_dx = int(args[-2])
+            sum_dy = int(args[-1])
+            pending_cmd = None
+            while True:
+                try:
+                    nxt = _xdotool_queue.get_nowait()
+                except _queue.Empty:
+                    break
+                if nxt and nxt[0] == 'mousemove_relative':
+                    sum_dx += int(nxt[-2])
+                    sum_dy += int(nxt[-1])
+                else:
+                    pending_cmd = nxt
+                    break
+            # Execute the accumulated movement as a single command
+            _run_xdotool(env, ['mousemove_relative', '--', str(sum_dx), str(sum_dy)])
+            # If we pulled a non-move command out of the queue, execute it too
+            if pending_cmd is not None:
+                _run_xdotool(env, [str(a) for a in pending_cmd])
+        else:
+            _run_xdotool(env, [str(a) for a in args])
+
+def _run_xdotool(env, argv):
+    try:
+        subprocess.run(
+            ['xdotool'] + argv,
+            env=env,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=2,
+        )
+    except Exception as e:
+        log.error(f'xdotool error: {e}')
+
+def xdotool(*args):
+    _xdotool_queue.put(args)
 
 # Track held keys so we can release them all on controller change
 _held_keys = set()
@@ -225,8 +274,15 @@ class CSSStreamAgent:
                      GstWebRTC.WebRTCICEConnectionState.COMPLETED):
             self._ice_ready.set()
         elif state == GstWebRTC.WebRTCICEConnectionState.FAILED:
-            log.error('ICE failed')
-            self.loop.quit()
+            log.error('ICE failed — exiting process so start.sh can restart it')
+            # os._exit forces immediate termination of the entire process,
+            # including the main thread blocked in asyncio. self.loop.quit()
+            # alone only stops the GLib thread, leaving the asyncio signaling
+            # loop running forever — which is why it got stuck instead of
+            # restarting.
+            os._exit(1)
+        elif state == GstWebRTC.WebRTCICEConnectionState.DISCONNECTED:
+            log.warning('ICE disconnected — monitoring for recovery')
 
     def set_remote_description(self, sdp_text, sdp_type='answer'):
         _, sdp = GstSdp.SDPMessage.new_from_text(sdp_text)
@@ -261,6 +317,60 @@ class CSSStreamAgent:
         finally:
             self.stop()
 
+# ── CSS process watchdog ──────────────────────────────────────────────────────
+def css_watchdog():
+    # If cstrike_win64.exe crashes (e.g. Wine c0000409 stack overrun), the
+    # stream keeps running but shows a black screen forever with no recovery.
+    # This watchdog detects the crash and relaunches CSS directly via revLoader,
+    # the same way start.sh does on initial boot. gst_agent keeps running and
+    # will simply start capturing the relaunched game once it renders.
+    #
+    # We only start watching once CSS has been seen alive at least once, to
+    # avoid racing its initial startup. Require several consecutive misses to
+    # avoid false positives during a normal controller-triggered relaunch
+    # (the signaling server kills+relaunches CSS on controller change).
+    home = os.path.expanduser('~')
+    revloader = os.path.join(home, 'Counter-Strike Source', 'revLoader.exe')
+    seen_alive = False
+    missing_count = 0
+    while True:
+        time.sleep(5)
+        try:
+            result = subprocess.run(
+                ['pgrep', '-f', 'cstrike_win64.exe'],
+                capture_output=True, text=True
+            )
+            alive = result.returncode == 0
+        except Exception:
+            alive = True  # on error, assume alive (don't false-trigger)
+
+        if alive:
+            seen_alive = True
+            missing_count = 0
+            continue
+
+        if not seen_alive:
+            continue
+
+        missing_count += 1
+        log.warning(f'CSS process not found ({missing_count}/4)')
+        # 4 consecutive misses (~20s) — long enough to not collide with the
+        # signaling server's own kill+relaunch cycle on controller change.
+        if missing_count >= 4:
+            log.error('CSS appears crashed — relaunching revLoader')
+            try:
+                env = dict(os.environ, DISPLAY=DISPLAY)
+                subprocess.Popen(['wine', revloader], env=env,
+                                 stdout=subprocess.DEVNULL,
+                                 stderr=subprocess.DEVNULL)
+                log.info('revLoader relaunched by watchdog')
+            except Exception as e:
+                log.error(f'watchdog relaunch failed: {e}')
+            # Reset so we give the relaunched game time to come up before
+            # counting misses again.
+            seen_alive = False
+            missing_count = 0
+
 # ── Signaling ─────────────────────────────────────────────────────────────────
 async def run_signaling(session_id, video_track_name, audio_track_name):
     log.info(f'Connecting to signaling server {SIG_URL}')
@@ -290,6 +400,10 @@ def main():
     Gst.init(None)
     log.info(f'Starting — {CAP_W}x{CAP_H}@{CAP_FPS}fps display={DISPLAY}')
 
+    # Start the serialized xdotool input worker (guarantees keydown/keyup order)
+    xw = threading.Thread(target=_xdotool_worker, daemon=True)
+    xw.start()
+
     agent = CSSStreamAgent()
     agent.build_pipeline()
     agent.start()
@@ -342,6 +456,10 @@ def main():
     log.info(f'Audio track: {audio_track_name}')
     log.info(f'Streaming — sessionId={session_id}')
 
+    # Start CSS crash watchdog (relaunches the game if it dies unexpectedly)
+    wd = threading.Thread(target=css_watchdog, daemon=True)
+    wd.start()
+
     asyncio.run(run_signaling(session_id, video_track_name, audio_track_name))
 
 if __name__ == '__main__':