Session_runner cleanup and webrtc refresh fix

flashdreams/flashdreams/api_v2/README.md

@@ -12,7 +12,7 @@ Protocols for the FlashDreams API.
   in `OutputSink.open`.
 - `user_input_event_data.py`: base type for event payloads.
 
-`flashdreams.runtime_v2.session_runner.run_session` drives a session against a
+`flashdreams.runtime_v2.session_runner.SessionRunner.run_session` drives a session against a
 window until the session reports `is_finished` or the window reports a close, or
 for a fixed number of steps a caller asks for. A caller holding an application
 uses `flashdreams.runtime_v2.application_runner.ApplicationRunner` to get there,
@@ -66,7 +66,7 @@ Agreed design decisions. Change them by discussion.
 Threading
 ---------
 
-`run_session` uses two threads, and every window runs that way. Generation is on
+`SessionRunner.run_session` uses two threads, and every window runs that way. Generation is on
 the calling thread; the window gets a thread of its own, ticking at
 `frames_per_second_for_ui` to read input, call `ISession.step_ui`, and write
 finished results. A step that takes longer than one of those ticks does not hold
@@ -84,7 +84,7 @@ a native window needs, and it keeps `IClientWindow` implementations free of
 locking.
 
 Writing happens on that thread too, so a window slower than generation leaves
-results waiting. `run_session` bounds how many wait, with `max_pending`, and
+results waiting. `SessionRunner` bounds how many wait, with `max_pending`, and
 `when_full` decides the rest: `WhenFull.BLOCK` holds generation back so every
 result is presented, which is what a file output wants, and `WhenFull.DROP_OLDEST`
 skips frames to keep latency down, which is what a realtime one wants. The caller
@@ -113,7 +113,7 @@ Not built yet
   fast steps can finish several of them between polls and hand them all the same
   batch. Pacing generation is what would fix it.
 - `ApplicationRunner`: takes an `IApplication` and an `IClientWindow` and drives
-  the main loop. `run_session` is what exists today; it drives a session the
+  the main loop. `SessionRunner` is what exists today; it drives a session the
   caller already created.
 - `flashdreams-run`: a CLI that creates the requested kind of client window,
   loads an application module, and hands both to `ApplicationRunner`. Until it

flashdreams/flashdreams/runtime_v2/application_runner.py

@@ -10,7 +10,7 @@
 from flashdreams.api_v2.application import IApplication
 from flashdreams.api_v2.client_window import IClientWindow
 from flashdreams.runtime_v2.session_desc import SessionDesc
-from flashdreams.runtime_v2.session_runner import run_session
+from flashdreams.runtime_v2.session_runner import SessionRunner
 
 _LOGGER = logging.getLogger(__name__)
 """Logger for an application that could not be closed."""
@@ -45,7 +45,7 @@ def run(
         try:
             self._application.init(commandline_args)
             session = self._application.create_session(session_desc)
-            run_session(session, self._client_window)
+            SessionRunner(session, self._client_window).run_session()
         finally:
             _close_application(
                 self._application, run_failed=sys.exc_info()[0] is not None

flashdreams/flashdreams/runtime_v2/serving/web/app.js

@@ -5,9 +5,11 @@ const peer = new RTCPeerConnection();
 const controls = peer.createDataChannel("controls");
 peer.addTransceiver("video", {direction: "recvonly"});
 
+const video = document.getElementById("video");
+
 peer.ontrack = event => {
-  document.getElementById("video").srcObject =
-    event.streams[0] ?? new MediaStream([event.track]);
+  video.srcObject = event.streams[0] ?? new MediaStream([event.track]);
+  video.play().catch(error => console.error("Unable to play WebRTC video", error));
 };
 
 const send = payload => {
@@ -36,7 +38,19 @@ document.getElementById("reset").onclick = () => {
   send({type: "reset"});
 };
 
-window.addEventListener("beforeunload", () => send({type: "close"}));
+
+async function waitForIceGathering() {
+  if (peer.iceGatheringState === "complete") {
+    return;
+  }
+  await new Promise(resolve => {
+    peer.addEventListener("icegatheringstatechange", () => {
+      if (peer.iceGatheringState === "complete") {
+        resolve();
+      }
+    });
+  });
+}
 
 async function connect() {
   while (true) {
@@ -47,6 +61,7 @@ async function connect() {
     await new Promise(resolve => setTimeout(resolve, 100));
   }
   await peer.setLocalDescription(await peer.createOffer());
+  await waitForIceGathering();
   const response = await fetch("/api/webrtc/offer", {
     method: "POST",
     headers: {"content-type": "application/json"},

flashdreams/flashdreams/runtime_v2/serving/webrtc_server.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import json
+import logging
 import socket
 import threading
 import time
@@ -35,6 +36,8 @@
 _WEB_RESOURCES = files("flashdreams.runtime_v2.serving").joinpath("web")
 _BROWSER_PAGE = _WEB_RESOURCES.joinpath("index.html").read_text(encoding="utf-8")
 _BROWSER_SCRIPT = _WEB_RESOURCES.joinpath("app.js").read_text(encoding="utf-8")
+_LOGGER = logging.getLogger(__name__)
+_DISCONNECT_GRACE_SECONDS = 5.0
 
 
 class _VideoTrack(MediaStreamTrack):
@@ -130,6 +133,7 @@ def __init__(
         self._runner: web.AppRunner | None = None
         self._peer_connection: RTCPeerConnection | None = None
         self._video_track: _VideoTrack | None = None
+        self._disconnect_task: asyncio.Task[None] | None = None
         self._session_desc: SessionDesc | None = None
         self._session_start_ns: int | None = None
         self._closed = False
@@ -159,8 +163,14 @@ def port(self) -> int:
 
     @property
     def url(self) -> str:
-        """Return the browser URL for this server."""
-        return f"http://{self._host}:{self._port}/"
+        """Return a browser URL for this server's local host.
+
+        ``0.0.0.0`` and ``::`` listen on every interface but are not browser
+        destinations. Map them to the local loopback address for the URL shown
+        by standalone applications.
+        """
+        host = "127.0.0.1" if self._host in {"0.0.0.0", "::"} else self._host
+        return f"http://{host}:{self._port}/"
 
     def open(self, session_desc: SessionDesc) -> None:
         """Configure the server for one session's generated video.
@@ -278,14 +288,17 @@ async def _start_server(self) -> None:
 
     async def _serve_browser(self, _: web.Request) -> web.Response:
         """Return the minimal browser client."""
+        _LOGGER.info("WebRTC endpoint called: GET /")
         return web.Response(text=_BROWSER_PAGE, content_type="text/html")
 
     async def _serve_browser_script(self, _: web.Request) -> web.Response:
         """Return the browser client's JavaScript."""
+        _LOGGER.info("WebRTC endpoint called: GET /app.js")
         return web.Response(text=_BROWSER_SCRIPT, content_type="text/javascript")
 
     async def _health(self, _: web.Request) -> web.Response:
         """Report whether the server has an open session and client."""
+        _LOGGER.info("WebRTC endpoint called: GET /healthz")
         return web.json_response(
             {
                 "open": self._session_desc is not None,
@@ -295,14 +308,12 @@ async def _health(self, _: web.Request) -> web.Response:
 
     async def _offer(self, request: web.Request) -> web.Response:
         """Negotiate one browser peer connection."""
+        _LOGGER.info("WebRTC endpoint called: %s %s", request.method, request.path)
         if self._closed:
             raise web.HTTPServiceUnavailable(reason="WebRTC server is closed.")
         session_desc = self._session_desc
         if session_desc is None:
             raise web.HTTPConflict(reason="WebRTC server is not open.")
-        if self._peer_connection is not None:
-            raise web.HTTPConflict(reason="A WebRTC client is already connected.")
-
         try:
             payload = await request.json()
         except (json.JSONDecodeError, web.HTTPException) as error:
@@ -319,12 +330,11 @@ async def _offer(self, request: web.Request) -> web.Response:
         peer_connection = RTCPeerConnection()
         video_track = _VideoTrack(session_desc.frames_per_second_for_ui)
         peer_connection.addTrack(video_track)
-        self._peer_connection = peer_connection
-        self._video_track = video_track
 
         @peer_connection.on("datachannel")
         def on_datachannel(channel: Any) -> None:
-            self._client_connected = True
+            if self._peer_connection is peer_connection:
+                self._client_connected = True
 
             @channel.on("message")
             def on_message(message: Any) -> None:
@@ -335,12 +345,12 @@ def on_message(message: Any) -> None:
 
             @channel.on("close")
             def on_close() -> None:
-                self._record_client_disconnect()
+                _LOGGER.info("WebRTC browser controls channel closed.")
 
         @peer_connection.on("connectionstatechange")
         async def on_connectionstatechange() -> None:
             if peer_connection.connectionState in {"failed", "disconnected", "closed"}:
-                self._record_client_disconnect()
+                await self._release_peer_connection(peer_connection)
 
         try:
             await peer_connection.setRemoteDescription(
@@ -350,17 +360,23 @@ async def on_connectionstatechange() -> None:
                 await peer_connection.createAnswer()
             )
         except Exception:
-            self._peer_connection = None
-            self._video_track = None
             await video_track.close()
             await peer_connection.close()
             raise
 
         local_description = peer_connection.localDescription
         if local_description is None:
+            await video_track.close()
+            await peer_connection.close()
             raise web.HTTPInternalServerError(
                 reason="WebRTC peer did not create an answer."
             )
+        active_peer = self._peer_connection
+        if active_peer is not None:
+            await self._release_peer_connection(active_peer, close_peer=True)
+        self._cancel_disconnect_timer()
+        self._peer_connection = peer_connection
+        self._video_track = video_track
         return web.json_response(
             {"sdp": local_description.sdp, "type": local_description.type}
         )
@@ -377,6 +393,7 @@ def _buffer_browser_message(self, raw_message: object) -> None:
             raise ValueError("Browser event must be a JSON object.")
 
         event_type = payload.get("type")
+        _LOGGER.info("WebRTC browser event received: %s", event_type)
         if event_type == "keyboard":
             key = payload.get("key")
             pressed = payload.get("pressed")
@@ -416,14 +433,49 @@ def _append_event(
         #  The callback stores it in WebRTCClientWindow’s thread-safe queue.
         callback(event)
 
-    def _record_client_disconnect(self) -> None:
-        """Buffer one close event when the active browser disconnects."""
-        if not self._client_connected:
+    async def _release_peer_connection(
+        self,
+        peer_connection: RTCPeerConnection,
+        *,
+        close_peer: bool = False,
+    ) -> None:
+        """Release the active peer so a browser can reconnect.
+
+        A browser refresh closes its data channel and peer connection. That ends
+        only this transport connection; it does not end the session running on
+        the server.
+        """
+        if self._peer_connection is not peer_connection:
             return
         self._client_connected = False
+        self._peer_connection = None
+        track = self._video_track
+        self._video_track = None
+        if track is not None:
+            await track.close()
+        if close_peer:
+            await peer_connection.close()
         if not self._closed:
+            self._disconnect_task = asyncio.create_task(
+                self._close_session_after_disconnect()
+            )
+
+    async def _close_session_after_disconnect(self) -> None:
+        """End a session only when no browser reconnects during the grace period."""
+        try:
+            await asyncio.sleep(_DISCONNECT_GRACE_SECONDS)
+        except asyncio.CancelledError:
+            return
+        if self._peer_connection is None and not self._closed:
+            _LOGGER.info("WebRTC reconnect grace period expired; closing session.")
             self._append_event(CloseUserInputEventData())
 
+    def _cancel_disconnect_timer(self) -> None:
+        """Cancel a pending session close after a browser reconnects."""
+        if self._disconnect_task is not None:
+            self._disconnect_task.cancel()
+            self._disconnect_task = None
+
     async def _enqueue_frames(
         self, frames: tuple[np.ndarray[Any, np.dtype[np.uint8]], ...]
     ) -> None:
@@ -434,6 +486,7 @@ async def _enqueue_frames(
 
     async def _shutdown(self) -> None:
         """Release async server resources on their owning loop."""
+        self._cancel_disconnect_timer()
         peer_connection = self._peer_connection
         self._peer_connection = None
         track = self._video_track