feat(web): PR 4 - E2E

.github/workflows/release-build-check.yml

@@ -9,6 +9,8 @@ on:
       - .github/workflows/release.yml
       - .github/workflows/release-build-check.yml
       - pyproject.toml
+      - setup.py
+      - MANIFEST.in
 
 permissions: {}
 

MANIFEST.in

@@ -24,6 +24,14 @@ recursive-exclude dimos/web/websocket_vis/node_modules *
 recursive-exclude dimos/web/dimos_interface *
 recursive-include dimos/web/dimos_interface/api *.py *.html
 
+# Deno relay sources (repo-root web/): shipped inside the wheel by the
+# build_py hook in setup.py (copied to dimos/web/relay_bridge/_relay_dist).
+# The graft keeps them in the sdist so sdist->wheel builds (uv build, pip)
+# can reproduce that copy.
+graft web
+prune web/cockpit/node_modules
+prune web/cockpit/dist
+
 # Exclude development files
 exclude .gitignore
 exclude .gitattributes

dimos/web/relay_bridge/demo_smoke.py

@@ -0,0 +1,180 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Manual smoke demo for the relay chain.
+
+Spawns the Deno relay (unless --url points at a running one), then drives a
+robot client pushing synthetic color_image JPEGs as fast as they encode
+(latest-wins) plus odom at 20 Hz (reliable), and a viewer client receiving
+both. Open the printed debug URL in Chrome/Firefox to watch the same stream.
+
+Run: uv run python -m dimos.web.relay_bridge.demo_smoke [--secs 20] [--url https://...]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import contextlib
+import json
+import math
+import time
+from typing import Any
+from urllib.parse import urlparse
+
+import cv2
+import numpy as np
+
+from dimos.web.relay_bridge.protocol import DataFrame
+from dimos.web.relay_bridge.relay_process import RelayProcess
+from dimos.web.relay_bridge.wt_client import RelayClient
+
+WIDTH, HEIGHT = 640, 480
+
+
+def make_jpeg(seq: int) -> bytes:
+    """Synthetic camera frame: moving gradient + seq/timestamp overlay."""
+    ramp = np.linspace(0, 255, WIDTH, dtype=np.uint8)
+    gray = np.roll(np.tile(ramp, (HEIGHT, 1)), (seq * 7) % WIDTH, axis=1)
+    image = cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)
+    cv2.putText(
+        image,
+        f"seq {seq}  {time.strftime('%H:%M:%S')}",
+        (20, 60),
+        cv2.FONT_HERSHEY_SIMPLEX,
+        1.2,
+        (0, 255, 0),
+        2,
+    )
+    ok, encoded = cv2.imencode(".jpg", image, [int(cv2.IMWRITE_JPEG_QUALITY), 75])
+    assert ok
+    return encoded.tobytes()
+
+
+class ViewerStats:
+    def __init__(self) -> None:
+        self.channels: dict[str, dict[str, Any]] = {}
+
+    def on_frame(self, frame: DataFrame) -> None:
+        ch = self.channels.setdefault(
+            frame.header.ch,
+            {"frames": 0, "bytes": 0, "seqs": set(), "last": -1, "ooo": 0, "lat_ms": 0.0},
+        )
+        ch["frames"] += 1
+        ch["bytes"] += len(frame.payload)
+        ch["seqs"].add(frame.header.seq)
+        if frame.header.seq < ch["last"]:
+            ch["ooo"] += 1
+        ch["last"] = max(ch["last"], frame.header.seq)
+        ch["lat_ms"] = (time.time() - frame.header.ts) * 1000
+
+    def line(self) -> str:
+        parts = []
+        for name, ch in sorted(self.channels.items()):
+            seqs = ch["seqs"]
+            span_loss = (max(seqs) - min(seqs) + 1 - len(seqs)) if seqs else 0
+            parts.append(
+                f"{name}: {ch['frames']}f span_loss={span_loss} ooo={ch['ooo']} "
+                f"lat={ch['lat_ms']:.1f}ms"
+            )
+        return " | ".join(parts) or "(nothing received yet)"
+
+
+async def run(url: str, secs: float) -> None:
+    stats = ViewerStats()
+    async with (
+        await RelayClient.connect(url, "robot") as robot,
+        await RelayClient.connect(url, "viewer") as viewer,
+    ):
+        await robot.hello()
+        await viewer.hello()
+        rtt = await viewer.ping()
+        print(f"connected; datagram RTT {rtt * 1000:.1f} ms")
+
+        deadline = time.monotonic() + secs if secs > 0 else math.inf
+        image_writer = robot.latest_writer("color_image")
+        odom_sent = 0
+
+        async def image_pump() -> None:
+            seq = 0
+            while time.monotonic() < deadline:
+                image_writer.offer(make_jpeg(seq), meta={"w": WIDTH, "h": HEIGHT})
+                seq += 1
+                await asyncio.sleep(0)  # flat out: paced by encode + delivery
+
+        async def odom_pump() -> None:
+            nonlocal odom_sent
+            while time.monotonic() < deadline:
+                t = time.time()
+                payload = json.dumps(
+                    {"x": 3 * math.sin(t / 3), "y": 2 * math.sin(t / 2), "yaw": t % 6.28, "ts": t}
+                ).encode()
+                robot.send_frame("odom", payload, delivery="reliable")
+                odom_sent += 1
+                await asyncio.sleep(1 / 20)
+
+        async def viewer_pump() -> None:
+            async for frame in viewer.frames():
+                stats.on_frame(frame)
+
+        async def report() -> None:
+            while time.monotonic() < deadline:
+                await asyncio.sleep(2)
+                print(
+                    f"tx img {image_writer.sent} (dropped {image_writer.dropped}, "
+                    f"resets {image_writer.resets}) odom {odom_sent} | rx {stats.line()}"
+                )
+
+        viewer_task = asyncio.create_task(viewer_pump())
+        try:
+            await asyncio.gather(image_pump(), odom_pump(), report())
+        finally:
+            await asyncio.sleep(0.3)  # let the tail of the stream arrive
+            viewer_task.cancel()
+
+        print("\nsummary:")
+        print(
+            f"  sent: color_image {image_writer.sent} (+{image_writer.dropped} shed "
+            f"at source), odom {odom_sent}"
+        )
+        print(f"  received: {stats.line()}")
+        odom = stats.channels.get("odom", {})
+        odom_ok = odom and len(odom["seqs"]) == odom_sent and odom["last"] == odom_sent - 1
+        img_frames = stats.channels.get("color_image", {}).get("frames", 0)
+        print(
+            f"  verdict: odom {'complete' if odom_ok else 'INCOMPLETE'}, "
+            f"color_image {img_frames} frames delivered"
+        )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--url", default=None, help="attach to a running relay (wtUrl)")
+    parser.add_argument("--secs", type=float, default=0, help="run time; 0 = until Ctrl-C")
+    args = parser.parse_args()
+
+    if args.url is not None:
+        # /api/info's wtUrl ends in /viewer; strip the path so each role picks its own.
+        parsed = urlparse(args.url)
+        url = f"{parsed.scheme}://{parsed.netloc}" if parsed.netloc else args.url
+        asyncio.run(run(url, args.secs))
+        return
+    with RelayProcess() as info:
+        print(f"relay up; open {info.debug_url} in Chrome/Firefox to watch")
+        with contextlib.suppress(KeyboardInterrupt):
+            asyncio.run(run(info.wt_url, args.secs))
+
+
+if __name__ == "__main__":
+    main()

dimos/web/relay_bridge/relay_process.py

@@ -0,0 +1,154 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Spawn the Deno relay as a child process (used by tests and the smoke demo;
+T2 grows this into the RelayBridge module's managed relay process)."""
+
+from __future__ import annotations
+
+import collections
+from dataclasses import dataclass
+import json
+import os
+from pathlib import Path
+import queue
+import subprocess
+import threading
+from typing import IO
+
+from dimos.utils.deno import ensure_deno
+from dimos.utils.logging_config import setup_logger
+from dimos.web.relay_bridge.locate import find_web_dir, relay_run_cmd
+
+logger = setup_logger()
+
+_STDERR_TAIL_LINES = 60
+
+
+@dataclass
+class RelayReadyInfo:
+    http_port: int
+    wt_url: str
+    cert_hash: str
+    v: int
+
+    @property
+    def debug_url(self) -> str:
+        return f"http://127.0.0.1:{self.http_port}/debug.html"
+
+
+class RelayProcess:
+    """Relay child process with a parsed ready line and clean teardown."""
+
+    def __init__(
+        self,
+        *,
+        port: int = 0,
+        host: str = "127.0.0.1",
+        web_dir: Path | None = None,
+        timeout: float = 20.0,
+    ) -> None:
+        self._port = port
+        self._host = host
+        self._web_dir = web_dir
+        self._timeout = timeout
+        self._process: subprocess.Popen[str] | None = None
+        self._threads: list[threading.Thread] = []
+        self._ready_queue: queue.Queue[RelayReadyInfo] = queue.Queue(maxsize=1)
+        self._stderr_tail: collections.deque[str] = collections.deque(maxlen=_STDERR_TAIL_LINES)
+        self.info: RelayReadyInfo | None = None
+
+    def start(self) -> RelayReadyInfo:
+        deno = ensure_deno()
+        web_dir = self._web_dir or find_web_dir()
+        cmd = relay_run_cmd(deno, web_dir, "--port", str(self._port), "--host", self._host)
+        logger.info(f"starting relay: {' '.join(cmd)}")
+        env = os.environ | {"NO_COLOR": "1"}
+        self._process = subprocess.Popen(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env
+        )
+        assert self._process.stdout is not None and self._process.stderr is not None
+        self._threads = [
+            threading.Thread(target=self._read_stdout, args=(self._process.stdout,), daemon=True),
+            threading.Thread(target=self._read_stderr, args=(self._process.stderr,), daemon=True),
+        ]
+        for thread in self._threads:
+            thread.start()
+        try:
+            self.info = self._ready_queue.get(timeout=self._timeout)
+        except queue.Empty:
+            code = self._process.poll()
+            self.stop()
+            stderr = "\n".join(self._stderr_tail)
+            state = f"exited with {code}" if code is not None else "still running"
+            raise RuntimeError(
+                f"relay produced no ready line within {self._timeout} s ({state}); "
+                f"stderr tail:\n{stderr}"
+            ) from None
+        logger.info(f"relay ready: {self.info}")
+        return self.info
+
+    def stop(self) -> None:
+        if self._process is None:
+            return
+        process, self._process = self._process, None
+        process.terminate()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=2)
+        # The child is dead: its pipes are at EOF, so the reader threads have
+        # finished. Join them and close the pipes so no file object leaks.
+        for thread in self._threads:
+            thread.join(timeout=1)
+        self._threads.clear()
+        for stream in (process.stdout, process.stderr):
+            if stream is not None:
+                stream.close()
+
+    def __enter__(self) -> RelayReadyInfo:
+        return self.start()
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.stop()
+
+    def _read_stdout(self, stream: IO[str]) -> None:
+        for line in stream:
+            line = line.rstrip()
+            if not line:
+                continue
+            if self.info is None and line.startswith("{"):
+                try:
+                    data = json.loads(line)
+                except ValueError:
+                    data = None
+                if isinstance(data, dict) and data.get("event") == "ready":
+                    self._ready_queue.put(
+                        RelayReadyInfo(
+                            http_port=int(data["httpPort"]),
+                            wt_url=str(data["wtUrl"]),
+                            cert_hash=str(data["certHash"]),
+                            v=int(data["v"]),
+                        )
+                    )
+                    continue
+            logger.debug(f"[relay stdout] {line}")
+
+    def _read_stderr(self, stream: IO[str]) -> None:
+        for line in stream:
+            line = line.rstrip()
+            if line:
+                self._stderr_tail.append(line)
+                logger.debug(f"[relay stderr] {line}")