perf: 桥变化才写、stdin 优先、stream 50ms 轮询

1. 桥写放大：旧代码每 ~50ms 轮询都无条件重写 state 和帧
   文件（约 40 次原子 temp+replace/s 走 runsc gofer，
   泵持久返回同一帧、updated_at 每轮都变，字节其实不变）。
   现帧字节不变不写；state 只在 url/title/started/holder/
   streaming 变化时写，或 0.5s 心跳一次（判死阈值 30s，
   绰绰有余）；静态页写量从 ~40次/s 降到 ~2次/s。
2. MCP 主循环 stdin 优先：旧循环先 drain 再读 stdin，
   每次工具调用都先付一轮文件 IO。现 pending 请求先处理，
   空闲轮询 drain，忙时 200ms 保底 drain 一次防饿死。
3. stream 轮询 200ms → 50ms，画面延迟下限随之下降。
附回归测试：无变化轮询不重写、帧变化立即写、holder
变化绕过节流立即写。

cloud/tests/test_playwright_bridge.py

@@ -116,6 +116,41 @@ class BridgeTestCase(unittest.TestCase):
                   "rb") as fh:
             self.assertEqual(fh.read(), b"FRAMEBYTES")
 
+    def test_writes_are_change_gated(self):
+        """Unchanged polls must not rewrite the bridge files.
+
+        The old code rewrote both files every ~50ms iteration
+        (~40 atomic temp+replace round-trips per second through
+        the runsc gofer mount for identical bytes). Now only
+        changed bytes and the heartbeat get written.
+        """
+        state_path = os.path.join(
+            self.tmp.name, ".browser-remote-state.json")
+        frame_path = os.path.join(
+            self.tmp.name, ".browser-remote-frame.jpg")
+        self.browser.frame = "FRAMEBYTES"
+        self.bridge.poll()
+        old = time.time() - 100
+        os.utime(state_path, (old, old))
+        os.utime(frame_path, (old, old))
+        # Unchanged second poll: neither file rewritten.
+        self.bridge.poll()
+        self.assertEqual(os.stat(state_path).st_mtime, old)
+        self.assertEqual(os.stat(frame_path).st_mtime, old)
+        # Changed frame: rewritten at once.
+        self.browser.frame = "NEWBYTES"
+        self.bridge.poll()
+        with open(frame_path, "rb") as fh:
+            self.assertEqual(fh.read(), b"NEWBYTES")
+        # Changed holder: state rewritten at once, bypassing
+        # the heartbeat throttle.
+        os.utime(state_path, (old, old))
+        write_control(self.tmp.name, "user")
+        self.bridge.poll()
+        state, err = read_state(self.tmp.name)
+        self.assertIsNone(err)
+        self.assertEqual(state["holder"], "user")
+
     def test_read_state_missing_and_stale(self):
         state, err = read_state(self.tmp.name)
         self.assertEqual(err[0], 503)