feat(web): acknowledged shared tx publishing

.github/workflows/ci.yml

@@ -482,7 +482,8 @@ jobs:
         run: >-
           uv run pytest -m web_browser dimos/e2e_tests/test_cockpit_browser.py
           dimos/e2e_tests/test_sdk_browser.py
-          dimos/e2e_tests/test_custom_channel_browser.py --no-cov
+          dimos/e2e_tests/test_custom_channel_browser.py
+          dimos/e2e_tests/test_publish_browser.py --no-cov
 
   tests:
     if: |

dimos/e2e_tests/test_publish_browser.py

@@ -0,0 +1,107 @@
+# 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.
+
+"""Shared-publish browser e2e (the W7 acceptance demo, in CI).
+
+Starts a cockpit(channels=[...]) bridge with
+Channel("human_input", str, dir="tx", encoding="text.json.v1",
+publish="shared") - a generated RelayBridgeModule subclass with a typed
+human_input Out port and the registry-resolved text.json.v1 decoder - serving
+web/examples/chat-input from its own local relay. The page calls
+session.publish() and renders the settled outcome: an ack proves the bridge
+called Out.publish() (so the DimOS consumer already holds the exact string by
+the time the browser promise resolves), and a bridge-rejected value renders
+its correlated error code without touching anything else.
+
+The bridge runs in-process (like the relay_bridge e2e) on an ephemeral port,
+so it cannot clash with test_sdk_browser's stack on :7780 in the same CI job.
+
+Marked web_browser: excluded from the default suite (needs the
+`browser-tests` dependency group, Playwright browsers, and built web dists).
+Locally:
+`uv run --group browser-tests pytest -m web_browser dimos/e2e_tests/test_publish_browser.py`.
+"""
+
+from collections.abc import Iterator
+
+import pytest
+
+from dimos.web.cockpit import Channel, cockpit
+from dimos.web.relay_bridge.e2e_support import stop_module
+from dimos.web.relay_bridge.locate import find_web_dir
+
+pytest.importorskip("playwright")
+
+from playwright.sync_api import Page, expect, sync_playwright
+
+pytestmark = pytest.mark.web_browser
+
+TEXT = "salut din browser β"
+
+
+@pytest.fixture(scope="module")
+def chat_bridge() -> Iterator[tuple[str, list[str]]]:
+    blueprint = cockpit(
+        channels=[Channel("human_input", str, dir="tx", encoding="text.json.v1", publish="shared")]
+    )
+    (atom,) = blueprint.blueprints
+    module = atom.module(
+        local_port=0,
+        open_browser=False,
+        robot_id="chat-input-e2e",
+        serve_dir=str(find_web_dir() / "examples" / "chat-input"),
+        **atom.kwargs,
+    )
+    received: list[str] = []
+    module.human_input.subscribe(received.append)
+    try:
+        module.start()  # builds web dists if stale, spawns the relay, registers
+        relay = module._relay
+        assert relay is not None and relay.info is not None
+        yield relay.info.open_url, received
+    finally:
+        stop_module(module)
+
+
+@pytest.fixture
+def chromium_page() -> Iterator[Page]:
+    with sync_playwright() as p:
+        browser = p.chromium.launch()
+        try:
+            yield browser.new_page()
+        finally:
+            browser.close()
+
+
+def test_publish_acks_and_rejections_render(
+    chat_bridge: tuple[str, list[str]], chromium_page: Page
+) -> None:
+    url, received = chat_bridge
+    chromium_page.goto(url)
+    expect(chromium_page.locator("#status")).to_contain_text("connected", timeout=120_000)
+
+    chromium_page.fill("#text", TEXT)
+    chromium_page.click("#send")
+    # The ack proves the whole chain: relay validation, the carrier tx frame,
+    # the registry-resolved decoder, Out.publish(), and the @control ack.
+    expect(chromium_page.locator("#log")).to_contain_text("acked human_input", timeout=15_000)
+    # The bridge acks only after Out.publish() returned, so the DimOS
+    # consumer already held the exact string when the promise resolved.
+    assert received == [TEXT]
+
+    # A value the decoder refuses settles as a definite rejection with the
+    # bridge's correlated code, and the accepted publish above is unaffected.
+    chromium_page.click("#send-bad")
+    expect(chromium_page.locator("#log")).to_contain_text("rejected decode_failed", timeout=15_000)
+    assert received == [TEXT]

dimos/web/cockpit.py

@@ -46,6 +46,7 @@
     RESERVED_CHANNEL_PREFIX,
     Delivery,
     Dir,
+    Publish,
     parse_manifest,
 )
 
@@ -136,6 +137,8 @@ class ChannelRequest:
     max_hz: float
     params: Mapping[str, Any] = field(default_factory=dict)
     delivery: Delivery = field(default="reliable", kw_only=True)
+    publish: Publish = field(default="none", kw_only=True)
+    required_scope: str | None = field(default=None, kw_only=True)
 
 
 @dataclass(frozen=True)
@@ -149,8 +152,16 @@ class Channel:
 
     `encoding` names a codec: a registered @web_encoder (dimos.web.codecs)
     whose message type must match `message_type`, or the generic "json.v1"
-    for JSON-shaped types and dataclasses. dir/publish/required_scope beyond
-    the rx defaults arrive with the publish ticket (W7).
+    for JSON-shaped types and dataclasses (rx) / JSON scalars, lists and
+    dicts (tx decode).
+
+    A dir="tx" channel is a generic browser publish input: it must declare
+    publish="shared" (any authorized viewer may publish; the bridge decodes
+    with the registered @web_decoder and publishes on a generated Out port).
+    publish="none" tx streams stay reserved for specialized protocol paths
+    (the teleop panel); publish="exclusive" arrives with the lease ticket
+    (W8). `required_scope` names the operator scope a remote relay demands
+    (the local relay never checks scopes).
     """
 
     stream: str
@@ -178,11 +189,6 @@ def __post_init__(self) -> None:
             raise TypeError(f"message_type must be a class, got {self.message_type!r}")
         if self.dir not in ("rx", "tx"):
             raise ValueError(f"dir must be 'rx' or 'tx', got {self.dir!r}")
-        if self.dir == "tx":
-            raise ValueError(
-                'dir="tx" channels are not available yet: generic browser-to-robot '
-                "publish is enabled by the publish ticket (W7)"
-            )
         if not isinstance(self.encoding, str) or not 1 <= len(self.encoding) <= MAX_MANIFEST_ID_LEN:
             raise ValueError(
                 f"encoding must be 1..{MAX_MANIFEST_ID_LEN} chars, got {self.encoding!r}"
@@ -194,12 +200,33 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f"publish must be 'none', 'shared', or 'exclusive', got {self.publish!r}"
             )
-        if self.publish != "none":
-            raise ValueError("rx channels must use publish='none' (generic publish is tx-only, W7)")
-        if self.required_scope is not None:
+        if self.publish == "exclusive":
+            raise ValueError(
+                "publish='exclusive' channels are enabled by the exclusive publisher "
+                "lease ticket (W8); use publish='shared' for interleavable input"
+            )
+        if self.dir == "rx" and self.publish != "none":
+            raise ValueError("rx channels must use publish='none'")
+        if self.dir == "tx" and self.publish == "none":
             raise ValueError(
-                "rx channels cannot set required_scope (reserved for tx publish channels, W7)"
+                'dir="tx" channels must declare publish="shared": publish="none" tx '
+                "streams are reserved for specialized protocol paths (the teleop panel)"
             )
+        if self.publish != "none" and self.delivery != "reliable":
+            raise ValueError(
+                f"publish channels must use delivery='reliable', got {self.delivery!r}"
+            )
+        if self.required_scope is not None:
+            if self.publish == "none":
+                raise ValueError("required_scope needs a publish policy (publish='shared')")
+            if (
+                not isinstance(self.required_scope, str)
+                or not 1 <= len(self.required_scope) <= MAX_MANIFEST_ID_LEN
+            ):
+                raise ValueError(
+                    f"required_scope must be 1..{MAX_MANIFEST_ID_LEN} chars, "
+                    f"got {self.required_scope!r}"
+                )
         if self.params is not None and not isinstance(self.params, Mapping):
             raise ValueError(f"params must be a mapping or None, got {self.params!r}")
         params = {} if self.params is None else dict(self.params)
@@ -408,8 +435,9 @@ def build_manifest_data(
     requested the stream.
 
     Advertisement order: rx built-ins in registry order, then rx customs in
-    first-declaration order, then tx in tx_registry order - so manifests
-    without custom channels keep their historical channel order.
+    first-declaration order, then tx in tx_registry order, then declared
+    publish tx channels in first-declaration order - so manifests without
+    custom channels keep their historical channel order.
     """
     tx_registry = {} if tx_registry is None else tx_registry
     merged: dict[str, ChannelRequest] = {}
@@ -420,11 +448,20 @@ def merge(request: ChannelRequest) -> None:
         if previous is None:
             merged[request.stream] = request
             return
-        same = (previous.dir, previous.encoding, previous.delivery, dict(previous.params)) == (
+        same = (
+            previous.dir,
+            previous.encoding,
+            previous.delivery,
+            dict(previous.params),
+            previous.publish,
+            previous.required_scope,
+        ) == (
             request.dir,
             request.encoding,
             request.delivery,
             dict(request.params),
+            request.publish,
+            request.required_scope,
         )
         if not same:
             raise ValueError(
@@ -491,6 +528,14 @@ def emit(node: Panel | _Split) -> Any:
                     f"unknown stream {stream!r}; this robot bridge supports: "
                     f"{', '.join(sorted(set(registry) | declared))}"
                 )
+        elif request.publish != "none":
+            # Generic-publish channels are declaration-driven: cockpit()
+            # generates their Out port and resolves their decoder, so the
+            # static tx tables have nothing to cross-check.
+            if stream not in declared:
+                raise ValueError(
+                    f"unknown publish stream {stream!r}; declare it in cockpit(channels=[...])"
+                )
         else:
             if stream not in tx_streams:
                 raise ValueError(
@@ -524,6 +569,14 @@ def emit(node: Panel | _Split) -> Any:
         if request.dir == "rx" and stream not in registry
     ]
     ordered += [merged[stream] for stream in tx_registry if stream in merged]
+    ordered += [
+        request
+        for stream, request in merged.items()
+        if request.dir == "tx" and stream not in tx_registry
+    ]
+    # Fully explicit (normalized) emission, publish/requiredScope included:
+    # parse_manifest(emitted).model_dump() == emitted must hold (idempotence),
+    # and plain model_dump() always carries every field.
     return {
         "version": MANIFEST_VERSION,
         "channels": [
@@ -534,6 +587,8 @@ def emit(node: Panel | _Split) -> Any:
                 "delivery": request.delivery,
                 "maxHz": request.max_hz,
                 "params": dict(request.params),
+                "publish": request.publish,
+                "requiredScope": request.required_scope,
             }
             for request in ordered
         ],
@@ -553,13 +608,14 @@ def cockpit(
 
     Walks the tree, merges panel requests with the `channels` declarations,
     compiles the manifest, validates it eagerly, resolves every rx channel's
-    codec, and returns a relay bridge blueprint carrying it all; compose
-    onto a robot with `autoconnect(robot_blueprint, cockpit(...))`. Channels
-    whose stream is not a built-in bridge port get a generated
-    RelayBridgeModule subclass with matching typed ports (autoconnected by
-    name + message type). Streams whose producer never publishes are still
-    advertised: their panels show "waiting for data" (no runtime stream
-    probing).
+    encoder and every publish tx channel's decoder, and returns a relay
+    bridge blueprint carrying it all; compose onto a robot with
+    `autoconnect(robot_blueprint, cockpit(...))`. Channels whose stream is
+    not a built-in bridge port get a generated RelayBridgeModule subclass
+    with matching typed ports (In for rx, Out for publish tx, autoconnected
+    by name + message type). Streams whose producer never publishes are
+    still advertised: their panels show "waiting for data" (no runtime
+    stream probing).
 
     The default preset layout applies only when neither a layout nor
     channels are given; cockpit(channels=[...]) alone compiles with no
@@ -585,7 +641,7 @@ def cockpit(
             "the cockpit blueprint needs the web extra: `uv sync --extra web --inexact`"
         ) from e
     from dimos.core.coordination.blueprints import autoconnect
-    from dimos.web.codecs import resolve_encoder
+    from dimos.web.codecs import resolve_decoder, resolve_encoder
 
     atom = RelayBridgeModule.blueprint().blueprints[0]
     port_types = {s.name: s.type for s in atom.streams}
@@ -606,6 +662,8 @@ def cockpit(
                 # (frozen) authoring record's nested values.
                 _thaw_params(c.params or {}),
                 delivery=c.delivery,
+                publish=c.publish,
+                required_scope=c.required_scope,
             )
             for c in declared.values()
         ),
@@ -617,10 +675,33 @@ def cockpit(
     specs: list[RuntimeChannelSpec] = []
     ports: list[DynamicPortSpec] = []
     for wire in data["channels"]:
-        if wire["dir"] != "rx":
-            continue
         ch = wire["ch"]
         explicit = declared.get(ch)
+        if wire["dir"] != "rx":
+            if wire["publish"] == "none":
+                continue  # specialized tx (teleop): no runtime spec, no decoder
+            assert explicit is not None  # build_manifest_data rejected the rest
+            ports.append(DynamicPortSpec(ch, explicit.message_type, "tx"))
+            try:
+                decoder = resolve_decoder(wire["encoding"], explicit.message_type)
+            except ValueError as e:
+                raise ValueError(f"channel {ch!r}: {e}") from e
+            specs.append(
+                RuntimeChannelSpec(
+                    ch=ch,
+                    message_type=explicit.message_type,
+                    dir="tx",
+                    encoding=wire["encoding"],
+                    delivery=wire["delivery"],
+                    max_hz=float(wire["maxHz"]),
+                    params=dict(wire["params"]),
+                    publish=wire["publish"],
+                    required_scope=wire["requiredScope"],
+                    decoder=decoder.decode,
+                    decoder_takes_context=decoder.takes_context,
+                )
+            )
+            continue
         builtin = builtin_by_ch.get(ch)
         if builtin is not None:
             message_type = port_types[ch]