host
962bb50a1c
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.
225 lines
9.0 KiB
Python
225 lines
9.0 KiB
Python
"""Browser endpoints over the file bridge: /tasks/{id}/browser/
|
|
{state,input,control,stream}.
|
|
|
|
The control plane never talks to the MCP process directly: it
|
|
reads/writes the bridge files in the task worktree (runsc
|
|
containers cannot publish a Unix socket to the host). Here the
|
|
worktree files are written by hand; no Chromium, no worker.
|
|
"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
|
|
|
|
class BrowserProxyTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.auth = AuthStore(self.db_path)
|
|
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
|
self.outsider_id, self.outsider_key = self.auth.create_user("o")
|
|
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
|
self.worktrees = os.path.join(self.tmp.name, "host-worktrees")
|
|
os.makedirs(self.worktrees)
|
|
self.server.host_worktrees_dir = self.worktrees
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(
|
|
target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
status, self.project = self._req(
|
|
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
|
self.assertEqual(status, 201)
|
|
self._task("task-b1")
|
|
self.wt = os.path.join(self.worktrees, "task-b1")
|
|
os.makedirs(self.wt)
|
|
self._write_state("agent")
|
|
self._task("task-nosock")
|
|
os.makedirs(os.path.join(self.worktrees, "task-nosock"))
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
try:
|
|
self.auth.conn.close()
|
|
except Exception:
|
|
pass
|
|
self.tmp.cleanup()
|
|
|
|
def _write_state(self, holder, fresh=True):
|
|
doc = {"url": "https://example.com", "title": "T",
|
|
"started": True, "holder": holder,
|
|
"streaming": False,
|
|
"updated_at": time.time() if fresh else time.time() - 120}
|
|
Path(self.wt, ".browser-remote-state.json").write_text(
|
|
json.dumps(doc), encoding="utf-8")
|
|
|
|
def _task(self, task_id):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO tasks(id, project_id, prompt, status)"
|
|
" VALUES(?, ?, ?, ?)",
|
|
(task_id, self.project["id"], "do it", "running"))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _req(self, method, path, body=None, key=None, raw=False):
|
|
conn = http.client.HTTPConnection("127.0.0.1", self.port,
|
|
timeout=10)
|
|
hdrs = {}
|
|
payload = None
|
|
if body is not None:
|
|
payload = json.dumps(body).encode("utf-8")
|
|
hdrs["Content-Type"] = "application/json"
|
|
if key is not None:
|
|
hdrs["X-API-Key"] = key
|
|
try:
|
|
conn.request(method, path, body=payload, headers=hdrs)
|
|
resp = conn.getresponse()
|
|
data = resp.read()
|
|
finally:
|
|
conn.close()
|
|
if raw:
|
|
return resp.status, data
|
|
try:
|
|
parsed = json.loads(data.decode("utf-8")) if data else None
|
|
except ValueError:
|
|
parsed = None
|
|
return resp.status, parsed
|
|
|
|
def test_state_input_control_flow(self):
|
|
status, body = self._req(
|
|
"GET", "/tasks/task-b1/browser/state", key=self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["holder"], "agent")
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/input",
|
|
{"type": "mouse_click", "x": 1, "y": 2}, self.owner_key)
|
|
self.assertEqual(status, 409, body)
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/input",
|
|
{"type": "mouse_click", "x": 9999, "y": 2}, self.owner_key)
|
|
self.assertEqual(status, 409, body) # holder checked first
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/control",
|
|
{"holder": "user"}, self.owner_key)
|
|
self.assertEqual((status, body["holder"]), (200, "user"))
|
|
ctrl = json.loads(Path(
|
|
self.wt, ".browser-remote-control.json").read_text())
|
|
self.assertEqual(ctrl["holder"], "user")
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/input",
|
|
{"type": "key", "key": "Enter"}, self.owner_key)
|
|
self.assertEqual((status, body["queued"]), (200, True))
|
|
lines = Path(self.wt,
|
|
".browser-remote-input.jsonl").read_text()
|
|
self.assertEqual(json.loads(lines.strip()),
|
|
{"type": "key", "key": "Enter"})
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/input",
|
|
{"type": "mouse_click", "x": 9999, "y": 2}, self.owner_key)
|
|
self.assertEqual(status, 400, body)
|
|
status, body = self._req(
|
|
"POST", "/tasks/task-b1/browser/control",
|
|
{"holder": "agent"}, self.owner_key)
|
|
self.assertEqual(body["holder"], "agent")
|
|
|
|
def test_forbidden_stale_and_no_session(self):
|
|
status, _ = self._req(
|
|
"GET", "/tasks/task-b1/browser/state", key=self.outsider_key)
|
|
self.assertEqual(status, 403)
|
|
self._write_state("agent", fresh=False)
|
|
status, body = self._req(
|
|
"GET", "/tasks/task-b1/browser/state", key=self.owner_key)
|
|
self.assertEqual(status, 503, body)
|
|
status, body = self._req(
|
|
"GET", "/tasks/task-nosock/browser/state",
|
|
key=self.owner_key)
|
|
self.assertEqual(status, 503, body)
|
|
|
|
def test_stream_emits_frame(self):
|
|
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=10)
|
|
try:
|
|
conn.request("GET", "/tasks/task-b1/browser/stream",
|
|
headers={"X-API-Key": self.owner_key})
|
|
resp = conn.getresponse()
|
|
self.assertEqual(resp.status, 200)
|
|
line = resp.readline().decode("utf-8")
|
|
self.assertTrue(line.startswith("data: "), line)
|
|
payload = json.loads(line[6:])
|
|
self.assertEqual(payload["image"], "JPEG1")
|
|
self.assertEqual(payload["url"], "https://example.com")
|
|
self.assertEqual(payload["title"], "T")
|
|
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()
|