serve the web console from the node and drop the websocket bridge

The browser reached this node through `dora-websocket-transceiver`: a second
node, a second port, and a round trip through the dataflow for every button
press, to carry a UI only this node can interpret. The node now serves the
page and holds the browsers' sockets itself on one port, so the console is
one URL and a consumer wires nothing for it.

`dora_teleop/server.py` runs aiohttp on a thread of its own and meets the
control loop at two non-blocking points: a bounded queue the tick drains, and
a snapshot handed back. Both queues are bounded deliberately — a flood is
dropped rather than grown, and a slow browser loses stale snapshots rather
than the newest — because nothing a browser does may stall the thread that
stops the arm. A port already in use stops the node rather than leaving it
announcing a console nobody can open.

The web deadman moved but did not change: it was a `clients` message from the
bridge and is now the open-socket count read each tick. One hop fewer, same
meaning.

Contract, so 0.2.0 -> 0.3.0: `ui_command` and `clients` are gone. `ui_state`
stays published for `dora topic echo` and the bench, with no browser reading
it. New WEB_HOST and WEB_PORT (7000; 0 opens no port).

The console itself is laid out like the arm's own pendant — two rings
quartered into arc sectors you hold, each with its out-of-plane axis as pills
above. Directions and signs match xArm Studio on the controller, `Y+` on the
left included: two consoles driving one arm that disagree about which way is
+Y is a trap, not a preference.

Also sends a velocity only when it changes, repeating an unchanged one on
JOG_KEEPALIVE_S (0.2 s) purely to beat the consumer's own expiry. Repeating
every tick made `dora-ulite6` re-plan the same motion twenty times a second.
That turned out not to be the whole cause of the stepping an operator felt —
the rest was fixed in dora-ulite6 v0.3.1, pinned here — but sending 18
messages where 83 went before is worth keeping on its own.

`dora-joy` moves to v0.1.1: v0.1 is an annotated tag, and dora 1.0.1 records
the tag object's id as the commit hash and then rejects its own checkout for
sitting on the peeled commit. Neither dataflow could be built before this.

tests/conftest.py

@@ -101,9 +101,12 @@ def clean_env(monkeypatch):
         "DEADZONE",
         "EXPO",
         "JOG_TTL_S",
+        "JOG_KEEPALIVE_S",
         "JOY_TTL_S",
         "GAMEPAD_IDLE_RELEASE_S",
         "STATE_POLL_S",
+        "WEB_HOST",
+        "WEB_PORT",
     ):
         monkeypatch.delenv(name, raising=False)
 

tests/test_config.py

@@ -47,6 +47,26 @@ def test_a_joy_ttl_at_or_above_the_jog_ttl_is_refused_at_startup(monkeypatch):
         Config.from_env()
 
 
+def test_a_keepalive_at_or_above_the_jog_ttl_is_refused_at_startup(monkeypatch):
+    # The repeat has to outpace the arm's own velocity expiry. It cannot be read
+    # from here, but JOG_TTL_S already sits below it, so a repeat slower than
+    # JOG_TTL_S is a repeat the arm stops between — an arm that stutters under a
+    # held stick, with nothing in the logs to say why.
+    monkeypatch.setenv("JOG_KEEPALIVE_S", "0.4")
+
+    with pytest.raises(ValueError, match="JOG_KEEPALIVE_S"):
+        Config.from_env()
+
+
+def test_a_keepalive_of_zero_is_refused_at_startup(monkeypatch):
+    # Zero would mean repeating every tick, which is the behaviour the keepalive
+    # exists to replace.
+    monkeypatch.setenv("JOG_KEEPALIVE_S", "0")
+
+    with pytest.raises(ValueError, match="JOG_KEEPALIVE_S"):
+        Config.from_env()
+
+
 @pytest.mark.parametrize("value", ["1.0", "-0.1", "2"])
 def test_a_deadzone_outside_the_stick_is_refused_at_startup(monkeypatch, value):
     # A deadzone of 1 would silently make the pad inert; catching it here beats

tests/test_console.py

@@ -11,7 +11,7 @@ from pathlib import Path
 
 from dora_teleop import ui
 
-PAGE = Path(__file__).resolve().parent.parent / "web" / "index.html"
+PAGE = Path(__file__).resolve().parent.parent / "dora_teleop" / "web" / "index.html"
 
 
 def page():

tests/test_deadman.py

@@ -1,8 +1,5 @@
 """The deadman: every way a source can go away, and the stop that follows it."""
 
-import json
-
-import pyarrow as pa
 import pytest
 
 from dora_teleop import main as loop
@@ -59,8 +56,7 @@ def drive(node, state, sample, count=1):
 
 def web_jog(node, state, axes):
     """Deliver one console jog message, in the normalised units the page speaks."""
-    payload = json.dumps({"type": "jog", "axes": axes})
-    loop._accept_ui_command(node, state, input_event("ui_command", pa.array([payload])))
+    loop._accept_ui_command(node, state, {"type": "jog", "axes": axes})
 
 
 # --- the gamepad ---
@@ -112,7 +108,7 @@ def test_a_joy_node_that_keeps_publishing_keeps_the_arm_moving(clock):
         clock.advance(0.05)
         sent = drive(node, state, stick(axes=PUSHED))
 
-    assert len(sent) == 4
+    assert sent
     assert all(jog[0] > 0 for jog in sent)
 
 
@@ -138,15 +134,45 @@ def test_a_stick_at_rest_sends_one_zero_not_a_stream(clock):
     assert jogs == [STOP]
 
 
-def test_a_held_stick_is_resent_every_tick(clock):
-    # The arm's controller expires a velocity on its own; the re-send is what
-    # keeps that backstop from firing mid-motion.
+def test_a_held_stick_is_not_resent_every_tick(clock):
+    # A velocity is a set-point the arm holds, not a frame in a sequence.
+    # Repeating it every tick made the controller re-plan the same motion twenty
+    # times a second, which an operator feels as the arm stepping rather than
+    # travelling.
     node, state = build()
 
     jogs = drive(node, state, stick(axes=PUSHED), count=4)
 
-    assert len(jogs) == 4
-    assert all(jog[0] > 0 for jog in jogs)
+    assert len(jogs) == 1
+    assert jogs[0][0] > 0
+
+
+def test_a_held_stick_is_repeated_before_the_arm_would_expire_it(clock):
+    # The repeat exists for one reason: the arm's controller drops a velocity on
+    # its own, and the repeat is what keeps that backstop from firing mid-motion.
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
+    node.clear()
+
+    clock.advance(state.config.deadman.jog_keepalive_s)
+    jogs = drive(node, state, stick(axes=PUSHED))
+
+    assert len(jogs) == 1
+    assert jogs[0][0] > 0
+
+
+def test_a_change_of_direction_never_waits_for_the_repeat(clock):
+    # Sending only on change is worth nothing if the change itself is delayed:
+    # a press or a release has to be acted on the tick it happens, not up to a
+    # keepalive later.
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
+    node.clear()
+
+    jogs = drive(node, state, stick(axes=[0.0, 1.0]))
+
+    assert len(jogs) == 1
+    assert jogs[0][0] < 0
 
 
 def test_a_pad_that_was_never_wired_up_leaves_the_arm_alone(clock):
@@ -172,7 +198,7 @@ def test_the_last_browser_leaving_stops_the_arm(clock):
     ticks(node, state)
     node.clear()
 
-    loop._accept_clients(state, input_event("clients", pa.array([{"connected": 0, "event": "close"}])))
+    loop._set_clients(state, 0)
     jogs = ticks(node, state)
 
     assert jogs == [STOP]
@@ -231,8 +257,7 @@ def test_estop_is_never_gated_by_the_lease(clock):
     state.clients = 1
     state.arbiter.take(GAMEPAD, clock.t)
 
-    payload = json.dumps({"type": "estop"})
-    loop._accept_ui_command(node, state, input_event("ui_command", pa.array([payload])))
+    loop._accept_ui_command(node, state, {"type": "estop"})
 
     assert node.jogs() == [STOP]
     assert "stop" in node.services()
@@ -321,8 +346,7 @@ def test_a_latched_estop_is_cleared_by_reset(clock):
     node, state = build()
     state.view.estop = True
 
-    payload = json.dumps({"type": "reset"})
-    loop._accept_ui_command(node, state, input_event("ui_command", pa.array([payload])))
+    loop._accept_ui_command(node, state, {"type": "reset"})
 
     assert state.view.estop is False
     assert "clear" in node.services()

tests/test_server.py

@@ -0,0 +1,223 @@
+"""The embedded console server: a real socket, and the deadman it feeds.
+
+These start the server for real rather than faking it. The thing worth checking
+is the boundary between the two threads — a browser's message reaching the tick,
+and a browser going away reaching the deadman — and a fake on either side of
+that boundary would check nothing.
+"""
+
+import asyncio
+import json
+import socket
+import time
+
+import aiohttp
+import pytest
+
+from dora_teleop import main as loop
+from dora_teleop.arbiter import WEB
+from dora_teleop.config import Config, ConsoleConfig
+from dora_teleop.server import Console
+from tests.conftest import RecordingNode
+
+
+def free_port():
+    """Ask the OS for a port nobody is using."""
+    with socket.socket() as held:
+        held.bind(("127.0.0.1", 0))
+        return held.getsockname()[1]
+
+
+@pytest.fixture
+def console():
+    """Run a real server on a real port, and stop it however the test ends."""
+    running = Console("127.0.0.1", free_port())
+    running.start()
+    try:
+        yield running
+    finally:
+        running.stop()
+
+
+def visit(console, exchange):
+    """Open one browser's socket against the server and run `exchange` on it."""
+
+    async def session():
+        async with aiohttp.ClientSession() as client:
+            url = f"http://{console.host}:{console.port}"
+            async with client.ws_connect(f"{url}/ws") as socket_:
+                return await exchange(client, url, socket_)
+
+    return asyncio.run(session())
+
+
+def wait_for(predicate):
+    """Block until the server thread catches up, rather than guessing at a sleep."""
+    for _ in range(200):
+        if predicate():
+            return
+        time.sleep(0.01)
+    raise AssertionError("the server never got there")
+
+
+def test_the_page_is_served_at_the_root(console):
+    async def exchange(client, url, socket_):
+        async with client.get(url) as answer:
+            return answer.status, await answer.text()
+
+    status, body = visit(console, exchange)
+
+    assert status == 200
+    assert "teleop console" in body
+
+
+def test_a_message_from_a_browser_reaches_the_tick(console):
+    # The console's half of the boundary: what the page sends has to be waiting
+    # on the queue the next tick drains, in the shape the loop reads.
+    async def exchange(client, url, socket_):
+        await socket_.send_str(json.dumps({"type": "jog", "axes": [1.0, 0, 0, 0, 0, 0]}))
+        await asyncio.sleep(0.05)
+
+    visit(console, exchange)
+
+    assert console.take() == [{"type": "jog", "axes": [1.0, 0, 0, 0, 0, 0]}]
+    assert console.take() == []
+
+
+def test_a_snapshot_reaches_the_browser(console):
+    # The other half: a telemetry object handed over from the dora thread comes
+    # out of the socket as the page's own JSON, with no envelope around it.
+    async def exchange(client, url, socket_):
+        console.broadcast({"type": "telemetry", "arm_state": "idle"})
+        return json.loads(await socket_.receive_str())
+
+    assert visit(console, exchange) == {"type": "telemetry", "arm_state": "idle"}
+
+
+def test_a_browser_that_is_not_talking_json_is_ignored(console):
+    # A page from a newer build saying something we cannot parse must not take
+    # the arm down, and must not stop the messages after it being read.
+    async def exchange(client, url, socket_):
+        await socket_.send_str("{not json")
+        await socket_.send_str(json.dumps({"type": "estop"}))
+        await asyncio.sleep(0.05)
+
+    visit(console, exchange)
+
+    assert console.take() == [{"type": "estop"}]
+
+
+def test_the_server_counts_the_browsers_on_it(console):
+    assert console.clients == 0
+    seen = []
+
+    async def exchange(client, url, socket_):
+        await asyncio.sleep(0.05)
+        seen.append(console.clients)
+
+    visit(console, exchange)
+
+    assert seen == [1]
+
+    # And the socket closing is what the deadman below reads.
+    wait_for(lambda: console.clients == 0)
+
+
+def test_a_flood_from_one_browser_cannot_grow_without_bound(console):
+    # An unbounded queue would answer a flood by growing until the node died,
+    # and the node dying is the one thing that must not stop the arm from being
+    # stopped.
+    async def exchange(client, url, socket_):
+        for index in range(600):
+            await socket_.send_str(json.dumps({"type": "jog_stop", "n": index}))
+        await asyncio.sleep(0.2)
+
+    visit(console, exchange)
+
+    assert 0 < len(console.take()) <= 256
+
+
+# --- where the server meets the deadman ---
+
+
+def build():
+    config = Config.from_env()
+    config.expo = 0.0
+    return RecordingNode(), loop.Teleop(config)
+
+
+def test_the_last_browser_leaving_takes_the_arm_back(console):
+    # The web deadman used to arrive as a `clients` message from the bridge. It
+    # is now read straight off the server, and it has to mean the same thing.
+    node, state = build()
+    state.console = console
+    state.arbiter.take(WEB, 1000.0)
+
+    loop._drain_console(node, state)
+
+    assert state.clients == 0
+    assert state.arbiter.owner is None
+    assert state.web_axes == [0.0] * 6
+
+
+def test_a_node_with_no_console_is_left_alone():
+    # A pad-only dataflow opens no port, and the tick must not read a lease
+    # release out of a server that was never started.
+    node, state = build()
+    state.clients = 1
+    state.arbiter.take(WEB, 1000.0)
+
+    loop._drain_console(node, state)
+
+    assert state.arbiter.owner == WEB
+
+
+# --- the port itself ---
+
+
+def test_the_console_is_on_by_default(monkeypatch):
+    monkeypatch.delenv("WEB_PORT", raising=False)
+    settings = ConsoleConfig.from_env()
+
+    assert settings.enabled is True
+    # Pinned because it is the number in the README, the manifest and every
+    # dataflow: a default that drifts from the docs sends an operator to a port
+    # nothing is listening on.
+    assert settings.port == 7000
+
+
+def test_port_zero_means_no_port_is_opened(monkeypatch):
+    # How `dataflow-gamepad.yml` asks for a pad and nothing else.
+    monkeypatch.setenv("WEB_PORT", "0")
+    assert ConsoleConfig.from_env().enabled is False
+
+
+def test_a_port_that_is_not_a_port_stops_the_node(monkeypatch):
+    monkeypatch.setenv("WEB_PORT", "99999")
+    with pytest.raises(ValueError, match="WEB_PORT"):
+        ConsoleConfig.from_env()
+
+
+def test_a_port_that_is_not_a_number_stops_the_node(monkeypatch):
+    monkeypatch.setenv("WEB_PORT", "eight-thousand")
+    with pytest.raises(ValueError, match="WEB_PORT"):
+        ConsoleConfig.from_env()
+
+
+def test_a_port_already_in_use_stops_the_node(console):
+    # Found the hard way: the bind failed on the server thread, `start` waited
+    # out its timeout and then announced a console anyway, and the node ran on
+    # serving nothing. An operator opening the URL got silence while the log
+    # said it was up.
+    second = Console(console.host, console.port)
+
+    with pytest.raises(OSError):
+        second.start()
+
+    second.stop()
+
+
+def test_stopping_a_console_that_never_started_is_harmless():
+    # `main` stops the console in a `finally`, which runs whether or not the
+    # server ever came up.
+    Console("127.0.0.1", 0).stop()