feat: browser stream waits through round gaps instead of disconnecting

Stale bridge state used to end the stream on the first stale read,
so every round gap and every worker restart killed the watch (the
2026-10-30 demo watch died this way when a 15:03 deploy restarted
the worker). Now a stale state switches to a waiting heartbeat
every 5s while the task is still alive (running/queued/paused/
interrupted), and frames resume when the next round's MCP starts
polling. The stream ends only on a terminal task state, a missing
task row, or 10 minutes of continuous staleness. The panel shows
a waiting status and keeps the last frame. Also force-close the
SSE connection on stream end: HTTP/1.1 keep-alive otherwise left
terminal streams hanging with no EOF.

cloud/tests/test_server_browser_proxy.py

@@ -174,6 +174,51 @@ class BrowserProxyTestCase(unittest.TestCase):
         finally:
             conn.close()
 
+    def test_stream_waits_through_stale_state(self):
+        """A stale state must not drop a live task's watch.
+
+        Regression for the 2026-10-30 demo watch: the old loop
+        broke on the first stale read, so every round gap (and
+        every worker restart) killed the stream. Now a stale
+        state switches to a waiting heartbeat while the task is
+        still alive, and only a terminal task status ends it.
+        """
+        Path(self.wt, ".browser-remote-frame.jpg").write_bytes(b"JPEG1")
+        self._write_state("agent")
+        conn = http.client.HTTPConnection("127.0.0.1", self.port,
+                                          timeout=30)
+        try:
+            conn.request("GET", "/tasks/task-b1/browser/stream",
+                         headers={"X-API-Key": self.owner_key})
+            resp = conn.getresponse()
+            self.assertEqual(resp.status, 200)
+            # Each SSE event is two lines: data + blank separator.
+            line = resp.readline().decode("utf-8")
+            payload = json.loads(line[6:])
+            self.assertEqual(payload["image"], "JPEG1")
+            self.assertEqual(resp.readline().decode("utf-8"), "\n")
+            # Round gap: state goes stale. The stream must stay
+            # open and emit a waiting heartbeat within ~5s.
+            self._write_state("agent", fresh=False)
+            line = resp.readline().decode("utf-8")
+            self.assertTrue(line.startswith("data: "), line)
+            payload = json.loads(line[6:])
+            self.assertTrue(payload.get("waiting"))
+            self.assertEqual(payload.get("task_status"), "running")
+            self.assertEqual(resp.readline().decode("utf-8"), "\n")
+            # Terminal task status ends the stream.
+            db = sqlite3.connect(self.db_path)
+            try:
+                db.execute("UPDATE tasks SET status = 'completed'"
+                           " WHERE id = 'task-b1'")
+                db.commit()
+            finally:
+                db.close()
+            line = resp.readline().decode("utf-8")
+            self.assertEqual(line, "")
+        finally:
+            conn.close()
+
 
 if __name__ == "__main__":
     unittest.main()