A channel can be given a region its messages are flooded into

The firmware has no per-channel scope, so a channel send is now a window
on the companion's session scope: set it, send, restore it, under a
transmit lock every flood-capable send also takes (DMs, adverts, admin
commands, logins, regions/neighbour requests), so nothing interleaves.
Ack waits stay outside the lock. A scope the radio can't take stops the
message (FloodScopeError naming the firmware it needs) rather than sending
it unscoped or under the default; * checks the firmware version, since
before 1.16 the override would quietly fall back to the default scope.
A failed restore is repaired by the next transmission.

Channel scopes are set from the channel page's Send scope picker (known
regions with their carriers, or a typed name) and from the CLI (channels
scope, chat send --scope). The chat's title carries the scope as a status
atom, ^R/Resend sends the newest scoped message again unscoped after an
amber confirm, and the scope each sent message went under is recorded
(messages.scope, schema v17) so the paths dialog can say what it went
under and when no known repeater carries that region. Device config now
teaches the region store the default scope it reads or sets.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

meshterm/core/connection.py

@@ -17,9 +17,11 @@ from __future__ import annotations
 import asyncio
 import logging
 import random
+import re
 import sys
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable
+from collections.abc import AsyncIterator, Callable, Iterable
+from contextlib import asynccontextmanager
 from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -48,6 +50,7 @@ from .regions import normalize as normalize_region
 from .regions import parse_region_list, region_key
 from .regions import validate as validate_region
 from .tracing import path_hash_flags, trace_timeout
+from .transmit_lock import TransmitLock
 
 if TYPE_CHECKING:
     from .config import SpiWiring
@@ -225,6 +228,89 @@ class DeviceAuthenticationError(DeviceCommandError):
     """
 
 
+class FloodScopeError(DeviceCommandError):
+    """The companion could not send under the scope a channel send asked for — so it didn't.
+
+    A scoped send is set-scope, send, restore (see :meth:`Device.send_channel_in_scope`),
+    and when the first step fails the message is **not** sent at all: going out unscoped
+    instead would reach every repeater the reader meant to keep it from, and going out
+    under the device's default scope would reach a region they never picked. Either would
+    be a silent substitute for what was asked. The message names the firmware each kind of
+    scope needs, since an old companion is by far the likeliest reason.
+
+    Attributes:
+        scope: The scope that was asked for — a region name, or ``*`` for unscoped.
+    """
+
+    def __init__(self, scope: str, reason: str) -> None:
+        """Explain which scope could not be set and why.
+
+        Args:
+            scope: The region name, or :data:`~meshterm.core.regions.WILDCARD`.
+            reason: The sentence to show (it already says nothing was sent).
+        """
+        super().__init__(reason)
+        self.scope = scope
+
+
+#: The companion firmware that first takes a session scope (``CMD_SET_FLOOD_SCOPE``), and
+#: the one that first takes the explicit-unscoped override on it (``*``, sent as flag byte
+#: ``0x01``) — before that release the flag frame is not understood, so an "unscoped" send
+#: would fall back to the default scope instead.
+SCOPE_FIRMWARE = (1, 10)
+UNSCOPED_FIRMWARE = (1, 16)
+
+#: The firmware that first has a persisted default scope (``CMD_SET_DEFAULT_FLOOD_SCOPE``).
+#: On anything older a plain flood is unscoped by construction.
+DEFAULT_SCOPE_FIRMWARE = (1, 15)
+
+
+def firmware_version(info: dict | None) -> tuple[int, int, int] | None:
+    """The companion's release as a comparable tuple, read from its device-query ``ver``.
+
+    Firmware reports its build as a string — ``v1.15.0``, ``1.16.0-dev``, a vendor's own
+    suffix — and only the leading ``major.minor[.patch]`` means anything here.
+
+    Args:
+        info: A :meth:`Device.get_device_info` payload.
+
+    Returns:
+        ``(major, minor, patch)``, or ``None`` where nothing version-shaped was reported
+        (the simulator, or firmware predating the device query) — a caller then trusts the
+        command itself to say whether it is understood.
+    """
+    match = _VERSION.search(str((info or {}).get("ver") or ""))
+    if match is None:
+        return None
+    return int(match.group(1)), int(match.group(2)), int(match.group(3) or 0)
+
+
+#: The leading ``major.minor[.patch]`` of a firmware build string.
+_VERSION = re.compile(r"(\d+)\.(\d+)(?:\.(\d+))?")
+
+
+def _version_text(version: tuple[int, ...]) -> str:
+    """``(1, 16)`` as ``1.16`` for a sentence."""
+    return ".".join(str(part) for part in version[:2])
+
+
+def _scope_refusal(scope: str, exc: BaseException) -> str:
+    """The sentence for a companion that refused to set a send scope.
+
+    Args:
+        scope: The region name, or ``*``.
+        exc: What the command raised, kept in the sentence for whoever is debugging it.
+
+    Returns:
+        A sentence naming the firmware the scope needs, and that nothing was sent.
+    """
+    if scope == REGION_WILDCARD:
+        need = f"sending unscoped needs firmware {_version_text(UNSCOPED_FIRMWARE)} or newer"
+    else:
+        need = f"region scopes need firmware {_version_text(SCOPE_FIRMWARE)} or newer"
+    return f"The radio refused scope {scope} ({exc}) — {need}. Nothing was sent."
+
+
 #: ``ERR_CODE_NOT_FOUND`` — the companion has no entry matching what a command addressed.
 _ERR_NOT_FOUND = 2
 
@@ -813,6 +899,140 @@ class Device(ABC):
             DeviceCommandError: If the companion rejected the send.
         """
 
+    # -- transmitting under a scope -----------------------------------------------------
+
+    @property
+    def transmit_lock(self) -> TransmitLock:
+        """The lock every transmission holds while it hands its frame to the companion.
+
+        Created on first use rather than in ``__init__`` so an implementation need not
+        remember to chain up for it. See :mod:`~meshterm.core.transmit_lock` for why it
+        exists: a channel scope is a window on the companion's one session scope, and
+        nothing else may be sent through that window.
+        """
+        lock = getattr(self, "_transmit_lock", None)
+        if lock is None:
+            lock = TransmitLock()
+            self._transmit_lock = lock
+        return lock
+
+    @asynccontextmanager
+    async def transmitting(self) -> AsyncIterator[None]:
+        """Hold the transmit lock for one hand-over, repairing a scope a failed restore left.
+
+        Every send an implementation makes that could go out as a flood takes this around
+        the command that transmits — only that command: an ack or reply *wait* afterwards
+        is not a transmission and must not hold a scoped channel send up behind it.
+
+        A scoped send whose restore failed (see :meth:`send_channel_in_scope`) leaves the
+        companion's session scope pointing at a channel's region, where every later flood
+        would follow it. The first transmission to come through here afterwards puts it
+        back before it sends anything — and if the companion still won't take the reset,
+        the send goes ahead under the scope it had rather than not at all, logged, since
+        refusing every transmission from then on would be the worse failure.
+        """
+        async with self.transmit_lock.held():
+            leaked = getattr(self, "_scope_leaked", None)
+            if leaked is not None and not getattr(self, "_scope_active", False):
+                try:
+                    await self.set_flood_scope(None)
+                except Exception as exc:  # noqa: BLE001 - see the docstring: send regardless
+                    _log.warning("send scope %r is still set on the radio: %s", leaked, exc)
+                else:
+                    self._scope_leaked = None
+            yield
+
+    async def send_channel_in_scope(self, index: int, text: str, scope: str | None) -> None:
+        """Broadcast on a channel under a given scope, and leave the session scope as found.
+
+        The one way MeshTerm sends a channel message. ``None`` is the plain send, going out
+        under whatever the companion's default scope is; a region name or ``*`` is the
+        three-step window the firmware makes a per-channel scope out of — set the session
+        scope, send, restore it (``None``, back to the default) — all under
+        :attr:`transmit_lock`, so no other flood can go out between the steps.
+
+        What is refused is refused *before* anything is sent (:class:`FloodScopeError`):
+        a scoped message never quietly goes out unscoped, or under the default scope.
+
+        * A region name needs firmware :data:`SCOPE_FIRMWARE`; older firmware answers the
+          command with an error, which is taken as the refusal it is.
+        * ``*`` needs firmware :data:`UNSCOPED_FIRMWARE` to override a default scope.
+          Older firmware can still send unscoped when there is no default to override —
+          anything before :data:`DEFAULT_SCOPE_FIRMWARE` has none at all, and later ones
+          may have none set — so then the plain send *is* the unscoped one and goes out;
+          only a default that is actually set is refused.
+
+        Args:
+            index: Zero-based channel slot to transmit on.
+            text: The message body.
+            scope: A region name, :data:`~meshterm.core.regions.WILDCARD` for unscoped, or
+                ``None`` for the device's default.
+
+        Raises:
+            FloodScopeError: If the companion can't send under ``scope``; nothing was sent.
+            ~meshterm.core.regions.RegionNameError: If ``scope`` is not a region name the
+                firmware could hold.
+            DeviceCommandError: If the send itself was rejected.
+        """
+        bare = normalize_region(scope) if scope else ""
+        if bare and bare != REGION_WILDCARD:
+            bare = validate_region(bare)
+        async with self.transmitting():
+            if not bare:
+                await self.send_channel_message(index, text)
+                return
+            if bare == REGION_WILDCARD and await self._unscoped_is_plain():
+                await self.send_channel_message(index, text)
+                return
+            try:
+                await self.set_flood_scope(bare)
+            except DeviceCommandError as exc:
+                raise FloodScopeError(bare, _scope_refusal(bare, exc)) from exc
+            self._scope_active = True
+            try:
+                await self.send_channel_message(index, text)
+            finally:
+                self._scope_active = False
+                try:
+                    await self.set_flood_scope(None)
+                except Exception as exc:  # noqa: BLE001 - the message went; repair next time
+                    _log.warning("couldn't clear send scope %r after a channel send: %s", bare, exc)
+                    self._scope_leaked = bare
+
+    async def _unscoped_is_plain(self) -> bool:
+        """Whether a plain flood is already unscoped, so ``*`` needs no override — or refuse.
+
+        Only firmware too old to take the override is asked anything; firmware that reports
+        no version is trusted to answer the override command itself.
+
+        Returns:
+            ``True`` when the plain send is unscoped; ``False`` when the override is
+            available and should be used.
+
+        Raises:
+            FloodScopeError: When a default scope is set and the firmware cannot override it.
+        """
+        try:
+            version = firmware_version(await self.get_device_info())
+        except Exception:  # noqa: BLE001 - an unreadable version is an unknown one
+            version = None
+        if version is None or version >= UNSCOPED_FIRMWARE:
+            return False
+        if version < DEFAULT_SCOPE_FIRMWARE:
+            return True
+        try:
+            default = await self.get_default_flood_scope()
+        except Exception:  # noqa: BLE001 - can't see the default: can't promise unscoped
+            default = "?"
+        if not normalize_region(default or ""):
+            return True
+        raise FloodScopeError(
+            REGION_WILDCARD,
+            f"This radio floods everything under its default scope {default}, and only "
+            f"firmware {_version_text(UNSCOPED_FIRMWARE)} or newer can send one message "
+            "unscoped over it. Nothing was sent.",
+        )
+
     # -- configuration: extra reads ---------------------------------------------
 
     @abstractmethod
@@ -2219,7 +2439,8 @@ class MeshCoreDevice(Device):
         budget = max(trace_timeout(hops), trace_timeout(0))
         started = loop.time()
         try:
-            sent = await mc.commands.send_login_sync(pub, password)
+            async with self.transmitting():
+                sent = await mc.commands.send_login_sync(pub, password)
             # The budget times the *node's* answer, so it is spent from the moment the
             # request is on the air — not from the moment we began queuing it. Everything
             # before that belongs to the companion, and it can be most of a minute:
@@ -2308,7 +2529,8 @@ class MeshCoreDevice(Device):
 
         mc = self._require()
         pub = self._node_pubkey(node)
-        sent = await mc.commands.send_cmd(pub, cmd)
+        async with self.transmitting():
+            sent = await mc.commands.send_cmd(pub, cmd)
         if sent is not None and getattr(sent, "is_error", lambda: False)():
             raise DeviceCommandError(
                 f"couldn't send admin command {cmd!r} to {node.name}: {reject_reason(sent)}"
@@ -2340,7 +2562,8 @@ class MeshCoreDevice(Device):
         # The library pages through the table (one binary request per ~25 entries) and
         # concatenates; ``min_timeout`` keeps slow multi-hop replies from being cut off
         # at the companion's optimistic suggested timeout.
-        result = await mc.commands.fetch_all_neighbours(pub, min_timeout=20)
+        async with self.transmitting():
+            result = await mc.commands.fetch_all_neighbours(pub, min_timeout=20)
         if result is None:
             raise DeviceCommandError(
                 f"{node.name!r} did not answer the neighbour request. Firmware ignores "
@@ -2373,7 +2596,8 @@ class MeshCoreDevice(Device):
         transmit_gate.mark()
         # ``min_timeout`` for the reason the neighbour request carries one: the companion's
         # suggested timeout is optimistic for anything but an adjacent node.
-        text = await mc.commands.req_regions_sync(pub, min_timeout=10)
+        async with self.transmitting():
+            text = await mc.commands.req_regions_sync(pub, min_timeout=10)
         if text is None:
             raise DeviceCommandError(
                 f"{node.name!r} did not answer the regions request. A repeater answers it "
@@ -2720,7 +2944,9 @@ class MeshCoreDevice(Device):
 
         mc = self._require()
         pub = self._node_pubkey(contact)
-        result = await mc.commands.send_msg(pub, text)
+        # Only the hand-over holds the transmit lock; the ack wait below is not a send.
+        async with self.transmitting():
+            result = await mc.commands.send_msg(pub, text)
         if result is None or getattr(result, "is_error", lambda: False)():
             # A recipient the firmware has no entry for is the one rejection with an obvious
             # fix, so it gets its own class for the chat screen to offer that fix on; see
@@ -2746,7 +2972,8 @@ class MeshCoreDevice(Device):
         self, index: int, text: str
     ) -> None:
         transmit_gate.mark()
-        self._ok(await self._require().commands.send_chan_msg(index, text))
+        async with self.transmitting():
+            self._ok(await self._require().commands.send_chan_msg(index, text))
 
     @staticmethod
     def _ok(event):  # type: ignore[no-untyped-def]
@@ -3020,7 +3247,8 @@ class MeshCoreDevice(Device):
 
     async def send_advert(self, flood: bool = False) -> None:  # noqa: D102
         transmit_gate.mark(flood_advert=flood)
-        self._ok(await self._require().commands.send_advert(flood))
+        async with self.transmitting():
+            self._ok(await self._require().commands.send_advert(flood))
 
     async def reboot(self) -> None:  # noqa: D102 - inherited docstring
         await self._require().commands.reboot()  # device reboots; no OK reply expected
@@ -3176,6 +3404,8 @@ class MockDevice(Device):
         self._flood_scope = ""
         #: The session send scope (``""`` follows the default, ``"*"`` is forced unscoped).
         self._send_scope = ""
+        #: Every channel message handed over: ``(slot, text, session scope at the time)``.
+        self.sent_channel: list[tuple[int, str, str]] = []
         # Simulated clock skew (seconds behind the host), so the sync-clock flow has a
         # visible drift to correct until set_time is called.
         self._clock_offset: int | None = -125
@@ -3247,7 +3477,8 @@ class MockDevice(Device):
         self, contact: Contact, text: str
     ) -> Ack | None:
         transmit_gate.mark()
-        await asyncio.sleep(0)
+        async with self.transmitting():
+            await asyncio.sleep(0)
         # Firmware can only address a contact it holds, so a recipient this simulated device
         # doesn't have is refused exactly as hardware refuses one — which is what makes the
         # chat screen's "add it back and send" offer walkable on the simulator.
@@ -3261,7 +3492,11 @@ class MockDevice(Device):
         self, index: int, text: str
     ) -> None:
         transmit_gate.mark()
-        await asyncio.sleep(0)
+        async with self.transmitting():
+            await asyncio.sleep(0)
+            #: What each channel message went out under — the session scope at the moment
+            #: it was handed over (``""`` the default, ``"*"`` forced unscoped).
+            self.sent_channel.append((index, text, self._send_scope))
 
     async def admin_login(self, node: Contact, password: str) -> LoginResult:  # noqa: D102
         await asyncio.sleep(0)
@@ -3589,7 +3824,8 @@ class MockDevice(Device):
 
     async def send_advert(self, flood: bool = False) -> None:  # noqa: D102
         transmit_gate.mark(flood_advert=flood)
-        await asyncio.sleep(0)
+        async with self.transmitting():
+            await asyncio.sleep(0)
 
     async def reboot(self) -> None:  # noqa: D102 - inherited docstring
         await asyncio.sleep(0)

meshterm/core/models.py

@@ -405,6 +405,10 @@ class ChatMessage:
         created_at: When the message was sent or received.
         row_id: The ``messages`` table primary key once persisted, used to update an
             outbound message's delivery state in place on retry; ``None`` until stored.
+        scope: What an outbound channel message was flooded under: the bare region name,
+            :data:`~meshterm.core.regions.WILDCARD` (``*``) for unscoped, or ``None`` where
+            it isn't known — every inbound or direct message, and anything sent before the
+            scope was recorded.
     """
 
     text: str
@@ -418,6 +422,7 @@ class ChatMessage:
     acked: bool | None = None
     created_at: datetime = field(default_factory=utcnow)
     row_id: int | None = None
+    scope: str | None = None
 
     @property
     def key(self) -> str:

meshterm/core/transmit_lock.py

@@ -0,0 +1,81 @@
+# SPDX-License-Identifier: Apache-2.0
+"""The transmit lock: what keeps a channel's scope from leaking onto someone else's flood.
+
+The firmware has no per-channel scope. What it has is a *session* scope — one setting on
+the companion that every flood it sends goes out under, channel messages, flood DMs,
+adverts and requests alike (see :meth:`~meshterm.core.connection.Device.set_flood_scope`).
+A channel scope is therefore three commands, not one: set the session scope, send the
+channel message, set it back. Anything else the app transmits *between* the first and the
+third goes out under that channel's region too — a courier retry, a DM typed in another
+screen, the weekly advert — and would be relayed only by the repeaters carrying a region
+its sender never chose.
+
+So every transmission MeshTerm asks of the companion holds this lock for the moment it
+takes to hand the frame over, and a scoped channel send holds it across all three commands.
+Nothing can be interleaved into the window, and the window is as short as three round
+trips; an acknowledgement *wait* is not a transmission and is done outside the lock, so a
+DM waiting eight seconds for its ack never holds a channel message up behind it.
+
+The lock is **re-entrant per task**: the scoped send takes it, then calls the ordinary
+channel send, which takes it again on its own account — and must not wait for itself.
+
+One leak is out of any client's reach: a DM *received* inside the window is acknowledged
+by the firmware on its own, and that ack is a flood under whatever the session scope is
+at that instant. The window being three round trips long is the whole of the mitigation,
+and it is what the official apps live with too.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+
+class TransmitLock:
+    """A task-re-entrant :class:`asyncio.Lock` for "something is going on the air now".
+
+    Interact through :meth:`held`, an async context manager. A task already holding the
+    lock enters again at once; any other task waits for the holder to leave.
+    """
+
+    def __init__(self) -> None:
+        """Start unheld."""
+        self._lock = asyncio.Lock()
+        self._owner: asyncio.Task | None = None
+        self._depth = 0
+
+    @property
+    def locked(self) -> bool:
+        """Whether any task holds the lock right now."""
+        return self._lock.locked()
+
+    def held_by_me(self) -> bool:
+        """Whether the calling task is the one holding the lock."""
+        try:
+            task = asyncio.current_task()
+        except RuntimeError:  # no running loop: nobody here holds anything
+            return False
+        return task is not None and self._owner is task
+
+    @asynccontextmanager
+    async def held(self) -> AsyncIterator[None]:
+        """Hold the lock for the body; re-enter at once if this task already holds it."""
+        if self.held_by_me():
+            self._depth += 1
+            try:
+                yield
+            finally:
+                self._depth -= 1
+            return
+        async with self._lock:
+            self._owner = asyncio.current_task()
+            self._depth = 1
+            try:
+                yield
+            finally:
+                self._owner = None
+                self._depth = 0
+
+
+__all__ = ["TransmitLock"]

meshterm/persistence/db.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import sqlite3
 from pathlib import Path
 
-SCHEMA_VERSION = 16
+SCHEMA_VERSION = 17
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_meta (
@@ -161,7 +161,9 @@ CREATE TABLE IF NOT EXISTS messages (
     text        TEXT    NOT NULL,
     snr         REAL,
     acked       INTEGER,
-    created_at  TEXT    NOT NULL
+    created_at  TEXT    NOT NULL,
+    scope       TEXT              -- outbound floods: the region it was sent under, '*' for
+                                  --   unscoped, NULL where it was never known
 );
 
 CREATE INDEX IF NOT EXISTS idx_traces_run ON traces(run_id);
@@ -348,3 +350,16 @@ def _migrate(conn: sqlite3.Connection) -> None:
         # raw frames they arrived in were never persisted, so their scope is lost.
         conn.execute("ALTER TABLE observations ADD COLUMN transport_code TEXT")
         conn.execute("ALTER TABLE observations ADD COLUMN scope_body TEXT")
+    if "scope" not in message_cols:
+        # v16 -> v17: the scope a message we sent went out under. A channel can be given a
+        # region its messages are flooded into (a scope is a *setting on the companion*,
+        # made just before the send — see Device.send_channel_in_scope), and the scope can
+        # change between one message and the next, or be dropped for one resend. So the
+        # transcript cannot work out afterwards what a message went out under from what the
+        # channel's scope is *now*; it is kept with the message, the moment it is sent.
+        # Its first reader is the message paths dialog: a scoped message nothing relayed is
+        # most often one no repeater in earshot carries the region of, and saying so needs
+        # the region. Stored as the bare name, '*' for a flood sent unscoped, and NULL for
+        # anything else — inbound messages, direct messages, and every row sent before this
+        # column existed, whose scope was never known and cannot be recovered.
+        conn.execute("ALTER TABLE messages ADD COLUMN scope TEXT")

meshterm/persistence/repository.py

@@ -1921,7 +1921,7 @@ class Repository:
         cur = self._conn.execute(
             "INSERT INTO messages "
             "(run_id, outbound, is_channel, channel_id, channel_idx, peer, peer_name, text, "
-            "snr, acked, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            "snr, acked, created_at, scope) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
             (
                 run_id,
                 int(msg.outbound),
@@ -1934,6 +1934,7 @@ class Repository:
                 msg.snr,
                 None if msg.acked is None else int(msg.acked),
                 msg.created_at.isoformat(),
+                msg.scope,
             ),
         )
         self._conn.commit()
@@ -2227,4 +2228,5 @@ class Repository:
             acked=None if row["acked"] is None else bool(row["acked"]),
             created_at=datetime.fromisoformat(row["created_at"]),
             row_id=row["id"],
+            scope=row["scope"] if "scope" in row.keys() else None,
         )

meshterm/services/chat_service.py

@@ -30,6 +30,7 @@ from ..core.models import (
     is_direct_messageable,
     utcnow,
 )
+from ..core.regions import WILDCARD, normalize
 
 if TYPE_CHECKING:
     from ..context import AppContext
@@ -509,28 +510,86 @@ class ChatService:
             self._ctx.repo.update_chat_ack(message.row_id, message.acked)
         return message
 
-    async def send_channel(self, index: int, text: str, *, label: str | None = None) -> ChatMessage:
-        """Broadcast a message on a channel and record it in history.
+    def channel_scope(self, channel_id: str | None) -> str | None:
+        """The region a channel's messages are sent under, or ``None`` for the device default.
+
+        Read from :class:`~meshterm.core.region_store.RegionStore` by the channel's intrinsic
+        identity, so the scope follows the channel across slots. A context with no store (a
+        test's, or a surface that never built one) has no channel scopes at all.
+        """
+        store = getattr(self._ctx, "region_store", None)
+        return store.channel_scope(channel_id) if store is not None else None
+
+    async def send_channel(
+        self,
+        index: int,
+        text: str,
+        *,
+        label: str | None = None,
+        scope: str | None = None,
+    ) -> ChatMessage:
+        """Broadcast a message on a channel, under the channel's scope, and record it.
+
+        The channel's identity is resolved *before* the send, because it is what names the
+        scope the message goes out under (:meth:`channel_scope`). The send itself is
+        :meth:`~meshterm.core.connection.Device.send_channel_in_scope` — set the session
+        scope, send, restore, with no other flood let in between — and a scope the radio
+        can't take stops the message rather than sending it some other way.
 
         Args:
             index: The channel slot to transmit on.
             text: The message body.
             label: A display label for the channel (e.g. ``#general``), stored for the
                 transcript.
+            scope: Overrides the channel's own scope for this one message: a region name,
+                or :data:`~meshterm.core.regions.WILDCARD` (``*``) to send it unscoped —
+                the resend for a scoped message nothing relayed. ``None`` uses the
+                channel's scope, or the device default when the channel has none.
 
         Returns:
-            The recorded outbound :class:`ChatMessage`.
+            The recorded outbound :class:`ChatMessage`, its :attr:`~ChatMessage.scope` the
+            one it went out under.
+
+        Raises:
+            ~meshterm.core.connection.FloodScopeError: If the radio can't send under the
+                scope; nothing was sent and nothing is recorded.
         """
         device = await self._ctx.device()
-        await device.send_channel_message(index, text)
+        channel_id = await self.channel_id_for(index)
+        asked = scope if scope is not None else self.channel_scope(channel_id)
+        await device.send_channel_in_scope(index, text, asked)
         chat = ChatMessage(
             text=text,
             outbound=True,
             is_channel=True,
-            channel_id=await self.channel_id_for(index),
+            channel_id=channel_id,
             channel_idx=index,
             peer_name=label,
             created_at=utcnow(),
+            scope=await self._sent_under(asked),
         )
-        self._ctx.repo.record_chat_message(chat, run_id=self._run_id)
+        chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id)
         return chat
+
+    async def _sent_under(self, asked: str | None) -> str | None:
+        """What a channel send asked to go out under, as the transcript records it.
+
+        A scope that was asked for is what it went under — the send refuses rather than
+        substitute. A plain send went under the device's default scope, which is read (once
+        a session, through the devstate cache): its name when one is set, ``*`` when none
+        is — a plain flood with no default *is* unscoped. A default that can't be read
+        (firmware before 1.15 has none to read; a link blip) records ``None``: unknown is
+        the honest answer, and the transcript never claims a scope it didn't see.
+        """
+        bare = normalize(asked) if asked else ""
+        if bare:
+            return bare
+        devstate = getattr(self._ctx, "devstate", None)
+        if devstate is None:
+            return None
+        try:
+            default = await devstate.default_scope()
+        except Exception as exc:  # noqa: BLE001 - unknown, not a failed send
+            self._ctx.log.debug("chat: default scope unreadable: %s", exc)
+            return None
+        return normalize(default) or WILDCARD

meshterm/services/device_state.py

@@ -99,6 +99,9 @@ class DeviceState:
         self._channels: list[ChannelSlot] | None = None
         self._channels_epoch = 0
         self._channel_capacity: int | None = None
+        #: The companion's default flood scope — ``""`` for none — once read (see
+        #: :meth:`default_scope`); ``None`` while unread.
+        self._default_scope: str | None = None
         # One lock per slow fetch so overlapping first-access callers (two screens opened in
         # quick succession) collapse onto a single round-trip instead of each firing their own.
         self._contacts_lock = asyncio.Lock()
@@ -309,6 +312,36 @@ class DeviceState:
             return 0
         return (mode + 1) if isinstance(mode, int) and 0 <= mode <= 3 else 0
 
+    # -- the default flood scope (held until a config write invalidates it) ------
+
+    async def default_scope(self) -> str:
+        """The companion's persisted default flood scope, ``""`` for none, read once.
+
+        What a plain channel send goes out under, so it is what a channel with no scope of
+        its own records its messages as sent under. One round trip, then held until the
+        config editor writes a setting (:meth:`invalidate_config`) or the link is rebuilt.
+
+        Returns:
+            The bare region name, or ``""`` when no default scope is set.
+
+        Raises:
+            Exception: Like :meth:`~meshterm.core.connection.Device.get_default_flood_scope`
+                — firmware older than 1.15 has no default scope to read. Not cached, so the
+                next call asks again.
+        """
+        if self._default_scope is None:
+            device = await self._ctx.device()
+            self._default_scope = str(await device.get_default_flood_scope() or "")
+        return self._default_scope
+
+    def note_default_scope(self, name: str | None) -> None:
+        """Record a default scope just read or written elsewhere (Device config).
+
+        Args:
+            name: The bare name, ``""`` for none, or ``None`` to forget it and re-read.
+        """
+        self._default_scope = None if name is None else str(name)
+
     # -- channel slots (held until the channel editor invalidates them) --------
 
     async def channel_slots(self) -> list[ChannelSlot]:
@@ -446,6 +479,7 @@ class DeviceState:
         """
         self.invalidate_self_info()
         self.invalidate_path_hash_mode()
+        self._default_scope = None
 
     def reset(self) -> None:
         """Clear the whole cache and cancel any in-flight background refresh.
@@ -457,6 +491,7 @@ class DeviceState:
         self.invalidate_self_info()
         self.invalidate_path_hash_mode()
         self.invalidate_channels()
+        self._default_scope = None
         # Capacity is a hardware constant (not touched by channel edits, so it has no per-write
         # invalidator), but a reconnect may be to a different device — drop it too.
         self._channel_capacity = None

meshterm/tools/channels.py

@@ -4,8 +4,8 @@
 Interactively it opens a full-screen channel manager (see :mod:`meshterm.ui.channels`) — a
 first-class, phone-app-style experience for creating private channels, adding public ``#``
 channels, joining with a key, importing a scanned ``meshcore://`` link, and sharing any
-channel as a QR code. On the CLI it exposes ``list``, ``add``, ``join``, ``import``, and
-``share`` subcommands for scripted use.
+channel as a QR code. On the CLI it exposes ``list``, ``add``, ``join``, ``import``,
+``share``, ``clear`` and ``scope`` subcommands for scripted use.
 
 Channels are also *listed* by the ``chat`` tool for picking a conversation; this tool owns
 everything to do with configuring the slots themselves.
@@ -81,6 +81,8 @@ class ChannelsTool(Tool):
             return await self._cli_share(ctx, params)
         if action == "clear":
             return await self._cli_clear(ctx, params)
+        if action == "scope":
+            return await self._cli_scope(ctx, params)
         return await self._cli_list(ctx)
 
     async def _cli_list(self, ctx: AppContext) -> ToolResult:
@@ -91,7 +93,7 @@ class ChannelsTool(Tool):
         slots = await read_channel_slots(device)
         return ToolResult(
             summary={"channels": len(slots)},
-            report=(_channel_listing(slots),),
+            report=(_channel_listing(slots, ctx.region_store),),
             exit_code=exitcodes.OK if slots else exitcodes.NO_RESULT,
         )
 
@@ -194,6 +196,46 @@ class ChannelsTool(Tool):
             report=(_cleared(idx, True),),
         )
 
+    async def _cli_scope(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Read, set or clear the region a channel's messages are sent under.
+
+        The scriptable face of the detail page's *Send scope…*. The scope is MeshTerm's to
+        keep (the firmware has none per channel; the send sets the companion's session
+        scope around each message), so setting one writes nothing to the radio — the slot
+        is read only to learn which channel is in it, since the scope is keyed by the
+        channel's identity and follows it to any slot.
+
+        With no region, the plain face prints the scope alone, the way ``config get`` prints
+        one value — and a channel with none prints nothing and exits 5, nothing to report;
+        a set or a clear prints nothing, and the document always carries the channel and
+        its scope (``null`` for none).
+        """
+        from ..core.channel_probe import read_channel_slots
+
+        device = await ctx.device()
+        idx = int(params["index"])
+        slot = next((s for s in await read_channel_slots(device) if s.idx == idx), None)
+        if slot is None:
+            return ToolResult(
+                summary={"index": idx, "scope": None},
+                report=(_scoped(idx, None, None, None, changed=False),),
+                exit_code=exitcodes.NO_RESULT,
+            )
+        store = ctx.region_store
+        region = params.get("region")
+        changed = bool(params.get("clear")) or region is not None
+        if params.get("clear"):
+            store.set_channel_scope(slot.identity, None)
+        elif region is not None:
+            store.set_channel_scope(slot.identity, region)
+        scope = store.channel_scope(slot.identity)
+        return ToolResult(
+            summary={"index": idx, "scope": scope},
+            report=(_scoped(idx, slot.name, slot.secret, scope, changed=changed),),
+            # A read of a channel with no scope found nothing to report; a write did its job.
+            exit_code=exitcodes.OK if scope or changed else exitcodes.NO_RESULT,
+        )
+
     @staticmethod
     async def _show_qr(ctx: AppContext, name: str, url: str) -> None:
         """Draw a channel's share link as a QR code — in the menu, and only there.
@@ -274,19 +316,48 @@ class ChannelsTool(Tool):
                 )
             run_tool_command(self, {"cli_action": "clear", "index": index})
 
+        @channels_app.command("scope", help="Show or set the region a channel sends into")
+        def _scope_cmd(
+            index: int = typer.Argument(..., help="Channel slot index"),
+            region: str | None = typer.Argument(
+                None, help="Region to send this channel's messages into (omit to show it)"
+            ),
+            clear: bool = typer.Option(
+                False, "--clear", help="Drop the channel's scope (send under the default)"
+            ),
+        ) -> None:
+            from ..core.regions import RegionNameError, validate
+
+            if clear and region is not None:
+                raise typer.BadParameter("Pass a region or --clear, not both.")
+            if region is not None:
+                try:
+                    region = validate(region)
+                except RegionNameError as exc:
+                    raise typer.BadParameter(str(exc)) from exc
+            run_tool_command(
+                self, {"cli_action": "scope", "index": index, "region": region, "clear": clear}
+            )
+
         app.add_typer(channels_app, name=self.name)
 
 
-def _channel_listing(slots: list) -> Listing:
+def _channel_listing(slots: list, store: object | None = None) -> Listing:
     """The configured slots, one record each.
 
     The one listing whose record *is* a shared shape rather than carrying one: a row that
     nested its only field under a ``channel`` key would make ``jq '.[].channel.name'``
     out of a listing whose every column is already the channel.
+
+    ``SCOPE`` is the region the channel's messages are sent under (``-`` and ``null`` for
+    the device default), read from the region store the send path reads.
     """
     from ..ui import fields
     from ..ui.report import Listing
 
+    def scope_of(slot) -> str | None:  # noqa: ANN001 - a ChannelSlot
+        return store.channel_scope(slot.identity) if store is not None else None
+
     return Listing(
         key="channels",
         columns=(
@@ -294,6 +365,7 @@ def _channel_listing(slots: list) -> Listing:
             fields.name("name", "NAME"),
             fields.word("type", "TYPE"),
             fields.hexid("hash", "HASH"),
+            fields.name("scope", "SCOPE"),
         ),
         rows=[
             {
@@ -301,12 +373,46 @@ def _channel_listing(slots: list) -> Listing:
                 "name": slot.name,
                 "type": "public" if slot.is_public else "private",
                 "hash": slot.hash,
+                "scope": scope_of(slot),
             }
             for slot in slots
         ],
     )
 
 
+def _scoped(
+    idx: int, name: str | None, secret: bytes | None, scope: str | None, *, changed: bool
+) -> Facts:
+    """What ``channels scope`` read or set: the channel and the region it sends into.
+
+    Args:
+        idx: The slot index.
+        name: The channel name, or ``None`` for a slot that turned out to be empty.
+        secret: The 16-byte key, or ``None`` for an empty slot.
+        scope: The channel's scope now, or ``None`` for the device default.
+        changed: Whether this run set or cleared it — then the plain face is silent, the
+            way every write's is; a read prints the scope alone.
+
+    Returns:
+        The facts block.
+    """
+    from ..ui import fields
+    from ..ui.fields import ChannelRef
+    from ..ui.report import BARE, SILENT, Facts
+
+    channel = None
+    if name is not None and secret is not None:
+        public = is_public_channel(name, secret)
+        channel = ChannelRef(slot=idx, name=name, public=public, hash=channel_hash(secret))
+    return Facts(
+        key="channel",
+        fields=(fields.channel("channel"), fields.name("scope", "scope")),
+        values={"channel": channel, "scope": scope},
+        shape=SILENT if changed else BARE,
+        bare="scope",
+    )
+
+
 def _written(idx: int, name: str | None, secret: bytes | None, *, shown: bool = True) -> Facts:
     """What a slot now holds, after ``add``, ``join``, ``import`` or ``share``.
 

meshterm/tools/chat.py

@@ -286,15 +286,22 @@ class ChatTool(Tool):
         message prints its ``acked`` state, which is the one thing the send does not
         already tell the caller: the radio accepted it either way, and whether the peer
         answered is a separate fact.
+
+        A channel message goes out under the channel's send scope, or under ``scope`` for
+        this one message (a region name, or ``*`` for unscoped); the document says which it
+        went under. A radio that can't send under the scope refuses before anything is
+        transmitted, and that is a failure (exit 1), never a quiet unscoped send.
         """
         device = await ctx.device()
         text = str(params["text"])
         channel = params.get("channel")
         if channel is not None:
-            await ctx.chat.send_channel(int(channel), text, label=f"#{channel}")
+            message = await ctx.chat.send_channel(
+                int(channel), text, label=f"#{channel}", scope=params.get("scope") or None
+            )
             return ToolResult(
-                summary={"channel": channel, "sent": True},
-                report=(_sent(channel=int(channel)),),
+                summary={"channel": channel, "sent": True, "scope": message.scope},
+                report=(_sent(channel=int(channel), scope=message.scope),),
             )
 
         contact = _resolve_contact(await device.get_contacts(), str(params["to"]))
@@ -433,11 +440,19 @@ class ChatTool(Tool):
             text: str = typer.Argument(..., help="The message body"),
             to: str | None = typer.Option(None, "--to", help="Contact name or key prefix"),
             channel: int | None = typer.Option(None, "--channel", help="Channel slot index"),
+            scope: str | None = typer.Option(
+                None,
+                "--scope",
+                help="Send under this region instead of the channel's scope (* = unscoped)",
+            ),
         ) -> None:
             if (to is None) == (channel is None):
                 raise typer.BadParameter("Pass exactly one of --to / --channel.")
+            if scope is not None:
+                scope = _cli_scope(scope, channel)
             run_tool_command(
-                self, {"cli_action": "send", "to": to, "channel": channel, "text": text}
+                self,
+                {"cli_action": "send", "to": to, "channel": channel, "text": text, "scope": scope},
             )
 
         @chat_app.command("history", help="Show a conversation's stored history")
@@ -476,6 +491,28 @@ class ChatTool(Tool):
 # -- helpers ------------------------------------------------------------------
 
 
+def _cli_scope(scope: str, channel: int | None) -> str:
+    """Check ``--scope`` before anything is sent: a region the firmware could hold, or ``*``.
+
+    A usage error (exit 2), not a device failure: nothing has been transmitted yet. A
+    direct message has no scope to give it here — the firmware floods a DM under the
+    session scope like anything else, but only a channel send sets one for itself.
+
+    Raises:
+        typer.BadParameter: For a direct send, or a name the firmware would refuse.
+    """
+    from ..core.regions import WILDCARD, RegionNameError, normalize, validate
+
+    if channel is None:
+        raise typer.BadParameter("--scope applies to a channel send (--channel).")
+    if normalize(scope) == WILDCARD:
+        return WILDCARD
+    try:
+        return validate(scope)
+    except RegionNameError as exc:
+        raise typer.BadParameter(str(exc)) from exc
+
+
 def _enable_receive_debug() -> None:
     """Route the message-pull and library debug logs to the console for diagnosis.
 
@@ -1092,6 +1129,7 @@ def _sent(
     contact: object | None = None,
     channel: int | None = None,
     acked: bool | None = None,
+    scope: str | None = None,
 ) -> Facts:
     """What one ``chat send`` did.
 
@@ -1100,10 +1138,24 @@ def _sent(
     answered is a separate fact. A channel broadcast prints nothing, because there *is* no
     acknowledgement on a channel — and the document says that in the one way the plain
     face never could, with ``null`` rather than ``false``.
+
+    The document also carries the ``scope`` a channel message went out under, in the same
+    shape a received frame's scope takes (:func:`~meshterm.ui.fields.scope`): the region,
+    ``unscoped``, or ``null`` where it isn't known (a direct message, or a default scope
+    that couldn't be read).
     """
+    from dataclasses import replace
+
+    from ..core.regions import UNSCOPED, WILDCARD, Scope
     from ..ui import fields
     from ..ui.report import PAIRS, SILENT, Facts
 
+    sent_scope = None
+    if scope == WILDCARD:
+        sent_scope = UNSCOPED
+    elif scope:
+        sent_scope = Scope("scoped", scope)
+
     node = (
         NodeRef(
             name=contact.name,
@@ -1125,6 +1177,9 @@ def _sent(
             fields.channel("channel", lanes=()),
             fields.hidden("sent"),
             fields.flag("acked", "acked"),
+            # Document only: a direct message has no scope of its own to print, and a
+            # channel send prints nothing at all.
+            replace(fields.scope("scope", "scope"), lanes=()),
         ),
         values={
             "kind": "channel" if channel is not None else "direct",
@@ -1134,6 +1189,7 @@ def _sent(
             ),
             "sent": True,
             "acked": acked,
+            "scope": sent_scope,
         },
         shape=SILENT if channel is not None else PAIRS,
     )

meshterm/tools/config.py

@@ -78,6 +78,7 @@ class ConfigTool(Tool):
         device = await ctx.device()
         ops: list[tuple] = list(params.get("ops") or [])
         snapshot = await build_snapshot(device)
+        learn_default_scope(ctx, snapshot)
         changes, artifacts, report = await apply_ops(ctx, device, snapshot, ops)
 
         plural = "" if changes == 1 else "s"
@@ -348,10 +349,42 @@ async def _apply_setting(
     # meshterm.core.settings_store). Provenance-gated: only values changed through MeshTerm.
     if ctx.settings_store is not None:
         ctx.settings_store.remember(str(snapshot.get("public_key") or ""), key, value)
+    if key == "flood_scope":
+        learn_default_scope(ctx, snapshot)
     ctx.ui.ack(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}")
     return {"key": key, "previous": previous, "value": value}
 
 
+def learn_default_scope(ctx: AppContext, snapshot: dict) -> None:
+    """Take in the default flood scope a config read or write has just seen.
+
+    Two things want it. The region store learns the name (source ``default``): it is the
+    region this station's own plain floods carry, so it is the first name any of them will
+    be resolved against. And the session cache holds it (``devstate``), since a channel with
+    no scope of its own records its messages as sent under the default, and this saves that
+    record a round trip.
+
+    A snapshot whose read of the default failed (firmware before 1.15) carries no
+    ``flood_scope`` key at all, and teaches nothing.
+
+    Args:
+        ctx: Shared application context.
+        snapshot: A :func:`~meshterm.core.device_config.build_snapshot` result, or one an
+            apply has just updated.
+    """
+    if "flood_scope" not in snapshot:
+        return
+    from ..core.regions import normalize
+
+    name = normalize(str(snapshot.get("flood_scope") or ""))
+    devstate = getattr(ctx, "devstate", None)
+    if devstate is not None:
+        devstate.note_default_scope(name)
+    store = getattr(ctx, "region_store", None)
+    if store is not None and name:
+        store.learn(name, "default")
+
+
 async def _show(ctx: AppContext, device: Device, snapshot: dict) -> Listing | None:
     """Print every current setting, plus any custom variables.
 

meshterm/ui/channels.py

@@ -16,6 +16,14 @@ recorded to history. The mute lives in :class:`~meshterm.core.mute_store.MuteSto
 the channel's intrinsic identity so it follows the channel across slot moves, and the list
 row shows a ``🔕`` in its (then always-empty) unread lane to mark it.
 
+A channel's **send scope** is the other: the region its messages are flooded into, so only
+the repeaters carrying that region relay them. The firmware has no such thing per channel —
+the chat sets the companion's session scope around each send instead (see
+:meth:`~meshterm.core.connection.Device.send_channel_in_scope`) — so it is kept by MeshTerm,
+in :class:`~meshterm.core.region_store.RegionStore`, keyed the same way a mute is. The
+detail page's *Send scope…* row picks it from the regions already known here, or takes a
+typed name.
+
 The list is laid out like the config editor: fixed, column-aligned lanes under one header
 line — the openness glyph and name, unread badge, total messages, last-message age, and a
 braille sparkline of the trailing two hours' traffic — so a glance shows not just *which*
@@ -55,6 +63,8 @@ from ..core.channels import (
     share_url,
 )
 from ..core.connection import Device
+from ..core.regions import RegionNameError
+from ..core.regions import validate as validate_region
 from ..persistence.repository import ACTIVITY_DRAWN_BUCKETS
 from ..platforms import get_platform
 from .braillechart import activity_peak, activity_sparkline
@@ -93,7 +103,15 @@ _QR = "qr"
 _KEY = "key"
 _CHAT = "chat"
 _MUTE = "mute"
+_SCOPE = "scope"
 _EDIT = "edit"
+
+# Send-scope picker sentinels: clear the channel's scope, or type a region by name.
+_NO_SCOPE = "__no_scope__"
+_TYPE_REGION = "__type_region__"
+
+#: The scope picker's footer: a value picker over a list that can grow, so it filters.
+_SCOPE_HINT = "↑↓ move · type to filter · Enter set · Esc keep"
 _CLEAR = "clear"
 
 
@@ -668,6 +686,11 @@ def _detail_summary(ctx: AppContext, slot: ChannelSlot, stats: _LiveStats) -> st
     unread = ctx.chat.unread(slot.conversation.key)
     kind = "public" if slot.is_public else "private"
     parts = [kind, f"hash {slot.hash}", f"slot {slot.idx}"]
+    # The scope rides with the channel's identity rather than its traffic: it decides who
+    # can hear the next message, so it is kept ahead of the counts when the line sheds.
+    scope = _channel_scope(ctx, slot)
+    if scope:
+        parts.append(f"scope {scope}")
     if st is None or not st.total:
         parts.append("no messages yet")
     else:
@@ -698,7 +721,7 @@ def _detail_items(ctx: AppContext, slot: ChannelSlot) -> list:
     # ✎ and 🗑 are one cell where 📱 🔑 💬 🔔 🔕 are two, so the column is measured once and
     # every mark padded out to it — otherwise the edit and clear rows start their labels a
     # column left of the rows above them.
-    lane = icon_lane(("📱", "🔑", "💬", "🔔", "🔕", "✎", "🗑"))
+    lane = icon_lane(("📱", "🔑", "💬", "🔔", "🔕", "🔖", "✎", "🗑"))
     chat_label = marked_label("💬", "Open in chat", "", lane=lane)
     if unread:
         chat_label.append("  ●", style="err")
@@ -729,6 +752,7 @@ def _detail_items(ctx: AppContext, slot: ChannelSlot) -> list:
             ),
             (chat_label, "Read and send messages on this channel", _CHAT),
             mute_row,
+            _scope_row(ctx, slot, lane),
             (
                 marked_label("✎", "Rename / change key…", "", lane=lane),
                 "Edit the name or paste a different key",
@@ -794,6 +818,10 @@ async def _channel_detail(
             # A preference, not a slot-config change: toggle in place and loop back to the
             # detail (whose rows re-render to the new state) without counting a channel change.
             _toggle_mute(ctx, slot)
+        elif choice == _SCOPE:
+            # MeshTerm's to keep, like the mute: nothing is written to the radio until a
+            # message is sent, so it is not a channel change either.
+            await _pick_scope(ctx, slot)
         elif choice == _EDIT and await _edit(ctx, device, slot):
             changes += 1
             if not await reread():  # pragma: no cover - the slot we just wrote is there
@@ -829,6 +857,106 @@ async def _channel_detail(
             )
 
 
+# --- send scope ----------------------------------------------------------------
+
+
+def _channel_scope(ctx: AppContext, slot: ChannelSlot) -> str | None:
+    """The region this channel's messages are sent under, or ``None`` for the device default."""
+    store = getattr(ctx, "region_store", None)
+    return store.channel_scope(slot.identity) if store is not None else None
+
+
+def _scope_row(ctx: AppContext, slot: ChannelSlot, lane: int) -> tuple:
+    """The detail page's *Send scope…* row, its description saying what is set now."""
+    scope = _channel_scope(ctx, slot)
+    description = f"Messages flood in {scope} only" if scope else "Keep messages in one region"
+    return (marked_label("🔖", "Send scope…", "", lane=lane), description, _SCOPE)
+
+
+def _scope_items(ctx: AppContext, current: str | None) -> list:
+    """The scope picker's rows: no scope, every region known here, then typing one in.
+
+    A region reads in the ``scope`` style (a region is not a node, so no node hue), with
+    how many repeaters were heard to carry it — the thing that decides whether a message
+    scoped to it goes anywhere at all. The one in force is marked ``current``.
+    """
+    store = getattr(ctx, "region_store", None)
+    none_row = Text("No scope — the device default")
+    if not current:
+        none_row.append("  · current", style="muted")
+    items: list = [Choice(title=none_row, value=_NO_SCOPE)]
+    for name in store.names() if store is not None else []:
+        row = Text(name, style="scope")
+        carriers = len(store.carriers(name))
+        if carriers:
+            row.append(f"  · {carriers} repeater{'s' if carriers != 1 else ''}", style="muted")
+        if name == current:
+            row.append("  · current", style="muted")
+        items.append(Choice(title=row, value=name))
+    items.append(Choice(title=Text("Type a region name…"), value=_TYPE_REGION))
+    return items
+
+
+async def _pick_scope(ctx: AppContext, slot: ChannelSlot) -> bool:
+    """Pick the region a channel's messages are sent under; return whether it changed.
+
+    A value picker floated over the detail page, opening on the scope in force. Typing a
+    new name is a second step stacked on the first (the ``run_steps`` shape): Esc on the
+    name field comes back to the list rather than abandoning the pick. A typed name is
+    learned (source ``typed``) as well as set, so the next channel can pick it from the
+    list.
+
+    Args:
+        ctx: Shared application context.
+        slot: The channel whose scope is being set.
+
+    Returns:
+        ``True`` when the channel's scope changed.
+    """
+    store = getattr(ctx, "region_store", None)
+    if store is None:  # pragma: no cover - every real context builds one
+        return False
+    current = store.channel_scope(slot.identity)
+    session = getattr(ctx.ui, "session", None)
+    while True:
+        items = _scope_items(ctx, current)
+        title = f"Send scope — {slot.name}"
+        prompt = "The region this channel's messages flood into:"
+        default = current or _NO_SCOPE
+        if session is not None:
+            picked = await session.select(
+                title, items, prompt=prompt, default=default, footer_hint=_SCOPE_HINT
+            )
+        else:
+            picked = await ctx.ui.select(title, items, prompt=prompt, default=default)
+        if picked is None:
+            return False
+        if picked == _TYPE_REGION:
+            typed = await ctx.ui.text(
+                "Send scope",
+                prompt="Region name, as its repeaters list it:",
+                default=current or "",
+                validate=_valid_region,
+            )
+            if typed is None:
+                continue  # Esc on the name steps back to the list
+            picked = store.learn(typed, "typed") or typed
+        wanted = None if picked == _NO_SCOPE else str(picked)
+        if wanted == current:
+            return False
+        store.set_channel_scope(slot.identity, wanted)
+        return True
+
+
+def _valid_region(text: str) -> bool | str:
+    """Require a region name the firmware could hold (see :func:`regions.validate`)."""
+    try:
+        validate_region(text)
+    except RegionNameError as exc:
+        return str(exc)
+    return True
+
+
 # --- create / join flows -----------------------------------------------------
 
 

meshterm/ui/chat.py

@@ -27,6 +27,7 @@ from ..core.channels import MENTION, split_channel_sender
 from ..core.connection import ContactNotOnDeviceError
 from ..core.events import EventKind, MeshEvent
 from ..core.models import ChatMessage, Contact, Conversation, Message, utcnow
+from ..core.regions import WILDCARD
 from .theme import name_style, snr_style
 from .tui.prompt import (
     CHANNEL_BYTE_LIMIT,
@@ -63,6 +64,11 @@ class ChatScreen(Screen):
     opens the picked message's paths in either kind — a path belongs to one message, so
     with nothing picked there is nothing to show. Esc leaves the chat once nothing is
     picked.
+
+    A channel with a send scope says so in the title (``#ops · scope yul``), and ^R there
+    resends the newest message *unscoped* — the way out for a scoped message no repeater
+    in earshot carries — after an amber confirm, since an unscoped flood reaches every
+    repeater the scope was keeping it from.
     """
 
     floating = False
@@ -85,8 +91,11 @@ class ChatScreen(Screen):
 
         The F3 pair follows the lane's two claims. *Retry* is absent in a channel — a
         channel message is never acknowledged, so there is no such thing to retry there —
-        and merely dim in a direct chat with nothing outstanding. *Paths* needs a picked
-        message to have paths of, and both nav slots need a transcript to walk.
+        and merely dim in a direct chat with nothing outstanding. A channel's Shift slot is
+        *Resend* instead — the newest message again, unscoped (see :meth:`_retry_target`)
+        — present where a resend is wired and dim until the newest message went out under
+        a region. *Paths* needs a picked message to have paths of, and both nav slots need
+        a transcript to walk.
 
         F1/F2 carry the **day** jump, the transcript's own section step (its dividers are
         days) — the same claim a grouped select list makes with ``Sect ↑``/``Sect ↓``, on
@@ -111,7 +120,12 @@ class ChatScreen(Screen):
         lane[1] = FPair("Day ↓", "ctrl_pagedown", enabled=days)
         lane[3] = FPair("Page ↓", "pagedown", "Latest", "ctrl_end", enabled=live, opp_enabled=live)
         lane[4] = FPair("Page ↑", "pageup", "Oldest", "ctrl_home", enabled=live, opp_enabled=live)
-        retry = ("Retry", "retry") if not self._is_channel else ("", "")
+        if not self._is_channel:
+            retry = ("Retry", "retry")
+        elif self._resend_unscoped is not None:
+            retry = ("Resend", "retry")
+        else:
+            retry = ("", "")
         lane[2] = FPair(
             "Paths",
             "paths",
@@ -132,6 +146,8 @@ class ChatScreen(Screen):
         resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None,
         paths: Callable[[ChatMessage], Awaitable[None]] | None = None,
         key_of: Callable[[str], str | None] | None = None,
+        scope: str | None = None,
+        resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None,
     ) -> None:
         """Build the chat screen.
 
@@ -152,9 +168,18 @@ class ChatScreen(Screen):
                 :func:`~meshterm.services.trace_runner.make_name_key_resolver`), the seed
                 of the sender's hue; ``None`` (or a name it can't place) leaves senders
                 muted — colour is reserved for keyed identities.
+            scope: The channel's send scope, stated in the title as a ``·`` atom
+                (``#ops · scope yul``); ``None`` for a channel sending under the device
+                default, and always for a direct chat.
+            resend_unscoped: Async callable that sends a channel message's text again,
+                unscoped, and returns the recorded message (channels only; ``None`` leaves
+                ^R inert there).
         """
         super().__init__()
-        self.title = conversation.label
+        self._scope = scope if conversation.is_channel else None
+        self.title = (
+            f"{conversation.label} · scope {self._scope}" if self._scope else conversation.label
+        )
         self._is_channel = conversation.is_channel
         self._key_of: Callable[[str], str | None] = key_of or (lambda name: None)
         # A direct thread's one remote sender is the peer; its key colours the header
@@ -168,6 +193,7 @@ class ChatScreen(Screen):
         self._messages = list(messages)
         self._send = send
         self._resend = resend
+        self._resend_unscoped = resend_unscoped if conversation.is_channel else None
         self._paths = paths
         self._names = names
         self._session = session
@@ -181,6 +207,7 @@ class ChatScreen(Screen):
         self._stick = True  # keep the newest message in view until the user scrolls up
         self._paths_open = False  # one paths dialog at a time
         self._paste_open = False  # one paste-confirm dialog at a time
+        self._resend_open = False  # one resend-unscoped confirm at a time
         # The pick: index of the highlighted message (or None when the compose line is
         # focused), plus the body line it rendered on so the frame keeps it in view.
         self._selected: int | None = None
@@ -215,6 +242,8 @@ class ChatScreen(Screen):
                 return "Enter reply (@mention) · ^P paths · ↑↓ pick · ^End/Esc cancel"
             return "Enter paths · ↑↓ pick · ^End/Esc cancel"
         if self._retry_target() is not None:
+            if self._is_channel:
+                return "Enter send · ↑ pick a message · ^R resend unscoped · Esc back"
             return "Enter send · ↑ pick a message · ^R retry failed · Esc back"
         return "Enter send · ↑ pick a message · Esc back"
 
@@ -496,6 +525,7 @@ class ChatScreen(Screen):
             if message.is_channel:
                 if message.acked is False:
                     text.append("  ⚠ no ack", style="warn")
+                text.append_text(self._scope_note(message))
             else:
                 text.append("  ")
                 text.append_text(self._delivery_glyph(message.acked))
@@ -503,6 +533,20 @@ class ChatScreen(Screen):
             text.append(f"  {message.snr:+.0f} dB", style=snr_style(message.snr))
         return text
 
+    def _scope_note(self, message: ChatMessage) -> Text:
+        """A muted tail on a sent message that went out under a scope other than the title's.
+
+        Only in a channel whose title names a scope, and only for the exceptions to it — an
+        unscoped resend, or a message sent before the scope was set or changed. Every other
+        line would repeat the title; and in a channel with no scope of its own, the default
+        its messages went under is the device's business, not news on every line.
+        """
+        if not self._scope or not message.scope or message.scope == self._scope:
+            return Text()
+        if message.scope == WILDCARD:
+            return Text("  · unscoped", style="muted")
+        return Text(f"  · scope {message.scope}", style="muted")
+
     def _render_mentions(self, body: str, *, selected: bool) -> Text:
         """Render body text, rewriting each ``@[Name]`` token to a ``@Name`` in its hue.
 
@@ -625,7 +669,9 @@ class ChatScreen(Screen):
         elif action == "paths":
             self._open_paths(self._selected)
         elif action == "retry":
-            if not self._is_channel:
+            if self._is_channel:
+                self._begin_resend_unscoped()
+            else:
                 self._retry()
         elif action == "escape":
             if self._selected is not None:
@@ -905,10 +951,20 @@ class ChatScreen(Screen):
     def _retry_target(self) -> ChatMessage | None:
         """The message ^R would re-send, or ``None`` when there is nothing to retry.
 
-        The newest outbound direct message that went out and was never acknowledged —
-        and only while a retry could actually start: no send already in flight, a resend
-        path wired (channels have none, since a channel message is never acked).
+        In a direct chat, the newest outbound message that went out and was never
+        acknowledged. In a channel — where nothing is ever acknowledged — the newest
+        message *we* sent, when it went out under a region: ^R sends it again unscoped. The
+        newest, not any scoped one: once the unscoped copy has gone it is the newest, and
+        there is nothing left to offer. Either way only while a resend could actually
+        start — no send already in flight, and a resend path wired.
         """
+        if self._is_channel:
+            if self._sending or self._resend_unscoped is None:
+                return None
+            newest = next((m for m in reversed(self._messages) if m.outbound), None)
+            if newest is None or not newest.scope or newest.scope == WILDCARD:
+                return None
+            return newest
         if self._sending or self._resend is None:
             return None
         return next(
@@ -928,6 +984,59 @@ class ChatScreen(Screen):
         self._session.invalidate()
         self._spawn(self._resend_message(target))
 
+    def _begin_resend_unscoped(self) -> None:
+        """Confirm, then send the newest scoped channel message again unscoped (^R).
+
+        Amber (``danger``): nothing is lost, but an unscoped flood is relayed by every
+        repeater that allows one — the whole mesh the scope was keeping the message out of —
+        so it is a choice made on purpose, with the region it is leaving named. The resend
+        is a new message on the air (a channel message has no identity to re-deliver), so
+        it appends as one; a radio that can't send unscoped refuses before anything goes
+        out, and the status line says why.
+        """
+        target = self._retry_target()
+        if target is None or self._resend_open:
+            return
+        self._resend_open = True
+        preview = target.text if len(target.text) <= 32 else target.text[:31] + "…"
+
+        async def run() -> None:
+            try:
+                confirmed = await self._session.button_dialog(
+                    Text(
+                        f"Send “{preview}” again, unscoped? Every repeater that relays "
+                        f"unscoped floods will carry it, not only those in {target.scope}.",
+                        style="warn",
+                    ),
+                    [("Cancel", False), ("Resend", True)],
+                    title="Resend unscoped",
+                    default=1,
+                    border_style="warn",
+                    footer_hint="←→ choose · Enter select · Esc cancel",
+                )
+            finally:
+                self._resend_open = False
+            if not confirmed or self._sending or self._resend_unscoped is None:
+                self._session.invalidate()
+                return
+            self._sending = True
+            self._status = "resending unscoped…"
+            self._stick = True
+            self._session.invalidate()
+            try:
+                message = await self._resend_unscoped(target)
+                if message is not None:
+                    self._messages.append(message)
+                self._status = ""
+            except Exception as exc:  # noqa: BLE001 - report inline, keep the chat alive
+                self._status = f"resend failed: {exc}"
+            finally:
+                self._sending = False
+                self._stick = True
+                self._session.invalidate()
+
+        self._spawn(run())
+
     async def _resend_message(self, message: ChatMessage) -> None:
         """Drive a retry to completion, refreshing the message's delivery state in place."""
         assert self._resend is not None
@@ -1010,7 +1119,18 @@ async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
         return await _with_restore(ctx, lambda: ctx.chat.send_direct(conversation.contact, text))
 
     resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None
-    if not conversation.is_channel:
+    resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None
+    scope: str | None = None
+    if conversation.is_channel:
+        scope = ctx.chat.channel_scope(conversation.channel_id)
+
+        async def resend_unscoped(message: ChatMessage) -> ChatMessage | None:
+            assert conversation.channel_idx is not None
+            return await ctx.chat.send_channel(
+                conversation.channel_idx, message.text, label=conversation.label, scope=WILDCARD
+            )
+
+    else:
 
         async def resend(message: ChatMessage) -> ChatMessage:
             assert conversation.contact is not None
@@ -1028,6 +1148,8 @@ async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
         resend=resend,
         paths=paths,
         key_of=key_of,
+        scope=scope,
+        resend_unscoped=resend_unscoped,
     )
     ctx.chat.set_active(conversation.key)
 
@@ -1152,6 +1274,49 @@ def _belongs(message: Message, conversation: Conversation) -> bool:
 # --- message paths (the ^P view) -----------------------------------------------------
 
 
+def _sent_scope_line(ctx: AppContext, message: ChatMessage, *, relayed: bool) -> Text | None:
+    """What a message we sent went out under, for the paths dialog — and why it may be lost.
+
+    Built from the scope recorded on the message at send time (:attr:`ChatMessage.scope`),
+    never from the channel's scope now, which may have moved on since. When a scoped
+    message has no relayed copy on record *and* no repeater was ever heard here to carry
+    its region (:meth:`~meshterm.core.region_store.RegionStore.carriers`), the likeliest
+    reason is stated plainly: a scoped flood is relayed only by repeaters carrying its
+    region, and nothing known in earshot does.
+
+    Args:
+        ctx: Shared application context (for the region store).
+        message: The message whose paths are open.
+        relayed: Whether any copy of it was heard coming back.
+
+    Returns:
+        The line, or ``None`` for a message with no recorded scope (inbound, direct, or
+        sent before scopes were recorded).
+    """
+    from ..core.regions import UNSCOPED, Scope
+    from .widgets import scope_text
+
+    if not message.outbound or not message.scope:
+        return None
+    line = Text("sent ", style="muted")
+    if message.scope == WILDCARD:
+        line.append_text(scope_text(UNSCOPED))
+        return line
+    line.append("under ", style="muted")
+    line.append_text(scope_text(Scope("scoped", message.scope)))
+    store = getattr(ctx, "region_store", None)
+    carriers = len(store.carriers(message.scope)) if store is not None else 0
+    if not relayed and not carriers:
+        line.append(" — no repeater known here carries it", style="warn")
+    elif carriers:
+        line.append(
+            f" · {carriers} known repeater{'s' if carriers != 1 else ''} carr"
+            f"{'y' if carriers != 1 else 'ies'} it",
+            style="muted",
+        )
+    return line
+
+
 async def _make_paths_presenter(
     ctx: AppContext,
     conversation: Conversation,
@@ -1291,6 +1456,7 @@ async def _make_paths_presenter(
                 type_of=type_of,
                 key_of=key_of,
                 scope=message_scope(arrivals, ctx.region_store.scope_of),
+                sent_scope=_sent_scope_line(ctx, message, relayed=bool(arrivals)),
             )
         )
 

meshterm/ui/config_editor.py

@@ -123,7 +123,11 @@ async def cached_snapshot(ctx: AppContext, device: Device) -> dict:
         path_hash_mode = await ctx.devstate.path_hash_mode()
     except Exception:  # noqa: BLE001 - fall back to the raw reads below
         pass
-    return await build_snapshot(device, self_info=self_info, path_hash_mode=path_hash_mode)
+    from ..tools.config import learn_default_scope
+
+    snapshot = await build_snapshot(device, self_info=self_info, path_hash_mode=path_hash_mode)
+    learn_default_scope(ctx, snapshot)
+    return snapshot
 
 
 async def edit_config(ctx: AppContext) -> dict[str, Any] | None:
@@ -1024,10 +1028,13 @@ async def _reread(ctx: AppContext, device: Device) -> tuple[dict, dict[str, str]
     Raw rather than through the session cache, which is exactly what just went stale — and
     the cache is dropped on the way, so the next screen that trusts it reads the truth too.
     """
+    from ..tools.config import learn_default_scope
+
     ctx.devstate.invalidate_config()
     async with ctx.ui.busy_overlay():
         snapshot = await build_snapshot(device)
         custom = await device.get_custom_vars()
+    learn_default_scope(ctx, snapshot)
     return snapshot, custom
 
 

meshterm/ui/message_paths_screen.py

@@ -120,6 +120,7 @@ class MessagePathsScreen(Screen):
         type_of: TypeOf | None = None,
         key_of: NameKeyResolver | None = None,
         scope: Scope | None = None,
+        sent_scope: Text | None = None,
     ) -> None:
         """Build the dialog over one message's matched arrivals.
 
@@ -150,6 +151,9 @@ class MessagePathsScreen(Screen):
             scope: The region the message was flooded into, read off its arrivals'
                 frames (:func:`~meshterm.services.message_paths.message_scope`); ``None``
                 where no copy was a flood, and the title then states nothing.
+            sent_scope: For a message we sent, the line saying what it was flooded under
+                (``sent under scope yul``, and why nothing relayed it where that is
+                known), drawn under the summary; ``None`` draws nothing.
         """
         super().__init__()
         self.title = self.titled(scope)
@@ -164,6 +168,7 @@ class MessagePathsScreen(Screen):
         self._destination = destination
         self._type_of = type_of
         self._key_of = key_of
+        self._sent_scope = sent_scope
         self._index = 0
         #: Cells the selected row is shifted left by (reset whenever ↑↓ move), and
         #: how far it *can* shift, measured against the width of the last render.
@@ -287,6 +292,9 @@ class MessagePathsScreen(Screen):
             render_to_ansi(Text(f"“{quoted}”"), width, no_wrap=True),
             render_to_ansi(stamp, width, no_wrap=True),
         ]
+        if self._sent_scope is not None:
+            # Wrapped rather than cut: its tail is the reason a message went nowhere.
+            lines.extend(render_lines(self._sent_scope, width))
         self._cursor = None
         self._list_hidden = False
         if not self._arrivals:
@@ -297,7 +305,9 @@ class MessagePathsScreen(Screen):
                 if self._matched
                 else "No direct-message frames logged in the window."
             )
-            lines.append(render_to_ansi(Text(note, style="muted"), width))
+            # One entry per drawn row: a wrapped render_to_ansi is one string holding a
+            # newline, which the frame counts as a single line and measures as one.
+            lines.extend(render_lines(Text(note, style="muted"), width))
             self._scroll_total = len(lines)
             return lines
 

tests/test_gallery.py

@@ -427,6 +427,39 @@ def _chat(cols: int, rows: int) -> Screen:
     )
 
 
+def _chat_channel_scoped(cols: int, rows: int) -> Screen:
+    """A channel with a send scope: the title's ``· scope`` atom, and a resend under it.
+
+    The widest honest case: a region name at the firmware's 30-byte ceiling in the title,
+    a message that went out under it, and an unscoped one wearing the muted tail — with
+    ^R's hint and the lane's *Resend* chip both live on the scoped newest message.
+    """
+    region = "lakeside-north-shore-emergency"  # 30 bytes, the firmware's longest
+    conv = Conversation(label="Lakeside emergency", is_channel=True, channel_idx=2)
+    messages = [
+        ChatMessage(text="Alice: anyone on the north shore?", is_channel=True),
+        ChatMessage(
+            text="here — relaying for the south side", outbound=True, is_channel=True, scope="*"
+        ),
+        ChatMessage(
+            text="road closed past the marina", outbound=True, is_channel=True, scope=region
+        ),
+    ]
+
+    async def resend(message: ChatMessage) -> ChatMessage:  # pragma: no cover - not pressed
+        return message
+
+    return ChatScreen(
+        conv,
+        messages,
+        send=None,
+        names={},
+        session=_GallerySession(cols, rows),
+        scope=region,
+        resend_unscoped=resend,
+    )
+
+
 class _PickerChat:
     """The unread counter the picker rows read live."""
 
@@ -527,13 +560,13 @@ def _channels_manager(cols: int, rows: int) -> Screen:
     return SelectScreen(title, items, footer_hint=_MANAGER_HINT)
 
 
-def _channel_detail(cols: int, rows: int) -> Screen:
+def _channel_detail(cols: int, rows: int, *, scope: str | None = None) -> Screen:
     """One channel's action page, its vital-signs line above the rows."""
     from meshterm.core.channel_probe import ChannelSlot
     from meshterm.ui.channels import _detail_items, _detail_summary, _LiveStats
 
     slot = ChannelSlot(idx=2, name="Lakeside emergency", secret=bytes(range(16)))
-    ctx = _ChannelsCtx(muted=set())
+    ctx = _ChannelsCtx(muted=set(), scopes={slot.identity: scope} if scope else None)
     stats = _LiveStats(ctx)
     return SelectScreen(
         f"Channel — {slot.name}",
@@ -543,6 +576,28 @@ def _channel_detail(cols: int, rows: int) -> Screen:
     )
 
 
+def _channel_detail_scoped(cols: int, rows: int) -> Screen:
+    """The same page with a send scope at the 30-byte ceiling — the summary's atom and row."""
+    return _channel_detail(cols, rows, scope="lakeside-north-shore-emergency")
+
+
+def _scope_picker(cols: int, rows: int) -> Screen:
+    """The send-scope value picker: no scope, known regions with their carriers, typing one."""
+    from meshterm.ui.channels import _SCOPE_HINT, _scope_items
+
+    ctx = _ChannelsCtx(
+        muted=set(),
+        regions={"lakeside": 3, "lakeside-north-shore-emergency": 1, "harbour": 0},
+    )
+    return SelectScreen(
+        "Send scope — Lakeside emergency",
+        _scope_items(ctx, "lakeside"),
+        prompt="The region this channel's messages flood into:",
+        default="lakeside",
+        footer_hint=_SCOPE_HINT,
+    )
+
+
 class _ChannelsChat:
     """The chat service surface the channel rows read: unread counts and mute state."""
 
@@ -557,11 +612,26 @@ class _ChannelsChat:
 class _ChannelsCtx:
     """The minimal AppContext surface the channel rows and detail page read."""
 
-    def __init__(self, muted: set[str]) -> None:
+    def __init__(
+        self,
+        muted: set[str],
+        *,
+        scopes: dict[str, str] | None = None,
+        regions: dict[str, int] | None = None,
+    ) -> None:
         self.chat = _ChannelsChat(muted)
         self.repo = _ChannelsRepo()
         self.preferences = Preferences()
         self._muted = muted
+        scopes = scopes or {}
+        regions = regions or {}
+        # The region store's reading surface: each channel's scope, and the known regions
+        # with how many repeaters were heard to carry each.
+        self.region_store = SimpleNamespace(
+            channel_scope=lambda identity: scopes.get(identity),
+            names=lambda: list(regions),
+            carriers=lambda name: ("3d63c6429436",) * regions.get(name, 0),
+        )
 
     @property
     def mute_store(self):  # noqa: ANN201 - a stand-in for the real store
@@ -729,6 +799,31 @@ def _message_paths_unknown_scope(cols: int, rows: int) -> Screen:
     )
 
 
+def _message_paths_sent_scope(cols: int, rows: int) -> Screen:
+    """Our scoped send that nothing relayed: the line saying no known repeater carries it."""
+    from meshterm.ui.chat import _sent_scope_line
+
+    message = ChatMessage(
+        text="road closed past the marina",
+        outbound=True,
+        is_channel=True,
+        created_at=utcnow(),
+        scope="lakeside-north-shore-emergency",
+    )
+    ctx = SimpleNamespace(region_store=SimpleNamespace(carriers=lambda name: ()))
+    return MessagePathsScreen(
+        message,
+        [],
+        matched=True,
+        resolve=lambda h: h,
+        prefix_bytes=1,
+        self_name="Homestead",
+        summary="no copies in the packet log",
+        source="Homestead",
+        sent_scope=_sent_scope_line(ctx, message, relayed=False),
+    )
+
+
 def _remote_cli(cols: int, rows: int) -> Screen:
     return RemoteCliScreen(
         node_label="Hilltop-Repeater",
@@ -1195,9 +1290,12 @@ _ENTRIES: list[_Entry] = [
     _Entry("map_panned", _map_panned),
     _Entry("map_find", _map_find),
     _Entry("chat", _chat),
+    _Entry("chat_channel_scoped", _chat_channel_scoped),
     _Entry("chat_picker", _chat_picker),
     _Entry("channels_manager", _channels_manager),
     _Entry("channel_detail", _channel_detail),
+    _Entry("channel_detail_scoped", _channel_detail_scoped),
+    _Entry("scope_picker", _scope_picker),
     _Entry("livefeed", _livefeed),
     _Entry("livefeed_scopes", _livefeed_scopes),
     _Entry("walk", _walk),
@@ -1205,6 +1303,7 @@ _ENTRIES: list[_Entry] = [
     _Entry("message_paths", _message_paths),
     _Entry("message_paths_scoped", _message_paths_scoped),
     _Entry("message_paths_unknown_scope", _message_paths_unknown_scope),
+    _Entry("message_paths_sent_scope", _message_paths_sent_scope),
     _Entry("remote_cli", _remote_cli),
     _Entry("path_composer", _path_composer),
     _Entry("courier_outbox", _courier_outbox),

tests/test_send_scope.py

@@ -0,0 +1,667 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for the sending side of region scopes.
+
+A channel's scope is a window on the companion's one *session* scope: set it, send, put it
+back, with nothing else let through in between. These pin that ordering against the
+simulator — including what happens when a step fails — then everything built on it: the
+recorded scope, the chat's title and its unscoped resend, the paths dialog's line, the
+channel page's picker, the CLI, and Device config teaching the default.
+
+Everything runs against :class:`~meshterm.core.connection.MockDevice` and scratch files;
+nothing transmits.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from meshterm.core.channel_probe import ChannelSlot
+from meshterm.core.config import Settings
+from meshterm.core.connection import (
+    DeviceCommandError,
+    FloodScopeError,
+    MockDevice,
+    firmware_version,
+)
+from meshterm.core.models import ChatMessage, Conversation
+from meshterm.core.preferences import Preferences
+from meshterm.core.region_store import RegionStore
+from meshterm.core.regions import RegionNameError
+from meshterm.core.transmit_lock import TransmitLock
+from meshterm.persistence import db as db_module
+from meshterm.persistence.repository import Repository
+from meshterm.services.chat_service import ChatService
+from meshterm.services.event_hub import EventHub
+from meshterm.ui.chat import ChatScreen, _sent_scope_line
+from meshterm.ui.tui import Choice, fkeys
+
+# -- a device that writes down what it was asked, in order --------------------------------
+
+
+class _TracingDevice(MockDevice):
+    """The simulator, keeping a log of every scope change and send in the order made.
+
+    ``refuse`` makes the named scope command fail the way old firmware does; ``version``
+    is what the device query reports as its firmware build.
+    """
+
+    def __init__(self, *, refuse: set[str] | None = None, version: str = "mock") -> None:
+        super().__init__()
+        self.trail: list[tuple] = []
+        self._refuse = refuse or set()
+        self._version = version
+        self.fail_restore = False
+
+    async def set_flood_scope(self, region: str | None) -> None:
+        key = region if region else None
+        if key in self._refuse or (key is None and self.fail_restore):
+            raise DeviceCommandError("ERR_CODE_UNSUPPORTED_CMD")
+        await super().set_flood_scope(region)
+        self.trail.append(("scope", key))
+
+    async def send_channel_message(self, index: int, text: str) -> None:
+        await super().send_channel_message(index, text)
+        self.trail.append(("channel", text, self._send_scope))
+
+    async def send_direct_message(self, contact, text):  # noqa: ANN001, ANN201
+        ack = await super().send_direct_message(contact, text)
+        self.trail.append(("dm", text, self._send_scope))
+        return ack
+
+    async def get_device_info(self) -> dict:
+        return {**await super().get_device_info(), "ver": self._version}
+
+
+# -- the transmit lock -----------------------------------------------------------------
+
+
+async def test_transmit_lock_reenters_for_its_holder_and_blocks_everyone_else() -> None:
+    """The scoped send re-enters its own lock; another task waits for the whole window."""
+    lock = TransmitLock()
+    order: list[str] = []
+
+    async def other() -> None:
+        async with lock.held():
+            order.append("other")
+
+    async with lock.held():
+        async with lock.held():  # the same task: must not deadlock
+            order.append("inner")
+        task = asyncio.ensure_future(other())
+        await asyncio.sleep(0.01)
+        assert order == ["inner"]  # still waiting behind us
+        order.append("outer-done")
+    await task
+    assert order == ["inner", "outer-done", "other"]
+    assert not lock.locked
+
+
+# -- the three-step window ---------------------------------------------------------------
+
+
+async def test_scoped_send_sets_then_sends_then_restores() -> None:
+    """Scope set → channel send (under it) → restored to the default, in that order."""
+    device = _TracingDevice()
+    await device.send_channel_in_scope(0, "hello", "lakeside")
+    assert device.trail == [
+        ("scope", "lakeside"),
+        ("channel", "hello", "lakeside"),
+        ("scope", None),
+    ]
+    assert device._send_scope == ""
+
+
+async def test_plain_send_never_touches_the_session_scope() -> None:
+    """No scope asked: the message goes out under the default and nothing is set."""
+    device = _TracingDevice()
+    await device.send_channel_in_scope(0, "hello", None)
+    assert device.trail == [("channel", "hello", "")]
+
+
+async def test_a_dm_cannot_interleave_into_the_window() -> None:
+    """A DM started mid-window waits until the scope is restored, then goes out unscoped."""
+    device = _TracingDevice()
+    await device.connect()
+    alice = next(c for c in await device.get_contacts() if c.name == "Alice")
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    original = device.send_channel_message
+
+    async def slow_send(index: int, text: str) -> None:
+        entered.set()
+        await release.wait()
+        await original(index, text)
+
+    device.send_channel_message = slow_send  # type: ignore[method-assign]
+    scoped = asyncio.ensure_future(device.send_channel_in_scope(0, "scoped", "lakeside"))
+    await entered.wait()
+    dm = asyncio.ensure_future(device.send_direct_message(alice, "dm"))
+    await asyncio.sleep(0.01)
+    assert not dm.done()  # held behind the window
+    release.set()
+    await asyncio.gather(scoped, dm)
+    assert device.trail == [
+        ("scope", "lakeside"),
+        ("channel", "scoped", "lakeside"),
+        ("scope", None),
+        ("dm", "dm", ""),
+    ]
+
+
+async def test_a_refused_scope_sends_nothing() -> None:
+    """Old firmware refuses the scope: FloodScopeError, and no message went out at all."""
+    device = _TracingDevice(refuse={"lakeside"})
+    with pytest.raises(FloodScopeError) as caught:
+        await device.send_channel_in_scope(0, "hello", "lakeside")
+    assert caught.value.scope == "lakeside"
+    assert "1.10" in str(caught.value) and "Nothing was sent" in str(caught.value)
+    assert device.trail == []
+
+
+async def test_a_send_that_fails_still_restores_the_scope() -> None:
+    """The channel send raising must not leave the session scope on the channel's region."""
+    device = _TracingDevice()
+
+    async def boom(index: int, text: str) -> None:
+        raise DeviceCommandError("rejected")
+
+    device.send_channel_message = boom  # type: ignore[method-assign]
+    with pytest.raises(DeviceCommandError):
+        await device.send_channel_in_scope(0, "hello", "lakeside")
+    assert device.trail == [("scope", "lakeside"), ("scope", None)]
+
+
+async def test_a_failed_restore_is_repaired_before_the_next_transmission() -> None:
+    """A restore that failed is retried by the next send, before that send goes out."""
+    device = _TracingDevice()
+    device.fail_restore = True
+    await device.send_channel_in_scope(0, "first", "lakeside")  # the message did go
+    assert device._send_scope == "lakeside"
+    device.fail_restore = False
+    await device.send_advert()
+    assert device.trail[-1] == ("scope", None)
+    assert device._send_scope == ""
+
+
+async def test_an_invalid_region_is_refused_before_the_radio_is_asked() -> None:
+    """A name the firmware couldn't hold never reaches the device."""
+    device = _TracingDevice()
+    with pytest.raises(RegionNameError):
+        await device.send_channel_in_scope(0, "hello", "two words")
+    assert device.trail == []
+
+
+# -- unscoped (*) and the firmware that can or can't do it ---------------------------------
+
+
+def test_firmware_version_reads_the_leading_release() -> None:
+    """``v1.15.0``, ``1.16.2-dev`` and ``1.9`` all compare; a non-version is None."""
+    assert firmware_version({"ver": "v1.15.0"}) == (1, 15, 0)
+    assert firmware_version({"ver": "1.16.2-dev"}) == (1, 16, 2)
+    assert firmware_version({"ver": "1.9"}) == (1, 9, 0)
+    assert firmware_version({"ver": "mock"}) is None
+    assert firmware_version({}) is None
+
+
+async def test_unscoped_uses_the_override_on_new_firmware() -> None:
+    """1.16+: ``*`` is set as the session scope, the message goes, the scope is restored."""
+    device = _TracingDevice(version="v1.16.0")
+    await device.send_channel_in_scope(0, "hello", "*")
+    assert device.trail == [("scope", "*"), ("channel", "hello", "*"), ("scope", None)]
+
+
+async def test_unscoped_on_old_firmware_without_a_default_is_the_plain_send() -> None:
+    """1.15 with no default scope: a plain flood already is unscoped, so it just goes."""
+    device = _TracingDevice(version="v1.15.0")
+    await device.send_channel_in_scope(0, "hello", "*")
+    assert device.trail == [("channel", "hello", "")]
+
+
+async def test_unscoped_on_old_firmware_over_a_default_is_refused() -> None:
+    """1.15 with a default set can't override it: refused, and nothing was sent."""
+    device = _TracingDevice(version="v1.15.0")
+    await device.set_default_flood_scope("lakeside")
+    with pytest.raises(FloodScopeError) as caught:
+        await device.send_channel_in_scope(0, "hello", "*")
+    assert "1.16" in str(caught.value) and "lakeside" in str(caught.value)
+    assert device.trail == []
+
+
+# -- the chat service: which scope, and what gets recorded ----------------------------------
+
+
+class _Ctx:
+    """The slice of AppContext the chat service reads, with a real region store."""
+
+    def __init__(self, device: MockDevice, repo: Repository, tmp: Path) -> None:
+        self._device = device
+        self.repo = repo
+        self.log = logging.getLogger("test.scope")
+        self.profile_name = None
+        self.settings = Settings()
+        self.preferences = Preferences()
+        self.events = EventHub(self)
+        self.region_store = RegionStore(tmp / "regions.json")
+        self.devstate = SimpleNamespace(default_scope=self._default_scope)
+        self.default = ""
+
+    async def _default_scope(self) -> str:
+        return self.default
+
+    async def device(self) -> MockDevice:
+        await self._device.connect()
+        return self._device
+
+
+@pytest.fixture()
+def repo(tmp_path: Path) -> Repository:
+    """A Repository on a scratch database, closed when the test finishes."""
+    r = Repository(tmp_path / "scope.db")
+    yield r
+    r.close()
+
+
+async def test_a_channel_send_goes_under_its_scope_and_records_it(
+    tmp_path: Path, repo: Repository
+) -> None:
+    """The channel's scope is looked up by identity, sent under, and kept with the message."""
+    device = _TracingDevice()
+    ctx = _Ctx(device, repo, tmp_path)
+    chat = ChatService(ctx)
+    channel_id = await chat.channel_id_for(0)
+    ctx.region_store.set_channel_scope(channel_id, "lakeside")
+
+    sent = await chat.send_channel(0, "hello", label="Public")
+    assert sent.scope == "lakeside"
+    assert ("channel", "hello", "lakeside") in device.trail
+    stored = repo.recent_chat_messages(is_channel=True, channel_id=channel_id)
+    assert stored[-1].scope == "lakeside" and stored[-1].row_id == sent.row_id
+
+
+async def test_a_channel_without_a_scope_records_the_default_or_unscoped(
+    tmp_path: Path, repo: Repository
+) -> None:
+    """No channel scope: recorded as the device default, or ``*`` when there is none."""
+    ctx = _Ctx(_TracingDevice(), repo, tmp_path)
+    chat = ChatService(ctx)
+    assert (await chat.send_channel(0, "a")).scope == "*"
+    ctx.default = "harbour"
+    assert (await chat.send_channel(0, "b")).scope == "harbour"
+
+
+async def test_the_one_shot_override_sends_unscoped(tmp_path: Path, repo: Repository) -> None:
+    """``scope="*"`` overrides the channel's scope for one message and is recorded."""
+    device = _TracingDevice()
+    ctx = _Ctx(device, repo, tmp_path)
+    chat = ChatService(ctx)
+    ctx.region_store.set_channel_scope(await chat.channel_id_for(0), "lakeside")
+    sent = await chat.send_channel(0, "again", scope="*")
+    assert sent.scope == "*"
+    assert ("channel", "again", "*") in device.trail
+
+
+async def test_a_refused_scope_records_nothing(tmp_path: Path, repo: Repository) -> None:
+    """A message the radio wouldn't send under its scope is not in the transcript either."""
+    ctx = _Ctx(_TracingDevice(refuse={"lakeside"}), repo, tmp_path)
+    chat = ChatService(ctx)
+    channel_id = await chat.channel_id_for(0)
+    ctx.region_store.set_channel_scope(channel_id, "lakeside")
+    with pytest.raises(FloodScopeError):
+        await chat.send_channel(0, "hello")
+    assert repo.recent_chat_messages(is_channel=True, channel_id=channel_id) == []
+
+
+# -- the schema --------------------------------------------------------------------------
+
+
+def test_a_v16_database_gains_the_scope_column(tmp_path: Path) -> None:
+    """Opening a database from before v17 adds ``messages.scope``, old rows NULL."""
+    path = tmp_path / "old.db"
+    conn = db_module.connect(path)
+    conn.execute("ALTER TABLE messages DROP COLUMN scope")
+    conn.execute(
+        "INSERT INTO messages (outbound, is_channel, text, created_at) "
+        "VALUES (1, 1, 'old', '2026-09-01T00:00:00+00:00')"
+    )
+    conn.commit()
+    conn.close()
+
+    conn = db_module.connect(path)
+    cols = {row["name"] for row in conn.execute("PRAGMA table_info(messages)")}
+    assert "scope" in cols
+    assert conn.execute("SELECT scope FROM messages").fetchone()["scope"] is None
+    version = conn.execute("SELECT value FROM schema_meta WHERE key = 'version'").fetchone()
+    assert version["value"] == "17"
+    conn.close()
+
+
+# -- the chat screen -----------------------------------------------------------------------
+
+
+class _Session:
+    def __init__(self, answer: bool = True) -> None:
+        self.answer = answer
+        self.dialogs: list[str] = []
+
+    def invalidate(self) -> None:
+        pass
+
+    def run_detached(self, work):  # noqa: ANN001, ANN201
+        return asyncio.ensure_future(work)
+
+    async def button_dialog(self, prompt, buttons, **kwargs):  # noqa: ANN001, ANN003, ANN201
+        self.dialogs.append(kwargs.get("title", ""))
+        return self.answer
+
+
+def _channel_chat(session, messages, *, scope="lakeside", resend=None) -> ChatScreen:  # noqa: ANN001
+    conv = Conversation(label="#ops", is_channel=True, channel_idx=1, channel_id="cid")
+    return ChatScreen(
+        conv,
+        messages,
+        send=None,
+        names={},
+        session=session,
+        scope=scope,
+        resend_unscoped=resend,
+    )
+
+
+def test_the_chat_title_carries_the_scope_as_a_status_atom() -> None:
+    """``#ops · scope lakeside``; no atom at all when the channel has no scope."""
+    assert _channel_chat(_Session(), []).title == "#ops · scope lakeside"
+    assert _channel_chat(_Session(), [], scope=None).title == "#ops"
+
+
+async def test_resend_unscoped_targets_only_a_newest_scoped_message() -> None:
+    """^R offers the newest sent message when it went under a region, and nothing else."""
+
+    async def resend(message: ChatMessage) -> ChatMessage:
+        return ChatMessage(text=message.text, outbound=True, is_channel=True, scope="*")
+
+    scoped = ChatMessage(text="hi", outbound=True, is_channel=True, scope="lakeside")
+    screen = _channel_chat(_Session(), [scoped], resend=resend)
+    assert screen._retry_target() is scoped
+    assert "^R resend unscoped" in screen.footer_hint
+    lane = screen.fkey_lane
+    assert lane[2].opp_label == "Resend" and lane[2].opp_enabled
+
+    screen.handle("retry")
+    await asyncio.sleep(0.05)
+    assert screen._messages[-1].scope == "*"
+    assert screen._retry_target() is None  # the unscoped copy is now the newest
+    assert "^R" not in screen.footer_hint
+    assert "unscoped" in _plain_lines(screen)
+
+
+async def test_a_declined_resend_sends_nothing() -> None:
+    """Cancel on the amber confirm leaves the transcript as it was."""
+    calls: list[str] = []
+
+    async def resend(message: ChatMessage) -> ChatMessage:
+        calls.append(message.text)
+        return message
+
+    scoped = ChatMessage(text="hi", outbound=True, is_channel=True, scope="lakeside")
+    session = _Session(answer=False)
+    screen = _channel_chat(session, [scoped], resend=resend)
+    screen.handle("retry")
+    await asyncio.sleep(0.05)
+    assert session.dialogs == ["Resend unscoped"] and calls == []
+
+
+def test_a_channel_with_nothing_to_resend_dims_the_chip() -> None:
+    """An unscoped newest message leaves Resend on the lane, dim — present but unavailable."""
+    unscoped = ChatMessage(text="hi", outbound=True, is_channel=True, scope="*")
+
+    async def resend(message: ChatMessage) -> ChatMessage:  # pragma: no cover - never called
+        return message
+
+    screen = _channel_chat(_Session(), [unscoped], resend=resend)
+    lane = screen.fkey_lane
+    assert lane[2].opp_label == "Resend" and not lane[2].opp_enabled
+    assert fkeys.lane_text(lane, shifted=True).plain
+
+
+def _plain_lines(screen: ChatScreen) -> str:
+    from tests.conftest import plain
+
+    return plain(screen.render_body(72))
+
+
+# -- the paths dialog's line ----------------------------------------------------------------
+
+
+def test_the_paths_line_says_when_no_known_repeater_carries_the_scope(tmp_path: Path) -> None:
+    """Scoped, nothing relayed, no carrier known: said plainly; a carrier changes the tail."""
+    store = RegionStore(tmp_path / "regions.json")
+    ctx = SimpleNamespace(region_store=store)
+    sent = ChatMessage(text="hi", outbound=True, is_channel=True, scope="lakeside")
+
+    line = _sent_scope_line(ctx, sent, relayed=False)
+    assert line.plain == "sent under scope lakeside — no repeater known here carries it"
+
+    store.learn("lakeside", "repeater", repeater="3d63c6429436")
+    assert _sent_scope_line(ctx, sent, relayed=False).plain == (
+        "sent under scope lakeside · 1 known repeater carries it"
+    )
+    unscoped = ChatMessage(text="hi", outbound=True, is_channel=True, scope="*")
+    assert _sent_scope_line(ctx, unscoped, relayed=True).plain == "sent unscoped"
+    inbound = ChatMessage(text="hi", is_channel=True)
+    assert _sent_scope_line(ctx, inbound, relayed=True) is None
+
+
+# -- the channel page's picker ---------------------------------------------------------------
+
+
+class _PickerUi:
+    """Replays scripted select/text answers; records the rows each select offered."""
+
+    def __init__(self, selects: list, texts: list) -> None:
+        self._selects = list(selects)
+        self._texts = list(texts)
+        self.offered: list[list] = []
+
+    async def select(self, title, items, *, prompt="", default=None):  # noqa: ANN001, ANN201
+        self.offered.append(items)
+        return self._selects.pop(0)
+
+    async def text(self, title, *, prompt="", default="", validate=None):  # noqa: ANN001, ANN201
+        answer = self._texts.pop(0)
+        if answer is not None and validate is not None:
+            assert validate(answer) is True
+        return answer
+
+
+async def test_the_picker_sets_types_steps_back_and_clears(tmp_path: Path) -> None:
+    """Pick a known region; type one (Esc on the field comes back to the list); clear."""
+    from meshterm.ui.channels import _NO_SCOPE, _TYPE_REGION, _detail_summary, _pick_scope
+
+    store = RegionStore(tmp_path / "regions.json")
+    store.learn("harbour", "repeater", repeater="3d63c6429436")
+    slot = ChannelSlot(idx=1, name="Ops", secret=bytes(range(16)))
+
+    ui = _PickerUi(selects=["harbour"], texts=[])
+    ctx = SimpleNamespace(region_store=store, ui=ui)
+    assert await _pick_scope(ctx, slot) is True
+    assert store.channel_scope(slot.identity) == "harbour"
+    labels = [item.label.plain for item in ui.offered[0] if isinstance(item, Choice)]
+    assert labels[1] == "harbour  · 1 repeater"
+
+    ctx.ui = _PickerUi(selects=[_TYPE_REGION, _TYPE_REGION], texts=[None, "yul"])
+    assert await _pick_scope(ctx, slot) is True
+    assert store.channel_scope(slot.identity) == "yul"
+    assert "typed" in store.get("yul").sources
+
+    ctx.ui = _PickerUi(selects=[_NO_SCOPE], texts=[])
+    assert await _pick_scope(ctx, slot) is True
+    assert store.channel_scope(slot.identity) is None
+
+    ctx.ui = _PickerUi(selects=[None], texts=[])
+    assert await _pick_scope(ctx, slot) is False  # Esc keeps
+
+    store.set_channel_scope(slot.identity, "yul")
+    stats = SimpleNamespace(get=lambda identity: None)
+    summary_ctx = SimpleNamespace(
+        region_store=store,
+        chat=SimpleNamespace(unread=lambda key: 0),
+        mute_store=SimpleNamespace(is_muted=lambda identity: False),
+    )
+    assert "scope yul" in _detail_summary(summary_ctx, slot, stats)
+
+
+# -- the CLI ------------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):  # noqa: ANN201
+    """The real Typer app against the simulator, in a config directory of its own."""
+    from typer.testing import CliRunner
+
+    from meshterm.cli import app
+
+    monkeypatch.setenv("MESHTERM_HOME", str(tmp_path / "home"))
+    runner = CliRunner()
+
+    def invoke(*args: str):  # noqa: ANN202
+        return runner.invoke(app, ["--mock", "--db", str(tmp_path / "test.db"), *args])
+
+    return invoke
+
+
+async def test_cli_channel_scope_reads_sets_and_clears(tmp_path: Path) -> None:
+    """``channels scope`` reads (exit 5 for none), sets, clears — keyed by the channel."""
+    from io import StringIO
+
+    from rich.console import Console
+
+    from meshterm.context import AppContext
+    from meshterm.core.admin_store import AdminStore
+    from meshterm.core.device_store import DeviceStore
+    from meshterm.tools.channels import ChannelsTool
+
+    settings = Settings(config_dir=tmp_path, db_path=tmp_path / "chan.db")
+    ctx = AppContext(
+        console=Console(file=StringIO()),
+        settings=settings,
+        repo=Repository(settings.db_path),
+        device_store=DeviceStore(tmp_path / "devices.json"),
+        admin_store=AdminStore(tmp_path / "admin.json"),
+        mock=True,
+    )
+    tool = ChannelsTool()
+    try:
+        await tool.run(ctx, {"cli_action": "add", "index": 1, "name": "Ops", "secret": None})
+        read = {"cli_action": "scope", "index": 1, "region": None, "clear": False}
+        empty = await tool.run(ctx, read)
+        assert empty.summary["scope"] is None and empty.exit_code == 5
+
+        await tool.run(ctx, {**read, "region": "lakeside"})
+        again = await tool.run(ctx, read)
+        assert again.summary["scope"] == "lakeside" and again.exit_code == 0
+        assert again.report[0].values["channel"].name == "Ops"
+
+        listed = await tool.run(ctx, {"cli_action": "list"})
+        assert [row["scope"] for row in listed.report[0].rows] == ["lakeside"]
+
+        cleared = await tool.run(ctx, {**read, "clear": True})
+        assert cleared.summary["scope"] is None and cleared.exit_code == 0
+    finally:
+        ctx.repo.close()
+
+
+def test_cli_channel_scope_refuses_bad_arguments(run) -> None:  # noqa: ANN001
+    """A bad name or a name with --clear is a usage error; an empty slot is exit 5 + a doc."""
+    assert run("channels", "scope", "0", "two words").exit_code == 2
+    assert run("channels", "scope", "0", "yul", "--clear").exit_code == 2
+    result = run("--json", "channels", "scope", "0")
+    assert result.exit_code == 5
+    assert json.loads(result.stdout) == {"channel": None, "scope": None}
+
+
+def test_cli_send_scope_is_a_channel_option_and_reports_its_scope(run) -> None:  # noqa: ANN001
+    """``--scope *`` goes out unscoped and says so; on a DM or a bad name it's a usage error."""
+    doc = json.loads(run("--json", "chat", "send", "hi", "--channel", "0", "--scope", "*").stdout)
+    assert doc["scope"] == {"state": "unscoped", "region": None, "code": None}
+    doc = json.loads(run("--json", "chat", "send", "hi", "--channel", "0", "--scope", "yul").stdout)
+    assert doc["scope"]["region"] == "yul"
+    assert run("chat", "send", "hi", "--to", "Alice", "--scope", "yul").exit_code == 2
+    assert run("chat", "send", "hi", "--channel", "0", "--scope", "$x").exit_code == 2
+
+
+# -- Device config teaches the default ---------------------------------------------------------
+
+
+def test_a_config_read_learns_the_default_scope(tmp_path: Path) -> None:
+    """A snapshot carrying the default scope teaches the store and seeds the cache."""
+    from meshterm.tools.config import learn_default_scope
+
+    noted: list = []
+    store = RegionStore(tmp_path / "regions.json")
+    ctx = SimpleNamespace(
+        region_store=store, devstate=SimpleNamespace(note_default_scope=noted.append)
+    )
+    learn_default_scope(ctx, {"flood_scope": "#harbour"})
+    assert store.get("harbour").sources == ("default",)
+    assert noted == ["harbour"]
+    learn_default_scope(ctx, {})  # the read failed: nothing learned, nothing noted
+    assert noted == ["harbour"]
+    learn_default_scope(ctx, {"flood_scope": ""})
+    assert noted == ["harbour", ""]
+
+
+async def test_a_channel_chat_sends_under_its_scope_end_to_end(tmp_path: Path) -> None:
+    """Typing in a scoped channel's chat, through the real session, sends under the scope."""
+    from io import StringIO
+
+    from prompt_toolkit.input.defaults import create_pipe_input
+    from prompt_toolkit.output import DummyOutput
+    from rich.console import Console
+
+    from meshterm.context import AppContext
+    from meshterm.core.admin_store import AdminStore
+    from meshterm.core.device_store import DeviceStore
+    from meshterm.ui.chat import open_chat
+    from meshterm.ui.surface import TuiUi
+    from meshterm.ui.tui.session import TuiSession
+
+    settings = Settings(config_dir=tmp_path, db_path=tmp_path / "e2e.db")
+    ctx = AppContext(
+        console=Console(file=StringIO()),
+        settings=settings,
+        repo=Repository(settings.db_path),
+        device_store=DeviceStore(tmp_path / "devices.json"),
+        admin_store=AdminStore(tmp_path / "admin.json"),
+        mock=True,
+    )
+    try:
+        device = await ctx.device()
+        await device.set_channel(0, "Public", bytes(16))
+        channel_id = await ctx.chat.channel_id_for(0)
+        ctx.region_store.set_channel_scope(channel_id, "lakeside")
+        conv = Conversation(label="Public", is_channel=True, channel_idx=0, channel_id=channel_id)
+        with create_pipe_input() as inp:
+            session = TuiSession(input=inp, output=DummyOutput())
+            ctx.ui = TuiUi(session)
+
+            async def main() -> None:
+                inp.send_text("hi\r")
+                task = asyncio.ensure_future(open_chat(ctx, conv))
+                await asyncio.sleep(0.3)
+                inp.send_text("\x1b")
+                await task
+
+            await asyncio.wait_for(session.run(main()), timeout=5)
+
+        stored = ctx.repo.recent_chat_messages(is_channel=True, channel_id=channel_id)
+        assert [(m.text, m.scope) for m in stored] == [("hi", "lakeside")]
+        assert device.sent_channel == [(0, "hi", "lakeside")]
+        assert device._send_scope == ""  # restored after the send
+    finally:
+        await ctx.aclose()