fix: 浏览器远控传输层改文件桥（Unix socket 在 runsc 下宿主不可见）

根因（2026-10-30 活体实证）：#181 的 .browser-remote.sock 假设
宿主能连容器内 socket，但生产 Docker 任务走 runsc/gVisor，
其 gofer 挂载下容器内建的 socket 只在容器内可见，宿主目录
里只有普通文件（browser-live.png 一直能透传）。后果：server
代理对真实任务一律 503，会话根本没活过。

修法：传输层改四个普通文件（worktree 内，原子写 + 追加日志）：
- .browser-remote-state.json：MCP 每轮重写，mtime 即存活信号，
  server 判 30 秒无心跳为会话死亡（503）；
- .browser-remote-frame.jpg：最新 screencast 帧（base64 文本）；
- .browser-remote-input.jsonl：server 追加，MCP 按偏移量消费
  完整行，执行时再判 holder，半行等下一轮；
- .browser-remote-control.json：{holder, seq}，MCP 首轮即应用，
  接管可先于浏览器启动。
- holder 竞态：control 写下后 state 文件约 50ms 才刷新，server
  的 input/stream 改判 effective_holder（控制文件 mtime 新者胜），
  否则接管后立刻点会误 409。
- 线程契约与控制权模型不变；MCP 主循环仍是唯一触页线程。
- 播种刷新（#181）让老租户自动拿到新版脚本。

测试：socket 测试改文件桥测试（桥轮询/偏移量/半行/陈旧态/
effective_holder + server 代理全链），全量 1357 项通过（skipped 4）。

cloud/tests/test_playwright_bridge.py

@@ -0,0 +1,166 @@
+"""File-bridge remote control tests (no Chromium, no sockets).
+
+The bridge is four regular files in the worktree because runsc
+containers cannot publish a Unix socket to the host (verified
+2026-10-30). These tests drive RemoteBridge.poll() with a stub
+browser and plain files.
+"""
+
+import json
+import os
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+try:
+    from playwright_remote import (
+        RemoteBridge, append_input, effective_holder, read_state,
+        validate_input, write_control)
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.playwright_remote import (
+        RemoteBridge, append_input, effective_holder, read_state,
+        validate_input, write_control)
+
+
+class _StubBrowser:
+    def __init__(self):
+        self.calls = []
+        self.frame = ""
+
+    def cached_info(self):
+        return {"url": "https://example.com", "title": "T",
+                "started": True}
+
+    def click_at(self, x, y):
+        self.calls.append(("click_at", x, y))
+        return {"ok": True}
+
+    def mouse_move(self, x, y):
+        self.calls.append(("mouse_move", x, y))
+        return {"ok": True}
+
+    def key_press(self, key):
+        self.calls.append(("key", key))
+        return {"ok": True}
+
+    def type_text(self, text):
+        self.calls.append(("type", text))
+        return {"ok": True}
+
+    def pump_screencast(self):
+        return self.frame
+
+
+class BridgeTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.browser = _StubBrowser()
+        self.bridge = RemoteBridge(self.browser, self.tmp.name)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_control_and_input_flow(self):
+        # Agent holder: inputs are dropped, not executed.
+        append_input(self.tmp.name,
+                     {"type": "mouse_click", "x": 10, "y": 20})
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [])
+        # Takeover via control file, then a fresh input executes.
+        write_control(self.tmp.name, "user")
+        self.bridge.poll()
+        self.assertEqual(self.bridge.get_holder(), "user")
+        append_input(self.tmp.name,
+                     {"type": "mouse_click", "x": 30, "y": 40})
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [("click_at", 30, 40)])
+        # Second poll re-executes nothing (offset tracking).
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [("click_at", 30, 40)])
+        # Release drops later inputs until the next takeover.
+        write_control(self.tmp.name, "agent")
+        self.bridge.poll()
+        append_input(self.tmp.name, {"type": "key", "key": "Enter"})
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [("click_at", 30, 40)])
+
+    def test_partial_line_waits(self):
+        write_control(self.tmp.name, "user")
+        self.bridge.poll()
+        with open(os.path.join(self.tmp.name,
+                               ".browser-remote-input.jsonl"),
+                  "ab") as fh:
+            fh.write(b'{"type": "key", "key": "En')
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [])
+        with open(os.path.join(self.tmp.name,
+                               ".browser-remote-input.jsonl"),
+                  "ab") as fh:
+            fh.write(b'ter"}\n')
+        self.bridge.poll()
+        self.assertEqual(self.browser.calls, [("key", "Enter")])
+
+    def test_state_and_frame_files(self):
+        self.browser.frame = "FRAMEBYTES"
+        self.bridge.poll()
+        state, err = read_state(self.tmp.name)
+        self.assertIsNone(err)
+        self.assertEqual(state["holder"], "agent")
+        self.assertTrue(state["streaming"])
+        self.assertGreater(state["updated_at"], time.time() - 5)
+        with open(os.path.join(self.tmp.name,
+                               ".browser-remote-frame.jpg"),
+                  "rb") as fh:
+            self.assertEqual(fh.read(), b"FRAMEBYTES")
+
+    def test_read_state_missing_and_stale(self):
+        state, err = read_state(self.tmp.name)
+        self.assertEqual(err[0], 503)
+        self.bridge.poll()
+        path = os.path.join(self.tmp.name, ".browser-remote-state.json")
+        with open(path, "rb") as fh:
+            doc = json.loads(fh.read())
+        doc["updated_at"] = time.time() - 120
+        with open(path, "wb") as fh:
+            fh.write(json.dumps(doc).encode())
+        state, err = read_state(self.tmp.name)
+        self.assertEqual(err[0], 503)
+
+    def test_effective_holder_control_race(self):
+        # State file says agent; a newer control file wins.
+        self.bridge.poll()
+        state, _ = read_state(self.tmp.name)
+        self.assertEqual(effective_holder(self.tmp.name, state),
+                         "agent")
+        write_control(self.tmp.name, "user")
+        self.assertEqual(effective_holder(self.tmp.name, state),
+                         "user")
+        # Once the MCP polls, the state file catches up and both
+        # agree again.
+        self.bridge.poll()
+        state, _ = read_state(self.tmp.name)
+        self.assertEqual(state["holder"], "user")
+        self.assertEqual(effective_holder(self.tmp.name, state),
+                         "user")
+
+    def test_validate_input(self):
+        item, err = validate_input({"type": "mouse_click", "x": 1,
+                                    "y": 2})
+        self.assertIsNone(err)
+        _, err = validate_input({"type": "mouse_click", "x": 2000,
+                                 "y": 2})
+        self.assertEqual(err[0], 400)
+        _, err = validate_input({"type": "mouse_click", "x": 1.5,
+                                 "y": 2})
+        self.assertEqual(err[0], 400)
+        _, err = validate_input({"type": "teleport"})
+        self.assertEqual(err[0], 400)
+        _, err = validate_input("nope")
+        self.assertEqual(err[0], 400)
+
+
+if __name__ == "__main__":
+    unittest.main()