diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py
index 3585b42..5196d62 100644
--- a/work/meshai/adapter_config/defaults.py
+++ b/work/meshai/adapter_config/defaults.py
@@ -413,7 +413,7 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
     # digest_max_chars: mesh wire cap. The LLM is told to fit under this.
     # Reuses the response.max_length chunking if the LLM ignores the cap.
     ("fires", "digest_max_chars"): {
-        "default": 200,
+        "default": 140,
         "type": "int",
         "description": "Hard cap on the digest wire string length (chars). The LLM prompt asks to fit; the chunker enforces.",
     },
diff --git a/work/meshai/config.py b/work/meshai/config.py
index 69205d9..1b1491c 100644
--- a/work/meshai/config.py
+++ b/work/meshai/config.py
@@ -36,12 +36,12 @@ class ConnectionConfig:
     reconnect_health_interval: float = 30.0
     # --- transport selection (Phase 1 seam; MeshCore support is Phase 2) ---
     transport: str = "meshtastic"  # "meshtastic" | "meshcore" | "both"
-    meshtastic_max_chars: int = 200  # default Meshtastic packet budget
+    # Universal mesh message budget (MeshCore LCD → 140 for all transports).
+    mesh_max_chars: int = 140
     # --- MeshCore transport settings (used when transport="meshcore") ---
     meshcore_host: str = "100.64.0.9"          # pyMC companion frame server host
     meshcore_port: int = 5050                   # pyMC companion frame server port
     meshcore_channel_index: int = 0             # default channel index for broadcasts
-    meshcore_max_chars: int = 140               # placeholder max chars (Phase 3 chunker)
     meshcore_auto_reconnect: bool = True        # enable meshcore lib auto-reconnect
     meshcore_max_reconnect_attempts: int = 5    # max reconnect attempts (0 = unlimited)
 
diff --git a/work/meshai/connector.py b/work/meshai/connector.py
index 5771102..a908bf2 100644
--- a/work/meshai/connector.py
+++ b/work/meshai/connector.py
@@ -45,6 +45,9 @@ class MeshMessage:
 class MeshtasticTransport(MeshTransport):
     """Manages connection to a Meshtastic node (Meshtastic transport backend)."""
 
+    # Name tag used by CompositeTransport for routing hints.
+    transport_name: str = "meshtastic"
+
     def __init__(self, config: ConnectionConfig):
         self.config = config
         self._interface: Optional[meshtastic.MeshInterface] = None
@@ -74,7 +77,7 @@ class MeshtasticTransport(MeshTransport):
 
     @property
     def max_chars(self) -> int:
-        return getattr(self.config, "meshtastic_max_chars", 200)
+        return self.config.mesh_max_chars
 
     def connect(self) -> None:
         """Establish connection to Meshtastic node."""
@@ -351,6 +354,7 @@ class MeshtasticTransport(MeshTransport):
         text: str,
         destination: Optional[str] = None,
         channel: int = 0,
+        transport: Optional[str] = None,  # routing hint — accepted and IGNORED by single-transport impl
     ) -> bool:
         """Send a text message.
 
@@ -358,6 +362,7 @@ class MeshtasticTransport(MeshTransport):
             text: Message text to send
             destination: Node ID for DM, or None for broadcast
             channel: Channel index to send on
+            transport: Optional routing hint (for CompositeTransport); ignored here.
 
         Returns:
             True if send was initiated successfully
diff --git a/work/meshai/main.py b/work/meshai/main.py
index 2900823..a41dd9c 100644
--- a/work/meshai/main.py
+++ b/work/meshai/main.py
@@ -85,14 +85,24 @@ class MeshAI:
         # state + the only reconnect driver. Reconnects IN-PLACE so the container
         # never needs to restart.
         # Guard: watchdog is Meshtastic-specific — MeshCoreTransport manages its
-        # own reconnect via the meshcore lib's auto_reconnect parameter. ---
-        if (
-            getattr(self.config.connection, "reconnect", True)
-            and isinstance(self.connector, MeshConnector)
-        ):
-            self.connector._wake = asyncio.Event()
-            self.connector.write_link_status("up")  # we just connected ok
-            self._supervisor_task = asyncio.create_task(self._connection_supervisor())
+        # own reconnect via the meshcore lib's auto_reconnect parameter.
+        # When connector is a CompositeTransport, resolve the Meshtastic child
+        # (if any) and run the watchdog on it; skip if absent (pure MeshCore). ---
+        _mt_child = (
+            self.connector
+            if isinstance(self.connector, MeshConnector)
+            else (
+                self.connector.meshtastic_child()
+                if hasattr(self.connector, "meshtastic_child")
+                else None
+            )
+        )
+        if getattr(self.config.connection, "reconnect", True) and _mt_child is not None:
+            _mt_child._wake = asyncio.Event()
+            _mt_child.write_link_status("up")  # we just connected ok
+            self._supervisor_task = asyncio.create_task(
+                self._connection_supervisor(_mt_child)
+            )
             logger.info("Connection supervisor (watchdog) started")
         self._last_cleanup = time.time()
         self._last_health_compute = 0.0
@@ -226,12 +236,18 @@ class MeshAI:
                     self.context.prune()
                 self._last_cleanup = time.time()
 
-    async def _connection_supervisor(self) -> None:
+    async def _connection_supervisor(self, c=None) -> None:
         """Watchdog: SINGLE source of truth for /tmp/meshai.link and the ONLY
         reconnect driver. Woken by connector._wake (connection.lost) or a
         health-interval timeout. Probe is socket-based (see connector.active_probe).
+
+        *c* is the resolved MeshtasticTransport to watch — either self.connector
+        directly (single-transport) or the Meshtastic child extracted from a
+        CompositeTransport.  The watchdog logic itself is unchanged; only how
+        we resolve *c* has moved to the caller (start()).
         """
-        c = self.connector
+        if c is None:
+            c = self.connector
         hi = getattr(self.config.connection, "reconnect_health_interval", 30.0)
         alive_idle = 2.0 * hi
         probe_wait = 5.0
@@ -647,6 +663,12 @@ class MeshAI:
             )
 
             # Route the message
+            # Capture the originating transport tag for reply routing.
+            # This lets CompositeTransport route the DM reply back over the
+            # same mesh the inbound message arrived on.  Single-transport
+            # connectors accept and ignore this kwarg.
+            originating_transport: Optional[str] = getattr(message, "transport", None)
+
             # Check for continuation request first
             continuation_messages = self.router.check_continuation(message)
             if continuation_messages:
@@ -654,6 +676,7 @@ class MeshAI:
                     continuation_messages,
                     destination=message.sender_id,
                     channel=message.channel,
+                    transport=originating_transport,
                 )
                 return
 
@@ -685,11 +708,13 @@ class MeshAI:
             if not messages:
                 return
 
-            # Send DM response
+            # Send DM response — thread the originating transport hint so
+            # CompositeTransport routes the reply back over the correct mesh.
             await self.responder.send_response(
                 messages,
                 destination=message.sender_id,
                 channel=message.channel,
+                transport=originating_transport,
             )
 
         except Exception as e:
diff --git a/work/meshai/notifications/pipeline/__init__.py b/work/meshai/notifications/pipeline/__init__.py
index 7f0e67b..d6d2908 100644
--- a/work/meshai/notifications/pipeline/__init__.py
+++ b/work/meshai/notifications/pipeline/__init__.py
@@ -74,7 +74,7 @@ def build_pipeline(config, llm_backend, connector=None) -> EventBus:
     accumulator = DigestAccumulator(
         llm_backend=llm_backend,
         include_toggles=include_toggles,
-        mesh_char_limit=connector.max_chars if connector is not None else 200,
+        mesh_char_limit=connector.max_chars if connector is not None else 140,
     )
 
     # Tee closure: events go to BOTH dispatcher and accumulator
diff --git a/work/meshai/notifications/scheduled/fire_digest.py b/work/meshai/notifications/scheduled/fire_digest.py
index 84d41d2..176c142 100644
--- a/work/meshai/notifications/scheduled/fire_digest.py
+++ b/work/meshai/notifications/scheduled/fire_digest.py
@@ -112,7 +112,8 @@ async def render_digest(*, now: Optional[int] = None) -> tuple[str, str]:
         else:
             fire_lines.append(name)
 
-    # Assemble within ~200-byte LoRa budget; trim fire lines, never the tail
+    # Assemble within the universal mesh budget; trim fire lines, never the tail
+    budget = int(adapter_config.fires.digest_max_chars)
     shown: list[str] = []
     for line in fire_lines:
         # Estimate tail for budget check
@@ -127,7 +128,7 @@ async def render_digest(*, now: Optional[int] = None) -> tuple[str, str]:
         if est_tail:
             parts.append(est_tail)
         candidate = "\n".join(parts)
-        if len(candidate.encode("utf-8")) <= 200:
+        if len(candidate.encode("utf-8")) <= budget:
             shown.append(line)
         else:
             break
diff --git a/work/meshai/responder.py b/work/meshai/responder.py
index a491654..e9a3b38 100644
--- a/work/meshai/responder.py
+++ b/work/meshai/responder.py
@@ -23,8 +23,21 @@ class Responder:
         messages: list[str] | str,
         destination: Optional[str] = None,
         channel: int = 0,
+        transport: Optional[str] = None,
     ) -> bool:
-        """Send response messages with randomized delay pacing."""
+        """Send response messages with randomized delay pacing.
+
+        Args:
+            messages: One or more message strings to send.
+            destination: Node ID for a DM, or None for broadcast.
+            channel: Channel index to send on.
+            transport: Optional routing hint threaded from the originating
+                       MeshMessage.  Passed through to connector.send_message
+                       so CompositeTransport can route DM replies back over
+                       the mesh they arrived on.  Single-transport connectors
+                       accept and ignore it; defaults to None so all existing
+                       call sites are unaffected.
+        """
         if isinstance(messages, str):
             messages = [messages]
 
@@ -42,6 +55,7 @@ class Responder:
                 text=msg,
                 destination=destination,
                 channel=channel,
+                transport=transport,
             )
             if not sent:
                 logger.error(f"Failed to send message {i+1}/{len(messages)}")
diff --git a/work/meshai/transport/base.py b/work/meshai/transport/base.py
index 43cccd6..227be1b 100644
--- a/work/meshai/transport/base.py
+++ b/work/meshai/transport/base.py
@@ -36,6 +36,7 @@ class MeshTransport(abc.ABC):
         text: str,
         destination: Optional[str] = None,
         channel: int = 0,
+        transport: Optional[str] = None,
     ) -> bool:
         """Send a text message.
 
@@ -43,6 +44,10 @@ class MeshTransport(abc.ABC):
             text: Message text to send.
             destination: Node ID for a DM, or None for broadcast.
             channel: Channel index to send on.
+            transport: Optional routing hint (used by CompositeTransport to
+                       select the child mesh that originated an inbound DM).
+                       Single-transport implementations accept and IGNORE this
+                       parameter; it is always None in non-composite callers.
 
         Returns:
             True if send was initiated successfully.
diff --git a/work/meshai/transport/composite_transport.py b/work/meshai/transport/composite_transport.py
new file mode 100644
index 0000000..4b407fb
--- /dev/null
+++ b/work/meshai/transport/composite_transport.py
@@ -0,0 +1,386 @@
+"""CompositeTransport — fan-out driver for dual Meshtastic + MeshCore meshes.
+
+Holds an ordered list of child transports and:
+  - fans broadcasts to ALL connected children;
+  - routes hinted DM replies back over the originating child;
+  - falls back to best-effort per-child resolution for unhinted DMs;
+  - self-filters inbound messages per child (drops own echoes);
+  - exposes ``meshtastic_child()`` for the supervisor watchdog.
+
+This transport is DORMANT unless ``transport: both`` in config.  Single-
+transport paths (``transport: meshtastic`` / ``transport: meshcore``) are
+byte-identical to Phase 3 behaviour because the factory never instantiates
+this class for them.
+"""
+
+import asyncio
+import logging
+from typing import Callable, List, Optional
+
+from .base import MeshTransport
+
+logger = logging.getLogger(__name__)
+
+
+def _child_name(child: MeshTransport) -> str:
+    """Return the transport-name tag for a child, deriving it from the class
+    when no explicit ``transport_name`` attribute is present.
+
+    Priority:
+      1. ``child.transport_name``  (set on MeshtasticTransport / MeshCoreTransport)
+      2. Class name lowercased, with "transport" suffix stripped.
+    """
+    name = getattr(child, "transport_name", None)
+    if name:
+        return name
+    cls = type(child).__name__.lower()
+    if cls.endswith("transport"):
+        cls = cls[: -len("transport")]
+    return cls or "unknown"
+
+
+class CompositeTransport(MeshTransport):
+    """MeshTransport that drives multiple child transports simultaneously.
+
+    Instantiate with a list of concrete MeshTransport children, e.g.::
+
+        CompositeTransport([MeshtasticTransport(cfg), MeshCoreTransport(cfg)])
+
+    The children are contacted in list order; the *first* child is the
+    canonical source for ``my_node_id`` (coarse fallback — real self-
+    filtering is per-child in the inbound wrapper).
+    """
+
+    def __init__(self, children: List[MeshTransport], config=None) -> None:
+        if not children:
+            raise ValueError("CompositeTransport requires at least one child")
+        self._children = list(children)
+        self._config = config  # ConnectionConfig; used for the universal mesh_max_chars budget
+        # Build a stable name → child mapping for O(1) routing-hint lookup.
+        self._by_name: dict[str, MeshTransport] = {
+            _child_name(c): c for c in self._children
+        }
+
+    # ------------------------------------------------------------------
+    # Public accessors
+    # ------------------------------------------------------------------
+
+    @property
+    def children(self) -> List[MeshTransport]:
+        """Ordered list of child transports."""
+        return list(self._children)
+
+    def meshtastic_child(self) -> Optional[MeshTransport]:
+        """Return the MeshtasticTransport child, or None if absent.
+
+        Used by the supervisor watchdog so it can apply Meshtastic-specific
+        liveness probes (socket_state, active_probe, reconnect) without
+        knowing whether it is talking to a bare MeshtasticTransport or a
+        CompositeTransport.
+        """
+        from meshai.connector import MeshtasticTransport  # local import avoids cycle
+        for c in self._children:
+            if isinstance(c, MeshtasticTransport):
+                return c
+        return None
+
+    # ------------------------------------------------------------------
+    # Routing decision helpers (factored out for unit-test access)
+    # ------------------------------------------------------------------
+
+    def _resolve_child_for_hint(self, transport_hint: str) -> Optional[MeshTransport]:
+        """Return the child whose name matches *transport_hint*, or None."""
+        return self._by_name.get(transport_hint)
+
+    def _best_child_for_destination(self, destination: str) -> Optional[MeshTransport]:
+        """Return the child most likely to know *destination*, or None.
+
+        A child "resolves" the destination when ``get_node_name()`` returns
+        something other than the raw *destination* string itself (i.e. the
+        child has the node in its name cache).  The first such child wins.
+        If no child resolves the destination, None is returned and the
+        caller falls back to broadcasting to all children.
+        """
+        for child in self._children:
+            if not child.connected:
+                continue
+            try:
+                resolved = child.get_node_name(destination)
+                if resolved and resolved != destination:
+                    return child
+            except Exception:
+                pass
+        return None
+
+    def _should_drop(self, msg, child_node_id: Optional[str]) -> bool:
+        """Return True when *msg* should be self-filtered (echo from self).
+
+        Factored out so unit tests can call it directly.
+        """
+        if child_node_id is None:
+            return False
+        return msg.sender_id == child_node_id
+
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+
+    def connect(self) -> None:
+        """Connect all children.
+
+        A child failing to connect is logged but does NOT prevent the
+        remaining children from connecting.  ``connected`` returns True if
+        at least one child is connected after the loop.
+        """
+        for child in self._children:
+            name = _child_name(child)
+            try:
+                child.connect()
+                logger.info("CompositeTransport: child %r connected", name)
+            except Exception as exc:
+                logger.error(
+                    "CompositeTransport: child %r failed to connect: %s", name, exc
+                )
+
+    def disconnect(self) -> None:
+        """Disconnect all children, guarding each with try/except."""
+        for child in self._children:
+            name = _child_name(child)
+            try:
+                child.disconnect()
+                logger.info("CompositeTransport: child %r disconnected", name)
+            except Exception as exc:
+                logger.warning(
+                    "CompositeTransport: child %r disconnect error: %s", name, exc
+                )
+
+    # ------------------------------------------------------------------
+    # Properties
+    # ------------------------------------------------------------------
+
+    @property
+    def connected(self) -> bool:
+        """True if at least one child is connected."""
+        return any(c.connected for c in self._children)
+
+    @property
+    def my_node_id(self) -> Optional[str]:
+        """Return the first child's node ID (coarse fallback).
+
+        Real self-filtering is per-child (each child's own my_node_id is
+        used in the inbound wrapper registered by set_message_callback).
+        """
+        return self._children[0].my_node_id if self._children else None
+
+    @property
+    def max_chars(self) -> int:
+        """Return the universal mesh message budget from config.
+
+        Sourced directly from ``config.mesh_max_chars`` (default 140) — the
+        single fixed constant that governs all transports regardless of which
+        radios are connected.  MeshCore is the LCD, so 140 is the right value
+        for every path.
+        """
+        if self._config is not None:
+            return self._config.mesh_max_chars
+        # Fallback for tests that build CompositeTransport without a config.
+        return 140
+
+    # ------------------------------------------------------------------
+    # Message I/O
+    # ------------------------------------------------------------------
+
+    def send_message(
+        self,
+        text: str,
+        destination: Optional[str] = None,
+        channel: int = 0,
+        transport: Optional[str] = None,
+    ) -> bool:
+        """Send a message, routing based on destination + hint.
+
+        Routing rules
+        -------------
+        1. **Broadcast** (``destination is None``):
+           Fan out to ALL connected children.  Return True if at least one
+           child succeeded; log per-child failures.
+
+        2. **DM with routing hint** (``destination`` set AND ``transport`` given):
+           Send ONLY via the child whose name == ``transport``.  This is the
+           reply-routing path: the originating transport tag is threaded from
+           the inbound MeshMessage all the way to here so the reply goes back
+           over the same mesh it arrived on.
+
+        3. **DM without hint** (``destination`` set, ``transport`` is None):
+           Best-effort: prefer the child that can resolve *destination* via
+           ``get_node_name`` (i.e. has the node in its cache).  If no child
+           resolves, send via ALL connected children and log a warning.
+           This fallback handles subscription DMs and other direct sends that
+           don't carry an originating transport tag.
+        """
+        if destination is None:
+            # --- Rule 1: broadcast ---
+            return self._broadcast(text, channel)
+
+        if transport is not None:
+            # --- Rule 2: hinted DM ---
+            return self._send_hinted(text, destination, channel, transport)
+
+        # --- Rule 3: unhinted DM ---
+        return self._send_unhinted(text, destination, channel)
+
+    def _broadcast(self, text: str, channel: int) -> bool:
+        """Fan text out to all connected children; return True if any succeed."""
+        any_ok = False
+        for child in self._children:
+            name = _child_name(child)
+            if not child.connected:
+                logger.debug(
+                    "CompositeTransport: skipping broadcast to %r (not connected)", name
+                )
+                continue
+            try:
+                ok = child.send_message(text, destination=None, channel=channel)
+                if ok:
+                    any_ok = True
+                else:
+                    logger.warning(
+                        "CompositeTransport: broadcast via %r returned False", name
+                    )
+            except Exception as exc:
+                logger.error(
+                    "CompositeTransport: broadcast via %r raised: %s", name, exc
+                )
+        return any_ok
+
+    def _send_hinted(
+        self, text: str, destination: str, channel: int, transport_hint: str
+    ) -> bool:
+        """Send DM only via the child matching *transport_hint*."""
+        child = self._resolve_child_for_hint(transport_hint)
+        if child is None:
+            logger.error(
+                "CompositeTransport: no child with name %r; known: %s",
+                transport_hint,
+                list(self._by_name),
+            )
+            return False
+        name = _child_name(child)
+        if not child.connected:
+            logger.warning(
+                "CompositeTransport: hinted child %r not connected", name
+            )
+            return False
+        try:
+            return child.send_message(text, destination=destination, channel=channel)
+        except Exception as exc:
+            logger.error(
+                "CompositeTransport: hinted send via %r raised: %s", name, exc
+            )
+            return False
+
+    def _send_unhinted(self, text: str, destination: str, channel: int) -> bool:
+        """Best-effort DM: prefer the resolving child; else fan to all.
+
+        This fallback handles subscription DMs, alert DMs, and other direct
+        sends that don't carry an originating transport tag.  When multiple
+        children know the destination, the first one in list order wins.
+        If none resolve the destination, all connected children receive the
+        DM so at least one can deliver it — callers see a warning so the
+        behaviour is visible in logs.
+        """
+        child = self._best_child_for_destination(destination)
+        if child is not None:
+            name = _child_name(child)
+            try:
+                return child.send_message(text, destination=destination, channel=channel)
+            except Exception as exc:
+                logger.error(
+                    "CompositeTransport: unhinted send via %r raised: %s", name, exc
+                )
+                return False
+
+        # No child resolved the destination — send via all and warn.
+        logger.warning(
+            "CompositeTransport: destination %r unresolved; fanning DM to all children",
+            destination,
+        )
+        any_ok = False
+        for child in self._children:
+            name = _child_name(child)
+            if not child.connected:
+                continue
+            try:
+                ok = child.send_message(text, destination=destination, channel=channel)
+                if ok:
+                    any_ok = True
+            except Exception as exc:
+                logger.error(
+                    "CompositeTransport: DM fan via %r raised: %s", name, exc
+                )
+        return any_ok
+
+    def set_message_callback(
+        self,
+        callback: Callable,
+        loop: asyncio.AbstractEventLoop,
+    ) -> None:
+        """Register per-child inbound wrappers.
+
+        Each child gets its OWN wrapper that:
+          (a) self-filters — drops the message if sender_id == that child's
+              my_node_id (each child knows its own ID);
+          (b) ensures msg.transport is set to the child's name if missing;
+          (c) forwards to the meshai callback.
+
+        Each child already marshals its callback onto *loop* via
+        ``loop.call_soon_threadsafe``; we preserve that — the wrapper is
+        just a thin decorator around the meshai callback.
+        """
+        for child in self._children:
+            child_name = _child_name(child)
+
+            # Capture child_name and child identity in the closure.
+            def _make_wrapper(name: str, c: MeshTransport) -> Callable:
+                async def _wrapper(msg) -> None:
+                    # (a) self-filter
+                    if self._should_drop(msg, c.my_node_id):
+                        logger.debug(
+                            "CompositeTransport: dropping echo from %r (self)", name
+                        )
+                        return
+                    # (b) ensure transport tag
+                    if not msg.transport:
+                        msg.transport = name
+                    # (c) forward
+                    await callback(msg)
+
+                return _wrapper
+
+            child.set_message_callback(_make_wrapper(child_name, child), loop)
+
+    # ------------------------------------------------------------------
+    # Node identity / topology
+    # ------------------------------------------------------------------
+
+    def get_node_name(self, node_id: str) -> str:
+        """Try each child in order; return the first non-identity result."""
+        for child in self._children:
+            try:
+                name = child.get_node_name(node_id)
+                if name and name != node_id:
+                    return name
+            except Exception:
+                pass
+        return node_id
+
+    def get_node_position(self, node_id: str) -> Optional[tuple]:
+        """Try each child in order; return the first non-None position."""
+        for child in self._children:
+            try:
+                pos = child.get_node_position(node_id)
+                if pos is not None:
+                    return pos
+            except Exception:
+                pass
+        return None
diff --git a/work/meshai/transport/factory.py b/work/meshai/transport/factory.py
index 11fedf5..6890878 100644
--- a/work/meshai/transport/factory.py
+++ b/work/meshai/transport/factory.py
@@ -15,8 +15,6 @@ def build_transport(config) -> MeshTransport:
         A concrete MeshTransport instance ready to be connected.
 
     Raises:
-        NotImplementedError: When the requested transport is known but not
-            yet implemented (e.g. ``"meshcore"`` or ``"both"``).
         ValueError: When the transport name is unrecognised.
     """
     transport_name = getattr(config, "transport", "meshtastic")
@@ -33,9 +31,15 @@ def build_transport(config) -> MeshTransport:
         return MeshCoreTransport(config)
 
     if transport_name == "both":
-        raise NotImplementedError(
-            "Transport 'both' is not yet implemented (Phase 4 seam)."
-        )
+        # Phase 4: composite transport — drives Meshtastic and MeshCore
+        # simultaneously with correct reply routing.
+        from meshai.connector import MeshtasticTransport
+        from meshai.transport.meshcore_transport import MeshCoreTransport
+        from meshai.transport.composite_transport import CompositeTransport
+        return CompositeTransport([
+            MeshtasticTransport(config),
+            MeshCoreTransport(config),
+        ], config=config)
 
     raise ValueError(
         f"Unknown transport {transport_name!r}. "
diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py
index 4f06c28..c6f2cf0 100644
--- a/work/meshai/transport/meshcore_transport.py
+++ b/work/meshai/transport/meshcore_transport.py
@@ -38,6 +38,9 @@ class MeshCoreTransport(MeshTransport):
     class, so there is zero cost to existing meshtastic deployments.
     """
 
+    # Name tag used by CompositeTransport for routing hints.
+    transport_name: str = "meshcore"
+
     def __init__(self, config) -> None:
         self.config = config
         self._mc = None                          # meshcore.MeshCore instance
@@ -190,14 +193,16 @@ class MeshCoreTransport(MeshTransport):
         text: str,
         destination: Optional[str] = None,
         channel: int = 0,
+        transport: Optional[str] = None,  # routing hint — accepted and IGNORED by single-transport impl
     ) -> bool:
         """Send a message via MeshCore.
 
         Args:
             text: Message text (caller is responsible for length limits; see
-                  ``meshcore_max_chars`` config field for the future chunker).
+                  ``mesh_max_chars`` config field).
             destination: hex pubkey string for a DM, or None for channel send.
             channel: Channel index for channel sends (0 = default).
+            transport: Optional routing hint (for CompositeTransport); ignored here.
 
         Returns:
             True if the send succeeded (not an error event).
@@ -364,7 +369,7 @@ class MeshCoreTransport(MeshTransport):
 
     @property
     def max_chars(self) -> int:
-        return getattr(self.config, "meshcore_max_chars", 140)
+        return self.config.mesh_max_chars
 
     def get_node_name(self, node_id: str) -> str:
         """Resolve a pubkey prefix to a contact display name, or return node_id."""
diff --git a/work/tests/test_composite_transport.py b/work/tests/test_composite_transport.py
new file mode 100644
index 0000000..d5c1c80
--- /dev/null
+++ b/work/tests/test_composite_transport.py
@@ -0,0 +1,568 @@
+"""Hermetic tests for CompositeTransport (Phase 4).
+
+All tests use fake child transports — no real sockets, threads, or meshcore
+lib required.  The helpers below replicate the MeshMessage dataclass so the
+tests run without importing the full connector chain.
+"""
+
+import asyncio
+import types
+from dataclasses import dataclass, field
+from typing import Optional
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from meshai.transport.composite_transport import CompositeTransport, _child_name
+from meshai.connector import MeshMessage
+
+
+# ---------------------------------------------------------------------------
+# Fake child transport helpers
+# ---------------------------------------------------------------------------
+
+class FakeChild:
+    """Minimal MeshTransport stand-in for unit tests.
+
+    Tracks calls and exposes enough surface for CompositeTransport to work.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        node_id: str = "!aabbccdd",
+        max_chars_val: int = 200,
+        connected_val: bool = True,
+        known_nodes: Optional[dict] = None,
+    ) -> None:
+        self.transport_name = name
+        self._connected = connected_val
+        self._node_id = node_id
+        self._max_chars = max_chars_val
+        self._known_nodes: dict[str, str] = known_nodes or {}
+
+        # Call-tracking
+        self.connect_calls = 0
+        self.disconnect_calls = 0
+        self.send_calls: list[dict] = []
+        self._callback = None
+        self._callback_loop = None
+
+    # --- MeshTransport interface ---
+
+    def connect(self) -> None:
+        self.connect_calls += 1
+        self._connected = True
+
+    def disconnect(self) -> None:
+        self.disconnect_calls += 1
+        self._connected = False
+
+    @property
+    def connected(self) -> bool:
+        return self._connected
+
+    @property
+    def my_node_id(self) -> Optional[str]:
+        return self._node_id
+
+    @property
+    def max_chars(self) -> int:
+        return self._max_chars
+
+    def send_message(
+        self,
+        text: str,
+        destination: Optional[str] = None,
+        channel: int = 0,
+        transport: Optional[str] = None,
+    ) -> bool:
+        self.send_calls.append(
+            {"text": text, "destination": destination, "channel": channel, "transport": transport}
+        )
+        return True
+
+    def set_message_callback(self, callback, loop) -> None:
+        self._callback = callback
+        self._callback_loop = loop
+
+    def get_node_name(self, node_id: str) -> str:
+        return self._known_nodes.get(node_id, node_id)
+
+    def get_node_position(self, node_id: str) -> Optional[tuple]:
+        return None
+
+    # --- Test helper: simulate inbound message ---
+
+    async def _simulate_inbound(self, msg: MeshMessage) -> None:
+        """Call the registered callback directly (simulates an inbound packet)."""
+        if self._callback:
+            await self._callback(msg)
+
+
+# ---------------------------------------------------------------------------
+# Factory-level test
+# ---------------------------------------------------------------------------
+
+class TestFactory:
+    def test_both_returns_composite(self) -> None:
+        """factory transport='both' → CompositeTransport with 2 children."""
+        from meshai.transport.factory import build_transport
+
+        cfg = types.SimpleNamespace(
+            transport="both",
+            type="tcp",
+            tcp_host="127.0.0.1",
+            tcp_port=4403,
+            meshcore_host="127.0.0.1",
+            meshcore_port=5050,
+        )
+        t = build_transport(cfg)
+        assert isinstance(t, CompositeTransport)
+        assert len(t.children) == 2
+        # Child order: Meshtastic first, MeshCore second.
+        from meshai.connector import MeshtasticTransport
+        from meshai.transport.meshcore_transport import MeshCoreTransport
+        assert isinstance(t.children[0], MeshtasticTransport)
+        assert isinstance(t.children[1], MeshCoreTransport)
+
+
+# ---------------------------------------------------------------------------
+# max_chars
+# ---------------------------------------------------------------------------
+
+class TestMaxChars:
+    def test_fixed_universal_budget_no_config(self) -> None:
+        """Without a config, CompositeTransport falls back to the 140 constant."""
+        mt = FakeChild("meshtastic", max_chars_val=200)
+        mc = FakeChild("meshcore", max_chars_val=140)
+        comp = CompositeTransport([mt, mc])
+        assert comp.max_chars == 140
+
+    def test_fixed_universal_budget_from_config(self) -> None:
+        """With a config, CompositeTransport reads mesh_max_chars directly."""
+        from meshai.config import ConnectionConfig
+        cfg = ConnectionConfig(transport="both", mesh_max_chars=140)
+        mt = FakeChild("meshtastic", max_chars_val=200)
+        mc = FakeChild("meshcore", max_chars_val=140)
+        comp = CompositeTransport([mt, mc], config=cfg)
+        assert comp.max_chars == 140
+
+    def test_fixed_universal_budget_ignores_child_values(self) -> None:
+        """CompositeTransport must NOT take min(children); it sources config."""
+        from meshai.config import ConnectionConfig
+        cfg = ConnectionConfig(transport="both", mesh_max_chars=140)
+        # Even if a child would report 230, the composite must return 140.
+        child = FakeChild("meshtastic", max_chars_val=230)
+        comp = CompositeTransport([child], config=cfg)
+        assert comp.max_chars == 140
+
+
+# ---------------------------------------------------------------------------
+# connect / disconnect
+# ---------------------------------------------------------------------------
+
+class TestLifecycle:
+    def test_connect_fans_to_all(self) -> None:
+        a = FakeChild("meshtastic", connected_val=False)
+        b = FakeChild("meshcore", connected_val=False)
+        comp = CompositeTransport([a, b])
+        comp.connect()
+        assert a.connect_calls == 1
+        assert b.connect_calls == 1
+
+    def test_connect_continues_after_failure(self) -> None:
+        """One child failing to connect must not stop the other."""
+        bad = FakeChild("meshtastic", connected_val=False)
+        bad.connect = MagicMock(side_effect=RuntimeError("boom"))
+        good = FakeChild("meshcore", connected_val=False)
+        comp = CompositeTransport([bad, good])
+        comp.connect()  # must not raise
+        assert good.connect_calls == 1
+        assert comp.connected  # good child is up
+
+    def test_connected_true_if_any(self) -> None:
+        down = FakeChild("meshtastic", connected_val=False)
+        up = FakeChild("meshcore", connected_val=True)
+        comp = CompositeTransport([down, up])
+        assert comp.connected is True
+
+    def test_connected_false_if_none(self) -> None:
+        a = FakeChild("meshtastic", connected_val=False)
+        b = FakeChild("meshcore", connected_val=False)
+        comp = CompositeTransport([a, b])
+        assert comp.connected is False
+
+    def test_disconnect_fans_to_all(self) -> None:
+        a = FakeChild("meshtastic")
+        b = FakeChild("meshcore")
+        comp = CompositeTransport([a, b])
+        comp.disconnect()
+        assert a.disconnect_calls == 1
+        assert b.disconnect_calls == 1
+
+    def test_disconnect_guarded(self) -> None:
+        """disconnect() must not raise even if a child raises."""
+        bad = FakeChild("meshtastic")
+        bad.disconnect = MagicMock(side_effect=RuntimeError("oops"))
+        good = FakeChild("meshcore")
+        comp = CompositeTransport([bad, good])
+        comp.disconnect()  # must not raise
+        assert good.disconnect_calls == 1
+
+
+# ---------------------------------------------------------------------------
+# send_message — broadcast
+# ---------------------------------------------------------------------------
+
+class TestBroadcast:
+    def test_broadcast_sends_to_all(self) -> None:
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("hello mesh")
+        assert result is True
+        assert len(mt.send_calls) == 1
+        assert len(mc.send_calls) == 1
+        assert mt.send_calls[0]["destination"] is None
+        assert mc.send_calls[0]["destination"] is None
+
+    def test_broadcast_skips_disconnected_child(self) -> None:
+        mt = FakeChild("meshtastic", connected_val=False)
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("hi")
+        assert result is True
+        assert len(mt.send_calls) == 0
+        assert len(mc.send_calls) == 1
+
+    def test_broadcast_true_if_at_least_one_ok(self) -> None:
+        mt = FakeChild("meshtastic")
+        mt.send_message = MagicMock(return_value=False)
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("test")
+        assert result is True
+
+
+# ---------------------------------------------------------------------------
+# send_message — hinted DM
+# ---------------------------------------------------------------------------
+
+class TestHintedDM:
+    def test_hinted_meshcore_only(self) -> None:
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("reply", destination="abc123", transport="meshcore")
+        assert result is True
+        assert len(mc.send_calls) == 1
+        assert len(mt.send_calls) == 0
+        assert mc.send_calls[0]["destination"] == "abc123"
+
+    def test_hinted_meshtastic_only(self) -> None:
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("reply", destination="!deadbeef", transport="meshtastic")
+        assert result is True
+        assert len(mt.send_calls) == 1
+        assert len(mc.send_calls) == 0
+
+    def test_unknown_hint_returns_false(self) -> None:
+        mt = FakeChild("meshtastic")
+        comp = CompositeTransport([mt])
+        result = comp.send_message("x", destination="y", transport="nonexistent")
+        assert result is False
+        assert len(mt.send_calls) == 0
+
+    def test_hinted_child_disconnected_returns_false(self) -> None:
+        mt = FakeChild("meshtastic", connected_val=False)
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("x", destination="y", transport="meshtastic")
+        assert result is False
+
+
+# ---------------------------------------------------------------------------
+# send_message — unhinted DM
+# ---------------------------------------------------------------------------
+
+class TestUnhintedDM:
+    def test_prefers_resolving_child(self) -> None:
+        mt = FakeChild("meshtastic", known_nodes={"!aabbccdd": "NodeA"})
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("hello", destination="!aabbccdd")
+        assert result is True
+        assert len(mt.send_calls) == 1
+        assert len(mc.send_calls) == 0
+
+    def test_fans_to_all_when_none_resolves(self) -> None:
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        result = comp.send_message("hello", destination="!unknown")
+        assert result is True
+        assert len(mt.send_calls) == 1
+        assert len(mc.send_calls) == 1
+
+
+# ---------------------------------------------------------------------------
+# set_message_callback — inbound forwarding and self-filter
+# ---------------------------------------------------------------------------
+
+class TestInboundCallback:
+    """Use asyncio.run() for each coroutine so the test always gets a fresh
+    event loop regardless of what earlier tests in the suite may have done.
+    FakeChild._simulate_inbound calls the wrapper directly (no
+    call_soon_threadsafe), so we don't actually need the loop we pass to
+    set_message_callback — we just need *some* AbstractEventLoop object to
+    satisfy the call signature.
+    """
+
+    @staticmethod
+    def _make_loop():
+        """Return a new event loop used only as a placeholder for the callback
+        registration (FakeChild doesn't schedule on it)."""
+        return asyncio.new_event_loop()
+
+    def test_inbound_forwarded_with_correct_transport(self) -> None:
+        mt = FakeChild("meshtastic", node_id="!aaaaaaaa")
+        mc = FakeChild("meshcore", node_id="!bbbbbbbb")
+        comp = CompositeTransport([mt, mc])
+
+        received: list[MeshMessage] = []
+
+        async def run():
+            async def cb(msg):
+                received.append(msg)
+
+            loop = self._make_loop()
+            comp.set_message_callback(cb, loop)
+            loop.close()
+
+            inbound = MeshMessage(
+                sender_id="!cccccccc",
+                sender_name="Other",
+                text="hello",
+                channel=0,
+                is_dm=True,
+                transport="meshcore",
+            )
+            await mc._simulate_inbound(inbound)
+
+        asyncio.run(run())
+        assert len(received) == 1
+        assert received[0].transport == "meshcore"
+
+    def test_self_filter_drops_own_message(self) -> None:
+        mt = FakeChild("meshtastic", node_id="!selfid1")
+        comp = CompositeTransport([mt])
+
+        received: list[MeshMessage] = []
+
+        async def run():
+            async def cb(msg):
+                received.append(msg)
+
+            loop = self._make_loop()
+            comp.set_message_callback(cb, loop)
+            loop.close()
+
+            # sender_id matches the child's own node ID → should be dropped
+            echo = MeshMessage(
+                sender_id="!selfid1",
+                sender_name="Self",
+                text="echo",
+                channel=0,
+                is_dm=False,
+            )
+            await mt._simulate_inbound(echo)
+
+        asyncio.run(run())
+        assert len(received) == 0
+
+    def test_non_self_not_dropped(self) -> None:
+        mt = FakeChild("meshtastic", node_id="!selfid1")
+        comp = CompositeTransport([mt])
+
+        received: list[MeshMessage] = []
+
+        async def run():
+            async def cb(msg):
+                received.append(msg)
+
+            loop = self._make_loop()
+            comp.set_message_callback(cb, loop)
+            loop.close()
+
+            msg = MeshMessage(
+                sender_id="!otherid",
+                sender_name="Friend",
+                text="hi",
+                channel=0,
+                is_dm=True,
+            )
+            await mt._simulate_inbound(msg)
+
+        asyncio.run(run())
+        assert len(received) == 1
+
+    def test_transport_tag_backfilled_when_missing(self) -> None:
+        """If msg.transport is empty/falsy the wrapper sets it to the child name."""
+        mc = FakeChild("meshcore", node_id="!mc00")
+        comp = CompositeTransport([mc])
+
+        received: list[MeshMessage] = []
+
+        async def run():
+            async def cb(msg):
+                received.append(msg)
+
+            loop = self._make_loop()
+            comp.set_message_callback(cb, loop)
+            loop.close()
+
+            msg = MeshMessage(
+                sender_id="!other",
+                sender_name="X",
+                text="ping",
+                channel=0,
+                is_dm=True,
+                transport="",  # missing tag
+            )
+            await mc._simulate_inbound(msg)
+
+        asyncio.run(run())
+        assert len(received) == 1
+        assert received[0].transport == "meshcore"
+
+
+# ---------------------------------------------------------------------------
+# Reply routing — responder seam
+# ---------------------------------------------------------------------------
+
+class TestReplyRouting:
+    """Verify that a meshcore-tagged inbound message causes send_message to be
+    called with transport='meshcore' via the Responder."""
+
+    def test_responder_threads_transport(self) -> None:
+        from meshai.responder import Responder
+        from meshai.config import ResponseConfig
+
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+
+        cfg = ResponseConfig(delay_min=0.0, delay_max=0.0)
+        responder = Responder(cfg, comp)
+
+        asyncio.run(
+            responder.send_response(
+                "pong",
+                destination="abc123",
+                channel=0,
+                transport="meshcore",
+            )
+        )
+
+        # Only meshcore child should have been called.
+        assert len(mc.send_calls) == 1
+        assert mc.send_calls[0]["destination"] == "abc123"
+        assert len(mt.send_calls) == 0
+
+
+# ---------------------------------------------------------------------------
+# meshtastic_child helper
+# ---------------------------------------------------------------------------
+
+class TestMeshtasticChild:
+    def test_returns_meshtastic_transport(self) -> None:
+        from meshai.connector import MeshtasticTransport
+        from meshai.transport.meshcore_transport import MeshCoreTransport
+
+        cfg = types.SimpleNamespace(
+            type="tcp",
+            tcp_host="127.0.0.1",
+            tcp_port=4403,
+            meshcore_host="127.0.0.1",
+            meshcore_port=5050,
+        )
+        mt = MeshtasticTransport(cfg)
+        mc = MeshCoreTransport(cfg)
+        comp = CompositeTransport([mt, mc])
+        child = comp.meshtastic_child()
+        assert child is mt
+
+    def test_returns_none_when_absent(self) -> None:
+        from meshai.transport.meshcore_transport import MeshCoreTransport
+
+        cfg = types.SimpleNamespace(
+            meshcore_host="127.0.0.1",
+            meshcore_port=5050,
+        )
+        mc = MeshCoreTransport(cfg)
+        comp = CompositeTransport([mc])
+        assert comp.meshtastic_child() is None
+
+
+# ---------------------------------------------------------------------------
+# _child_name helper
+# ---------------------------------------------------------------------------
+
+class TestChildName:
+    def test_uses_transport_name_attr(self) -> None:
+        child = FakeChild("foobar")
+        assert _child_name(child) == "foobar"
+
+    def test_derives_from_class_name(self) -> None:
+        class MyTransport:
+            pass
+
+        obj = MyTransport()
+        assert _child_name(obj) == "my"
+
+    def test_strips_transport_suffix(self) -> None:
+        class SomeTransport:
+            pass
+
+        assert _child_name(SomeTransport()) == "some"
+
+
+# ---------------------------------------------------------------------------
+# should_drop helper (unit test for routing logic)
+# ---------------------------------------------------------------------------
+
+class TestShouldDrop:
+    def test_drops_own_sender_id(self) -> None:
+        comp = CompositeTransport([FakeChild("x")])
+        msg = MeshMessage("!abc", "Self", "hi", 0, False)
+        assert comp._should_drop(msg, "!abc") is True
+
+    def test_keeps_other_sender(self) -> None:
+        comp = CompositeTransport([FakeChild("x")])
+        msg = MeshMessage("!xyz", "Other", "hi", 0, False)
+        assert comp._should_drop(msg, "!abc") is False
+
+    def test_none_node_id_no_drop(self) -> None:
+        comp = CompositeTransport([FakeChild("x")])
+        msg = MeshMessage("!abc", "X", "hi", 0, False)
+        assert comp._should_drop(msg, None) is False
+
+
+# ---------------------------------------------------------------------------
+# _resolve_child_for_hint (unit test for routing logic)
+# ---------------------------------------------------------------------------
+
+class TestResolveChildForHint:
+    def test_resolves_known_name(self) -> None:
+        mt = FakeChild("meshtastic")
+        mc = FakeChild("meshcore")
+        comp = CompositeTransport([mt, mc])
+        assert comp._resolve_child_for_hint("meshcore") is mc
+        assert comp._resolve_child_for_hint("meshtastic") is mt
+
+    def test_returns_none_for_unknown(self) -> None:
+        comp = CompositeTransport([FakeChild("meshtastic")])
+        assert comp._resolve_child_for_hint("lora") is None
diff --git a/work/tests/test_fire_digest_recency.py b/work/tests/test_fire_digest_recency.py
index b1a30ec..1c85f00 100644
--- a/work/tests/test_fire_digest_recency.py
+++ b/work/tests/test_fire_digest_recency.py
@@ -2,7 +2,7 @@
 
 Validates recency ordering, contained/tombstoned exclusion, the
 "Name in County Co, ST" line format, correct tail count after budget
-trimming, singular/plural grammar, and the 200-byte LoRa budget.
+trimming, singular/plural grammar, and the 140-byte universal mesh budget.
 """
 from __future__ import annotations
 
@@ -62,28 +62,34 @@ class TestFireDigestRecency:
     """Deterministic fire digest renderer tests."""
 
     def test_top_2_listed_in_recency_order(self):
-        """The 2 most recent active fires are listed in order."""
+        """The most recent active fire is listed first.
+
+        With the universal 140-byte budget, the header + 1 fire line + tail
+        fits; the second fire line does not.  Alpha (most recent) must appear;
+        Bravo belongs to the tail count.
+        """
         conn = get_db()
         now = int(time.time())
         _seed_scenario(conn)
         from meshai.notifications.scheduled.fire_digest import render_digest
         wire, source = asyncio.run(render_digest(now=now))
         assert source == "deterministic"
+        # Most recent fire must appear in the wire body.
         assert "Alpha Fire" in wire
-        assert "Bravo Fire" in wire
-        pos_alpha = wire.index("Alpha Fire")
-        pos_bravo = wire.index("Bravo Fire")
-        assert pos_alpha < pos_bravo, "Alpha Fire (most recent) should appear before Bravo Fire"
+        # Bravo does not fit within 140 bytes alongside the header + tail.
+        assert "Bravo Fire" not in wire
 
     def test_line_format_name_in_county_co_state(self):
-        """Fire lines render as 'Name in County Co, ST'."""
+        """Fire lines render as 'Name in County Co, ST'.
+
+        Only the most recent fire fits within the 140-byte budget.
+        """
         conn = get_db()
         now = int(time.time())
         _seed_scenario(conn)
         from meshai.notifications.scheduled.fire_digest import render_digest
         wire, _ = asyncio.run(render_digest(now=now))
         assert "Alpha Fire in Ada Co, ID" in wire
-        assert "Bravo Fire in Boise Co, ID" in wire
 
     def test_missing_county_renders_name_in_state(self):
         """Fire with no county renders as 'Name in ST'."""
@@ -123,17 +129,25 @@ class TestFireDigestRecency:
         """N == 1 renders 'There is 1 additional wildfire.'."""
         conn = get_db()
         now = int(time.time())
-        _seed_scenario(conn)
+        # 3 fires; short names + no county → lines are short enough that
+        # 2 fit within the 140-byte budget, leaving 1 in the tail.
+        for irwin, name, offset in [
+            ("SG-01", "A", 3600),
+            ("SG-02", "B", 7200),
+            ("SG-03", "C", 10800),
+        ]:
+            _seed_fire(conn, irwin_id=irwin, name=name, acres=100,
+                       contained=None, last_event_at=now - offset,
+                       county=None, state="ID")
         from meshai.notifications.scheduled.fire_digest import render_digest
         wire, _ = asyncio.run(render_digest(now=now))
-        # 3 active total, 2 shown, 1 remaining
+        # 3 active total, 2 shown (budget), 1 remaining
         assert "There is 1 additional wildfire. DM me for the full list." in wire
 
     def test_n_plural_grammar(self):
         """N > 1 renders 'There are N additional wildfires.'."""
         conn = get_db()
         now = int(time.time())
-        day = 86400
         for i in range(4):
             _seed_fire(conn, irwin_id=f"PL-{i:02d}", name=f"Fire {i}",
                        acres=100 + i, contained=None,
@@ -141,8 +155,9 @@ class TestFireDigestRecency:
                        county="Ada", state="ID")
         from meshai.notifications.scheduled.fire_digest import render_digest
         wire, _ = asyncio.run(render_digest(now=now))
-        # 4 active, 2 shown, 2 remaining
-        assert "There are 2 additional wildfires. DM me for the full list." in wire
+        # 4 active; "Fire N in Ada Co, ID" lines total ~142 bytes for 2 lines +
+        # header + tail → only 1 line fits in the 140-byte budget → 3 remaining.
+        assert "There are 3 additional wildfires. DM me for the full list." in wire
 
     def test_n_zero_omits_sentence(self):
         """When N == 0, the tail sentence is omitted entirely."""
@@ -186,7 +201,7 @@ class TestFireDigestRecency:
         wire, source = asyncio.run(render_digest(now=now))
         assert source == "deterministic"
         byte_len = len(wire.encode("utf-8"))
-        assert byte_len <= 200, f"Wire is {byte_len} bytes, exceeds 200"
+        assert byte_len <= 140, f"Wire is {byte_len} bytes, exceeds 140"
         # If both long lines fit, remaining = 1; if only one fits, remaining = 2
         # Either way the tail count must match (total - shown)
         lines = wire.split("\n")
@@ -205,7 +220,7 @@ class TestFireDigestRecency:
         from meshai.notifications.scheduled.fire_digest import render_digest
         wire, _ = asyncio.run(render_digest(now=now))
         byte_len = len(wire.encode("utf-8"))
-        assert byte_len <= 200, f"Digest is {byte_len} bytes, exceeds 200-byte budget"
+        assert byte_len <= 140, f"Digest is {byte_len} bytes, exceeds 140-byte budget"
 
     def test_no_fires_returns_empty(self):
         """No active fires -> empty wire, 'no_fires' source."""
diff --git a/work/tests/test_fire_tracker_phase4.py b/work/tests/test_fire_tracker_phase4.py
index ddcbbd5..0408b9b 100644
--- a/work/tests/test_fire_tracker_phase4.py
+++ b/work/tests/test_fire_tracker_phase4.py
@@ -76,7 +76,7 @@ def test_adapter_config_seeds_digest_keys():
     assert rows[("fires", "digest_enabled")] == "true"
     assert rows[("fires", "digest_schedule")] == '["06:00", "18:00"]'
     assert rows[("fires", "digest_timezone")] == '"America/Boise"'
-    assert rows[("fires", "digest_max_chars")] == "200"
+    assert rows[("fires", "digest_max_chars")] == "140"
 
 
 # ===========================================================================
@@ -107,7 +107,7 @@ def test_render_digest_terse_fallback_when_no_llm():
     assert source == "deterministic"
     assert wire
     assert "Cache Peak" in wire
-    assert len(wire) <= 200
+    assert len(wire) <= 140
 
 
 def test_render_digest_uses_llm_when_available():
diff --git a/work/tests/test_transport_abstraction.py b/work/tests/test_transport_abstraction.py
index 91f8234..962d5f8 100644
--- a/work/tests/test_transport_abstraction.py
+++ b/work/tests/test_transport_abstraction.py
@@ -79,10 +79,13 @@ class TestBuildTransport:
         assert isinstance(transport, MeshCoreTransport)
         assert isinstance(transport, MeshTransport)
 
-    def test_both_raises_not_implemented(self):
+    def test_both_returns_composite(self):
+        """Phase 4: transport='both' now returns a CompositeTransport (seam filled)."""
+        from meshai.transport.composite_transport import CompositeTransport
         cfg = self._config_with("both")
-        with pytest.raises(NotImplementedError):
-            build_transport(cfg)
+        t = build_transport(cfg)
+        assert isinstance(t, CompositeTransport)
+        assert len(t.children) == 2
 
     def test_unknown_transport_raises_value_error(self):
         cfg = self._config_with("unknown_transport_xyz")
diff --git a/work/tests/test_uniform_sizing.py b/work/tests/test_uniform_sizing.py
index e2f3a92..7a48414 100644
--- a/work/tests/test_uniform_sizing.py
+++ b/work/tests/test_uniform_sizing.py
@@ -1,8 +1,8 @@
-"""Tests for Phase 3 uniform mesh-packet sizing.
+"""Tests for uniform mesh-packet sizing.
 
-Verifies that every size-sensitive code path uses the active transport's
-max_chars rather than a hardcoded 200.  Meshtastic default stays 200
-(byte-identical); MeshCore uses 140.
+Verifies that every size-sensitive code path uses the single universal
+mesh_max_chars constant (140 = MeshCore LCD).  All transports — Meshtastic,
+MeshCore, and the composite — report the same fixed value.
 """
 
 import sys
@@ -127,59 +127,90 @@ def _minimal_config():
 # ---------------------------------------------------------------------------
 
 class TestTransportMaxChars:
-    def test_meshtastic_default_is_200(self):
+    def test_meshtastic_default_is_140(self):
+        """Meshtastic now uses the universal mesh_max_chars constant (140)."""
         cfg = _mt_config()
         t = MeshtasticTransport(cfg)
-        assert t.max_chars == 200
-
-    def test_meshtastic_respects_config_field(self):
-        cfg = _mt_config(meshtastic_max_chars=160)
-        t = MeshtasticTransport(cfg)
-        assert t.max_chars == 160
+        assert t.max_chars == 140
 
     def test_meshcore_default_is_140(self):
         cfg = _mc_config()
         t = MeshCoreTransport(cfg)
         assert t.max_chars == 140
 
-    def test_meshcore_respects_config_field(self):
-        cfg = _mc_config(meshcore_max_chars=120)
-        t = MeshCoreTransport(cfg)
-        assert t.max_chars == 120
-
     def test_build_transport_meshtastic_max_chars(self):
+        """build_transport(meshtastic) → 140 (universal constant)."""
         t = build_transport(_mt_config())
-        assert t.max_chars == 200
+        assert t.max_chars == 140
 
     def test_build_transport_meshcore_max_chars(self):
         t = build_transport(_mc_config())
         assert t.max_chars == 140
 
+    def test_mesh_max_chars_config_field_governs_all(self):
+        """A non-default mesh_max_chars propagates to both transport types."""
+        cfg_mt = _mt_config(mesh_max_chars=120)
+        assert MeshtasticTransport(cfg_mt).max_chars == 120
+
+        cfg_mc = _mc_config(mesh_max_chars=120)
+        assert MeshCoreTransport(cfg_mc).max_chars == 120
+
+    def test_all_three_transports_and_composite_report_140(self):
+        """Sanity: all transports + CompositeTransport report mesh_max_chars=140.
+
+        This is the canonical single-budget guarantee: Meshtastic, MeshCore,
+        and the composite (transport=both) all resolve to the same constant.
+        """
+        from meshai.config import ConnectionConfig
+        from meshai.transport.composite_transport import CompositeTransport
+
+        # Meshtastic
+        assert MeshtasticTransport(_mt_config()).max_chars == 140
+
+        # MeshCore
+        assert MeshCoreTransport(_mc_config()).max_chars == 140
+
+        # Composite (built via factory with transport="both")
+        cfg_both = ConnectionConfig(
+            transport="both",
+            type="tcp",
+            tcp_host="127.0.0.1",
+            tcp_port=4403,
+            meshcore_host="127.0.0.1",
+            meshcore_port=5050,
+            meshcore_channel_index=0,
+        )
+        comp = build_transport(cfg_both)
+        assert isinstance(comp, CompositeTransport)
+        assert comp.max_chars == 140
+
 
 # ---------------------------------------------------------------------------
 # 2. MeshRenderer char_limit propagation via channels
 # ---------------------------------------------------------------------------
 
 class TestChannelRendererBudget:
-    def test_broadcast_channel_with_meshcore_connector_uses_140(self):
+    def test_broadcast_channel_propagates_connector_max_chars(self):
+        """Channel renderer inherits max_chars from whatever the connector reports."""
         conn = _fake_connector(140)
         ch = MeshBroadcastChannel(connector=conn, channel_index=0)
         assert ch._renderer._limit == 140
 
-    def test_broadcast_channel_with_meshtastic_connector_uses_200(self):
-        conn = _fake_connector(200)
+    def test_broadcast_channel_with_meshtastic_connector_uses_140(self):
+        """After the universal budget refactor, a meshtastic connector reports 140."""
+        conn = _fake_connector(140)
         ch = MeshBroadcastChannel(connector=conn, channel_index=0)
-        assert ch._renderer._limit == 200
+        assert ch._renderer._limit == 140
 
-    def test_dm_channel_with_meshcore_connector_uses_140(self):
+    def test_dm_channel_propagates_connector_max_chars(self):
         conn = _fake_connector(140)
         ch = MeshDMChannel(connector=conn, node_ids=["!aabbccdd"])
         assert ch._renderer._limit == 140
 
-    def test_dm_channel_with_meshtastic_connector_uses_200(self):
-        conn = _fake_connector(200)
+    def test_dm_channel_with_meshtastic_connector_uses_140(self):
+        conn = _fake_connector(140)
         ch = MeshDMChannel(connector=conn, node_ids=["!aabbccdd"])
-        assert ch._renderer._limit == 200
+        assert ch._renderer._limit == 140
 
 
 # ---------------------------------------------------------------------------
@@ -187,11 +218,12 @@ class TestChannelRendererBudget:
 # ---------------------------------------------------------------------------
 
 class TestBuildPipelineMeshCharLimit:
-    def test_no_connector_uses_200(self):
+    def test_no_connector_uses_140(self):
+        """Without a connector the pipeline falls back to the universal constant (140)."""
         cfg = _minimal_config()
         bus = build_pipeline(cfg, llm_backend=None, connector=None)
         acc = bus._pipeline_components["accumulator"]
-        assert acc._mesh_char_limit == 200
+        assert acc._mesh_char_limit == 140
 
     def test_meshcore_connector_uses_140(self):
         cfg = _minimal_config()
@@ -200,12 +232,13 @@ class TestBuildPipelineMeshCharLimit:
         acc = bus._pipeline_components["accumulator"]
         assert acc._mesh_char_limit == 140
 
-    def test_meshtastic_connector_uses_200(self):
+    def test_meshtastic_connector_uses_140(self):
+        """After universal budget, a meshtastic connector also reports 140."""
         cfg = _minimal_config()
-        conn = _fake_connector(200)
+        conn = _fake_connector(140)
         bus = build_pipeline(cfg, llm_backend=None, connector=conn)
         acc = bus._pipeline_components["accumulator"]
-        assert acc._mesh_char_limit == 200
+        assert acc._mesh_char_limit == 140
 
 
 # ---------------------------------------------------------------------------