diff --git a/cloud/tests/test_coverage85_misc.py b/cloud/tests/test_coverage85_misc.py
new file mode 100644
index 0000000..1d2c08c
--- /dev/null
+++ b/cloud/tests/test_coverage85_misc.py
@@ -0,0 +1,827 @@
+"""覆盖率 85% 第 4 步第 5 批：workers / autoscaler / playwright 长尾。
+
+只动测试。workers 与 autoscaler 纯函数走真实 SQLite；
+playwright Browser 走类自带的 browser_factory 注入点（假
+playwright 对象），ensure_* 只写临时 codex_home，绝不启动
+真实浏览器。autoscaler 的进程操作以 Popen 替身覆盖启动，
+不拉真实 worker 子进程。
+"""
+
+import json
+import os
+import queue
+import sqlite3
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+from cloud import autoscaler as autoscaler_mod
+from cloud import playwright_browser as pw_mod
+from cloud import workers as workers_mod
+
+
+def _sqlite_db(path, tasks=True, workers=False):
+    conn = sqlite3.connect(path)
+    if tasks:
+        conn.execute(
+            "CREATE TABLE tasks(id TEXT PRIMARY KEY, status TEXT,"
+            " depends_on TEXT, executor TEXT, coordinator_parts TEXT,"
+            " last_heartbeat REAL)")
+    if workers:
+        conn.execute(
+            "CREATE TABLE workers(worker_id TEXT PRIMARY KEY,"
+            " executor TEXT, capacity INTEGER, running INTEGER,"
+            " last_seen REAL, created_at REAL)")
+    conn.commit()
+    return conn
+
+
+class WorkersTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = os.path.join(self.tmp.name, "w.db")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_threshold_override_and_env(self):
+        self.assertEqual(workers_mod.alive_threshold_seconds(30), 30)
+        self.assertEqual(workers_mod.alive_threshold_seconds("bogus"),
+                         120)
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_WORKER_ALIVE_SECONDS": "bogus"}):
+            self.assertEqual(workers_mod.alive_threshold_seconds(), 120)
+        with mock.patch.dict(os.environ,
+                             {"CLOUD_WORKER_ALIVE_SECONDS": "45"}):
+            self.assertEqual(workers_mod.alive_threshold_seconds(), 45)
+
+    def test_register_and_heartbeat_flows(self):
+        with self.assertRaises(ValueError):
+            workers_mod.register_worker(self.db, "")
+        with self.assertRaises(ValueError):
+            workers_mod.heartbeat_worker(self.db, None)
+        row = workers_mod.register_worker(self.db, "w1",
+                                          executor="docker", capacity=2,
+                                          running=1)
+        self.assertEqual(row["worker_id"], "w1")
+        self.assertEqual(row["executor"], "docker")
+        self.assertEqual(row["capacity"], 2)
+        self.assertEqual(row["running"], 1)
+        row = workers_mod.register_worker(self.db, "w1",
+                                          executor="local", capacity=3)
+        self.assertEqual(row["executor"], "local")
+        self.assertEqual(row["running"], 0)
+        # 心跳未给 executor/capacity 时保留旧值，只刷时间。
+        row = workers_mod.heartbeat_worker(self.db, "w1", running=2)
+        self.assertEqual(row["executor"], "local")
+        self.assertEqual(row["capacity"], 3)
+        self.assertEqual(row["running"], 2)
+        # 未注册 worker 的心跳即注册。
+        row = workers_mod.heartbeat_worker(self.db, "w2")
+        self.assertEqual(row["capacity"], 1)
+        self.assertEqual(row["executor"], None)
+
+    def test_list_alive_and_stale(self):
+        workers_mod.register_worker(self.db, "fresh")
+        workers_mod.register_worker(self.db, "stale")
+        conn = sqlite3.connect(self.db)
+        conn.execute("UPDATE workers SET last_seen = ?"
+                     " WHERE worker_id = 'stale'",
+                     (time.time() - 10000,))
+        conn.commit()
+        conn.close()
+        rows = workers_mod.list_workers(self.db, alive_seconds=120)
+        by_id = {r["worker_id"]: r for r in rows}
+        self.assertTrue(by_id["fresh"]["alive"])
+        self.assertFalse(by_id["stale"]["alive"])
+        self.assertEqual([r["worker_id"] for r in rows],
+                         ["fresh", "stale"])
+
+    def test_ensure_old_table_adds_running_column(self):
+        conn = sqlite3.connect(self.db)
+        conn.execute(
+            "CREATE TABLE workers(worker_id TEXT PRIMARY KEY,"
+            " executor TEXT, capacity INTEGER, last_seen REAL,"
+            " created_at REAL)")
+        conn.execute("INSERT INTO workers VALUES('old', 'docker',"
+                     " 1, 1000, 1000)")
+        conn.commit()
+        conn.close()
+        workers_mod.ensure_workers_table(self.db)
+        conn = sqlite3.connect(self.db)
+        cols = [r[1] for r in
+                conn.execute("PRAGMA table_info(workers)").fetchall()]
+        conn.close()
+        self.assertIn("running", cols)
+        # 幂等：已含 running 列时再跑不报错。
+        workers_mod.ensure_workers_table(self.db)
+
+    def test_worker_to_dict_missing_keys(self):
+        d = workers_mod._worker_to_dict({"worker_id": "w"})
+        self.assertEqual(d["worker_id"], "w")
+        self.assertIsNone(d["executor"])
+        self.assertEqual(d["running"], 0)
+        d = workers_mod._worker_to_dict(None)
+        self.assertIsNone(d["worker_id"])
+        self.assertEqual(d["running"], 0)
+
+
+class _FakeCursor:
+    def __init__(self, rows=None):
+        self.rows = rows or []
+
+    def fetchone(self):
+        return self.rows[0] if self.rows else None
+
+    def fetchall(self):
+        return self.rows
+
+
+class _FakeConn:
+    """workers 依赖桩：PRAGMA 抛错、information_schema 可控。"""
+
+    def __init__(self, info_row=None, count_row=None):
+        self.info_row = info_row
+        self.count_row = count_row
+        self.executed = []
+
+    def execute(self, sql, params=None):
+        self.executed.append(sql)
+        if "PRAGMA table_info" in sql:
+            raise RuntimeError("no pragma")
+        if "information_schema.columns" in sql and "COUNT" in sql:
+            return _FakeCursor([self.count_row] if self.count_row
+                               else [])
+        if "information_schema.columns" in sql:
+            if self.info_row is None:
+                raise RuntimeError("no information_schema")
+            return _FakeCursor([self.info_row])
+        if "ALTER TABLE" in sql:
+            raise RuntimeError("alter not supported here")
+        return _FakeCursor()
+
+    def commit(self):
+        pass
+
+    def rollback(self):
+        pass
+
+    def close(self):
+        pass
+
+
+class WorkersFakeConnTest(unittest.TestCase):
+    def test_ensure_running_column_pg_fallback(self):
+        conn = _FakeConn(count_row=(0,))
+        workers_mod._ensure_running_column(conn)
+        self.assertTrue(any("ADD COLUMN IF NOT EXISTS" in s
+                            for s in conn.executed))
+
+    def test_ensure_running_column_pg_query_fails(self):
+        conn = _FakeConn()
+        # information_schema 两次查询都抛错：函数只回滚不抛。
+        workers_mod._ensure_running_column(conn)
+
+    def test_ensure_table_pg_real_column_path(self):
+        conn = _FakeConn(info_row=("real",))
+        with mock.patch.object(workers_mod, "_db_connect",
+                               return_value=conn):
+            with mock.patch.object(workers_mod,
+                                   "_ensure_running_column"):
+                workers_mod.ensure_workers_table("fake.db")
+        self.assertTrue(any("ALTER TABLE workers ALTER COLUMN" in s
+                            for s in conn.executed))
+
+
+class AutoscalerPureTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = os.path.join(self.tmp.name, "a.db")
+        self.conn = _sqlite_db(self.db, workers=True)
+
+    def tearDown(self):
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _add_task(self, tid, status, deps=None, executor=None,
+                  parts=None, hb=None):
+        self.conn.execute(
+            "INSERT INTO tasks VALUES(?, ?, ?, ?, ?, ?)",
+            (tid, status,
+             json.dumps(deps) if deps is not None else None,
+             executor, parts, hb))
+        self.conn.commit()
+
+    def test_queue_depth_variants(self):
+        self._add_task("a", "queued")
+        self._add_task("b", "queued", deps=["a"])
+        self._add_task("c", "completed")
+        self._add_task("d", "queued", deps=["c"])
+        self._add_task("e", "queued", deps=None)
+        self.assertEqual(autoscaler_mod.queue_depth(self.db), 3)
+        # 坏 JSON 依赖当无依赖。
+        self.conn.execute(
+            "INSERT INTO tasks VALUES('f', 'queued', 'not-json',"
+            " NULL, NULL, NULL)")
+        self.conn.commit()
+        self.assertEqual(autoscaler_mod.queue_depth(self.db), 4)
+        self._add_task("i", "queued", executor="docker")
+        self.assertEqual(autoscaler_mod.queue_depth(
+            self.db, executor="local"), 4)
+        self.assertEqual(autoscaler_mod.queue_depth(
+            self.db, executor="docker"), 5)
+        # 成环任务永不就绪（依赖不在 completed 集里），但不影响
+        # 其他就绪任务的计数：仍为 a、d、e、f、i 共 5。
+        self.conn.execute(
+            "INSERT INTO tasks VALUES('g', 'queued', '[\"h\"]',"
+            " NULL, NULL, NULL)")
+        self.conn.execute(
+            "INSERT INTO tasks VALUES('h', 'queued', '[\"g\"]',"
+            " NULL, NULL, NULL)")
+        self.conn.commit()
+        self.assertEqual(autoscaler_mod.queue_depth(self.db), 5)
+
+    def test_desired_workers(self):
+        f = autoscaler_mod.desired_workers
+        self.assertEqual(f(0, 2, 1, 3), 1)
+        self.assertEqual(f(10, 0, 1, 3), 3)
+        self.assertEqual(f(10, 3, 1, 2), 2)
+        self.assertEqual(f(5, 2, 1, 10), 3)
+        self.assertEqual(f(-1, 2, 2, 5), 2)
+
+    def test_live_heartbeat(self):
+        f = autoscaler_mod._is_live_heartbeat
+        now = time.time()
+        self.assertFalse(f(None, now, 120))
+        self.assertFalse(f("bogus", now, 120))
+        self.assertTrue(f(now - 10, now, 120))
+        self.assertTrue(f(now + 60, now, 120))
+        self.assertFalse(f(now - 1000, now, 120))
+
+    def test_running_task_helpers(self):
+        now = time.time()
+        self._add_task("r1", "running",
+                       parts='[{"name": "a", "prompt": "do A"}]',
+                       hb=now - 5)
+        self._add_task("r2", "running", parts="bad-json",
+                       hb=now - 5)
+        self._add_task("r3", "running",
+                       parts='[{"name": "a", "prompt": "do A"}]',
+                       hb=now - 10000)
+        self.assertTrue(autoscaler_mod._has_live_running_coordinator(
+            self.db, now=now))
+        self.assertTrue(autoscaler_mod._has_live_running_task(
+            self.db, now=now))
+        # 全部心跳陈旧时两者皆 False。
+        self.conn.execute("UPDATE tasks SET last_heartbeat = ?",
+                          (now - 10000,))
+        self.conn.commit()
+        self.assertFalse(autoscaler_mod._has_live_running_coordinator(
+            self.db, now=now))
+        self.assertFalse(autoscaler_mod._has_live_running_task(
+            self.db, now=now))
+
+    def test_worker_registry_running(self):
+        now = time.time()
+        f = autoscaler_mod._worker_registry_running
+        self.assertEqual(f(self.db, "nope", now=now), 0)
+        self.conn.execute(
+            "INSERT INTO workers VALUES('w1', 'docker', 2, 3, ?, ?)",
+            (now - 5, now - 5))
+        self.conn.execute(
+            "INSERT INTO workers VALUES('w2', 'docker', 2, 9, ?, ?)",
+            (now - 10000, now - 10000))
+        self.conn.execute(
+            "INSERT INTO workers VALUES('w3', 'docker', 2, NULL, ?, ?)",
+            (now - 5, now - 5))
+        self.conn.commit()
+        self.assertEqual(f(self.db, "w1", now=now), 3)
+        self.assertEqual(f(self.db, "w2", now=now), 0)
+        self.assertEqual(f(self.db, "w3", now=now), 0)
+        # workers 表缺失时按 0 处理。
+        os.remove(self.db)
+        self.assertEqual(f(self.db, "w1", now=now), 0)
+
+    def test_victim_is_busy(self):
+        now = time.time()
+        self.conn.execute(
+            "INSERT INTO workers VALUES('w1', 'docker', 2, 1, ?, ?)",
+            (now - 5, now - 5))
+        self.conn.commit()
+        # 注册表报 busy 但 DB 无活任务：幽灵占位，不拦。
+        self.assertFalse(autoscaler_mod._victim_is_busy(
+            self.db, "w1", now=now))
+        self._add_task("r1", "running", hb=now - 5)
+        self.assertTrue(autoscaler_mod._victim_is_busy(
+            self.db, "w1", now=now))
+        # 注册表报 0：直接不忙。
+        self.assertFalse(autoscaler_mod._victim_is_busy(
+            self.db, "nobody", now=now))
+
+
+class _FakeProc:
+    def __init__(self, pid=4242, poll_result=None, wait_raises=False):
+        self.pid = pid
+        self._poll = poll_result
+        self.returncode = poll_result
+        self.signals = []
+        self.wait_calls = 0
+        self.wait_raises = wait_raises
+
+    def poll(self):
+        return self._poll
+
+    def send_signal(self, sig):
+        self.signals.append(sig)
+
+    def wait(self, timeout=None):
+        self.wait_calls += 1
+        if self.wait_raises and self.wait_calls == 1:
+            import subprocess as _sp
+            raise _sp.TimeoutExpired("cmd", timeout)
+        self._poll = 0
+        self.returncode = 0
+        return 0
+
+
+class AutoscalerProcTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = os.path.join(self.tmp.name, "a.db")
+        self.state = os.path.join(self.tmp.name, "state.json")
+        self.log_dir = os.path.join(self.tmp.name, "logs")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _scaler(self, **kw):
+        kw.setdefault("worker_args", [])
+        kw.setdefault("state_file", self.state)
+        kw.setdefault("log_dir", self.log_dir)
+        return autoscaler_mod.Autoscaler(self.db, **kw)
+
+    def test_worker_command_variants(self):
+        s = self._scaler(worker_cmd=["/bin/worker", "--x"])
+        self.assertEqual(s._worker_command("w1"),
+                         ["/bin/worker", "--x"])
+        s = self._scaler(worker_args=["--loop", "--db", "/tmp/d.db",
+                                      "--host-name", "h1"])
+        cmd = s._worker_command("w9")
+        self.assertIn("--worker-id", cmd)
+        self.assertEqual(cmd.count("--loop"), 1)
+        self.assertEqual(cmd.count("--db"), 1)
+        self.assertEqual(cmd.count("--host-name"), 1)
+        s = self._scaler(worker_args=[])
+        cmd = s._worker_command("w1")
+        self.assertIn("--loop", cmd)
+        self.assertIn("--db", cmd)
+        self.assertIn("--host-name", cmd)
+
+    def test_start_stop_reap_next_id(self):
+        s = self._scaler()
+        procs = []
+        with mock.patch.object(autoscaler_mod.subprocess, "Popen",
+                               side_effect=lambda *a, **k: procs.pop(0)):
+            procs.append(_FakeProc(pid=111))
+            p = s._start_worker("autoscale-1")
+            self.assertEqual(p.pid, 111)
+            self.assertTrue(os.path.exists(
+                os.path.join(self.log_dir, "autoscale-1.log")))
+        self.assertEqual(s._next_worker_id(), "autoscale-2")
+        # 停不存在的 worker 返回 True。
+        self.assertTrue(s._stop_worker("ghost"))
+        # 超时路径：wait 先 TimeoutExpired，再 killpg + 等到退出。
+        s.workers["autoscale-1"]["proc"] = _FakeProc(
+            pid=111, wait_raises=True)
+        with mock.patch.object(autoscaler_mod.os, "killpg"):
+            self.assertTrue(s._stop_worker("autoscale-1"))
+        self.assertNotIn("autoscale-1", s.workers)
+        # reap：已死 proc 被回收、日志句柄关闭。
+        dead = _FakeProc(pid=222, poll_result=1)
+        fh = open(os.path.join(self.tmp.name, "x.log"), "wb")
+        s.workers["autoscale-2"] = {"proc": dead, "started_at": 1.0,
+                                    "log": fh}
+        s._reap()
+        self.assertNotIn("autoscale-2", s.workers)
+        self.assertTrue(fh.closed)
+
+    def test_write_state_and_run_loop_stop(self):
+        s = self._scaler()
+        s.workers["autoscale-1"] = {
+            "proc": _FakeProc(pid=777), "started_at": 123.0,
+            "log": None}
+        s.last_depth = 4
+        s.last_desired = 2
+        s._write_state()
+        with open(self.state) as f:
+            state = json.load(f)
+        self.assertEqual(state["actual"], 1)
+        self.assertEqual(state["workers"][0]["pid"], 777)
+        self.assertEqual(state["depth"], 4)
+        # 无 state_file 时 _write_state 是空操作。
+        s2 = autoscaler_mod.Autoscaler(self.db, [])
+        s2._write_state()
+        # run_loop：stop_event 已置位则立即退出；finally 停掉
+        # 全部 worker 后再写 state，actual 回到 0。
+        import threading
+        ev = threading.Event()
+        ev.set()
+        s.run_loop(stop_event=ev)
+        with open(self.state) as f:
+            state = json.load(f)
+        self.assertEqual(state["actual"], 0)
+        s.request_stop()
+        self.assertTrue(s._stop)
+
+
+# ---------------------------------------------------------- playwright
+
+class _FakeMouse:
+    def __init__(self):
+        self.moves = []
+        self.clicks = []
+
+    def move(self, x, y):
+        self.moves.append((x, y))
+
+    def click(self, x, y):
+        self.clicks.append((x, y))
+
+
+class _FakeKeyboard:
+    def __init__(self):
+        self.pressed = []
+        self.typed = []
+
+    def press(self, key):
+        self.pressed.append(key)
+
+    def type(self, text):
+        self.typed.append(text)
+
+
+class _FakeLocator:
+    def __init__(self, page, selector):
+        self.page = page
+        self.selector = selector
+        self.clicked = False
+
+    def inner_text(self, timeout=None):
+        return self.page.body_text
+
+    @property
+    def first(self):
+        return self
+
+    def click(self, timeout=None):
+        self.clicked = True
+        self.page.locator_clicks.append(self.selector)
+
+
+class _FakeCDP:
+    def __init__(self):
+        self.handlers = {}
+        self.sent = []
+
+    def on(self, event, handler):
+        self.handlers[event] = handler
+
+    def send(self, method, params=None):
+        self.sent.append((method, params))
+        if method == "Page.captureScreenshot":
+            return {"data": "eager-b64"}
+        return {}
+
+
+class _FakeContext:
+    def __init__(self, page):
+        self.page = page
+        self.cdp = None
+
+    def new_cdp_session(self, page):
+        self.cdp = _FakeCDP()
+        return self.cdp
+
+
+class _FakePage:
+    def __init__(self):
+        self.url = "about:blank"
+        self._title = "Fake Title"
+        self.body_text = "hello world"
+        self.mouse = _FakeMouse()
+        self.keyboard = _FakeKeyboard()
+        self.locator_clicks = []
+        self.context = _FakeContext(self)
+        self.shots = []
+        self.goto_fail = False
+        self.eval_result = None
+
+    def title(self):
+        return self._title
+
+    def goto(self, url, timeout=None, wait_until=None):
+        if self.goto_fail:
+            raise RuntimeError("net fail")
+        self.url = url
+
+    def click(self, selector, timeout=None):
+        self.locator_clicks.append(selector)
+
+    def fill(self, selector, text, timeout=None):
+        self.filled = (selector, text)
+
+    def locator(self, selector):
+        return _FakeLocator(self, selector)
+
+    def evaluate(self, script):
+        if self.eval_result is not None:
+            return self.eval_result
+        return "eval-ok"
+
+    def screenshot(self, path=None):
+        with open(path, "wb") as f:
+            f.write(b"PNGDATA")
+        self.shots.append(path)
+
+    def close(self):
+        self.closed = True
+
+
+class _FakeBrowser:
+    def __init__(self, page):
+        self.page = page
+        self.closed = False
+
+    def new_page(self, viewport=None):
+        return self.page
+
+    def close(self):
+        self.closed = True
+
+
+class _FakePW:
+    def __init__(self, page, fail_launch=False):
+        self.page = page
+        self.fail_launch = fail_launch
+        self.stopped = False
+        self.launch_kwargs = None
+
+        class _Chromium:
+            def __init__(self, outer):
+                self.outer = outer
+
+            def launch(self, **kw):
+                self.outer.launch_kwargs = kw
+                if self.outer.fail_launch:
+                    raise RuntimeError("no chromium")
+                return _FakeBrowser(self.outer.page)
+
+        self.chromium = _Chromium(self)
+
+    def start(self):
+        return self
+
+    def stop(self):
+        self.stopped = True
+
+
+class PlaywrightBrowserTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.page = _FakePage()
+        self.env = mock.patch.dict(os.environ, {"HTTPS_PROXY": "",
+                                                "HTTP_PROXY": ""})
+        self.env.start()
+
+    def tearDown(self):
+        self.env.stop()
+        self.tmp.cleanup()
+
+    def _browser(self, proxy=None, fail_launch=False):
+        return pw_mod.Browser(
+            cwd=self.tmp.name, proxy=proxy,
+            browser_factory=lambda: _FakePW(self.page, fail_launch))
+
+    def test_navigate_and_info(self):
+        b = self._browser(proxy="http://proxy:1")
+        out = b.navigate("https://example.com")
+        self.assertTrue(out["ok"])
+        self.assertEqual(out["url"], "https://example.com")
+        self.assertEqual(out["title"], "Fake Title")
+        self.assertEqual(b.cached_info()["url"], "https://example.com")
+        self.assertTrue(b.cached_info()["started"])
+        with self.assertRaises(pw_mod.BrowserError):
+            b.navigate("   ")
+        self.page.goto_fail = True
+        with self.assertRaises(pw_mod.BrowserError):
+            b.navigate("https://example.com")
+
+    def test_launch_failure(self):
+        b = self._browser(fail_launch=True)
+        with self.assertRaises(pw_mod.BrowserError) as cm:
+            b.snapshot()
+        self.assertIn("browser launch failed", str(cm.exception))
+
+    def test_snapshot_truncation_and_failure(self):
+        b = self._browser()
+        out = b.snapshot()
+        self.assertEqual(out["text"], "hello world")
+        self.assertFalse(out["truncated"])
+        self.page.body_text = "x" * 9000
+        out = b.snapshot()
+        self.assertTrue(out["truncated"])
+        self.assertEqual(len(out["text"]), 8000)
+        orig = self.page.locator
+        self.page.locator = lambda s: (_ for _ in ()).throw(
+            RuntimeError("loc fail"))
+        with self.assertRaises(pw_mod.BrowserError):
+            b.snapshot()
+        self.page.locator = orig
+
+    def test_click_fill_key_type(self):
+        b = self._browser()
+        self.assertTrue(b.click("button.ok")["ok"])
+        self.assertIn("button.ok", self.page.locator_clicks)
+        with self.assertRaises(pw_mod.BrowserError):
+            b.click("")
+        self.assertTrue(b.fill("#q", None)["ok"])
+        self.assertEqual(self.page.filled, ("#q", ""))
+        with self.assertRaises(pw_mod.BrowserError):
+            b.fill("", "x")
+        self.assertTrue(b.key_press("Enter")["ok"])
+        self.assertEqual(self.page.keyboard.pressed, ["Enter"])
+        with self.assertRaises(pw_mod.BrowserError):
+            b.key_press(" ")
+        self.assertTrue(b.type_text("hi")["ok"])
+        self.assertEqual(self.page.keyboard.typed, ["hi"])
+        with self.assertRaises(pw_mod.BrowserError):
+            b.type_text(None)
+
+    def test_mouse_and_click_at(self):
+        b = self._browser()
+        out = b.mouse_move(10, 20)
+        self.assertEqual((out["x"], out["y"]), (10.0, 20.0))
+        self.assertEqual(self.page.mouse.moves, [(10.0, 20.0)])
+        with self.assertRaises(pw_mod.BrowserError):
+            b.mouse_move("a", 1)
+        with self.assertRaises(pw_mod.BrowserError):
+            b.mouse_move(2000, 10)
+        out = b.click_at(5, 6)
+        self.assertTrue(out["ok"])
+        self.assertEqual(self.page.mouse.clicks, [(5.0, 6.0)])
+        with self.assertRaises(pw_mod.BrowserError):
+            b.click_at("x", 1)
+        with self.assertRaises(pw_mod.BrowserError):
+            b.click_at(10, 900)
+
+    def test_snapshot_refs_and_click_ref(self):
+        b = self._browser()
+        with self.assertRaises(pw_mod.BrowserError):
+            b.click_ref("e0")
+        self.page.eval_result = [
+            {"tag": "button", "role": "", "name": "OK",
+             "path": "body > button"},
+            {"tag": "a", "role": "link", "name": "Home",
+             "path": "body > a"}]
+        out = b.snapshot_refs()
+        self.assertEqual(out["count"], 2)
+        self.assertEqual(out["elements"][0]["ref"], "e0")
+        out = b.click_ref("e1")
+        self.assertTrue(out["ok"])
+        self.assertEqual(out["ref"], "e1")
+        self.assertIn("body > a", self.page.locator_clicks)
+
+    def test_screencast_sink_and_pump(self):
+        b = self._browser()
+        self.assertEqual(b.pump_screencast(), "")
+        sink = queue.Queue()
+        b.set_frame_sink(sink)
+        b.navigate("https://example.com")
+        # navigate 后 sink 已接上 CDP（_start_screencast 在
+        # navigate 内跑）；再显式 set 一次走“已接同一页”短路。
+        b.set_frame_sink(sink)
+        cdp = self.page.context.cdp
+        self.assertIsNotNone(cdp)
+        cdp.handlers["Page.screencastFrame"](
+            {"sessionId": "s1", "data": "frame-b64"})
+        self.assertEqual(b.pump_screencast(), "frame-b64")
+        self.assertIn(("Page.screencastFrameAck",
+                       {"sessionId": "s1"}), cdp.sent)
+        # eager 帧（空 session）只存不 ack。
+        sink.put(("", "eager-b64"))
+        self.assertEqual(b.pump_screencast(), "eager-b64")
+        self.assertEqual(
+            [s for s in cdp.sent if s[0] == "Page.screencastFrameAck"],
+            [("Page.screencastFrameAck", {"sessionId": "s1"})])
+
+    def test_live_shot_screenshot_evaluate_close(self):
+        b = self._browser()
+        status = b.live_shot()
+        self.assertEqual(status["image"], "browser-live.png")
+        self.assertEqual(status["bytes"], 7)
+        with open(status["status_path"]) as f:
+            sidecar = json.load(f)
+        self.assertEqual(sidecar["image"], "browser-live.png")
+        shot = b.screenshot()
+        self.assertTrue(shot["filename"].startswith("shot-"))
+        self.assertEqual(shot["bytes"], 7)
+        shot2 = b.screenshot("my.png")
+        self.assertEqual(shot2["filename"], "my.png")
+        with self.assertRaises(pw_mod.BrowserError):
+            b.screenshot("../escape.png")
+        self.assertEqual(b.evaluate("1+1")["result"], "eval-ok")
+        with self.assertRaises(pw_mod.BrowserError):
+            b.evaluate("  ")
+        out = b.close()
+        self.assertTrue(out["closed"])
+        self.assertEqual(out["errors"], [])
+        # 已关闭后再次 close 仍返回 closed（各 closer 遇 None 短路）。
+        out = b.close()
+        self.assertTrue(out["closed"])
+
+
+class PlaywrightEnsureTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.src = os.path.join(self.tmp.name, "src")
+        self.home = os.path.join(self.tmp.name, "home")
+        os.makedirs(self.src)
+        for name in ("playwright_browser.py", "playwright_mcp.py",
+                     "playwright_remote.py"):
+            with open(os.path.join(self.src, name), "w") as f:
+                f.write("# %s v1\n" % name)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_seed_refresh_and_config(self):
+        out = pw_mod.ensure_codex_home_playwright(
+            self.home, script_dir=self.src)
+        self.assertTrue(out["script_changed"])
+        self.assertTrue(out["config_changed"])
+        with open(os.path.join(self.home, "config.toml")) as f:
+            cfg = f.read()
+        self.assertIn("[mcp_servers.playwright]", cfg)
+        # 第二次：脚本相同、配置已有 section，全部不变。
+        out = pw_mod.ensure_codex_home_playwright(
+            self.home, script_dir=self.src)
+        self.assertFalse(out["script_changed"])
+        self.assertFalse(out["config_changed"])
+        # 旧脚本被刷新上报。
+        with open(os.path.join(self.home,
+                               "playwright_mcp.py"), "w") as f:
+            f.write("# stale\n")
+        out = pw_mod.ensure_codex_home_playwright(
+            self.home, script_dir=self.src)
+        self.assertIn("playwright_mcp.py", out["script_refreshed"])
+        # 自定义 command_args 写进 section（删旧配置重写，
+        # 否则已有 section 不会被改写）。
+        os.remove(os.path.join(self.home, "config.toml"))
+        out = pw_mod.ensure_codex_home_playwright(
+            self.home, command_args=["py3", "mcp.py", "--x"],
+            script_dir=self.src)
+        self.assertTrue(out["config_changed"])
+        with open(os.path.join(self.home, "config.toml")) as f:
+            cfg = f.read()
+        self.assertIn('"py3"', cfg)
+
+    def test_config_no_trailing_newline(self):
+        os.makedirs(self.home)
+        with open(os.path.join(self.home, "config.toml"), "w") as f:
+            f.write("[other]")
+        pw_mod.ensure_codex_home_playwright(self.home,
+                                            script_dir=self.src)
+        with open(os.path.join(self.home, "config.toml")) as f:
+            cfg = f.read()
+        self.assertIn("[other]\n\n[mcp_servers.playwright]", cfg)
+
+    def test_missing_src_script_skipped(self):
+        os.remove(os.path.join(self.src, "playwright_remote.py"))
+        out = pw_mod.ensure_codex_home_playwright(
+            self.home, script_dir=self.src)
+        self.assertTrue(out["script_changed"])
+        self.assertFalse(os.path.exists(
+            os.path.join(self.home, "playwright_remote.py")))
+
+    def test_mcp_env_writer(self):
+        os.makedirs(self.home)
+        self.assertFalse(pw_mod.ensure_playwright_mcp_env(
+            self.home, {}))
+        self.assertTrue(pw_mod.ensure_playwright_mcp_env(
+            self.home, {"B": "2", "A": "1"}))
+        with open(os.path.join(self.home, "config.toml")) as f:
+            cfg = f.read()
+        self.assertIn("[mcp_servers.playwright.env]", cfg)
+        self.assertLess(cfg.index('A = '), cfg.index('B = '))
+        # 幂等替换：旧 env 段被新段取代，不残留旧 key。
+        self.assertTrue(pw_mod.ensure_playwright_mcp_env(
+            self.home, {"C": "3"}))
+        with open(os.path.join(self.home, "config.toml")) as f:
+            cfg = f.read()
+        self.assertIn('C = ', cfg)
+        self.assertNotIn('A = ', cfg)
+
+
+if __name__ == "__main__":
+    unittest.main()