fix(spa): a film playing waits for the session after a reconnect

Stream credit sent before the new handshake was refused, and the node's
"Handshake required" replaced the film.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

3 files changed, 45 insertions, 8 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js b/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js
index ca015b9b..91af0338 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport-media.js
@@ -254,7 +254,13 @@ extendTransport(class {
 
   /** Room for `n` more segments. */
   grantStreamCredit(n = 1) {
-    if (!this._connected) return;
+    // Only on a session that is up: the player goes on granting as it plays
+    // from its buffer through a reconnect, and a grant that reached the new
+    // channel before its handshake was answered "Handshake required", which
+    // the player showed in place of the film. The stream went with the lost
+    // session; the reconnect asks for it again (video-player.js
+    // `reconnectPlan`). The same goes for the diagnostic and the stop below.
+    if (!this._sessionUp) return;
     console.log('[stream] grant credit:', n);
     this._send({ type: 'stream_more', v: '0.1', n });
   }
@@ -267,7 +273,7 @@ extendTransport(class {
    * the two halves in one file. The node only logs it.
    */
   sendStreamDiag(diag) {
-    if (!this._connected) return;
+    if (!this._sessionUp) return;
     try { this._send({ type: 'client_diag', v: '0.1', ...diag }); } catch { /* gone */ }
   }
 
@@ -279,7 +285,7 @@ extendTransport(class {
    * the next video answered "server busy".
    */
   stopStream() {
-    if (!this._connected) return;
+    if (!this._sessionUp) return;
     try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ }
   }
 });
diff --git a/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs
index 0f1cd979..606598da 100644
--- a/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs
+++ b/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs
@@ -30,6 +30,7 @@ const {
   rttMs = 300,            // one request → its reply, during the handshake
   lateCloseMs = 0,        // >0: the replaced channel's close event fires this late
   runUntil = 300000,
+  stream = false,         // a film playing instead of a download (see below)
 } = JSON.parse(process.argv[4] || '{}');
 
 for (const level of ['log', 'warn', 'error', 'info', 'debug']) {
@@ -106,6 +107,7 @@ globalThis.fetch = async (url) => ({
 const OPENING = new Set(['handshake', 'handshake_response', 'keypair_bundle_fetch',
   'gek_bundle_fetch', 'join_request', 'device_add_request', 'device_hello']);
 let refusedBeforeHandshake = 0;
+const refusedTypes = new Set();
 let served = 0;
 const pcs = [];
 
@@ -127,6 +129,7 @@ class NodeSession {
     const req_id = msg.req_id;
     if (!this.authed && !OPENING.has(msg.type)) {
       refusedBeforeHandshake += 1;
+      refusedTypes.add(msg.type);
       return this.reply({ type: 'error', detail: 'Handshake required', req_id }, 1);
     }
     switch (msg.type) {
@@ -210,10 +213,26 @@ main.addReconnectListener(() => { reconnected += 1; });
 const written = [];
 const writable = { write: async (bytes) => { written.push(bytes[0]); } };
 
-main.connect('node', 'token', 'group', new Uint8Array(32), identity, null, 'u', 'uid',
-             undefined, undefined, '', 'pk')
-  .then(() => pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, ''))
-  .then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; });
+// A film playing (`stream`): the player goes on granting the node room for
+// segments as it plays from its buffer, and on telling it what it sees, through
+// the connection loss and the reconnect, as video-player.js does. A refusal of
+// either is a node error the player shows in place of the film.
+const opened = main.connect('node', 'token', 'group', new Uint8Array(32), identity, null,
+                            'u', 'uid', undefined, undefined, '', 'pk');
+if (stream) {
+  opened.then(() => {
+    main.requestStream('f'.repeat(64));
+    setInterval(() => {
+      main.grantStreamCredit(1);
+      main.sendStreamDiag({ buffered: 1 });
+    }, 50);
+    outcome = 'done';
+  }, (err) => { outcome = `failed: ${err.message}`; });
+} else {
+  opened
+    .then(() => pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, ''))
+    .then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; });
+}
 
 await run(hangAt);
 pcs[0].channel.session.hung = true;
@@ -223,7 +242,7 @@ pcs[0].onconnectionstatechange();
 await run(runUntil);
 
 process.stdout.write(JSON.stringify({
-  outcome, reconnected, refusedBeforeHandshake, served,
+  outcome, reconnected, refusedBeforeHandshake, refusedTypes: [...refusedTypes], served,
   written: written.length,
   inOrder: written.every((b, i) => b === (i & 255)),
   connections: pcs.length,
diff --git a/packages/meshbay-hub/tests/test_reconnect_race.py b/packages/meshbay-hub/tests/test_reconnect_race.py
index 01c51824..539ae268 100644
--- a/packages/meshbay-hub/tests/test_reconnect_race.py
+++ b/packages/meshbay-hub/tests/test_reconnect_race.py
@@ -75,6 +75,18 @@ def test_the_replaced_channels_close_event_is_ignored(late):
     assert out["outcome"] == "done" and out["written"] == 400
 
 
+@pytest.mark.parametrize("cfg", [{}, {"rttMs": 800}, {"openMs": 2500}])
+def test_a_film_playing_says_nothing_before_the_handshake(cfg):
+    # Reported live on a phone moving from mobile data to Wi-Fi: the player
+    # went on granting stream credit through the reconnect, one grant reached
+    # the new channel before its handshake, and the node's "Handshake required"
+    # replaced the film. A grant or a diagnostic for a stream the lost session
+    # took with it has nowhere to go; the reconnect asks for the stream again.
+    out = _run(stream=True, runUntil=60000, **cfg)
+    assert out["outcome"] == "done" and out["reconnected"] == 1
+    assert out["refusedTypes"] == [], out["refusedTypes"]
+
+
 def _node_opening_types() -> set[str]:
     """What the node answers before the handshake, read from its dispatch."""
     from meshbay_common.protocol import MNP