diff --git a/dimos/agents/skills/person_follow.py b/dimos/agents/skills/person_follow.py
index 29f81a0133..e9dc39802a 100644
--- a/dimos/agents/skills/person_follow.py
+++ b/dimos/agents/skills/person_follow.py
@@ -19,7 +19,6 @@
import numpy as np
from reactivex.disposable import Disposable
-from turbojpeg import TurboJPEG
from dimos.agents.annotation import skill
from dimos.agents.capabilities import CAP_MOVEMENT
@@ -40,6 +39,7 @@
from dimos.navigation.visual_servoing.detection_navigation import DetectionNavigation
from dimos.navigation.visual_servoing.visual_servoing_2d import VisualServoing2D
from dimos.utils.logging_config import setup_logger
+from dimos.utils.turbojpeg import get_turbojpeg
logger = setup_logger()
@@ -331,5 +331,5 @@ def _send_stop_reason(self, query: str, reason: str) -> None:
def _decode_base64_image(b64: str) -> Image:
- bgr_array = TurboJPEG().decode(base64.b64decode(b64))
+ bgr_array = get_turbojpeg().decode(base64.b64decode(b64))
return Image(data=bgr_array, format=ImageFormat.BGR)
diff --git a/dimos/e2e_tests/test_cockpit_browser.py b/dimos/e2e_tests/test_cockpit_browser.py
index 891d95480a..644593d1a2 100644
--- a/dimos/e2e_tests/test_cockpit_browser.py
+++ b/dimos/e2e_tests/test_cockpit_browser.py
@@ -12,15 +12,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""Cockpit browser e2e (the T3 acceptance demo, in CI).
+"""Cockpit browser e2e (the T3 + T5 acceptance demos, in CI).
Starts `dimos --robot-ip fake run unitree-go2-basic --local-relay` (the go2
blueprint on the go2_short replay dataset: real recorded odom + camera, no
MuJoCo and no hardware) and drives the served Cockpit with a real headless
-browser: live odom must tick, the session must stay connected (Firefox's
-tight QUIC stream credit found the relay's stream-per-frame overflow the
-first time), and killing + restarting the dimos process must take the page
-through reconnecting and back to live data without a reload.
+browser: live odom must tick, the video panel must decode real camera frames
+at rate, the session must stay connected (Firefox's tight QUIC stream credit
+found the relay's stream-per-frame overflow the first time; T5's video load
+is exactly the traffic that pressures it), and killing + restarting the
+dimos process must take the page through reconnecting and back to live data
+without a reload.
Runs in both Chromium and Firefox: their WebTransport implementations differ
enough that one browser staying green says little about the other.
@@ -111,6 +113,38 @@ def _assert_odom_ticks(page: Page) -> None:
expect(page.get_by_test_id("ch-odom-value")).to_contain_text("yaw")
+def _assert_video_plays(page: Page) -> None:
+ # The manifest-driven video panel is live: the badge reporting an fps
+ # proves frames arrive, the canvas leaving its 300 px HTML default width
+ # proves one frame decoded, and two successive pixel changes prove frames
+ # keep drawing (the replay dataset has genuinely varying content).
+ expect(page.get_by_test_id("video-color_image-badge")).to_contain_text("fps", timeout=60_000)
+ page.wait_for_function(
+ """() => {
+ const canvas = document.querySelector('[data-testid="video-color_image-canvas"]');
+ return canvas !== null && canvas.width !== 300;
+ }""",
+ timeout=30_000,
+ )
+
+ def wait_pixels_change() -> None:
+ baseline = page.evaluate(
+ """() =>
+ document.querySelector('[data-testid="video-color_image-canvas"]').toDataURL()"""
+ )
+ page.wait_for_function(
+ """(baseline) => {
+ const canvas = document.querySelector('[data-testid="video-color_image-canvas"]');
+ return canvas.toDataURL() !== baseline;
+ }""",
+ arg=baseline,
+ timeout=30_000,
+ )
+
+ wait_pixels_change()
+ wait_pixels_change()
+
+
def test_cockpit_live_data_and_reconnect(
start_go2_replay: Callable[[], DimosCliCall], page: Page
) -> None:
@@ -122,6 +156,7 @@ def test_cockpit_live_data_and_reconnect(
expect(status).to_have_attribute("data-phase", "connected", timeout=120_000)
expect(page.get_by_test_id("robot")).not_to_have_text("no robot", timeout=30_000)
_assert_odom_ticks(page)
+ _assert_video_plays(page)
# Stability: record every phase change and require none for the window.
page.evaluate("""() => {
@@ -134,6 +169,7 @@ def test_cockpit_live_data_and_reconnect(
flaps = page.evaluate("() => window.__phases")
assert flaps == [], f"session flapped during the stability window: {flaps}"
_assert_odom_ticks(page)
+ _assert_video_plays(page)
# Kill dimos (relay dies with it): the page must notice on its own.
call.stop()
@@ -144,3 +180,4 @@ def test_cockpit_live_data_and_reconnect(
_wait_for_relay(restarted, START_TIMEOUT_S)
expect(status).to_have_attribute("data-phase", "connected", timeout=120_000)
_assert_odom_ticks(page)
+ _assert_video_plays(page)
diff --git a/dimos/msgs/sensor_msgs/Image.py b/dimos/msgs/sensor_msgs/Image.py
index a95aad5cc6..05b0522575 100644
--- a/dimos/msgs/sensor_msgs/Image.py
+++ b/dimos/msgs/sensor_msgs/Image.py
@@ -26,9 +26,11 @@
import numpy as np
import reactivex as rx
from reactivex import operators as ops
+from turbojpeg import TJPF_RGB
from dimos.types.timestamped import Timestamped, TimestampedBufferCollection, to_human_readable
from dimos.utils.reactive import quality_barrier
+from dimos.utils.turbojpeg import get_turbojpeg
if TYPE_CHECKING:
from collections.abc import Callable
@@ -542,12 +544,9 @@ def to_jpeg_bytes(self, quality: int = 75) -> bytes:
Returns:
Raw JPEG bytes.
"""
- from turbojpeg import TJPF_RGB, TurboJPEG
-
- jpeg = TurboJPEG()
# Canonicalize to RGB so JPEG bytes are deterministic regardless of input format.
rgb_array = self.to_rgb().data
- return jpeg.encode(rgb_array, quality=quality, pixel_format=TJPF_RGB) # type: ignore[no-any-return]
+ return get_turbojpeg().encode(rgb_array, quality=quality, pixel_format=TJPF_RGB) # type: ignore[no-any-return]
def lcm_jpeg_encode(self, quality: int = 75, frame_id: str | None = None) -> bytes:
"""Convert to LCM Image message with JPEG-compressed data.
@@ -599,15 +598,12 @@ def lcm_jpeg_decode(cls, data: bytes, **kwargs: Any) -> Image:
Returns:
Image instance
"""
- from turbojpeg import TJPF_RGB, TurboJPEG
-
- jpeg = TurboJPEG()
msg = LCMImage.lcm_decode(data)
if msg.encoding != "jpeg":
raise ValueError(f"Expected JPEG encoding, got {msg.encoding}")
- rgb_array = jpeg.decode(msg.data, pixel_format=TJPF_RGB)
+ rgb_array = get_turbojpeg().decode(msg.data, pixel_format=TJPF_RGB)
return cls(
data=rgb_array,
diff --git a/dimos/protocol/pubsub/impl/jpeg_shm.py b/dimos/protocol/pubsub/impl/jpeg_shm.py
index b4d4d3e68e..649da4f9c9 100644
--- a/dimos/protocol/pubsub/impl/jpeg_shm.py
+++ b/dimos/protocol/pubsub/impl/jpeg_shm.py
@@ -14,17 +14,15 @@
from typing import Any
-from turbojpeg import TurboJPEG
-
from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
from dimos.protocol.pubsub.encoders import PubSubEncoderMixin
from dimos.protocol.pubsub.impl.shmpubsub import SharedMemoryPubSubBase
+from dimos.utils.turbojpeg import get_turbojpeg
class JpegSharedMemoryEncoderMixin(PubSubEncoderMixin[str, Image, bytes]):
def __init__(self, quality: int = 75, **kwargs) -> None: # type: ignore[no-untyped-def]
super().__init__(**kwargs)
- self.jpeg = TurboJPEG()
self.quality = quality
def encode(self, msg: Any, _topic: str) -> bytes:
@@ -32,10 +30,10 @@ def encode(self, msg: Any, _topic: str) -> bytes:
raise ValueError("Can only encode images.")
bgr_image = msg.to_bgr().to_opencv()
- return self.jpeg.encode(bgr_image, quality=self.quality) # type: ignore[no-any-return]
+ return get_turbojpeg().encode(bgr_image, quality=self.quality) # type: ignore[no-any-return]
def decode(self, msg: bytes, _topic: str) -> Image:
- bgr_array = self.jpeg.decode(msg)
+ bgr_array = get_turbojpeg().decode(msg)
return Image(data=bgr_array, format=ImageFormat.BGR)
diff --git a/dimos/robot/unitree/mujoco_connection.py b/dimos/robot/unitree/mujoco_connection.py
index 1ab492338a..2d1cd2f866 100644
--- a/dimos/robot/unitree/mujoco_connection.py
+++ b/dimos/robot/unitree/mujoco_connection.py
@@ -92,6 +92,7 @@ def __init__(self, global_config: GlobalConfig) -> None:
self._stop_timer: threading.Timer | None = None
self._stream_threads: list[threading.Thread] = []
+ self._output_thread: threading.Thread | None = None
self._stop_events: list[threading.Event] = []
self._is_cleaned_up = False
@@ -126,7 +127,8 @@ def start(self) -> None:
self.process = subprocess.Popen(
[executable, str(LAUNCHER_PATH), config_pickle, shm_names_json],
- stderr=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
env=env,
)
@@ -134,6 +136,17 @@ def start(self) -> None:
self.shm_data.cleanup()
raise RuntimeError(f"Failed to start MuJoCo subprocess: {e}") from e
+ # A captured pipe must always be drained. MuJoCo can emit a sustained
+ # stream of physics warnings; once an unread OS pipe fills, the child
+ # blocks in write(2) and silently stops updating every shared-memory
+ # sensor while this parent (and the relay) remain healthy.
+ self._output_thread = threading.Thread(
+ target=self._pump_subprocess_output,
+ name="mujoco-output-pump",
+ daemon=True,
+ )
+ self._output_thread.start()
+
# Wait for process to be ready
ready_timeout = 300.0
start_time = time.time()
@@ -164,19 +177,37 @@ def cleanup_on_exit(
self.stop()
raise RuntimeError("MuJoCo process failed to start (timeout)")
+ def _pump_subprocess_output(self) -> None:
+ """Drain MuJoCo output into DimOS logs so the child cannot pipe-block."""
+ process = self.process
+ if process is None or process.stdout is None:
+ return
+ try:
+ for raw in process.stdout:
+ line = raw.decode(errors="replace").rstrip()
+ if not line:
+ continue
+ lowered = line.lower()
+ if "warning" in lowered or "error" in lowered or "fatal" in lowered:
+ logger.warning(f"[mujoco] {line}")
+ else:
+ logger.info(f"[mujoco] {line}")
+
+ return_code = process.poll()
+ if return_code not in (None, 0) and not self._is_cleaned_up:
+ logger.error(f"[mujoco] subprocess exited unexpectedly with code {return_code}")
+ except (OSError, ValueError) as e:
+ # Teardown can race the final descriptor close; only an unexpected
+ # read failure while live is actionable.
+ if not self._is_cleaned_up:
+ logger.warning(f"MuJoCo output pump terminated: {e}")
+
def stop(self) -> None:
if self._is_cleaned_up:
return
self._is_cleaned_up = True
- # clean up open file descriptors
- if self.process:
- if self.process.stderr:
- self.process.stderr.close()
- if self.process.stdout:
- self.process.stdout.close()
-
# Cancel any pending timers
if self._stop_timer:
self._stop_timer.cancel()
@@ -197,7 +228,9 @@ def stop(self) -> None:
if self.shm_data:
self.shm_data.signal_stop()
- # Wait for process to finish
+ # Stop the child before joining/closing its output. BufferedReader.close()
+ # can wait on a pump thread blocked in read(), so closing the pipe while
+ # the child is still alive can deadlock shutdown.
if self.process:
try:
self.process.terminate()
@@ -210,7 +243,22 @@ def stop(self) -> None:
except Exception as e:
logger.error(f"Error stopping MuJoCo process: {e}")
+ if self._output_thread is not None and self._output_thread.is_alive():
+ self._output_thread.join(timeout=DEFAULT_THREAD_JOIN_TIMEOUT)
+ if self._output_thread.is_alive():
+ logger.warning("MuJoCo output thread did not stop gracefully")
+
+ # The exited child has closed its pipe end, so these closes cannot
+ # contend with a reader indefinitely.
+ if self.process.stderr:
+ self.process.stderr.close()
+ if self.process.stdout and not (
+ self._output_thread is not None and self._output_thread.is_alive()
+ ):
+ self.process.stdout.close()
+
self.process = None
+ self._output_thread = None
# Clean up shared memory
if self.shm_data:
diff --git a/dimos/robot/unitree/test_mujoco_connection.py b/dimos/robot/unitree/test_mujoco_connection.py
new file mode 100644
index 0000000000..0c385dc9d2
--- /dev/null
+++ b/dimos/robot/unitree/test_mujoco_connection.py
@@ -0,0 +1,186 @@
+# Copyright 2025-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.
+
+from collections.abc import Iterator
+import subprocess
+import sys
+import threading
+from types import ModuleType
+from typing import Any, cast
+
+from pytest import MonkeyPatch
+
+from dimos.core.global_config import GlobalConfig
+from dimos.robot.unitree import mujoco_connection
+from dimos.robot.unitree.mujoco_connection import MujocoConnection
+
+
+class _FakeShmNames:
+ def to_names(self) -> dict[str, str]:
+ return {}
+
+
+class _FakeShmWriter:
+ shm = _FakeShmNames()
+
+ def is_ready(self) -> bool:
+ return True
+
+ def signal_stop(self) -> None:
+ pass
+
+ def cleanup(self) -> None:
+ pass
+
+
+class _FakeLogger:
+ def info(self, _message: str) -> None:
+ pass
+
+ def warning(self, _message: str) -> None:
+ pass
+
+ def error(self, _message: str) -> None:
+ pass
+
+
+class _BlockingOutput:
+ """Pipe double whose close blocks until the child is terminated."""
+
+ def __init__(self) -> None:
+ self.read_started = threading.Event()
+ self.child_stopped = threading.Event()
+
+ def __iter__(self) -> Iterator[bytes]:
+ self.read_started.set()
+ self.child_stopped.wait()
+ return iter(())
+
+ def close(self) -> None:
+ self.child_stopped.wait()
+
+
+class _QuietProcess:
+ def __init__(self) -> None:
+ self.stdout = _BlockingOutput()
+ self.stderr = None
+ self.returncode: int | None = None
+ self.terminations = 0
+
+ def poll(self) -> int | None:
+ return self.returncode
+
+ def terminate(self) -> None:
+ self.terminations += 1
+ self.returncode = 0
+ self.stdout.child_stopped.set()
+
+ def wait(self, timeout: float) -> int:
+ assert timeout > 0
+ assert self.returncode is not None
+ return self.returncode
+
+ def kill(self) -> None:
+ self.terminate()
+
+
+def _bare_connection(monkeypatch: MonkeyPatch) -> MujocoConnection:
+ mjx_env = ModuleType("mujoco_playground._src.mjx_env")
+ mjx_env.ensure_menagerie_exists = lambda: None
+ playground_src = ModuleType("mujoco_playground._src")
+ playground_src.mjx_env = mjx_env
+ monkeypatch.setitem(sys.modules, "mujoco_playground._src", playground_src)
+ monkeypatch.setattr(mujoco_connection, "get_data", lambda _name: None)
+ return MujocoConnection(GlobalConfig())
+
+
+def test_start_drains_subprocess_output_larger_than_a_pipe(
+ monkeypatch: MonkeyPatch,
+) -> None:
+ """A noisy simulator must exit instead of blocking on an unread pipe."""
+ real_popen = subprocess.Popen
+ popen_kwargs: dict[str, Any] = {}
+
+ def noisy_child(_command: list[str], **kwargs: Any) -> subprocess.Popen[bytes]:
+ popen_kwargs.update(kwargs)
+ return real_popen(
+ [
+ sys.executable,
+ "-c",
+ "import sys; "
+ "sys.stderr.buffer.write(b'x' * 4_000_000); "
+ "sys.stderr.flush(); "
+ "sys.stdin.buffer.read(1)",
+ ],
+ stdin=subprocess.PIPE,
+ **kwargs,
+ )
+
+ connection = _bare_connection(monkeypatch)
+
+ monkeypatch.setattr(mujoco_connection, "ShmWriter", _FakeShmWriter)
+ monkeypatch.setattr("dimos.robot.unitree.mujoco_connection.subprocess.Popen", noisy_child)
+ monkeypatch.setattr(
+ "dimos.robot.unitree.mujoco_connection.atexit.register", lambda *_args: None
+ )
+ monkeypatch.setattr(mujoco_connection, "logger", _FakeLogger())
+
+ try:
+ connection.start()
+ process = connection.process
+ assert process is not None
+ assert process.stdin is not None
+ process.stdin.write(b"x")
+ process.stdin.close()
+ process.wait(timeout=5)
+ output_thread = connection._output_thread
+ assert output_thread is not None
+ output_thread.join(timeout=5)
+
+ assert not output_thread.is_alive()
+ assert popen_kwargs["stdout"] is subprocess.PIPE
+ assert popen_kwargs["stderr"] is subprocess.STDOUT
+ finally:
+ connection.stop()
+
+
+def test_stop_terminates_child_before_closing_pumped_output(
+ monkeypatch: MonkeyPatch,
+) -> None:
+ """Stopping a quiet child must not deadlock on the pump's read lock."""
+ monkeypatch.setattr(mujoco_connection, "logger", _FakeLogger())
+ connection = _bare_connection(monkeypatch)
+ process = _QuietProcess()
+ state = cast("Any", connection)
+ state.process = process
+ state.shm_data = _FakeShmWriter()
+ state._output_thread = threading.Thread(
+ target=connection._pump_subprocess_output,
+ name="mujoco-output-pump",
+ daemon=True,
+ )
+ state._output_thread.start()
+ assert process.stdout.read_started.wait(timeout=1)
+
+ stopper = threading.Thread(target=connection.stop)
+ stopper.start()
+ stopper.join(timeout=1)
+ if stopper.is_alive():
+ # Cleanup for the broken ordering: release close() without pretending
+ # terminate() was called, so both the leak and ordering assertions fail.
+ process.stdout.child_stopped.set()
+ stopper.join(timeout=1)
+
+ assert not stopper.is_alive()
+ assert process.terminations == 1
diff --git a/dimos/utils/turbojpeg.py b/dimos/utils/turbojpeg.py
new file mode 100644
index 0000000000..0d53a87551
--- /dev/null
+++ b/dimos/utils/turbojpeg.py
@@ -0,0 +1,26 @@
+# 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.
+
+
+from __future__ import annotations
+
+import functools
+
+from turbojpeg import TurboJPEG
+
+
+@functools.cache
+def get_turbojpeg() -> TurboJPEG:
+ """Return the shared TurboJPEG codec handle."""
+ return TurboJPEG()
diff --git a/dimos/web/relay_bridge/_wt_session.py b/dimos/web/relay_bridge/_wt_session.py
index a4300d2fea..ef7c3c8115 100644
--- a/dimos/web/relay_bridge/_wt_session.py
+++ b/dimos/web/relay_bridge/_wt_session.py
@@ -37,7 +37,7 @@
WebTransportStreamDataReceived,
)
from aioquic.quic.configuration import QuicConfiguration
-from aioquic.quic.events import ConnectionTerminated, QuicEvent
+from aioquic.quic.events import ConnectionTerminated, QuicEvent, StreamReset
from dimos.utils.logging_config import setup_logger
from dimos.web.relay_bridge.protocol import (
@@ -172,6 +172,12 @@ def open_session(self, authority: str, path: str) -> None:
def quic_event_received(self, event: QuicEvent) -> None:
if isinstance(event, ConnectionTerminated):
self.closed.set()
+ elif isinstance(event, StreamReset):
+ # aioquic's H3 layer ignores resets, and the relay ends every
+ # latest stream with one (reap, dispose, or teardown): without
+ # this pop the reader map grows by one entry per latest stream.
+ # A partial frame in that reader is stale by definition.
+ self._frame_readers.pop(event.stream_id, None)
for h3_event in self.h3.handle_event(event):
self._h3_event_received(h3_event)
diff --git a/dimos/web/relay_bridge/demo_smoke.py b/dimos/web/relay_bridge/demo_smoke.py
index 6cb7d397a0..49e0e467ec 100644
--- a/dimos/web/relay_bridge/demo_smoke.py
+++ b/dimos/web/relay_bridge/demo_smoke.py
@@ -17,7 +17,8 @@
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.
+both. Open the printed URL in Chrome/Firefox to watch the same stream in the
+Cockpit (build web/cockpit first if the page reports a missing dist).
Run: uv run python -m dimos.web.relay_bridge.demo_smoke [--secs 20] [--url https://...]
"""
@@ -39,6 +40,7 @@
ChannelSpec,
DataFrame,
Manifest,
+ PanelSpec,
RobotInfo,
RobotManifest,
Sub,
@@ -55,7 +57,9 @@
channels=[
ChannelSpec(ch="color_image", encoding="jpeg.v1", delivery="latest", maxHz=60.0),
ChannelSpec(ch="odom", encoding="pose.json.v1", delivery="reliable", maxHz=20.0),
- ]
+ ],
+ # The Cockpit only subscribes jpeg.v1 when a video panel binds it.
+ panels=[PanelSpec(id="color_image", kind="video", channels=["color_image"])],
)
@@ -235,7 +239,7 @@ def main() -> None:
asyncio.run(run(url, args.secs))
return
with RelayProcess() as info:
- print(f"relay up; open {info.debug_url} in Chrome/Firefox to watch")
+ print(f"relay up; open {info.open_url} in Chrome/Firefox to watch")
with contextlib.suppress(KeyboardInterrupt):
asyncio.run(run(info.wt_url, args.secs))
diff --git a/dimos/web/relay_bridge/manifest.py b/dimos/web/relay_bridge/manifest.py
index 9da2f656db..ee10ec76ba 100644
--- a/dimos/web/relay_bridge/manifest.py
+++ b/dimos/web/relay_bridge/manifest.py
@@ -17,8 +17,9 @@
Pinned by the golden vectors in web/shared/fixtures/manifests.json (tested
from both pytest and deno test). The transport (protocol.py) checks only
field shapes; this module owns the domain rules: bounded unique ids, positive
-rates, and panel/layout references that resolve. Panels and layout are
-minimal until T7 (the layout is a flat panel-id order, not a tree).
+rates, panel/layout references that resolve, and kind-specific panel rules
+(video). Panels and layout are minimal until T7 (the layout is a flat
+panel-id order, not a tree).
"""
import json
@@ -97,7 +98,7 @@ def parse_manifest(data: Any) -> Manifest:
except ValidationError as e:
raise ManifestError("invalid_shape", str(e)) from e
- ch_ids: set[str] = set()
+ ch_ids: dict[str, ChannelSpec] = {}
for spec in manifest.channels:
if not _bounded_id(spec.ch):
raise ManifestError(
@@ -105,7 +106,7 @@ def parse_manifest(data: Any) -> Manifest:
)
if spec.ch in ch_ids:
raise ManifestError("duplicate_channel_id", f"duplicate channel {spec.ch}")
- ch_ids.add(spec.ch)
+ ch_ids[spec.ch] = spec
if not _bounded_id(spec.encoding):
raise ManifestError(
"invalid_encoding", f"encoding must be 1..{MAX_MANIFEST_ID_LEN} chars"
@@ -131,6 +132,18 @@ def parse_manifest(data: Any) -> Manifest:
raise ManifestError(
"unknown_panel_channel", f"panel {panel.id} wants undeclared channel {ch}"
)
+ # Kind-specific rules; unknown kinds stay unvalidated (forward
+ # compatibility with newer bridges).
+ if panel.kind == "video":
+ if len(panel.channels) != 1:
+ raise ManifestError(
+ "invalid_video_panel", f"video panel {panel.id} must bind exactly one channel"
+ )
+ bound = ch_ids[panel.channels[0]]
+ if bound.encoding != "jpeg.v1" or bound.delivery != "latest":
+ raise ManifestError(
+ "invalid_video_panel", f"video panel {panel.id} needs a jpeg.v1 latest channel"
+ )
for panel_id in manifest.layout:
if panel_id not in panel_ids:
diff --git a/dimos/web/relay_bridge/protocol.py b/dimos/web/relay_bridge/protocol.py
index b0743bd0b7..be7cbd9cae 100644
--- a/dimos/web/relay_bridge/protocol.py
+++ b/dimos/web/relay_bridge/protocol.py
@@ -53,7 +53,12 @@
# Channel/manifest domain types live in manifest.py; re-exported here (the
# redundant aliases mark them as such for mypy) so protocol consumers keep a
# single import surface, mirroring protocol.ts.
-from dimos.web.relay_bridge.manifest import ChannelSpec as ChannelSpec, Delivery as Delivery
+from dimos.web.relay_bridge.manifest import (
+ MAX_MANIFEST_ID_LEN,
+ ChannelSpec as ChannelSpec,
+ Delivery as Delivery,
+ PanelSpec as PanelSpec,
+)
logger = setup_logger()
@@ -97,6 +102,9 @@ class RobotInfo(_WireModel):
class RobotManifest(_WireModel):
channels: list[ChannelSpec]
+ # Empty and absent are equivalent on the wire (TS omits undefined; the
+ # local default always serializes as []).
+ panels: list[PanelSpec] = Field(default_factory=list)
# Sentinel validation context passed by every wire-decode path: lets Hello
@@ -162,6 +170,7 @@ class Manifest(_WireModel):
t: Literal["manifest"] = "manifest"
robotId: str
channels: list[ChannelSpec]
+ panels: list[PanelSpec] = Field(default_factory=list)
class Sub(_WireModel):
@@ -232,7 +241,10 @@ class FrameHeader(_WireModel):
`meta` carries encoding-specific extras.
"""
- ch: str
+ # Bounded like manifest channel ids: the relay drops frames with oversize
+ # undeclared names, so local construction fails fast instead of emitting
+ # a frame the relay cannot route (the file's encode-fail-fast policy).
+ ch: str = Field(max_length=MAX_MANIFEST_ID_LEN)
seq: int | float
ts: int | float
delivery: Delivery
diff --git a/dimos/web/relay_bridge/relay_bridge_module.py b/dimos/web/relay_bridge/relay_bridge_module.py
index a805d5006c..583d288155 100644
--- a/dimos/web/relay_bridge/relay_bridge_module.py
+++ b/dimos/web/relay_bridge/relay_bridge_module.py
@@ -145,7 +145,10 @@ class RelayBridgeConfig(ModuleConfig):
"""Relay identity; empty falls back to g.robot_id, then the hostname."""
robot_name: str = ""
"""Display name; empty falls back to robot_id."""
- jpeg_quality: int = 75
+ jpeg_quality: int = Field(default=75, ge=0, le=100)
+ # MuJoCo publishes video at 20 Hz. Keep enough headroom for that source and
+ # for camera jitter: a cap close to the nominal rate aliases slightly early
+ # frames into an every-other-frame pattern.
image_max_hz: float = Field(default=30.0, gt=0.0)
odom_max_hz: float = Field(default=20.0, gt=0.0)
available_channels: tuple[str, ...] | None = None
@@ -180,6 +183,22 @@ class ChannelDef:
delivery: Delivery
max_hz: Callable[[RelayBridgeConfig], float]
encode: Callable[[RelayBridgeModule, Any], tuple[bytes, _FrameMeta]]
+ # Cockpit panel-component kind rendered for this channel (None: raw row
+ # only). One panel per channel until the T7 authoring API.
+ panel_kind: str | None = None
+
+
+def _passes_rate_gate(
+ last_input: dict[str, float],
+ ch: str,
+ now: float,
+ min_interval: float,
+) -> bool:
+ """Claim the current input when it is outside the channel's rate interval."""
+ if now - last_input.get(ch, 0.0) < min_interval:
+ return False
+ last_input[ch] = now
+ return True
@dataclass(slots=True)
@@ -193,7 +212,9 @@ class _Session:
# The v0 channel table; every entry needs a matching `In` on the module.
CHANNELS: tuple[ChannelDef, ...] = (
- ChannelDef("color_image", "jpeg.v1", "latest", lambda c: c.image_max_hz, _encode_image),
+ ChannelDef(
+ "color_image", "jpeg.v1", "latest", lambda c: c.image_max_hz, _encode_image, "video"
+ ),
ChannelDef("odom", "pose.json.v1", "reliable", lambda c: c.odom_max_hz, _encode_odom),
)
@@ -211,10 +232,15 @@ def build_manifest(config: RelayBridgeConfig, channels: tuple[ChannelDef, ...])
"maxHz": cd.max_hz(config),
}
for cd in channels
- ]
+ ],
+ "panels": [
+ {"id": cd.ch, "kind": cd.panel_kind, "channels": [cd.ch]}
+ for cd in channels
+ if cd.panel_kind is not None
+ ],
}
)
- return RobotManifest(channels=manifest.channels)
+ return RobotManifest(channels=manifest.channels, panels=manifest.panels)
def resolve_robot_info(config: RelayBridgeConfig) -> RobotInfo:
@@ -468,9 +494,8 @@ def _on_input(self, session: _Session, cd: ChannelDef, sender: _Sender, msg: Any
if session.retired.is_set():
return
now = time.monotonic()
- if now - self._last_input.get(cd.ch, 0.0) < self._min_interval[cd.ch]:
+ if not _passes_rate_gate(self._last_input, cd.ch, now, self._min_interval[cd.ch]):
return
- self._last_input[cd.ch] = now
try:
payload, meta = cd.encode(self, msg)
except Exception:
diff --git a/dimos/web/relay_bridge/relay_process.py b/dimos/web/relay_bridge/relay_process.py
index 8b084f6012..71d5ef1377 100644
--- a/dimos/web/relay_bridge/relay_process.py
+++ b/dimos/web/relay_bridge/relay_process.py
@@ -83,9 +83,9 @@ def ensure_cockpit_dist(
hash over the cockpit/shared sources, workspace config, and lockfile) no
longer matches; the build runs under a cross-process lock, into a
temporary directory published atomically, so concurrent starts serialize
- and a failed build leaves the served dist untouched (the relay still runs,
- and serves the debug page when there is no dist at all). Setting `cancel`
- kills the build child within a bounded grace.
+ and a failed build leaves the served dist untouched (the relay still runs;
+ with no dist at all it serves only /api plus a build hint at /). Setting
+ `cancel` kills the build child within a bounded grace.
"""
dist = find_cockpit_dist(web_dir)
cockpit = web_dir / "cockpit"
@@ -255,14 +255,11 @@ class RelayReadyInfo:
# True when the relay serves a built Cockpit at /; set by RelayProcess.
cockpit: bool = False
- @property
- def debug_url(self) -> str:
- return f"http://127.0.0.1:{self.http_port}/debug.html"
-
@property
def open_url(self) -> str:
- """What a browser should open: the Cockpit, or the debug page without one."""
- return f"http://127.0.0.1:{self.http_port}/" if self.cockpit else self.debug_url
+ """What a browser should open (without a cockpit dist the relay
+ answers it with a 404 build hint)."""
+ return f"http://127.0.0.1:{self.http_port}/"
class RelayProcess:
diff --git a/dimos/web/relay_bridge/test_protocol.py b/dimos/web/relay_bridge/test_protocol.py
index ae7558b7dc..22315945b4 100644
--- a/dimos/web/relay_bridge/test_protocol.py
+++ b/dimos/web/relay_bridge/test_protocol.py
@@ -31,6 +31,7 @@
DataFrameStreamReader,
FrameHeader,
Hello,
+ PanelSpec,
Ping,
ProtocolError,
RobotInfo,
@@ -268,6 +269,12 @@ def test_msg_from_dict_validates_nested_session_shapes():
)
# hello stays valid without the optional robot/manifest (viewer form).
assert msg_from_dict({"t": "hello", "v": 1, "role": "viewer"}) == Hello(v=1, role="viewer")
+ panel = {"id": "pose", "kind": "readout", "channels": ["odom"]}
+ expected_panel = PanelSpec(id="pose", kind="readout", channels=["odom"])
+ with_panels = {**full, "manifest": {"channels": [spec], "panels": [panel]}}
+ assert msg_from_dict(with_panels).manifest.panels == [expected_panel]
+ manifest_msg = {"t": "manifest", "robotId": "r", "channels": [spec], "panels": [panel]}
+ assert msg_from_dict(manifest_msg).panels == [expected_panel]
bad = [
# Optional means absent-or-valid: explicit null is rejected on the
# wire (local construction with robot=None stays fine: absent).
@@ -276,6 +283,9 @@ def test_msg_from_dict_validates_nested_session_shapes():
{**full, "manifest": {"channels": [{**spec, "maxHz": "20"}]}},
{**full, "manifest": {"channels": [{**spec, "delivery": "bogus"}]}},
{**full, "manifest": {"channels": robot}},
+ {**full, "manifest": {"channels": [spec], "panels": None}},
+ {**full, "manifest": {"channels": [spec], "panels": [{"id": "x"}]}},
+ {"t": "manifest", "robotId": "r", "channels": [spec], "panels": [{"kind": 5}]},
{"t": "robots", "robots": {}},
{"t": "robots", "robots": [{"id": "a", "name": "b"}]},
{"t": "robots"},
@@ -324,6 +334,20 @@ def test_data_frame_header_rejects_non_finite():
decode_data_frame(_raw_data_frame(hdr))
+def test_data_frame_header_bounds_ch_length():
+ # ch is bounded like manifest channel ids (64, mirrored in protocol.ts):
+ # oversize undeclared names are dropped before routing, and local
+ # construction fails fast too.
+ def hdr(ch):
+ return json.dumps({"ch": ch, "seq": 1, "ts": 1.0, "delivery": "latest"}).encode()
+
+ assert decode_data_frame(_raw_data_frame(hdr("c" * 64))).header.ch == "c" * 64
+ with pytest.raises(ProtocolError):
+ decode_data_frame(_raw_data_frame(hdr("c" * 65)))
+ with pytest.raises(ValueError):
+ FrameHeader(ch="c" * 65, seq=1, ts=1.0, delivery="latest")
+
+
def test_huge_int_is_a_valid_number():
# Arbitrary-precision ints are legal JSON and always finite; they must
# pass (a math.isfinite check would raise OverflowError on them).
diff --git a/dimos/web/relay_bridge/test_relay_bridge_module.py b/dimos/web/relay_bridge/test_relay_bridge_module.py
index be67c1fa49..1c16ed4211 100644
--- a/dimos/web/relay_bridge/test_relay_bridge_module.py
+++ b/dimos/web/relay_bridge/test_relay_bridge_module.py
@@ -40,6 +40,7 @@
from dimos.core.stream import Out
from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
from dimos.msgs.sensor_msgs.Image import Image
+from dimos.simulation.mujoco.constants import VIDEO_FPS
from dimos.web.relay_bridge import relay_bridge_module
from dimos.web.relay_bridge.e2e_support import stop_module
from dimos.web.relay_bridge.protocol import Msg, RobotManifest, Subs
@@ -250,6 +251,10 @@ def test_manifest_and_robot_info_content() -> None:
image, odom = manifest.channels
assert (image.encoding, image.delivery, image.maxHz) == ("jpeg.v1", "latest", 12.0)
assert (odom.encoding, odom.delivery, odom.maxHz) == ("pose.json.v1", "reliable", 20.0)
+ # One video panel for the camera; odom stays a raw channel row.
+ assert [(p.id, p.kind, p.channels) for p in manifest.panels] == [
+ ("color_image", "video", ["color_image"])
+ ]
info = resolve_robot_info(config)
assert (info.id, info.name) == ("go2-lab", "Lab")
@@ -266,9 +271,11 @@ def test_manifest_and_robot_info_content() -> None:
("image_max_hz", -1.0),
("odom_max_hz", 0.0),
("odom_max_hz", -1.0),
+ ("jpeg_quality", -1),
+ ("jpeg_quality", 101),
],
)
-def test_channel_rates_must_be_positive(field: str, value: float) -> None:
+def test_config_rejects_out_of_range_values(field: str, value: float) -> None:
with pytest.raises(ValidationError):
RelayBridgeConfig(**{field: value})
@@ -346,6 +353,49 @@ def test_encode_paths_and_max_hz_gate(bridge) -> None:
assert wait_until(lambda: module.encoded["odom"] == count + 2)
+def test_default_image_gate_preserves_mujoco_video_rate() -> None:
+ config = RelayBridgeConfig()
+ last_input: dict[str, float] = {}
+ times = [100.0 + frame / VIDEO_FPS for frame in range(VIDEO_FPS)]
+ accepted = [
+ now
+ for now in times
+ if relay_bridge_module._passes_rate_gate(
+ last_input,
+ "color_image",
+ now,
+ 1.0 / config.image_max_hz,
+ )
+ ]
+
+ assert accepted == times
+
+
+def test_no_jpeg_encode_while_unsubscribed(bridge, monkeypatch) -> None:
+ # The ticket-mandated spy: encode work must not happen without viewers,
+ # independent of the module.encoded bookkeeping.
+ module, clients = bridge
+ calls = {"n": 0}
+ real = Image.to_jpeg_bytes
+
+ def spy(self: Image, quality: int = 75) -> bytes:
+ calls["n"] += 1
+ return real(self, quality=quality)
+
+ monkeypatch.setattr(Image, "to_jpeg_bytes", spy)
+ image = Image.from_numpy(np.zeros((8, 12, 3), dtype=np.uint8))
+ image_transport(module).publish(image)
+ image_transport(module).publish(image)
+ flush_loop(module)
+ assert calls["n"] == 0
+ assert module.encoded["color_image"] == 0
+
+ push(module, clients[0], Subs(chs=["color_image"], n=1))
+ assert wait_until(lambda: image_transport(module).subscribers)
+ image_transport(module).publish(image)
+ assert calls["n"] == 1
+
+
def test_session_loss_stops_encoders_and_reconnects(bridge) -> None:
module, clients = bridge
push(module, clients[0], Subs(chs=["odom"], n=5))
@@ -571,6 +621,7 @@ def test_unwired_input_is_not_advertised_or_subscribed(monkeypatch) -> None:
_, manifest = clients[0].hello_args
assert isinstance(manifest, RobotManifest)
assert [channel.ch for channel in manifest.channels] == ["odom"]
+ assert manifest.panels == [] # the video panel drops with its channel
push(module, clients[0], Subs(chs=["color_image", "odom"], n=1))
assert wait_until(lambda: len(odom_transport(module).subscribers) == 1)
diff --git a/dimos/web/relay_bridge/test_relay_e2e.py b/dimos/web/relay_bridge/test_relay_e2e.py
index 00efbb51bb..ac54e98d8b 100644
--- a/dimos/web/relay_bridge/test_relay_e2e.py
+++ b/dimos/web/relay_bridge/test_relay_e2e.py
@@ -266,8 +266,20 @@ async def test_stats_reflect_traffic(
assert {"id": ROBOT.id, "name": ROBOT.name, "model": ROBOT.model} in stats["robots"]
assert stats["viewers"] >= 1
assert stats["perRobot"][ROBOT.id]["subs"] == ["odom"]
- assert stats["perRobot"][ROBOT.id]["channels"]["odom"]["framesIn"] >= 1
- assert stats["perRobot"][ROBOT.id]["channels"]["odom"]["delivery"] == "reliable"
+ # This module's robot declares no manifest, so its traffic lands in the
+ # one aggregate bucket: per-channel ingress stats exist only for declared
+ # channels (arbitrary ch strings must not grow the map).
+ assert stats["perRobot"][ROBOT.id]["channels"] == {}
+ undeclared = stats["perRobot"][ROBOT.id]["undeclared"]
+ assert undeclared["framesIn"] >= 1
+ assert undeclared["bytesIn"] >= 2
+ assert isinstance(undeclared["fps"], (int, float))
+ viewer_stats = next(v for v in stats["perViewer"] if v["watched"] == ROBOT.id)
+ odom = viewer_stats["channels"]["odom"]
+ assert odom["sent"] >= 1
+ assert odom["bytesOut"] >= 2
+ # A healthy reliable channel never resets anything.
+ assert (odom["aborted"], odom["expired"], odom["inflight"]) == (0, 0, 0)
async def test_duplicate_robot_id_is_terminal_until_first_disconnects(
diff --git a/dimos/web/relay_bridge/test_relay_process.py b/dimos/web/relay_bridge/test_relay_process.py
index 742f4f0c7d..cf7a19c840 100644
--- a/dimos/web/relay_bridge/test_relay_process.py
+++ b/dimos/web/relay_bridge/test_relay_process.py
@@ -85,9 +85,6 @@ def test_relay_serves_cockpit_dist(tmp_path: Path) -> None:
assert status == 200 and b"fake cockpit index" in index
status, asset = _fetch(f"{info.open_url}assets/app.js")
assert status == 200 and b"console.log" in asset
- # The debug page still resolves from relay/static/ behind the cockpit.
- status, debug = _fetch(info.debug_url)
- assert status == 200 and b"DimOS relay debug" in debug
status, body = _fetch(f"{info.open_url}api/info")
assert status == 200
@@ -97,14 +94,13 @@ def test_relay_serves_cockpit_dist(tmp_path: Path) -> None:
assert api_info["certHash"] == info.cert_hash
-def test_relay_without_cockpit_serves_debug(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_relay_without_cockpit_serves_build_hint(monkeypatch: pytest.MonkeyPatch) -> None:
# The checkout may have a built dist; force the "not built" path.
monkeypatch.setattr(relay_process, "find_cockpit_dist", lambda _: None)
with RelayProcess(port=0) as info:
assert info.cockpit is False
- assert info.open_url == info.debug_url
- status, index = _fetch(f"http://127.0.0.1:{info.http_port}/")
- assert status == 200 and b"DimOS relay debug" in index
+ status, index = _fetch(info.open_url)
+ assert status == 404 and b"cockpit dist not built" in index
class _FakeBuild:
diff --git a/dimos/web/relay_bridge/test_wt_client.py b/dimos/web/relay_bridge/test_wt_client.py
index c7f08c0070..635ef6396a 100644
--- a/dimos/web/relay_bridge/test_wt_client.py
+++ b/dimos/web/relay_bridge/test_wt_client.py
@@ -110,13 +110,62 @@ async def test_offer_survives_delivery_and_keeps_sending() -> None:
# A healthy pump keeps accepting; sanity check that the stub path works.
session = StubSession()
writer = _client(session).latest_writer("cam")
- for i in range(5):
- writer.offer(i.to_bytes(4, "little"))
- await asyncio.sleep(0)
- await asyncio.sleep(0.02)
- assert not writer._task.done()
- assert writer.sent >= 1
- writer.stop()
+ try:
+ for i in range(5):
+ writer.offer(i.to_bytes(4, "little"))
+ await asyncio.sleep(0)
+ await asyncio.sleep(0.02)
+ assert not writer._task.done()
+ assert writer.sent >= 1
+ finally:
+ writer.stop()
+
+
+async def test_mailbox_drops_stale_frames_under_slow_writer() -> None:
+ # A wedged transport must never queue stale frames: the 1-slot mailbox
+ # sheds everything but the newest, which is what goes out on unwedge.
+ session = StubSession()
+ payloads: list[bytes] = []
+ polls = 0
+
+ real_send = session.send_frame
+
+ def send_frame(header: FrameHeader, payload: bytes) -> int:
+ payloads.append(payload)
+ return real_send(header, payload)
+
+ def stream_in_flight(stream_id: int) -> bool:
+ nonlocal polls
+ polls += 1
+ return polls < 20 # first send stays in flight for ~20 pump polls
+
+ session.send_frame = send_frame # type: ignore[method-assign]
+ session.stream_in_flight = stream_in_flight # type: ignore[method-assign]
+
+ # stale_after high enough that the wedge never triggers the reset path:
+ # this test pins the mailbox semantics alone.
+ writer = _client(session).latest_writer("cam", stale_after=10.0)
+ try:
+ writer.offer(b"f0")
+ # Wait until the pump provably took f0 (send_frame ran): it is wedged
+ # in its in-flight poll, so the offers below only touch the mailbox.
+ for _ in range(500):
+ if payloads:
+ break
+ await asyncio.sleep(0.005)
+ assert payloads, "pump did not send f0 within 2.5 s"
+ for i in range(1, 6):
+ writer.offer(b"f%d" % i)
+ assert writer.dropped == 4 # f1..f4 displaced by their successors
+
+ for _ in range(500):
+ if writer.sent == 2:
+ break
+ await asyncio.sleep(0.005)
+ assert payloads == [b"f0", b"f5"] # only the newest frame followed the wedge
+ assert writer.sent == 2
+ finally:
+ writer.stop()
async def test_frames_cancel_does_not_steal_next_frame() -> None:
diff --git a/dimos/web/relay_bridge/test_wt_session.py b/dimos/web/relay_bridge/test_wt_session.py
index 2976d07740..d9b65bddb6 100644
--- a/dimos/web/relay_bridge/test_wt_session.py
+++ b/dimos/web/relay_bridge/test_wt_session.py
@@ -16,6 +16,7 @@
frames are fed straight into the protocol callbacks)."""
from aioquic.quic.connection import QuicConnection
+from aioquic.quic.events import StreamReset
from dimos.web.relay_bridge._wt_session import (
_FRAME_QUEUE_MAX,
@@ -111,3 +112,23 @@ async def test_per_encoding_payload_caps():
session._stream_data_received(12, _frame_bytes("mystery", 1, 9 * 1024 * 1024), True)
assert session.frames.qsize() == 3
assert session.frames_oversized == 1
+
+
+async def test_stream_reset_drops_the_frame_reader():
+ # The relay ends every latest stream with a reset (reap/dispose); the
+ # reader map must not grow one leaked entry per stream, and a partial
+ # frame is stale by definition.
+ session = _session()
+ full = _frame_bytes("cam", 1, 1024)
+ session._stream_data_received(4, full[: len(full) // 2], False)
+ assert 4 in session._frame_readers
+ session.quic_event_received(StreamReset(error_code=1, stream_id=4))
+ assert 4 not in session._frame_readers
+ assert session.frames.qsize() == 0
+
+ # A reset for a stream that already dispatched its frame is a no-op.
+ session._stream_data_received(8, _frame_bytes("cam", 2, 16), False)
+ assert session.frames.qsize() == 1
+ session.quic_event_received(StreamReset(error_code=1, stream_id=8))
+ assert 8 not in session._frame_readers
+ assert session.frames.qsize() == 1
diff --git a/setup.py b/setup.py
index 6bc932be8a..9cd153ebc5 100644
--- a/setup.py
+++ b/setup.py
@@ -97,20 +97,22 @@ def _copy_relay_dist(self):
if not (src / "relay" / "main.ts").is_file():
raise RuntimeError(f"relay sources missing at {src}; refusing to build the wheel")
if not (src / "cockpit" / "dist" / "index.html").is_file():
- # Release wheels must carry the Cockpit (the release workflow
- # builds it first). A plain `pip install .` from a checkout
- # without a built dist still works: the relay serves the debug
- # page instead.
- if os.environ.get("CIBUILDWHEEL") == "1":
+ # Wheels must carry the Cockpit: there is no fallback debug page,
+ # so a dist-less wheel's relay serves no UI at all. Building a
+ # deliberate python-only wheel (e.g. where deno is unavailable)
+ # requires the explicit env-var opt-out.
+ if os.environ.get("DIMOS_ALLOW_MISSING_COCKPIT") != "1":
raise RuntimeError(
f"cockpit dist missing at {src / 'cockpit' / 'dist'}; run "
- "`deno task build` in web/cockpit before building release wheels"
+ "`deno task build` in web/cockpit (or `dimos run --local-relay` "
+ "from a checkout builds it), or set DIMOS_ALLOW_MISSING_COCKPIT=1 "
+ "to build a UI-less wheel anyway"
)
- self.warn("cockpit dist missing; this wheel will serve the debug page only")
+ self.warn("cockpit dist missing; this wheel's relay will have no UI")
dst = Path(self.build_lib) / RELAY_DIST_TARGET
for name in RELAY_DIST_SOURCES:
entry = src / name
- if not entry.exists(): # only cockpit/dist may be absent (warned above)
+ if not entry.exists(): # only cockpit/dist may be absent (env-var opt-out above)
continue
for path in sorted(entry.rglob("*")) if entry.is_dir() else [entry]:
# Filter on the path below src: matching path.parts would also
diff --git a/web/README.md b/web/README.md
index aba3aeb27e..18e94bea55 100644
--- a/web/README.md
+++ b/web/README.md
@@ -9,7 +9,7 @@ node/npm anywhere: vite, vitest, and tsc run as npm packages under Deno (`nodeMo
and `dimos --local-relay` auto-downloads Deno via `ensure_deno()`.
```bash
-deno task dev # relay on http://127.0.0.1:7780 (debug page at /debug.html)
+deno task dev # relay on http://127.0.0.1:7780 (add --cockpit-dir cockpit/dist for the UI)
deno task test # relay + shared tests (unit + loopback e2e)
deno task check # type-check relay + shared; deno fmt + deno lint for style (all of web/)
```
@@ -91,8 +91,30 @@ Several choices are workarounds for upstream bugs, verified 2026-07-10..15 on De
FIN goes out lazily (bug 2), so with stream-per-frame the relay's
`createUnidirectionalStream({waitUntilAvailable})` hangs after ~100 frames, the reliable FIFO
overflows, and the relay kicks the viewer every ~8 s. Chromium's much larger window masks this.
- Latest channels keep per-frame streams (their reset semantics need them) and are the known
- remaining credit pressure for T5 video under Firefox.
+ Latest channels keep per-frame streams (their reset semantics need them); bug 12 is what keeps
+ their credit pressure bounded.
+12. **Relay->viewer latest streams are never FIN'd: every one ends in a RESET.** JS WebTransport
+ exposes no delivery signal (`getStats()` is a zeros stub in Deno 2.6.10) and quinn buffers
+ writes without bound, so "write accepted" says nothing about delivery - and a closed
+ WritableStream can no longer be aborted. The relay therefore keeps each latest stream open and
+ reaps it: streams older than `LATEST_STALE_MS` (500 ms, matching the Python leg's `stale_after`)
+ are reset, discarding buffered-but-undelivered bytes on both ends and returning Firefox's stream
+ credit far faster than the lazy FIN would. Reaping fires from newer offers AND from a periodic
+ reap every `LATEST_STALE_MS`: an idle input stops offering and would otherwise leave up to about
+ 100 open streams pinning Firefox's uni-stream credit. The one send still wedged in
+ `createUnidirectionalStream` is never reset - resets do not replenish stream credit while the
+ viewer's application is not reading (verified against Deno's client; a frozen tab behaves the
+ same) - instead newer offers supersede its payload in place (the payload binds only when the
+ write starts), so a resuming viewer receives the newest frame with zero stream churn. Once the
+ write has started the payload can no longer change, so a newer offer resets a write-wedged
+ carrier once it is `LATEST_STALE_MS` old and resends the newest on a fresh stream - bounded to
+ one reset per stale window, and once stream credit runs out the wedge moves back to creation,
+ where superseding is churn-free. In `/api/stats`, `aborted` counts backpressure resets - stale
+ accepted streams reaped while the newest send is unaccepted, plus stale write-wedged carriers
+ (the suspended-viewer signal; 0 when healthy); `expired` counts routine end-of-life resets (~=
+ `sent` on a healthy latest channel); superseded and displaced payloads count as `dropped`.
+ Receivers dispatch frames on byte count (bug 2) and treat the reset as end-of-stream; a reset
+ mid-frame drops a stale partial by design.
Latest streams deliver out of order by design; consumers keep the newest frame by `seq` (a reliable
channel's persistent stream is ordered) and loss metrics are span-based
diff --git a/web/cockpit/src/App.test.tsx b/web/cockpit/src/App.test.tsx
index c2e0736003..d4ab66f708 100644
--- a/web/cockpit/src/App.test.tsx
+++ b/web/cockpit/src/App.test.tsx
@@ -9,6 +9,7 @@ import { ChannelStore, StatusStore } from "./session/store.ts";
(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
const ODOM = { ch: "odom", encoding: "pose.json.v1", delivery: "reliable", maxHz: 20 } as const;
+const IMAGE = { ch: "color_image", encoding: "jpeg.v1", delivery: "latest", maxHz: 15 } as const;
describe("App session states", () => {
let container: HTMLElement;
@@ -111,6 +112,25 @@ describe("App session states", () => {
expect(container.querySelector('[data-testid="ch-odom-seq"]')!.textContent).toBe("9");
});
+ it("marks the jpeg channel unsubscribed until a video panel binds it", () => {
+ act(() => {
+ status.update({
+ robot: { id: "a", name: "A", model: "go2" },
+ robotCount: 1,
+ channels: [ODOM, IMAGE],
+ panels: [],
+ });
+ });
+ const value = () => container.querySelector('[data-testid="ch-color_image-value"]')!;
+ expect(value().textContent).toContain("not subscribed (no panel binds it)");
+
+ // The manifest gains a video panel: the session subscribes, the row waits.
+ act(() => {
+ status.update({ panels: [{ id: "cam", kind: "video", channels: ["color_image"] }] });
+ });
+ expect(value().textContent).toContain("waiting for data...");
+ });
+
it("shows the multi-robot notice instead of channels", () => {
act(() => status.update({ robotCount: 2, channels: [] }));
expect(container.textContent).toContain("2 robots connected");
diff --git a/web/cockpit/src/App.tsx b/web/cockpit/src/App.tsx
index 17d57d73b1..d43b49fd4c 100644
--- a/web/cockpit/src/App.tsx
+++ b/web/cockpit/src/App.tsx
@@ -1,3 +1,4 @@
+import { PanelGrid } from "./panels/PanelGrid.tsx";
import { useStatus } from "./session/hooks.ts";
import type { SessionHandle } from "./session/session.ts";
import { ChannelList } from "./ui/ChannelList.tsx";
@@ -19,7 +20,12 @@ export function App({ session }: { session: SessionHandle }) {
} else if (status.channels.length === 0) {
content = Waiting for a robot to register...
;
} else {
- content = ;
+ content = (
+ <>
+
+
+
+ );
}
return (
diff --git a/web/cockpit/src/main.tsx b/web/cockpit/src/main.tsx
index 7aa39339cc..bede506159 100644
--- a/web/cockpit/src/main.tsx
+++ b/web/cockpit/src/main.tsx
@@ -5,7 +5,7 @@ import "./index.css";
const root = createRoot(document.getElementById("root")!);
-// Same capability checks as the debug page: WebTransport needs a secure
+// Capability checks before anything mounts: WebTransport needs a secure
// context, and Safari has no WebTransport as of mid-2026.
if (!globalThis.isSecureContext) {
root.render(
diff --git a/web/cockpit/src/panels/PanelGrid.module.css b/web/cockpit/src/panels/PanelGrid.module.css
new file mode 100644
index 0000000000..d80281df1d
--- /dev/null
+++ b/web/cockpit/src/panels/PanelGrid.module.css
@@ -0,0 +1,6 @@
+.grid {
+ display: grid;
+ gap: 1rem;
+ grid-template-columns: repeat(auto-fit, minmax(320px, 1fr));
+ margin-bottom: 1rem;
+}
diff --git a/web/cockpit/src/panels/PanelGrid.tsx b/web/cockpit/src/panels/PanelGrid.tsx
new file mode 100644
index 0000000000..910b397e4a
--- /dev/null
+++ b/web/cockpit/src/panels/PanelGrid.tsx
@@ -0,0 +1,22 @@
+// Fixed layout v0: the manifest's panels in manifest order in a simple CSS
+// grid. The T7 layout tree (rows/cols with shares, pages) replaces this.
+
+import type { PanelSpec } from "@dimos/shared";
+import type { ChannelStore } from "../session/store.ts";
+import { getPanel } from "./registry.ts";
+import styles from "./PanelGrid.module.css";
+
+export function PanelGrid({ panels, store }: { panels: PanelSpec[]; store: ChannelStore }) {
+ const known = panels.flatMap((spec) => {
+ const Component = getPanel(spec.kind);
+ // Unknown kind: this build has no such panel component (newer bridge);
+ // the channel list below still shows the underlying channels.
+ return Component === undefined ? [] : [{ spec, Component }];
+ });
+ if (known.length === 0) return null;
+ return (
+
+ {known.map(({ spec, Component }) => )}
+
+ );
+}
diff --git a/web/cockpit/src/panels/VideoPanel.module.css b/web/cockpit/src/panels/VideoPanel.module.css
new file mode 100644
index 0000000000..933cd9e7c3
--- /dev/null
+++ b/web/cockpit/src/panels/VideoPanel.module.css
@@ -0,0 +1,50 @@
+.panel {
+ border: 1px solid #30363d;
+ border-radius: 6px;
+ display: flex;
+ flex-direction: column;
+ overflow: hidden;
+}
+
+.head {
+ align-items: center;
+ background: #1b1f24;
+ display: flex;
+ gap: 0.5rem;
+ justify-content: space-between;
+ padding: 0.3rem 0.6rem;
+}
+
+.title {
+ font-weight: 600;
+}
+
+.badge,
+.badgeStale {
+ color: #8b949e;
+ font-variant-numeric: tabular-nums;
+}
+
+.badgeStale {
+ color: #f85149;
+}
+
+.body {
+ align-items: center;
+ background: #000;
+ display: flex;
+ justify-content: center;
+ min-height: 120px;
+ position: relative;
+}
+
+.canvas {
+ display: block;
+ max-height: 70vh;
+ max-width: 100%;
+}
+
+.waiting {
+ color: #8b949e;
+ position: absolute;
+}
diff --git a/web/cockpit/src/panels/VideoPanel.tsx b/web/cockpit/src/panels/VideoPanel.tsx
new file mode 100644
index 0000000000..c03cd9ae3b
--- /dev/null
+++ b/web/cockpit/src/panels/VideoPanel.tsx
@@ -0,0 +1,178 @@
+// Live JPEG video: canvas drawing driven by the store's direct-subscribe path
+// (React is not involved at frame rate; the badge rides the 500 ms UI tick).
+
+import { useEffect, useRef } from "react";
+import { useChannel } from "../session/hooks.ts";
+import type { ChannelStore } from "../session/store.ts";
+import type { PanelProps } from "./registry.ts";
+import styles from "./VideoPanel.module.css";
+
+// A frame this far behind source time is flagged stale (frames are useless
+// the moment a newer one exists, so this only trips on silence or stalls).
+export const VIDEO_STALE_MS = 2000;
+
+/** Test seams: real decode is createImageBitmap; real hidden is the page's. */
+export interface VideoSinkDeps {
+ decode?: (payload: Uint8Array) => Promise;
+ hidden?: () => boolean;
+}
+
+/** Sink-side draw diagnostics, mutated in place and sampled by the badge. */
+export interface DrawHealth {
+ /** Browser ms of the last successful draw, stamped at sink start before the first. */
+ lastDrawOkAtMs: number;
+ /** Consecutive failed decode-or-draw attempts since the last success. */
+ failures: number;
+}
+
+/**
+ * Drive `canvas` from the channel's slot: at most one decode in flight, and
+ * on completion the pump re-checks the slot, so a burst of frames costs one
+ * decode of the newest (latest-wins, same shedding rule as everywhere else in
+ * the pipeline). Undecodable frames are skipped without spinning. While the
+ * document is hidden nothing decodes; visibilitychange catches back up.
+ * Returns the cleanup function.
+ */
+export function startVideoSink(
+ store: ChannelStore,
+ ch: string,
+ canvas: HTMLCanvasElement,
+ health: DrawHealth,
+ deps: VideoSinkDeps = {},
+): () => void {
+ const decode = deps.decode ??
+ ((payload: Uint8Array) =>
+ createImageBitmap(new Blob([payload as BlobPart], { type: "image/jpeg" })));
+ const hidden = deps.hidden ?? (() => document.hidden);
+ const ctx = canvas.getContext("2d");
+ let decoding = false;
+ let drawnVersion = -1;
+ let stopped = false;
+ // A fresh mount is never instantly "stalled".
+ health.lastDrawOkAtMs = Date.now();
+ health.failures = 0;
+
+ const pump = (): void => {
+ if (stopped || decoding || hidden()) return;
+ const slot = store.get(ch);
+ if (slot === null || slot.version === drawnVersion) return;
+ if (!(slot.value instanceof Uint8Array)) return; // undecoded channel: nothing to draw
+ const version = slot.version;
+ decoding = true;
+ decode(slot.value)
+ .then((bmp) => {
+ try {
+ if (!stopped) {
+ if (canvas.width !== bmp.width || canvas.height !== bmp.height) {
+ canvas.width = bmp.width;
+ canvas.height = bmp.height;
+ }
+ ctx?.drawImage(bmp, 0, 0);
+ health.lastDrawOkAtMs = Date.now();
+ health.failures = 0;
+ }
+ } finally {
+ bmp.close(); // a throwing draw must not leak the bitmap
+ }
+ })
+ .catch(() => {
+ // Decode rejection or draw throw: skip this frame but count it, so
+ // the badge can surface a pipeline that never draws.
+ health.failures += 1;
+ })
+ .finally(() => {
+ drawnVersion = version;
+ decoding = false;
+ pump(); // newer frames may have landed during the decode
+ });
+ };
+
+ const unsubscribe = store.subscribe(ch, pump);
+ const onVisibility = (): void => pump();
+ document.addEventListener("visibilitychange", onVisibility);
+ pump(); // a slot may predate the mount
+ return () => {
+ stopped = true;
+ unsubscribe();
+ document.removeEventListener("visibilitychange", onVisibility);
+ };
+}
+
+function Badge({ store, ch, health }: { store: ChannelStore; ch: string; health: DrawHealth }) {
+ // Re-rendered on the 500 ms UI tick via useChannel; `health` is mutated by
+ // the sink at draw rate and simply sampled here (intended coupling).
+ const { stats } = useChannel(store, ch);
+ let text: string;
+ let error = false;
+ let stale = false;
+ if (stats.frames === 0) {
+ // Nothing ever arrived; a corrupt first frame is an error, not "waiting".
+ text = "waiting";
+ } else if (stats.decodeFailing || health.failures > 0) {
+ // A single bad frame trips this; the next success clears it.
+ text = "decode failing";
+ error = true;
+ } else if (stats.ageMs !== null && stats.ageMs > VIDEO_STALE_MS) {
+ text = `stale ${(stats.ageMs / 1000).toFixed(1)} s`;
+ stale = true;
+ } else if (stats.lastFrameAtMs - health.lastDrawOkAtMs > VIDEO_STALE_MS) {
+ // Frames arrive but nothing draws (e.g. a decoder that never settles);
+ // both operands are browser milliseconds.
+ text = "stalled";
+ stale = true;
+ } else {
+ text = `${stats.hz.toFixed(1)} fps`;
+ }
+ return (
+
+ {text}
+
+ );
+}
+
+export function VideoPanel({ spec, store }: PanelProps) {
+ const ch = spec.channels[0] as string | undefined;
+ if (ch === undefined) {
+ // A video panel without a channel is a bridge authoring mistake; render
+ // it visibly instead of crashing the grid.
+ return video panel {spec.id}: no channel bound
;
+ }
+ return ;
+}
+
+function VideoCanvas({ spec, store, ch }: PanelProps & { ch: string }) {
+ const canvasRef = useRef(null);
+ const health = useRef({ lastDrawOkAtMs: Date.now(), failures: 0 }).current;
+ const { slot } = useChannel(store, ch);
+
+ useEffect(() => {
+ const canvas = canvasRef.current;
+ if (canvas === null) return;
+ return startVideoSink(store, ch, canvas, health);
+ }, [store, ch, health]);
+
+ return (
+
+
+ {spec.id}
+
+
+
+
+ {slot === null && waiting for data...}
+
+
@@ -58,7 +72,9 @@ export function ChannelList({ channels, store }: { channels: ChannelSpec[]; stor
- {channels.map((spec) => )}
+ {channels.map((spec) => (
+
+ ))}
);
diff --git a/web/cockpit/src/ui/StatusBar.test.tsx b/web/cockpit/src/ui/StatusBar.test.tsx
index ee0a77b92b..93d7c36791 100644
--- a/web/cockpit/src/ui/StatusBar.test.tsx
+++ b/web/cockpit/src/ui/StatusBar.test.tsx
@@ -13,6 +13,7 @@ function makeStatus(over: Partial = {}): SessionStatus {
robot: null,
robotCount: 0,
channels: [],
+ panels: [],
epoch: 0,
lastError: null,
...over,
diff --git a/web/relay/forward.ts b/web/relay/forward.ts
index 51467f54db..5118a16431 100644
--- a/web/relay/forward.ts
+++ b/web/relay/forward.ts
@@ -11,18 +11,47 @@ import {
} from "@dimos/shared";
// Reliable channels: a viewer this far behind is dead weight; kick it so it
-// reconnects with a clean slate (T5 hardens and tunes these).
+// reconnects with a clean slate.
const RELIABLE_MAX_QUEUE = 64;
const RELIABLE_MAX_BYTES = 16 * 1024 * 1024;
+// A latest stream still un-reset this long after its frame was accepted is
+// presumed stale: a newer offer resets it. Matches the Python robot leg's
+// stale_after. Healthy viewers read frames long before this; the reset then
+// only ends an already-consumed stream (and returns its Firefox uni-stream
+// credit far faster than Deno's ~1 s lazy FIN would).
+export const LATEST_STALE_MS = 500;
+
// One decoder for every robot frame header (fatal so corrupt UTF-8 drops the
// frame rather than routing a mangled channel name).
const headerDecoder = new TextDecoder("utf-8", { fatal: true });
+/** One latest-channel frame handed to the transport. */
+export interface FrameSend {
+ /** Settles when the transport ACCEPTED the whole frame into its send
+ * buffers (flow-control/credit acceptance, NOT delivery); rejects on
+ * abort() or transport failure. */
+ readonly done: Promise;
+ /** True once abort() was called; a rejection of `done` is then expected. */
+ readonly aborted: boolean;
+ /** Reset the frame's stream: quinn discards buffered unsent data and the
+ * receiver discards buffered unread data. Rejects `done` immediately even
+ * while the stream is still being created. Idempotent; never throws. */
+ abort(): void;
+ /** Replace the payload if the stream write has not started (the send is
+ * still wedged in stream creation); false once the write began or after
+ * abort(). */
+ supersede(bytes: Uint8Array): boolean;
+}
+
/** Transport surface a policy writes to. */
export interface ViewerSink {
- /** One uni stream per call (latest channels; reset semantics need it). */
- sendFrame(bytes: Uint8Array): Promise;
+ /** One uni stream per call, never FIN'd (a closed WritableStream cannot be
+ * aborted, per spec abort() on a closed stream is a no-op): every latest
+ * stream ends in a reset - reap, dispose, or session teardown. Receivers
+ * dispatch frames on byte count and treat the reset as end-of-stream.
+ * Returns synchronously; all async work rides `done`. */
+ sendFrame(bytes: Uint8Array): FrameSend;
/** One persistent uni stream (reliable channels pack frames onto it). */
openStream(): Promise;
kick(reason: string): void;
@@ -33,46 +62,206 @@ export interface FrameWriter {
abort(reason?: unknown): Promise;
}
+export interface PolicyOptions {
+ staleMs?: number;
+ now?: () => number;
+}
+
export interface ChannelPolicy {
readonly delivery: Delivery;
+ /** Frames accepted by the transport (acceptance, not delivery). */
sent: number;
+ /** Frames shed before reaching the wire (latest: pending-slot replacement,
+ * carrier supersede replacement, and the payload of a stale write-wedged
+ * carrier reset; reliable: never, it kicks instead). Outside dispose/kick:
+ * offers == sent + dropped + queued(). */
dropped: number;
+ /** Latest streams reset while the channel was backpressured (the newest
+ * send still unaccepted): stale accepted streams reaped, plus stale
+ * write-wedged carriers themselves. The stalled/suspended-viewer signal,
+ * 0 when healthy. */
+ aborted: number;
+ /** Latest streams reset as routine end-of-life (~= sent when healthy:
+ * every latest stream ends in a reset). */
+ expired: number;
+ /** Payload bytes accepted by the transport (a superseded carrier counts
+ * the payload it finally carried). */
+ bytesOut: number;
+ /** Acceptance rate over the trailing window. */
+ readonly rate: Rate;
queued(): number;
+ /** Latest streams handed to the transport and not yet reset. */
+ inflight(): number;
offer(bytes: Uint8Array): void;
- /** Discard queued frames and release any persistent stream; later offers
- * and in-flight drain completions become no-ops. Idempotent. */
+ /** Reset accepted latest streams older than staleMs; clock-free (callers
+ * pass nowMs). Never touches the unaccepted carrier. No-op for reliable. */
+ reap(nowMs: number): void;
+ /** Discard queued frames, reset in-flight sends, and release any
+ * persistent stream; later offers and in-flight drain completions become
+ * no-ops. Idempotent. */
dispose(): void;
}
/**
- * Latest-wins: a 1-slot pending buffer. A frame arriving while a write is in
- * flight replaces the pending one (newest wins); the final frame is always
- * eventually delivered. A slow viewer sheds its own frames and nothing else.
+ * Frames/bytes rate over a trailing window: a ring of fixed time buckets,
+ * advanced on touch. The current partial bucket is included, so a cold start
+ * under-reads briefly. No clock inside: callers pass nowMs (one Date.now()
+ * per offer/frame on the hot path; tests fabricate time).
+ */
+export class Rate {
+ static readonly BUCKET_MS = 500;
+ static readonly BUCKETS = 10; // 5 s window
+ #frames = new Array(Rate.BUCKETS).fill(0);
+ #bytes = new Array(Rate.BUCKETS).fill(0);
+ #head = -1; // absolute index of the newest bucket; -1 = empty
+
+ #advance(bucket: number): void {
+ if (this.#head === -1 || bucket - this.#head >= Rate.BUCKETS) {
+ this.#frames.fill(0);
+ this.#bytes.fill(0);
+ } else {
+ for (let b = this.#head + 1; b <= bucket; b++) {
+ this.#frames[b % Rate.BUCKETS] = 0;
+ this.#bytes[b % Rate.BUCKETS] = 0;
+ }
+ }
+ this.#head = bucket;
+ }
+
+ push(bytes: number, nowMs: number): void {
+ let bucket = Math.floor(nowMs / Rate.BUCKET_MS);
+ if (bucket > this.#head) this.#advance(bucket);
+ else bucket = this.#head; // clock wobble: count into the newest bucket
+ this.#frames[bucket % Rate.BUCKETS] += 1;
+ this.#bytes[bucket % Rate.BUCKETS] += bytes;
+ }
+
+ snapshot(nowMs: number): { fps: number; bps: number } {
+ const bucket = Math.floor(nowMs / Rate.BUCKET_MS);
+ if (bucket > this.#head && this.#head !== -1) this.#advance(bucket);
+ let frames = 0;
+ let bytes = 0;
+ for (const n of this.#frames) frames += n;
+ for (const n of this.#bytes) bytes += n;
+ const windowS = (Rate.BUCKETS * Rate.BUCKET_MS) / 1000;
+ return { fps: Math.round((frames / windowS) * 10) / 10, bps: Math.round(bytes / windowS) };
+ }
+}
+
+interface OutstandingSend {
+ send: FrameSend;
+ at: number;
+ size: number;
+ accepted: boolean;
+}
+
+/**
+ * Latest-wins: a 1-slot pending buffer feeds sequential sends; a frame
+ * arriving while a send is in flight replaces the pending one (newest wins),
+ * and the final frame is always eventually delivered. A slow viewer sheds
+ * its own frames and nothing else.
+ *
+ * Acceptance is not delivery (quinn buffers writes without bound and QUIC
+ * exposes no delivery signal to JS), so accepted streams are tracked and
+ * REAPED: newer offers (and the relay's periodic reap; see server.ts) reset
+ * streams older than staleMs, discarding their buffered-but-undelivered
+ * bytes on both ends. A suspended viewer therefore resumes to at most
+ * ~staleMs of backlog instead of a connection window's worth of stale video
+ * replayed oldest-first.
*/
export class LatestChannel implements ChannelPolicy {
readonly delivery: Delivery = "latest";
sent = 0;
dropped = 0;
+ aborted = 0;
+ expired = 0;
+ bytesOut = 0;
+ readonly rate = new Rate();
#pending: Uint8Array | null = null;
+ #outstanding: OutstandingSend[] = []; // append order = age order
#writing = false;
#disposed = false;
+ readonly #staleMs: number;
+ readonly #now: () => number;
- constructor(readonly sink: ViewerSink) {}
+ constructor(readonly sink: ViewerSink, opts: PolicyOptions = {}) {
+ this.#staleMs = opts.staleMs ?? LATEST_STALE_MS;
+ this.#now = opts.now ?? Date.now;
+ }
queued(): number {
return this.#pending ? 1 : 0;
}
+ inflight(): number {
+ return this.#outstanding.length;
+ }
+
offer(bytes: Uint8Array): void {
if (this.#disposed) return;
+ const now = this.#now();
+ this.reap(now);
+ const last = this.#outstanding[this.#outstanding.length - 1];
+ if (last !== undefined && !last.accepted) {
+ // The carrier (sends are sequential: at most one unaccepted entry,
+ // always the tail). While it is still wedged in stream creation the
+ // newest payload replaces its old one in place: no reset, no
+ // replacement stream (README bug 12).
+ if (last.send.supersede(bytes)) {
+ this.dropped++; // the displaced carrier payload
+ last.at = now;
+ last.size = bytes.byteLength;
+ // Provably already null (parking only happens after supersede has
+ // latched false for this carrier); kept as cheap insurance.
+ this.#pending = null;
+ return;
+ }
+ // The write has begun, so the payload can no longer change: reset a
+ // stale carrier and let the newest go out on a fresh stream.
+ if (now - last.at >= this.#staleMs) {
+ // Pop first so reap and inflight() never see a dead unaccepted entry.
+ this.#outstanding.pop();
+ last.send.abort();
+ this.aborted++;
+ this.dropped++; // its payload never reached the wire
+ }
+ }
if (this.#pending) this.dropped++;
this.#pending = bytes;
this.#drain();
}
+ /**
+ * Reset accepted sends older than staleMs. The unaccepted tail (at most
+ * one: sends are sequential) is never reaped: it is the drain's pacing
+ * carrier, and resetting a create-wedged send would only spawn a
+ * replacement wedged on the same exhausted stream credit - resets do not
+ * replenish credit while the viewer's application is not reading its
+ * incoming streams (verified against Deno's client; a frozen tab behaves
+ * the same way). Newer offers supersede its payload in place instead
+ * (details in README bug 12).
+ */
+ reap(nowMs: number): void {
+ const backpressured = this.#outstanding.length > 0 &&
+ !this.#outstanding[this.#outstanding.length - 1].accepted;
+ while (this.#outstanding.length > 0) {
+ const oldest = this.#outstanding[0];
+ if (!oldest.accepted || nowMs - oldest.at < this.#staleMs) break;
+ this.#outstanding.shift();
+ oldest.send.abort();
+ if (backpressured) this.aborted++;
+ else this.expired++;
+ }
+ }
+
dispose(): void {
+ if (this.#disposed) return;
+ // Set first: the in-flight rejection this triggers must classify as
+ // disposal, not viewer failure.
this.#disposed = true;
this.#pending = null;
+ for (const entry of this.#outstanding) entry.send.abort();
+ this.#outstanding.length = 0;
}
#drain(): void {
@@ -82,8 +271,29 @@ export class LatestChannel implements ChannelPolicy {
while (this.#pending) {
const bytes = this.#pending;
this.#pending = null;
- await this.sink.sendFrame(bytes);
- this.sent++;
+ const send = this.sink.sendFrame(bytes);
+ const entry: OutstandingSend = {
+ send,
+ at: this.#now(),
+ size: bytes.byteLength,
+ accepted: false,
+ };
+ this.#outstanding.push(entry);
+ try {
+ await send.done;
+ entry.accepted = true;
+ // Staleness is measured from acceptance (per the LATEST_STALE_MS
+ // doc): a long-wedged carrier accepted on resume must not be
+ // instantly reaped by the idle reap timer.
+ entry.at = this.#now();
+ this.sent++;
+ this.bytesOut += entry.size;
+ this.rate.push(entry.size, entry.at);
+ } catch (e) {
+ // dispose() or a stale-carrier reset in offer() aborted it (the
+ // entry is already off the list), or the transport failed.
+ if (!send.aborted && !this.#disposed) throw e;
+ }
}
})()
.catch(() => {
@@ -104,7 +314,9 @@ export class LatestChannel implements ChannelPolicy {
/**
* Reliable: bounded per-viewer FIFO, no drops, delivery order preserved. On
- * overflow the viewer is kicked (better a visible reconnect than silent loss).
+ * overflow the viewer is kicked once and the channel self-disposes (better a
+ * visible reconnect than silent loss); frames offered before transport
+ * teardown completes are ignored.
*
* All frames ride ONE persistent uni stream (opened on first use): QUIC
* streams deliver in order, and stream-per-frame exhausts Firefox's ~100
@@ -116,29 +328,47 @@ export class ReliableChannel implements ChannelPolicy {
readonly delivery: Delivery = "reliable";
sent = 0;
dropped = 0;
+ aborted = 0; // reliable never resets frames; fixed 0
+ expired = 0;
+ bytesOut = 0;
+ readonly rate = new Rate();
#fifo: Uint8Array[] = [];
#bytes = 0;
#writing = false;
#writer: FrameWriter | null = null;
#disposed = false;
+ readonly #now: () => number;
- constructor(readonly sink: ViewerSink) {}
+ constructor(readonly sink: ViewerSink, opts: PolicyOptions = {}) {
+ this.#now = opts.now ?? Date.now;
+ }
queued(): number {
return this.#fifo.length;
}
+ inflight(): number {
+ return 0; // one persistent stream, no per-frame streams to reset
+ }
+
offer(bytes: Uint8Array): void {
if (this.#disposed) return;
this.#fifo.push(bytes);
this.#bytes += bytes.byteLength;
if (this.#fifo.length > RELIABLE_MAX_QUEUE || this.#bytes > RELIABLE_MAX_BYTES) {
this.sink.kick("reliable channel overflow");
+ // wt.closed teardown is async; until it runs, later offers must be
+ // no-ops, not re-queue + re-kick.
+ this.dispose();
return;
}
this.#drain();
}
+ reap(_nowMs: number): void {
+ // no-op: one persistent stream, no per-frame streams to reset
+ }
+
dispose(): void {
this.#disposed = true;
this.#fifo.length = 0;
@@ -167,6 +397,8 @@ export class ReliableChannel implements ChannelPolicy {
this.#bytes -= bytes.byteLength;
await writer.write(bytes);
this.sent++;
+ this.bytesOut += bytes.byteLength;
+ this.rate.push(bytes.byteLength, this.#now());
}
})()
.catch(() => {
diff --git a/web/relay/forward_test.ts b/web/relay/forward_test.ts
index a8d08656be..bdcab8735d 100644
--- a/web/relay/forward_test.ts
+++ b/web/relay/forward_test.ts
@@ -1,22 +1,91 @@
-import { assertEquals, assertRejects } from "@std/assert";
+import { assert, assertEquals, assertRejects } from "@std/assert";
import { encodeDataFrame, type FrameHeader } from "@dimos/shared";
import {
+ type FrameSend,
type FrameWriter,
LatestChannel,
parseRobotFrameHeader,
+ Rate,
readDataFrameBytes,
readWebTransportPreamble,
ReliableChannel,
type ViewerSink,
} from "./forward.ts";
+class FakeSend implements FrameSend {
+ aborted = false;
+ settled = false;
+ readonly done: Promise;
+ #resolve!: () => void;
+ #reject!: (e: Error) => void;
+
+ constructor(
+ public bytes: Uint8Array,
+ readonly sink: FakeSink,
+ /** Position in sink.sent[], captured at sendFrame time. */
+ readonly index: number,
+ /** Mirrors the real sink: false only while wedged in stream creation. */
+ public writeStarted: boolean,
+ ) {
+ this.done = new Promise((resolve, reject) => {
+ this.#resolve = resolve;
+ this.#reject = reject;
+ });
+ }
+
+ accept(): void {
+ if (!this.settled) {
+ this.writeStarted = true;
+ this.settled = true;
+ this.#resolve();
+ }
+ }
+
+ /** The stream got created; the (still unaccepted) write is in progress. */
+ beginWrite(): void {
+ this.writeStarted = true;
+ }
+
+ fail(e: Error): void {
+ if (!this.settled) {
+ this.settled = true;
+ this.#reject(e);
+ }
+ }
+
+ abort(): void {
+ if (this.aborted) return;
+ this.aborted = true;
+ this.sink.sendsAborted++;
+ this.fail(new Error("frame send aborted"));
+ }
+
+ supersede(bytes: Uint8Array): boolean {
+ if (this.writeStarted || this.aborted) return false;
+ this.bytes = bytes;
+ // sent[] holds the latest binding, so deep-equals assertions read what
+ // would actually go on the wire.
+ this.sink.sent[this.index] = bytes;
+ return true;
+ }
+}
+
class FakeSink implements ViewerSink {
+ /** Every frame handed to the sink, in order (latest path). */
sent: Uint8Array[] = [];
+ sends: FakeSend[] = [];
+ /** FrameSend aborts (latest streams reset), settled or not. */
+ sendsAborted = 0;
kicked: string | null = null;
+ kicks = 0;
streamsOpened = 0;
+ /** Persistent-stream (reliable) aborts. */
streamsAborted = 0;
auto: boolean;
manualOpen = false;
+ /** Model sends stuck in createUnidirectionalStream (credit exhausted):
+ * their write has not started, so they stay supersedable. */
+ wedgeCreate = false;
#waiters: { resolve: () => void; reject: (e: Error) => void }[] = [];
#openWaiters: (() => void)[] = [];
@@ -24,8 +93,12 @@ class FakeSink implements ViewerSink {
this.auto = auto;
}
- sendFrame(bytes: Uint8Array): Promise {
- return this.#write(bytes);
+ sendFrame(bytes: Uint8Array): FrameSend {
+ const send = new FakeSend(bytes, this, this.sent.length, !this.wedgeCreate);
+ this.sent.push(bytes);
+ this.sends.push(send);
+ if (this.auto) send.accept();
+ return send;
}
openStream(): Promise {
@@ -47,12 +120,19 @@ class FakeSink implements ViewerSink {
return new Promise((resolve, reject) => this.#waiters.push({ resolve, reject }));
}
+ /** Accept the oldest unsettled send (latest) or write (reliable). */
release(n = 1): void {
- while (n-- > 0) this.#waiters.shift()?.resolve();
+ while (n-- > 0) {
+ const pending = this.sends.find((s) => !s.settled);
+ if (pending !== undefined) pending.accept();
+ else this.#waiters.shift()?.resolve();
+ }
}
rejectWrite(): void {
- this.#waiters.shift()?.reject(new Error("stream aborted"));
+ const pending = this.sends.find((s) => !s.settled);
+ if (pending !== undefined) pending.fail(new Error("stream aborted"));
+ else this.#waiters.shift()?.reject(new Error("stream aborted"));
}
releaseOpen(): void {
@@ -60,6 +140,7 @@ class FakeSink implements ViewerSink {
}
kick(reason: string): void {
+ this.kicks++;
this.kicked = reason;
}
}
@@ -150,37 +231,42 @@ Deno.test("reliable: drain pauses reuse the same persistent stream", async () =>
assertEquals(sink.streamsOpened, 1);
});
-Deno.test("reliable: queue overflow kicks the viewer", async () => {
+Deno.test("reliable: overflow kicks once and empties the FIFO", async () => {
const sink = new FakeSink(false);
const ch = new ReliableChannel(sink);
- // 1 in flight + 64 queued is accepted; the next one overflows.
- for (let i = 0; i < 66 && sink.kicked === null; i++) ch.offer(frame(i));
+ // Nothing drains (the persistent stream is still opening), so the 65th
+ // queued frame overflows; the channel kicks once, self-disposes, and
+ // ignores the rest of the burst instead of re-queueing + re-kicking.
+ for (let i = 0; i < 200; i++) ch.offer(frame(i));
await tick();
+ assertEquals(sink.kicks, 1);
assertEquals(sink.kicked, "reliable channel overflow");
+ assertEquals(ch.queued(), 0);
+ assertEquals(sink.streamsAborted, 1);
+ ch.offer(frame(200)); // still a no-op until transport teardown completes
+ assertEquals(ch.queued(), 0);
+ assertEquals(sink.kicks, 1);
});
-Deno.test("latest: dispose during an in-flight write discards the pending frame", async () => {
+Deno.test("latest: dispose resets every outstanding send without kicking", async () => {
const sink = new FakeSink(false);
const ch = new LatestChannel(sink);
- ch.offer(frame(1)); // write in flight
- ch.offer(frame(2)); // parked in the pending slot
+ ch.offer(frame(1)); // send in flight
+ sink.release();
+ await tick();
+ ch.offer(frame(2)); // second send in flight (first still outstanding)
+ ch.offer(frame(3)); // parked in the pending slot
ch.dispose();
assertEquals(ch.queued(), 0);
- sink.release(); // the in-flight write completes after disposal
- await tick();
- ch.offer(frame(3)); // ignored after dispose
+ assertEquals(ch.inflight(), 0);
+ // Both the accepted stream and the in-flight one are reset (an accepted
+ // stream may still hold undelivered buffered data).
+ assertEquals(sink.sendsAborted, 2);
+ assertEquals([ch.aborted, ch.expired], [0, 0]); // disposal resets are not counted
await tick();
- assertEquals(sink.sent, [frame(1)]);
- assertEquals(sink.kicked, null);
-});
-
-Deno.test("latest: an in-flight write failing after dispose does not kick", async () => {
- const sink = new FakeSink(false);
- const ch = new LatestChannel(sink);
- ch.offer(frame(1));
- ch.dispose();
- sink.rejectWrite(); // the stream died with the disposal
+ ch.offer(frame(4)); // ignored after dispose
await tick();
+ assertEquals(sink.sent, [frame(1), frame(2)]);
assertEquals(sink.kicked, null);
});
@@ -252,6 +338,189 @@ Deno.test("reliable: a frame offered in the drain-completion window still sends"
}
});
+// ---------------------------------------------------------------------------
+// Reaping (the T1-review issue-12 fix): accepted-but-possibly-undelivered
+// latest streams are reset once stale, with counters that tell a suspended
+// viewer (aborted) apart from routine end-of-life resets (expired).
+
+function timedChannel(
+ sink: FakeSink,
+ start = 100_000,
+): { ch: LatestChannel; tick(ms: number): void; now(): number } {
+ let now = start;
+ const ch = new LatestChannel(sink, { now: () => now });
+ return { ch, tick: (ms: number) => now += ms, now: () => now };
+}
+
+Deno.test("latest: a stale accepted send is reset by the next offer (expired)", async () => {
+ const sink = new FakeSink(false);
+ const { ch, tick: advance } = timedChannel(sink);
+ ch.offer(frame(1));
+ sink.release(); // accepted; its stream stays open (never FIN'd)
+ await tick();
+ assertEquals(ch.inflight(), 1);
+
+ advance(600); // past LATEST_STALE_MS
+ ch.offer(frame(2));
+ assertEquals(sink.sendsAborted, 1); // frame 1's stream reset
+ assertEquals(sink.sends[0].aborted, true);
+ assertEquals([ch.aborted, ch.expired], [0, 1]); // no backpressure: routine expiry
+ sink.release();
+ await tick();
+ assertEquals(ch.sent, 2);
+ assertEquals(ch.bytesOut, 2);
+ assertEquals(ch.inflight(), 1); // only frame 2's stream remains
+ assertEquals(sink.kicked, null);
+});
+
+Deno.test("latest: reaps under backpressure count as aborted", async () => {
+ const sink = new FakeSink(false);
+ const { ch, tick: advance } = timedChannel(sink);
+ ch.offer(frame(1));
+ sink.release(); // frame 1 accepted at t0
+ await tick();
+ advance(400);
+ ch.offer(frame(2)); // frame 2 in flight, never accepted: the viewer wedged
+ await tick();
+
+ advance(200); // frame 1's stream is now 600 ms old, the carrier only 200
+ ch.offer(frame(3));
+ // The stale accepted stream is reset and counted aborted (the channel is
+ // backpressured); the still-young write-wedged carrier is left alone.
+ assertEquals([ch.aborted, ch.expired], [1, 0]);
+ assertEquals(sink.sendsAborted, 1);
+ assertEquals(sink.sends[1].aborted, false);
+});
+
+Deno.test("latest: offers supersede the create-wedged carrier; release delivers the newest", async () => {
+ // Resetting a create-wedged send would only spawn a replacement wedged on
+ // the same exhausted stream credit (resets do not replenish credit while
+ // the viewer's app is not reading), growing an unbounded zombie chain.
+ // Instead newer offers replace its payload in place.
+ const sink = new FakeSink(false);
+ sink.wedgeCreate = true;
+ const { ch, tick: advance } = timedChannel(sink);
+ ch.offer(frame(1)); // handed to the transport, wedged in stream creation
+ await tick();
+ advance(60_000);
+ ch.offer(frame(2)); // supersedes the carrier's payload
+ ch.offer(frame(3)); // supersedes again
+ assertEquals(sink.sends.length, 1); // no zombie chain even 60 s stale
+ assertEquals(sink.sendsAborted, 0);
+ assertEquals(ch.dropped, 2);
+ assertEquals(ch.queued(), 0);
+ assertEquals([ch.aborted, ch.expired], [0, 0]);
+
+ sink.release(); // the viewer resumes: the carrier is accepted
+ await tick();
+ assertEquals(sink.sent, [frame(3)]); // exactly the newest payload went out
+ assertEquals(ch.sent, 1);
+ assertEquals(ch.bytesOut, frame(3).byteLength); // 3 offers = 1 sent + 2 dropped
+});
+
+Deno.test("latest: a stale write-wedged carrier is reset and the newest resent", async () => {
+ // Once the write has begun the payload can no longer change, so a stale
+ // carrier is reset and the newest goes out on a fresh stream.
+ const sink = new FakeSink(false); // default: the write is already in progress
+ const { ch, tick: advance } = timedChannel(sink);
+ ch.offer(frame(1)); // write begins, wedged in flow control
+ await tick();
+ advance(600); // past LATEST_STALE_MS
+ ch.offer(frame(2)); // supersede refused -> the stale carrier is reset
+ assertEquals(sink.sends[0].aborted, true);
+ assertEquals(ch.aborted, 1);
+ assertEquals(ch.dropped, 1); // frame 1's payload never reached the wire
+ await tick(); // the drain swallows the abort and picks up the parked frame
+ assertEquals(ch.queued(), 0);
+ assertEquals(ch.inflight(), 1); // only the fresh carrier
+
+ sink.release();
+ await tick();
+ assertEquals(ch.sent, 1); // frame 1 was handed to the transport, never accepted
+ assertEquals(sink.sent, [frame(1), frame(2)]);
+});
+
+Deno.test("latest: supersede stops once the write begins", async () => {
+ const sink = new FakeSink(false);
+ sink.wedgeCreate = true;
+ const ch = new LatestChannel(sink);
+ ch.offer(frame(1)); // wedged in stream creation
+ ch.offer(frame(2)); // supersedes in place
+ assertEquals(ch.dropped, 1);
+ sink.sends[0].beginWrite(); // stream created; the write is now in progress
+ ch.offer(frame(3)); // the payload can no longer change: parks instead
+ assertEquals(ch.queued(), 1);
+ assertEquals(ch.dropped, 1); // parking in an empty slot drops nothing
+
+ sink.release();
+ await tick();
+ sink.release();
+ await tick();
+ assertEquals(sink.sent, [frame(2), frame(3)]);
+ assertEquals(ch.sent, 2);
+});
+
+Deno.test("latest: a healthy viewer expires streams but never aborts", async () => {
+ const sink = new FakeSink();
+ const { ch, tick: advance, now } = timedChannel(sink);
+ for (let i = 0; i < 5; i++) {
+ ch.offer(frame(i)); // auto sink: accepted immediately
+ await tick();
+ advance(600);
+ }
+ assertEquals(ch.sent, 5);
+ assertEquals(ch.aborted, 0);
+ assertEquals(ch.expired, 4); // every reap saw an accepted newest send
+ assertEquals(ch.inflight(), 1);
+ assertEquals(sink.kicked, null);
+
+ // The input idled; the relay's periodic reap (registry.reapAll) ends the
+ // last stream too, still as routine expiry.
+ ch.reap(now()); // the loop's final advance left the last acceptance 600 ms old
+ assertEquals(ch.expired, 5);
+ assertEquals(ch.inflight(), 0);
+ assertEquals(ch.aborted, 0);
+});
+
+Deno.test("latest: a real transport failure still kicks exactly once", async () => {
+ const sink = new FakeSink(false);
+ const ch = new LatestChannel(sink);
+ ch.offer(frame(1));
+ sink.release();
+ await tick();
+ ch.offer(frame(2));
+ sink.rejectWrite(); // connection died: not an abort()
+ await tick();
+ assertEquals(sink.kicked, "write failed");
+ // The kick's dispose sweeps both outstanding streams (harmless on the dead
+ // frame-2 stream; the real sink swallows abort errors).
+ assertEquals(sink.sendsAborted, 2);
+ ch.offer(frame(3)); // ignored after the dispose
+ await tick();
+ assertEquals(sink.sent, [frame(1), frame(2)]);
+});
+
+Deno.test("rate: bucketed trailing window with idle decay and wraparound", () => {
+ const rate = new Rate();
+ const t0 = 1_000_000;
+ // 10 frames of 1000 B over one second.
+ for (let i = 0; i < 10; i++) rate.push(1000, t0 + i * 100);
+ const busy = rate.snapshot(t0 + 1000);
+ assertEquals(busy.fps, 2); // 10 frames / 5 s window
+ assertEquals(busy.bps, 2000);
+
+ // Enough silence slides the burst's first bucket out but keeps the tail.
+ const later = rate.snapshot(t0 + 5200);
+ assert(later.fps > 0 && later.fps < busy.fps, `partial decay, got ${later.fps}`);
+
+ // Beyond the window everything zeroes.
+ assertEquals(rate.snapshot(t0 + 20_000), { fps: 0, bps: 0 });
+
+ // Wraparound: pushes far apart still land in the right buckets.
+ rate.push(500, t0 + 30_000);
+ assertEquals(rate.snapshot(t0 + 30_000), { fps: 0.2, bps: 100 });
+});
+
Deno.test("parseRobotFrameHeader accepts valid frames and rejects junk", () => {
const good = dataFrame("odom", 4, "reliable");
assertEquals(parseRobotFrameHeader(good), { ch: "odom", seq: 4, ts: 4.5, delivery: "reliable" });
diff --git a/web/relay/main.ts b/web/relay/main.ts
index 6f2e58adee..80f8e358ee 100644
--- a/web/relay/main.ts
+++ b/web/relay/main.ts
@@ -1,4 +1,4 @@
-// Relay CLI. Run from web/: deno task dev (or --port/--host/--static-dir).
+// Relay CLI. Run from web/: deno task dev (or --port/--host/--cockpit-dir).
// Prints a single JSON ready line on stdout for parent processes to parse;
// everything else logs to stderr-adjacent console lines prefixed [relay].
import { parseArgs } from "@std/cli";
@@ -6,7 +6,7 @@ import { PROTOCOL_VERSION } from "@dimos/shared";
import { startRelay } from "./server.ts";
const args = parseArgs(Deno.args, {
- string: ["host", "static-dir", "cockpit-dir"],
+ string: ["host", "cockpit-dir"],
default: { port: 7780, host: "127.0.0.1" },
});
@@ -24,7 +24,6 @@ if (host !== "127.0.0.1" && host !== "localhost") {
const relay = await startRelay({
port: Number(args.port),
host,
- staticDir: args["static-dir"],
cockpitDir: args["cockpit-dir"],
});
@@ -38,8 +37,9 @@ console.log(JSON.stringify({
const pageHost = host === "0.0.0.0" ? "127.0.0.1" : host;
if (args["cockpit-dir"] !== undefined) {
console.log(`[relay] cockpit: http://${pageHost}:${relay.httpPort}/`);
+} else {
+ console.log("[relay] no cockpit dist configured; serving /api only");
}
-console.log(`[relay] debug page: http://${pageHost}:${relay.httpPort}/debug.html`);
for (const signal of ["SIGINT", "SIGTERM"] as const) {
try {
diff --git a/web/relay/registry.ts b/web/relay/registry.ts
index 2b144d2d1c..24f8b04620 100644
--- a/web/relay/registry.ts
+++ b/web/relay/registry.ts
@@ -14,6 +14,7 @@ import {
type Delivery,
encodeDatagram,
type Msg,
+ type PanelSpec,
PROTOCOL_VERSION,
type RobotInfo,
} from "@dimos/shared";
@@ -22,6 +23,7 @@ import {
type ChannelPolicy,
LatestChannel,
parseRobotFrameHeader,
+ Rate,
ReliableChannel,
type ViewerSink,
} from "./forward.ts";
@@ -36,6 +38,7 @@ export interface RobotPeer {
/** Set by the session once a valid robot hello arrived. */
readonly info: RobotInfo | null;
readonly channels: ChannelSpec[];
+ readonly panels: PanelSpec[];
/** Close reason once closed; a closed session must never (re)register. */
readonly closed: string | null;
/** Control message upstream to the bridge (datagram: lossy, never blocks). */
@@ -67,6 +70,7 @@ interface ChannelInStats {
delivery: Delivery;
framesIn: number;
bytesIn: number;
+ rate: Rate;
}
interface RobotEntry {
@@ -77,7 +81,12 @@ interface RobotEntry {
n: number;
/** Last snapshot content, for change detection and periodic resend. */
lastChs: string[];
+ /** Per-channel ingress stats, declared channels only. */
channelsIn: Map;
+ /** One aggregate bucket for frames on undeclared channels: arbitrary
+ * header ch strings must not grow channelsIn. A manifest-less robot's
+ * traffic lands entirely here. */
+ undeclared: { framesIn: number; bytesIn: number; rate: Rate };
}
export class Registry {
@@ -117,6 +126,7 @@ export class Registry {
n: 0,
lastChs: [],
channelsIn: new Map(),
+ undeclared: { framesIn: 0, bytesIn: 0, rate: new Rate() },
});
this.#pushRobots();
// Forced: gives a fresh bridge its baseline and reattaches surviving
@@ -189,7 +199,12 @@ export class Registry {
viewer.watched = msg.robotId;
if (previous !== null && previous !== msg.robotId) this.#syncSubs(previous);
this.#syncSubs(msg.robotId);
- reply({ t: "manifest", robotId: msg.robotId, channels: entry.peer.channels });
+ reply({
+ t: "manifest",
+ robotId: msg.robotId,
+ channels: entry.peer.channels,
+ panels: entry.peer.panels,
+ });
break;
}
case "sub":
@@ -249,13 +264,27 @@ export class Registry {
return;
}
const ch = header.ch;
- const delivery = entry.delivery.get(ch) ?? header.delivery;
+ const declared = entry.delivery.get(ch);
+ // Manifest delivery wins; the header's is the undeclared-channel fallback.
+ const delivery = declared ?? header.delivery;
- const stats = entry.channelsIn.get(ch) ?? { delivery, framesIn: 0, bytesIn: 0 };
- stats.delivery = delivery;
- stats.framesIn++;
- stats.bytesIn += bytes.byteLength;
- entry.channelsIn.set(ch, stats);
+ if (declared === undefined) {
+ // Aggregate bucket only: arbitrary header ch strings must not grow the
+ // per-channel map. Routing below still forwards to subscribed viewers
+ // (a robot that declared no manifest accepts any sub).
+ entry.undeclared.framesIn++;
+ entry.undeclared.bytesIn += bytes.byteLength;
+ entry.undeclared.rate.push(bytes.byteLength, Date.now());
+ } else {
+ // delivery is fixed at creation: the manifest cannot change within a
+ // registration.
+ const stats = entry.channelsIn.get(ch) ??
+ { delivery, framesIn: 0, bytesIn: 0, rate: new Rate() };
+ stats.framesIn++;
+ stats.bytesIn += bytes.byteLength;
+ stats.rate.push(bytes.byteLength, Date.now());
+ entry.channelsIn.set(ch, stats);
+ }
for (const viewer of this.#viewers) {
if (viewer.watched !== id || !viewer.subs.has(ch)) continue;
@@ -282,6 +311,15 @@ export class Registry {
}
}
+ /** Reap stale accepted latest streams on every viewer. Offers reap
+ * opportunistically, but an idle input stops offering; server.ts drives
+ * this on an interval. Clock passed in so tests fabricate time. */
+ reapAll(nowMs: number): void {
+ for (const viewer of this.#viewers) {
+ for (const policy of viewer.policies.values()) policy.reap(nowMs);
+ }
+ }
+
robotsMsg(): Msg {
return { t: "robots", robots: this.#robotInfos() };
}
@@ -295,6 +333,7 @@ export class Registry {
}
stats(): unknown {
+ const now = Date.now();
return {
robots: this.#robotInfos(),
viewers: this.#viewers.size,
@@ -303,7 +342,19 @@ export class Registry {
perRobot: Object.fromEntries(
[...this.#robots].map(([id, e]) => [id, {
subs: e.lastChs,
- channels: Object.fromEntries(e.channelsIn),
+ channels: Object.fromEntries(
+ [...e.channelsIn].map(([ch, s]) => [ch, {
+ delivery: s.delivery,
+ framesIn: s.framesIn,
+ bytesIn: s.bytesIn,
+ ...s.rate.snapshot(now),
+ }]),
+ ),
+ undeclared: {
+ framesIn: e.undeclared.framesIn,
+ bytesIn: e.undeclared.bytesIn,
+ ...e.undeclared.rate.snapshot(now),
+ },
}]),
),
perViewer: [...this.#viewers].map((v) => ({
@@ -311,10 +362,17 @@ export class Registry {
watched: v.watched,
subs: [...v.subs].sort(),
channels: Object.fromEntries(
- [...v.policies].map(([ch, p]) => [
- ch,
- { sent: p.sent, dropped: p.dropped, queued: p.queued() },
- ]),
+ [...v.policies].map(([ch, p]) => [ch, {
+ delivery: p.delivery,
+ sent: p.sent,
+ dropped: p.dropped,
+ queued: p.queued(),
+ aborted: p.aborted,
+ expired: p.expired,
+ inflight: p.inflight(),
+ bytesOut: p.bytesOut,
+ ...p.rate.snapshot(now),
+ }]),
),
})),
};
diff --git a/web/relay/registry_test.ts b/web/relay/registry_test.ts
index a699c1c64f..3e01c17337 100644
--- a/web/relay/registry_test.ts
+++ b/web/relay/registry_test.ts
@@ -6,14 +6,18 @@ import {
type ChannelSpec,
encodeDataFrame,
type FrameHeader,
+ type ManifestMsg,
type Msg,
+ type PanelSpec,
PROTOCOL_VERSION,
type RobotInfo,
type SubsMsg,
} from "@dimos/shared";
import {
type ChannelPolicy,
+ type FrameSend,
type FrameWriter,
+ LATEST_STALE_MS,
LatestChannel,
ReliableChannel,
type ViewerSink,
@@ -28,16 +32,40 @@ class FakeSink implements ViewerSink {
auto = true;
#waiters: (() => void)[] = [];
- sendFrame(bytes: Uint8Array): Promise {
+ sendFrame(bytes: Uint8Array): FrameSend {
this.sent.push(bytes);
- if (this.auto) return Promise.resolve();
- return new Promise((resolve) => this.#waiters.push(resolve));
+ let settle!: () => void;
+ let fail!: (e: Error) => void;
+ const done = new Promise((resolve, reject) => {
+ settle = resolve;
+ fail = reject;
+ });
+ if (this.auto) settle();
+ else this.#waiters.push(settle);
+ let aborted = false;
+ return {
+ done,
+ get aborted() {
+ return aborted;
+ },
+ abort() {
+ if (aborted) return;
+ aborted = true;
+ fail(new Error("frame send aborted"));
+ },
+ // Registry tests exercise the parking path, not the create wedge.
+ supersede: () => false,
+ };
}
openStream(): Promise {
this.streamsOpened++;
return Promise.resolve({
- write: (bytes: Uint8Array) => this.sendFrame(bytes),
+ write: (bytes: Uint8Array) => {
+ this.sent.push(bytes);
+ if (this.auto) return Promise.resolve();
+ return new Promise((resolve) => this.#waiters.push(resolve));
+ },
abort: () => {
this.streamsAborted++;
return Promise.resolve();
@@ -57,12 +85,14 @@ class FakeSink implements ViewerSink {
class FakeRobot implements RobotPeer {
info: RobotInfo | null;
channels: ChannelSpec[];
+ panels: PanelSpec[];
msgs: Msg[] = [];
closed: string | null = null;
- constructor(id: string, channels: ChannelSpec[] = []) {
+ constructor(id: string, channels: ChannelSpec[] = [], panels: PanelSpec[] = []) {
this.info = { id, name: id, model: "test" };
this.channels = channels;
+ this.panels = panels;
}
sendMsg(msg: Msg): void {
@@ -172,14 +202,16 @@ Deno.test("watch switch moves subscriptions between robots", () => {
Deno.test("re-watching the same robot keeps subscriptions", () => {
const reg = new Registry();
- const robot = new FakeRobot("r1", SPECS);
+ const panels: PanelSpec[] = [{ id: "color_image", kind: "video", channels: ["color_image"] }];
+ const robot = new FakeRobot("r1", SPECS, panels);
reg.registerRobot(robot);
const viewer = attach(reg, "r1", ["odom"]);
send(reg, viewer, { t: "watch", robotId: "r1" });
assertEquals(viewer.subs, new Set(["odom"]));
- const manifests = viewer.replies.filter((m) => m.t === "manifest");
+ const manifests = viewer.replies.filter((m): m is ManifestMsg => m.t === "manifest");
assertEquals(manifests.length, 2);
- assertEquals((manifests[1] as { channels: ChannelSpec[] }).channels, SPECS);
+ assertEquals(manifests[1].channels, SPECS);
+ assertEquals(manifests[1].panels, panels); // the manifest reply carries the panels
});
Deno.test("duplicate live robot id is rejected; reconnect works after close", () => {
@@ -462,3 +494,95 @@ Deno.test("resendSnapshots repeats the last set with a fresh n", () => {
reg.resendSnapshots();
assertEquals(robot.lastSubs(), { t: "subs", chs: last.chs, n: last.n + 1 });
});
+
+Deno.test("reapAll resets stale accepted latest streams", async () => {
+ // What server.ts drives on an interval: an idle input stops offering, so
+ // without this the last accepted stream would stay open indefinitely.
+ const reg = new Registry();
+ const robot = new FakeRobot("r1", SPECS);
+ reg.registerRobot(robot);
+ const viewer = attach(reg, "r1", ["color_image"]);
+ reg.onRobotFrame(robot, frame("color_image", 1, "latest"));
+ await tick();
+ const policy = viewer.policies.get("color_image")!;
+ assertEquals(policy.inflight(), 1);
+ // Entries carry real Date.now timestamps; fabricate a clock past staleMs.
+ reg.reapAll(Date.now() + LATEST_STALE_MS + 100);
+ assertEquals(policy.inflight(), 0);
+ assertEquals(policy.expired, 1);
+ assertEquals(policy.aborted, 0);
+});
+
+Deno.test("stats project exact per-channel key sets with counters and rates", async () => {
+ const reg = new Registry();
+ const robot = new FakeRobot("r1", SPECS);
+ reg.registerRobot(robot);
+ const viewer = attach(reg, "r1", ["color_image", "odom"]);
+ reg.onRobotFrame(robot, frame("color_image", 1, "latest"));
+ reg.onRobotFrame(robot, frame("odom", 1, "reliable"));
+ await tick();
+
+ const stats = reg.stats() as {
+ perRobot: Record>;
+ undeclared: Record;
+ }>;
+ perViewer: { channels: Record> }[];
+ };
+ assertEquals(Object.keys(stats.perRobot.r1).sort(), ["channels", "subs", "undeclared"]);
+ const inKeys = Object.keys(stats.perRobot.r1.channels.color_image).sort();
+ assertEquals(inKeys, ["bps", "bytesIn", "delivery", "fps", "framesIn"]);
+ // Declared-only traffic leaves the aggregate bucket zeroed.
+ assertEquals(stats.perRobot.r1.undeclared, { framesIn: 0, bytesIn: 0, fps: 0, bps: 0 });
+ const out = stats.perViewer[0].channels;
+ const outKeys = Object.keys(out.color_image).sort();
+ assertEquals(outKeys, [
+ "aborted",
+ "bps",
+ "bytesOut",
+ "delivery",
+ "dropped",
+ "expired",
+ "fps",
+ "inflight",
+ "queued",
+ "sent",
+ ]);
+ assertEquals(out.color_image.delivery, "latest");
+ assertEquals(out.color_image.sent, 1);
+ assertEquals(out.color_image.inflight, 1); // its stream stays open until reaped
+ assertEquals(out.odom.delivery, "reliable");
+ assertEquals(out.odom.inflight, 0); // one persistent stream, nothing to reset
+ assert((out.odom.bytesOut as number) > 0);
+ assertEquals(viewer.sink.kicked, null);
+});
+
+Deno.test("undeclared channel frames aggregate into one per-robot bucket", async () => {
+ const reg = new Registry();
+ const declared = new FakeRobot("r1", SPECS);
+ reg.registerRobot(declared);
+ // Novel header ch strings on a declared robot must not grow channels.
+ reg.onRobotFrame(declared, frame("novel_a", 1, "latest"));
+ reg.onRobotFrame(declared, frame("novel_b", 2, "latest"));
+ let stats = reg.stats() as {
+ perRobot: Record;
+ undeclared: { framesIn: number; bytesIn: number };
+ }>;
+ };
+ assertEquals(stats.perRobot.r1.channels, {});
+ assertEquals(stats.perRobot.r1.undeclared.framesIn, 2);
+ assert(stats.perRobot.r1.undeclared.bytesIn > 0);
+
+ // A manifest-less robot still forwards to subscribed viewers while all its
+ // traffic lands in the aggregate.
+ const bare = new FakeRobot("r2", []);
+ reg.registerRobot(bare);
+ const viewer = attach(reg, "r2", ["tele"]);
+ reg.onRobotFrame(bare, frame("tele", 1, "latest"));
+ await tick();
+ assertEquals(viewer.sink.sent.length, 1);
+ stats = reg.stats() as typeof stats;
+ assertEquals(stats.perRobot.r2.channels, {});
+ assertEquals(stats.perRobot.r2.undeclared.framesIn, 1);
+});
diff --git a/web/relay/server.ts b/web/relay/server.ts
index b90ec6538d..fe3c307885 100644
--- a/web/relay/server.ts
+++ b/web/relay/server.ts
@@ -6,6 +6,7 @@
import { PROTOCOL_VERSION } from "@dimos/shared";
import { fileURLToPath, pathToFileURL } from "node:url";
import { makeEphemeralCert } from "./cert.ts";
+import { LATEST_STALE_MS } from "./forward.ts";
import { Registry } from "./registry.ts";
import { RobotSession, ViewerSession } from "./session.ts";
@@ -18,12 +19,9 @@ export interface RelayOptions {
port?: number;
/** Bind host for both listeners. The default is the only secure-context-friendly choice. */
host?: string;
- /** Directory served over HTTP. Defaults to ./static next to this module. */
- staticDir?: string;
/**
- * Built Cockpit app (web/cockpit/dist). When set, / serves its index.html
- * and files resolve here first, with staticDir as the fallback (so
- * /debug.html keeps working). Without it, / serves the debug page.
+ * Built Cockpit app (web/cockpit/dist): / serves its index.html. Without it
+ * the relay has no UI (/ answers 404 with a build hint); only /api/* works.
*/
cockpitDir?: string;
}
@@ -117,14 +115,9 @@ export async function startRelay(options: RelayOptions = {}): Promise registry.resendSnapshots(), SNAPSHOT_RESEND_MS);
// A pending resend must not keep the Deno process alive after shutdown().
Deno.unrefTimer(resendTimer);
+ // Offers reap stale latest streams opportunistically, but an idle input
+ // stops offering; this interval bounds an idle stream's lifetime to just
+ // under 2x the stale window.
+ const reapTimer = setInterval(() => registry.reapAll(Date.now()), LATEST_STALE_MS);
+ Deno.unrefTimer(reapTimer);
(async () => {
for await (const incoming of listener) {
@@ -189,14 +187,15 @@ export async function startRelay(options: RelayOptions = {}): Promise {
clearInterval(resendTimer);
+ clearInterval(reapTimer);
for (const wt of sessions) {
try {
wt.close({ closeCode: 0, reason: "relay shutdown" });
diff --git a/web/relay/server_test.ts b/web/relay/server_test.ts
index d9c1eebbf4..9d77d8af21 100644
--- a/web/relay/server_test.ts
+++ b/web/relay/server_test.ts
@@ -14,6 +14,7 @@ import {
encodeDatagram,
type FrameHeader,
type Msg,
+ type PanelSpec,
PROTOCOL_VERSION,
type RobotInfo,
} from "@dimos/shared";
@@ -24,6 +25,7 @@ const CHANNELS: ChannelSpec[] = [
{ ch: "color_image", encoding: "jpeg.v1", delivery: "latest", maxHz: 15.5 },
{ ch: "odom", encoding: "pose.json.v1", delivery: "reliable", maxHz: 20.5 },
];
+const PANELS: PanelSpec[] = [{ id: "color_image", kind: "video", channels: ["color_image"] }];
function certOpts(hashB64: string): WebTransportOptions {
return {
@@ -140,7 +142,7 @@ Deno.test({
const relay = await startRelay({ port: 0 });
const httpBase = `http://127.0.0.1:${relay.httpPort}`;
- await t.step("/api/info matches the handle and the debug page serves", async () => {
+ await t.step("/api/info matches the handle; no cockpit dist -> 404 with a hint", async () => {
const info = await (await fetch(`${httpBase}/api/info`)).json();
assertEquals(info, {
wtUrl: `${relay.wtUrl}/viewer`,
@@ -148,25 +150,14 @@ Deno.test({
v: PROTOCOL_VERSION,
});
assert(relay.wtUrl.startsWith("https://127.0.0.1:"), relay.wtUrl);
- const page = await (await fetch(`${httpBase}/debug.html`)).text();
- assert(page.includes("DimOS relay debug"));
- const index = await (await fetch(`${httpBase}/`)).text();
- assert(index.includes("DimOS relay debug"));
- // Traversal probes. The client/URL parser normalizes these two away from
- // the tree before the guard sees them, so they 404 on absence.
- for (const path of ["/../etc/passwd", "/%2e%2e/etc/passwd"]) {
- const res = await fetch(`${httpBase}${path}`);
- await res.body?.cancel();
- assertEquals(res.status, 404, path);
- }
- // These survive normalization and must be rejected by the containment
- // check: a leading "//" makes new URL() jump to the filesystem root, and
- // encoded slashes let a "../" escape reassemble after decoding.
- for (const path of ["//etc/passwd", "/..%2f..%2f..%2f..%2fetc%2fpasswd"]) {
- const res = await fetch(`${httpBase}${path}`);
- await res.body?.cancel();
- assertEquals(res.status, 400, path);
- }
+ // Without a cockpit dist the relay serves no files at all: / explains how
+ // to get one (traversal guards are covered by the cockpit-dist test).
+ const index = await fetch(`${httpBase}/`);
+ assertEquals(index.status, 404);
+ assert((await index.text()).includes("cockpit dist not built"));
+ const stray = await fetch(`${httpBase}/debug.html`);
+ await stray.body?.cancel();
+ assertEquals(stray.status, 404);
});
const viewer = new WebTransport(`${relay.wtUrl}/viewer`, certOpts(relay.certHash));
@@ -213,7 +204,7 @@ Deno.test({
v: PROTOCOL_VERSION,
role: "robot",
robot: ROBOT,
- manifest: { channels: CHANNELS },
+ manifest: { channels: CHANNELS, panels: PANELS },
}),
);
// Registration and welcome are separate datagrams, so their relative
@@ -246,6 +237,7 @@ Deno.test({
t: "manifest",
robotId: ROBOT.id,
channels: CHANNELS,
+ panels: PANELS,
});
await controlWriter.write(encodeControlFrame({ t: "sub", ch: "odom" }));
await controlWriter.write(encodeControlFrame({ t: "sub", ch: "color_image" }));
@@ -352,6 +344,15 @@ Deno.test({
);
assertEquals(viewerStats.subs, ["odom"]);
assertEquals(viewerStats.channels.odom.sent, 3);
+ // Reset counters and rates (T5): a healthy reliable channel never resets.
+ assertEquals(viewerStats.channels.odom.delivery, "reliable");
+ assertEquals(viewerStats.channels.odom.aborted, 0);
+ assertEquals(viewerStats.channels.odom.expired, 0);
+ assertEquals(viewerStats.channels.odom.inflight, 0);
+ assert(viewerStats.channels.odom.bytesOut > 0);
+ assertEquals(typeof viewerStats.channels.odom.fps, "number");
+ assert(stats.perRobot[ROBOT.id].channels.odom.bytesIn > 0);
+ assertEquals(typeof stats.perRobot[ROBOT.id].channels.odom.bps, "number");
});
await t.step("robot hello without robot{} -> missing_robot_id + close", async () => {
@@ -471,6 +472,124 @@ Deno.test({
await relay.shutdown();
});
+// The T5 demo criterion as a test: one suspended viewer must cost only
+// itself. Its stale streams are reset ("aborted, not queued") while the
+// healthy viewer keeps receiving fresh frames at full rate. This is also the
+// permanent proof that reaping keeps working against real quinn streams.
+Deno.test({
+ name: "a viewer that stops reading is reset, not queued; others keep full rate",
+ sanitizeOps: false,
+ sanitizeResources: false,
+}, async () => {
+ const relay = await startRelay({ port: 0 });
+ const httpBase = `http://127.0.0.1:${relay.httpPort}`;
+ const robot = new WebTransport(`${relay.wtUrl}/robot`, certOpts(relay.certHash));
+ await within(robot.ready, "robot connect");
+ const robotDatagrams = datagramQueue(robot.datagrams.readable);
+ const robotDgWriter = robot.datagrams.writable.getWriter();
+ await robotDgWriter.write(
+ encodeDatagram({
+ t: "hello",
+ v: PROTOCOL_VERSION,
+ role: "robot",
+ robot: ROBOT,
+ manifest: {
+ channels: [{ ch: "color_image", encoding: "jpeg.v1", delivery: "latest", maxHz: 100 }],
+ },
+ }),
+ );
+ await within(robotDatagrams(), "robot hello reply");
+
+ async function attachViewer(name: string): Promise {
+ const wt = new WebTransport(`${relay.wtUrl}/viewer`, certOpts(relay.certHash));
+ await within(wt.ready, `${name} connect`);
+ const control = await wt.createBidirectionalStream();
+ const writer = control.writable.getWriter();
+ const next = controlQueue(control.readable);
+ await writer.write(encodeControlFrame({ t: "hello", v: PROTOCOL_VERSION, role: "viewer" }));
+ await writer.write(encodeControlFrame({ t: "watch", robotId: ROBOT.id }));
+ await writer.write(encodeControlFrame({ t: "sub", ch: "color_image" }));
+ let msg: Msg;
+ do {
+ msg = await within(next(), `${name} manifest`);
+ } while (msg.t !== "manifest");
+ return wt;
+ }
+
+ const healthy = await attachViewer("healthy");
+ const healthyFrames = frameQueue(healthy);
+ const stalled = await attachViewer("stalled");
+ void stalled; // never reads incomingUnidirectionalStreams: a suspended tab
+
+ const payload = new Uint8Array(8 * 1024).fill(9);
+ const fetchStalled = async () => {
+ const stats = await (await fetch(`${httpBase}/api/stats`)).json();
+ const viewers = stats.perViewer as {
+ channels: Record;
+ }[];
+ // The stalled viewer is the one whose acceptance count froze.
+ return viewers
+ .map((v) => v.channels.color_image)
+ .filter((c) => c !== undefined)
+ .sort((a, b) => a.sent - b.sent)[0];
+ };
+
+ // Pump frames until the stalled viewer's stale streams are being reset
+ // under backpressure. Onset needs its uni-stream credit (~100) exhausted
+ // plus one LATEST_STALE_MS window, so give it a generous frame budget.
+ let lastSeq = 0;
+ let stalledStats = await fetchStalled();
+ for (let seq = 1; seq <= 1200; seq++) {
+ await sendRobotFrame(
+ robot,
+ { ch: "color_image", seq, ts: seq / 100, delivery: "latest" },
+ payload,
+ );
+ lastSeq = seq;
+ if (seq % 50 === 0) {
+ stalledStats = await fetchStalled();
+ if (stalledStats !== undefined && stalledStats.aborted >= 3) break;
+ }
+ await new Promise((resolve) => setTimeout(resolve, 5));
+ }
+ assert(stalledStats !== undefined, "stalled viewer never got a policy");
+ assert(stalledStats.aborted >= 3, `expected aborted resets, got ${stalledStats.aborted}`);
+ assert(stalledStats.queued <= 1, `latest queue must stay 0|1, got ${stalledStats.queued}`);
+
+ // The healthy viewer kept receiving fresh frames the whole time: drain its
+ // queue and require a seq from the era after the stalled viewer wedged.
+ let newest = 0;
+ const drainUntil = Date.now() + 8000;
+ while (newest < lastSeq - 50 && Date.now() < drainUntil) {
+ newest = Math.max(newest, (await within(healthyFrames(), "healthy frame")).header.seq);
+ }
+ assert(newest >= lastSeq - 50, `healthy viewer stalled at seq ${newest} of ${lastSeq}`);
+
+ // The input has quiesced, so no newer offer will reap the healthy viewer's
+ // last accepted stream: only the relay's periodic reap timer can return it
+ // (against real quinn streams).
+ const reapDeadline = Date.now() + 5000;
+ let healthyInflight = -1;
+ while (Date.now() < reapDeadline) {
+ const stats = await (await fetch(`${httpBase}/api/stats`)).json();
+ const channels = (stats.perViewer as {
+ channels: Record;
+ }[])
+ .map((v) => v.channels.color_image)
+ .filter((c) => c !== undefined)
+ .sort((a, b) => b.sent - a.sent); // the healthy viewer accepted the most
+ healthyInflight = channels[0]?.inflight ?? -1;
+ if (healthyInflight === 0) break;
+ await new Promise((resolve) => setTimeout(resolve, 200));
+ }
+ assertEquals(healthyInflight, 0, "idle reap never reset the healthy viewer's last stream");
+
+ healthy.close();
+ stalled.close();
+ robot.close();
+ await relay.shutdown();
+});
+
Deno.test({
name: "relay serves the cockpit dist when configured",
sanitizeOps: false,
@@ -486,13 +605,21 @@ Deno.test({
assertEquals(asset.status, 200);
assertEquals(asset.headers.get("content-type"), "application/javascript");
await asset.body?.cancel();
- // The debug page still resolves from relay/static/ behind the cockpit.
- const page = await (await fetch(`${httpBase}/debug.html`)).text();
- assert(page.includes("DimOS relay debug"));
- // The traversal guard covers the cockpit root too.
- const res = await fetch(`${httpBase}//etc/passwd`);
- await res.body?.cancel();
- assertEquals(res.status, 400);
+ // Traversal probes. The client/URL parser normalizes these two away from
+ // the tree before the guard sees them, so they 404 on absence.
+ for (const path of ["/../etc/passwd", "/%2e%2e/etc/passwd"]) {
+ const probe = await fetch(`${httpBase}${path}`);
+ await probe.body?.cancel();
+ assertEquals(probe.status, 404, path);
+ }
+ // These survive normalization and must be rejected by the containment
+ // check: a leading "//" makes new URL() jump to the filesystem root, and
+ // encoded slashes let a "../" escape reassemble after decoding.
+ for (const path of ["//etc/passwd", "/..%2f..%2f..%2f..%2fetc%2fpasswd"]) {
+ const probe = await fetch(`${httpBase}${path}`);
+ await probe.body?.cancel();
+ assertEquals(probe.status, 400, path);
+ }
// A symlink whose target lies outside the root must not be followed
// (readFile follows symlinks; the containment check compares realpaths).
const escape = await fetch(`${httpBase}/escape.txt`);
@@ -511,9 +638,9 @@ Deno.test("startRelay rejects a bad served dir with a labeled error", async () =
// offending option, and a file (realpath-able, would 404 everything) is
// rejected too.
await assertRejects(
- () => startRelay({ staticDir: "/no/such/dir" }),
+ () => startRelay({ cockpitDir: "/no/such/dir" }),
Error,
- "staticDir does not exist: /no/such/dir",
+ "cockpitDir does not exist: /no/such/dir",
);
await assertRejects(
() => startRelay({ cockpitDir: fileURLToPath(import.meta.url) }),
diff --git a/web/relay/session.ts b/web/relay/session.ts
index 37156dbf17..6fff1752cc 100644
--- a/web/relay/session.ts
+++ b/web/relay/session.ts
@@ -16,12 +16,14 @@ import {
encodeControlFrame,
encodeDatagram,
type Msg,
+ type PanelSpec,
PROTOCOL_VERSION,
type RobotInfo,
} from "@dimos/shared";
import { parseManifest } from "@dimos/shared/manifest";
import {
type ChannelPolicy,
+ type FrameSend,
type FrameWriter,
readDataFrameBytes,
readWebTransportPreamble,
@@ -38,6 +40,11 @@ import type { Registry, RobotPeer, ViewerPeer } from "./registry.ts";
export const CONTROL_SEND_ORDER = 2;
export const RELIABLE_SEND_ORDER = 1;
+// Application error code carried by latest-stream resets (mirrors the Python
+// robot leg's STALE_STREAM_ERROR_CODE). Receivers do not act on it; it only
+// labels the reset for debugging.
+export const STALE_STREAM_ERROR_CODE = 0x01;
+
function closeAfterFlush(wt: WebTransport, reason: string): void {
// Session close discards queued stream/datagram data, so give a just-sent
// reply (e.g. the version_mismatch error) a moment to reach the wire.
@@ -53,6 +60,7 @@ function closeAfterFlush(wt: WebTransport, reason: string): void {
export class RobotSession implements RobotPeer {
info: RobotInfo | null = null;
channels: ChannelSpec[] = [];
+ panels: PanelSpec[] = [];
/** Close reason; set before transport close so rejected hello resends
* cannot register this session. */
closed: string | null = null;
@@ -137,6 +145,7 @@ export class RobotSession implements RobotPeer {
if (this.info === null) {
this.info = msg.robot;
this.channels = msg.manifest?.channels ?? [];
+ this.panels = msg.manifest?.panels ?? [];
}
if (!this.#registry.registerRobot(this)) {
return this.#reject(
@@ -187,14 +196,65 @@ export class ViewerSession implements ViewerPeer {
this.#registry = registry;
let latestOrder = 1;
this.sink = {
- async sendFrame(bytes: Uint8Array): Promise {
- const stream = await wt.createUnidirectionalStream({
- waitUntilAvailable: true,
- sendOrder: -(latestOrder++),
+ sendFrame(bytes: Uint8Array): FrameSend {
+ let aborted = false;
+ let writeStarted = false;
+ let writer: WritableStreamDefaultWriter | null = null;
+ let settle!: () => void;
+ let fail!: (e: unknown) => void;
+ const done = new Promise((resolve, reject) => {
+ settle = resolve;
+ fail = reject;
});
- const writer = stream.getWriter();
- await writer.write(bytes);
- await writer.close();
+ const reset = () => {
+ writer?.abort(
+ new WebTransportError("stale frame superseded", {
+ source: "stream",
+ streamErrorCode: STALE_STREAM_ERROR_CODE,
+ }),
+ ).catch(() => {});
+ };
+ (async () => {
+ const stream = await wt.createUnidirectionalStream({
+ waitUntilAvailable: true,
+ sendOrder: -(latestOrder++),
+ });
+ // Keep the writer for the stream's whole life: aborts go through it
+ // (the stream itself stays locked).
+ writer = stream.getWriter();
+ if (aborted) {
+ // abort() raced the create (done already rejected); release the
+ // just-granted stream credit.
+ reset();
+ return;
+ }
+ // Latching writeStarted and starting the write in one synchronous
+ // step is what makes supersede race-free: the payload can no
+ // longer change once bytes may have reached the transport.
+ writeStarted = true;
+ await writer.write(bytes);
+ settle();
+ // No close(): a closed stream cannot be aborted, and every latest
+ // stream ends in a reset (reap, dispose, or session teardown).
+ // Receivers dispatch on byte count and treat the reset as EOF.
+ })().catch((e) => fail(e)); // no-op if done already settled
+ return {
+ done,
+ get aborted() {
+ return aborted;
+ },
+ abort() {
+ if (aborted) return;
+ aborted = true;
+ fail(new Error("frame send aborted"));
+ reset();
+ },
+ supersede(newBytes: Uint8Array): boolean {
+ if (writeStarted || aborted) return false;
+ bytes = newBytes; // the write reads `bytes` only after create resolves
+ return true;
+ },
+ };
},
async openStream(): Promise {
// Persistent stream for a reliable channel.
diff --git a/web/relay/session_test.ts b/web/relay/session_test.ts
index 62f1ec7e4c..2ef66a8720 100644
--- a/web/relay/session_test.ts
+++ b/web/relay/session_test.ts
@@ -2,10 +2,15 @@
// scheme (control > reliable telemetry > latest video) must hold for every
// stream the relay creates or replies on. Wire-level scheduling itself is
// quinn's job; here we pin the orders the relay assigns.
-import { assert, assertEquals } from "@std/assert";
+import { assert, assertEquals, assertRejects } from "@std/assert";
import { ControlFrameReader, encodeControlFrame, type Msg, PROTOCOL_VERSION } from "@dimos/shared";
import { Registry } from "./registry.ts";
-import { CONTROL_SEND_ORDER, RELIABLE_SEND_ORDER, ViewerSession } from "./session.ts";
+import {
+ CONTROL_SEND_ORDER,
+ RELIABLE_SEND_ORDER,
+ STALE_STREAM_ERROR_CODE,
+ ViewerSession,
+} from "./session.ts";
function tick(): Promise {
return new Promise((resolve) => setTimeout(resolve, 0));
@@ -25,13 +30,142 @@ Deno.test("the sink assigns the documented sendOrder to each stream kind", async
},
} as unknown as WebTransport;
const session = new ViewerSession(wt, 1, new Registry());
- await session.sink.sendFrame(new Uint8Array([1]));
- await session.sink.sendFrame(new Uint8Array([2]));
+ await session.sink.sendFrame(new Uint8Array([1])).done;
+ await session.sink.sendFrame(new Uint8Array([2])).done;
await session.sink.openStream();
// Latest streams count down (oldest first, all below reliable).
assertEquals(orders, [-1, -2, RELIABLE_SEND_ORDER]);
});
+// ---------------------------------------------------------------------------
+// FrameSend contract of the real sink: latest streams are never closed (a
+// closed stream cannot be aborted) and abort() works in every phase - before
+// the stream exists, during a stalled write, and after acceptance.
+
+interface SpyStream {
+ stream: WritableStream;
+ writes: Uint8Array[];
+ aborts: unknown[];
+ closed: boolean;
+}
+
+function spyStream(opts: { stallWrite?: boolean } = {}): SpyStream {
+ const spy: SpyStream = { writes: [], aborts: [], closed: false, stream: null! };
+ spy.stream = new WritableStream({
+ write(chunk, controller) {
+ spy.writes.push(chunk);
+ if (!opts.stallWrite) return Promise.resolve();
+ // A stalled transport write must still honor an abort: the spec only
+ // invokes the underlying abort() after the pending write settles, and
+ // controller.signal is how the transport learns to give up.
+ return new Promise((_, reject) => {
+ controller.signal.addEventListener("abort", () => reject(controller.signal.reason));
+ });
+ },
+ abort(reason) {
+ spy.aborts.push(reason);
+ },
+ close() {
+ spy.closed = true;
+ },
+ });
+ return spy;
+}
+
+function sinkOver(create: () => Promise>) {
+ const wt = { createUnidirectionalStream: create } as unknown as WebTransport;
+ return new ViewerSession(wt, 1, new Registry()).sink;
+}
+
+Deno.test("sink: acceptance never closes the stream; abort after acceptance resets it", async () => {
+ const spy = spyStream();
+ const sink = sinkOver(() => Promise.resolve(spy.stream));
+ const send = sink.sendFrame(new Uint8Array([7]));
+ await send.done;
+ assertEquals(spy.writes.length, 1);
+ assertEquals(spy.closed, false); // never FIN'd: the stream must stay abortable
+
+ send.abort();
+ send.abort(); // idempotent
+ await tick();
+ assertEquals(spy.aborts.length, 1);
+ const reason = spy.aborts[0] as WebTransportError;
+ assertEquals(reason.streamErrorCode, STALE_STREAM_ERROR_CODE);
+ assertEquals(send.aborted, true);
+});
+
+Deno.test("sink: abort during a pending create rejects done and resets on arrival", async () => {
+ const spy = spyStream();
+ let releaseCreate!: () => void;
+ const sink = sinkOver(() =>
+ new Promise((resolve) => {
+ releaseCreate = () => resolve(spy.stream);
+ })
+ );
+ const send = sink.sendFrame(new Uint8Array([7]));
+ send.abort();
+ await assertRejects(() => send.done); // rejects while the create still hangs
+
+ releaseCreate(); // credit finally granted: the zombie stream must be reset
+ await tick();
+ assertEquals(spy.aborts.length, 1);
+ assertEquals(spy.writes.length, 0); // the stale frame was never written
+});
+
+Deno.test("sink: abort during a stalled write rejects done and resets the stream", async () => {
+ const spy = spyStream({ stallWrite: true });
+ const sink = sinkOver(() => Promise.resolve(spy.stream));
+ const send = sink.sendFrame(new Uint8Array([7]));
+ await tick();
+ assertEquals(spy.writes.length, 1); // write started, wedged in flow control
+
+ send.abort();
+ await assertRejects(() => send.done);
+ await tick();
+ assertEquals(spy.aborts.length, 1);
+});
+
+Deno.test("sink: a create failure rejects done without the aborted flag", async () => {
+ const sink = sinkOver(() => Promise.reject(new Error("connection lost")));
+ const send = sink.sendFrame(new Uint8Array([7]));
+ await assertRejects(() => send.done, Error, "connection lost");
+ assertEquals(send.aborted, false); // policies must treat this as a real failure
+});
+
+Deno.test("sink: supersede before the create resolves late-binds the payload", async () => {
+ const spy = spyStream();
+ let releaseCreate!: () => void;
+ const sink = sinkOver(() =>
+ new Promise((resolve) => {
+ releaseCreate = () => resolve(spy.stream);
+ })
+ );
+ const send = sink.sendFrame(new Uint8Array([1]));
+ assertEquals(send.supersede(new Uint8Array([2])), true);
+ assertEquals(send.supersede(new Uint8Array([3])), true); // repeatable while wedged
+
+ releaseCreate(); // credit finally granted: the write picks up the latest binding
+ await send.done;
+ assertEquals(spy.writes, [new Uint8Array([3])]);
+});
+
+Deno.test("sink: supersede is refused once the write starts and after abort", async () => {
+ const stalled = spyStream({ stallWrite: true });
+ const stalledSink = sinkOver(() => Promise.resolve(stalled.stream));
+ const inWrite = stalledSink.sendFrame(new Uint8Array([1]));
+ await tick();
+ assertEquals(stalled.writes.length, 1); // write started, wedged in flow control
+ assertEquals(inWrite.supersede(new Uint8Array([2])), false);
+ inWrite.abort(); // release the stalled write
+ await assertRejects(() => inWrite.done);
+
+ const wedgedSink = sinkOver(() => new Promise(() => {})); // create never resolves
+ const wedged = wedgedSink.sendFrame(new Uint8Array([3]));
+ wedged.abort(); // the send's stream is condemned: supersede must refuse too
+ assertEquals(wedged.supersede(new Uint8Array([4])), false);
+ await assertRejects(() => wedged.done);
+});
+
Deno.test("the viewer control stream is raised to CONTROL_SEND_ORDER", async () => {
const written: Uint8Array[] = [];
const controlWritable = new WritableStream({
diff --git a/web/relay/static/debug.html b/web/relay/static/debug.html
deleted file mode 100644
index 8834c4c897..0000000000
--- a/web/relay/static/debug.html
+++ /dev/null
@@ -1,461 +0,0 @@
-
-
-
-
-
- DimOS relay debug
-
-
-
- DimOS relay debug
- checking WebTransport support...
-
-
-
-
odom
-
-
-
round-trips
-
control: - datagram: -
-
-
-
robots
-
-
-
manifest
-
-
-
-
-
-
-
- | channel |
- frames |
- Hz |
- KB/s |
- span loss |
- out of order |
- last seq |
-
-
-
-
-