diff --git a/.gitignore b/.gitignore
index 32d46ef4..c7ee3ad0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,6 +12,7 @@
 
 # Used by dotenv library to load environment variables.
 .env
+.env-*
 
 # Ignore Byebug command history file.
 .byebug_history
diff --git a/data/config.sh b/data/config.sh
new file mode 100755
index 00000000..e72cec71
--- /dev/null
+++ b/data/config.sh
@@ -0,0 +1,37 @@
+#!/usr/bin/env bash
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -euo pipefail
+
+# Usage: config.sh [profile | PATH] [args for mesh_env...]
+# Default env file: repo-root .env. With profile: .env-<profile> (letters, digits, _, -).
+# Or pass a path to any .env file as the first argument (not a bare profile token).
+
+_script_dir="$(cd "$(dirname "$0")" && pwd)"
+cd "${_script_dir}"
+
+_repo_root="$(cd "${_script_dir}/.." && pwd)"
+# shellcheck source=potato_mesh_env.sh
+source "${_script_dir}/potato_mesh_env.sh"
+
+potato_mesh_resolve_env_file config "${_repo_root}" "$@"
+shift "${_potato_mesh_env_shift}"
+
+potato_mesh_source_env_if_exists "${_env_file}"
+
+potato_mesh_venv_and_requirements "${_script_dir}/requirements.txt"
+
+export PYTHONPATH="${_script_dir}"
+exec python -m mesh_env "${_env_file}" "$@"
diff --git a/data/mesh.sh b/data/mesh.sh
index 451a5775..bbe61e24 100755
--- a/data/mesh.sh
+++ b/data/mesh.sh
@@ -15,8 +15,21 @@
 
 set -euo pipefail
 
-python -m venv .venv
-source .venv/bin/activate
-pip install -U pip
-pip install -r "$(dirname "$0")/requirements.txt"
+# Usage: mesh.sh [profile]
+# Loads repo-root .env, or .env-<profile> when profile is given (letters, digits, _, -).
+
+_script_dir="$(cd "$(dirname "$0")" && pwd)"
+_repo_root="$(cd "${_script_dir}/.." && pwd)"
+# shellcheck source=potato_mesh_env.sh
+source "${_script_dir}/potato_mesh_env.sh"
+
+potato_mesh_resolve_env_file mesh "${_repo_root}" "$@"
+shift "${_potato_mesh_env_shift}"
+if [[ $# -gt 0 ]]; then
+  echo "mesh.sh: unexpected arguments (only optional profile name is supported): $*" >&2
+  exit 2
+fi
+
+potato_mesh_source_env_if_exists "${_env_file}"
+potato_mesh_venv_and_requirements "${_script_dir}/requirements.txt"
 exec python mesh.py
diff --git a/data/mesh_env/__init__.py b/data/mesh_env/__init__.py
new file mode 100644
index 00000000..2517dffc
--- /dev/null
+++ b/data/mesh_env/__init__.py
@@ -0,0 +1,15 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Interactive configuration helpers for the mesh ingestor (standalone from mesh_ingestor)."""
diff --git a/data/mesh_env/__main__.py b/data/mesh_env/__main__.py
new file mode 100644
index 00000000..83f929c5
--- /dev/null
+++ b/data/mesh_env/__main__.py
@@ -0,0 +1,451 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Interactive CLI to write a mesh ingestor ``.env`` file (path optional on the command line)."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import sys
+from pathlib import Path
+
+from . import env_file, meshcore_probe, meshtastic_probe, tui
+from .connection_parse import connection_kind
+from .devices import list_serial_paths, scan_ble_devices
+
+# Matches local `cd web && ./app.sh` / README examples (Sinatra default port).
+_LOCAL_INSTANCE_DOMAIN = "http://127.0.0.1:41447"
+_LOCAL_INSTANCE_NORMALIZED = frozenset(
+    {
+        "http://127.0.0.1:41447",
+        "http://localhost:41447",
+    }
+)
+
+
+def _normalized_instance_url(value: str) -> str:
+    return (value or "").strip().rstrip("/").lower()
+
+
+def _instance_domain_prefers_local_default(existing_domain: str) -> bool:
+    """True when *existing_domain* is empty or already the usual local dev URL."""
+
+    n = _normalized_instance_url(existing_domain)
+    if not n:
+        return True
+    return n in _LOCAL_INSTANCE_NORMALIZED
+
+
+def _default_env_path() -> Path:
+    """``<repository-root>/.env`` (``mesh_env/__main__.py`` → parents[2] is repo root)."""
+
+    return (Path(__file__).resolve().parents[2] / ".env").resolve()
+
+
+def _parse_env_file_arg(argv: list[str] | None) -> Path:
+    """Parse optional ``PATH`` positional; default :func:`_default_env_path`."""
+
+    parser = argparse.ArgumentParser(
+        description="Interactive wizard to write a potato-mesh ingestor .env file.",
+    )
+    parser.add_argument(
+        "env_file",
+        nargs="?",
+        default=None,
+        metavar="PATH",
+        help="Env file to read/write (default: <repository>/.env)",
+    )
+    ns = parser.parse_args(argv)
+    raw = (ns.env_file or "").strip()
+    if not raw:
+        return _default_env_path()
+    return Path(raw).expanduser().resolve()
+
+
+def _profile_label_from_env_path(path: Path) -> str | None:
+    """Return the profile segment for ``.env-<profile>`` filenames; ``None`` for default ``.env``."""
+
+    name = path.name
+    prefix = ".env-"
+    if not name.startswith(prefix):
+        return None
+    rest = name[len(prefix) :].strip()
+    return rest or None
+
+
+def _csv_casefold_frozen(raw: str) -> frozenset[str]:
+    return frozenset(p.casefold() for p in (x.strip() for x in raw.split(",")) if p)
+
+
+_HINT_BLE_ADDR = "Use the address your OS shows for the radio (MAC or UUID)."
+_HINT_SERIAL_PATH = (
+    "Path to the USB serial device (Linux often /dev/ttyACM0 or ttyUSB0)."
+)
+
+
+def _prompt_ble_address(message: str, existing: str) -> str:
+    return tui.text(message, existing, hint=_HINT_BLE_ADDR)
+
+
+def _prompt_serial_path(message: str, default: str) -> str:
+    return tui.text(message, default, hint=_HINT_SERIAL_PATH)
+
+
+def _pick_connection_string(existing: str) -> str:
+    kind = tui.select(
+        "How should the ingestor connect to the radio?",
+        [
+            ("Serial USB", "serial"),
+            ("Bluetooth (BLE)", "ble"),
+            ("TCP (host:port or [IPv6]:port)", "tcp"),
+        ],
+        default_value=connection_kind(existing),
+        hint=(
+            "Serial is a direct USB device. BLE uses a short scan to list nearby radios. "
+            "TCP targets an IP host:port (tunnel, proxy, or meshtasticd)."
+        ),
+    )
+    if kind == "ble":
+        tui.print_info("\nScanning for BLE devices (8s)…")
+        try:
+            found = asyncio.run(scan_ble_devices(8.0))
+        except Exception as exc:
+            tui.print_info(f"Bleak scan failed: {exc}")
+            return _prompt_ble_address("Enter BLE MAC or UUID", existing)
+        if not found:
+            tui.print_info("No devices found.")
+            return _prompt_ble_address("Enter BLE MAC or UUID", existing)
+        choices: list[tuple[str, str]] = [
+            (f"{name}  ({addr})", addr) for name, addr in found
+        ]
+        choices.append(("Type address manually…", "__manual__"))
+        default_ble = found[0][1]
+        exn = existing.strip()
+        if exn:
+            for _name, addr in found:
+                if addr.upper() == exn.upper() or addr == exn:
+                    default_ble = addr
+                    break
+        pick = tui.select(
+            "Choose a BLE device:",
+            choices,
+            default_value=default_ble,
+            hint="Pick your radio from the scan, or type an address if it is missing.",
+        )
+        if pick == "__manual__" or pick is None:
+            return _prompt_ble_address("BLE MAC or UUID", existing)
+        return pick
+    if kind == "tcp":
+        return tui.text(
+            "TCP target (e.g. 192.168.1.5:4403 or mesh.local:4403)",
+            existing,
+            hint="Host and port where the radio or bridge accepts a TCP connection (often :4403).",
+        )
+
+    paths = list_serial_paths()
+    if not paths:
+        return _prompt_serial_path("Serial device path", existing or "/dev/ttyACM0")
+    choices = [(p, p) for p in paths]
+    choices.append(("Other… type a custom path", "__custom__"))
+    default_val = existing if existing in paths else paths[0]
+    pick = tui.select(
+        "Choose a serial device:",
+        choices,
+        default_value=default_val,
+        hint="Choose the port your radio appears as, or type another path if needed.",
+    )
+    if pick == "__custom__" or pick is None:
+        return _prompt_serial_path("Serial device path", existing or paths[0])
+    return pick
+
+
+def _channel_filter_value(
+    rows: list[tuple[int, str]],
+    label_short: str,
+    *,
+    all_label: str,
+    existing: str,
+    field_hint: str,
+) -> str:
+    """Build ``ALLOWED_CHANNELS`` or ``HIDDEN_CHANNELS`` (comma-separated names).
+
+    *field_hint* is dim explanatory text for what this env var does in the ingestor.
+    """
+
+    if not rows:
+        return tui.text(
+            f"{label_short} — comma-separated channel names (leave empty for no filter)",
+            existing,
+            hint=field_hint,
+        )
+
+    ex_stripped = existing.strip()
+    if not ex_stripped:
+        mode_default = "all"
+    else:
+        parts = [x.strip() for x in existing.split(",") if x.strip()]
+        row_names_cf = {name.casefold() for _, name in rows}
+        mode_default = (
+            "pick"
+            if parts and all(p.casefold() in row_names_cf for p in parts)
+            else "type"
+        )
+
+    mode = tui.select(
+        f"Configure {label_short}",
+        [
+            (all_label, "all"),
+            ("Pick from discovered channels (checkboxes)", "pick"),
+            ("Type comma-separated names", "type"),
+        ],
+        default_value=mode_default,
+        hint=field_hint,
+    )
+    if mode == "type" or mode is None:
+        return tui.text(
+            "Comma-separated channel names",
+            existing,
+            hint="Names are matched case-insensitively against what the radio reports.",
+        )
+    if mode == "all":
+        return ""
+
+    choices: list[tuple[str, str]] = []
+    for idx, name in rows:
+        title = f"LoRa index {idx} — {name}"
+        choices.append((title, name))
+    picked = tui.checkbox(
+        "Toggle channels with ↑/↓ and Space; Enter confirms.",
+        choices,
+        prechecked_values=_csv_casefold_frozen(existing),
+        hint=(
+            "Leaving none checked clears this filter (same as choosing allow-all / hide-none above)."
+        ),
+    )
+    if not picked:
+        return ""
+    # Preserve stable order by first appearance in *rows*
+    order = [name for _, name in rows]
+    seen: set[str] = set()
+    ordered: list[str] = []
+    for n in order:
+        if n in picked and n not in seen:
+            ordered.append(n)
+            seen.add(n)
+    for n in picked:
+        if n not in seen:
+            ordered.append(n)
+            seen.add(n)
+    return ",".join(ordered)
+
+
+def main(argv: list[str] | None = None) -> int:
+    path = _parse_env_file_arg(sys.argv[1:] if argv is None else argv)
+    tui.show_welcome(
+        path,
+        profile_name=_profile_label_from_env_path(path),
+        is_new_file=not path.is_file(),
+    )
+
+    existing = env_file.load_managed_from_file(path)
+    parsed_env = (
+        env_file.parse_env_lines(path.read_text(encoding="utf-8", errors="replace"))
+        if path.is_file()
+        else {}
+    )
+
+    total_steps = 7
+    step_i = 0
+
+    def step(title: str) -> None:
+        nonlocal step_i
+        step_i += 1
+        tui.step_header(step_i, total_steps, title)
+
+    step("Backend provider")
+    prov_default = existing.get("PROVIDER", "meshtastic")
+    if prov_default not in ("meshtastic", "meshcore"):
+        prov_default = "meshtastic"
+    provider = tui.select(
+        "Backend provider",
+        [
+            ("Meshtastic", "meshtastic"),
+            ("MeshCore", "meshcore"),
+        ],
+        default_value=prov_default,
+        hint="Which firmware stack the ingestor speaks: Meshtastic is the common default; MeshCore uses the other backend.",
+    )
+    if provider not in ("meshtastic", "meshcore") or provider is None:
+        tui.print_error("Invalid provider.")
+        return 1
+
+    step("Connection")
+    conn_default = existing.get("CONNECTION", "")
+    conn_mode = tui.select(
+        "How do you want to set CONNECTION (where the ingestor reaches the radio)?",
+        [
+            (
+                "Guided — choose serial port, BLE device after scan, or enter TCP host:port",
+                "guided",
+            ),
+            (
+                "Manual — paste or type the full CONNECTION string (e.g. /dev/ttyACM0, "
+                "BLE MAC/UUID, or host:port)",
+                "manual",
+            ),
+        ],
+        default_value=("manual" if conn_default.strip() else "guided"),
+        hint="CONNECTION is the ingestor’s link to the radio: device path, BLE address, or TCP host:port.",
+    )
+    if conn_mode == "manual":
+        connection = tui.text(
+            "CONNECTION",
+            conn_default,
+            hint="Paste the same value you would set in `.env`: serial path, BLE MAC/UUID, or host:port.",
+        )
+    else:
+        # "guided", None (e.g. cancelled select), or unknown → use menus
+        connection = _pick_connection_string(conn_default)
+    if not connection:
+        tui.print_error("CONNECTION is required.")
+        return 1
+
+    step("Channel discovery")
+    rows: list[tuple[int, str]] = []
+    probe_note: str | None = None
+    tui.print_info(
+        "Connecting briefly to read channel names from the radio (skips ahead if unavailable)…"
+    )
+    if provider == "meshtastic":
+        fb = parsed_env.get("CHANNEL", "").strip() or None
+        rows, err = meshtastic_probe.probe_channels(connection, channel_fallback=fb)
+        probe_note = err
+    else:
+        rows, err = meshcore_probe.probe_channels(connection)
+        probe_note = err
+    if probe_note:
+        tui.print_dim(f"Channel probe: {probe_note}")
+    if rows:
+        tui.print_info(f"Discovered {len(rows)} channel(s).")
+    else:
+        tui.print_dim(
+            "No channel names from probe (you can still set allow/hide filters manually)."
+        )
+
+    step("Allow / hide channels by name")
+    allowed = _channel_filter_value(
+        rows,
+        "ALLOWED_CHANNELS",
+        all_label="Allow all channels (recommended unless you need a strict allowlist)",
+        existing=existing.get("ALLOWED_CHANNELS", ""),
+        field_hint=(
+            "If set, only traffic on these channel names is ingested; leave empty to allow every channel."
+        ),
+    )
+    hidden = _channel_filter_value(
+        rows,
+        "HIDDEN_CHANNELS",
+        all_label="Do not hide any channel by name",
+        existing=existing.get("HIDDEN_CHANNELS", ""),
+        field_hint="Traffic on these channel names is dropped; leave empty to hide nothing by name.",
+    )
+
+    if provider == "meshcore" and allowed.strip():
+        tui.print_warning(
+            "MeshCore packets often lack Meshtastic-style channel names in the ingestor; "
+            "a non-empty ALLOWED_CHANNELS can drop traffic until names match runtime metadata. "
+            "Prefer “allow all” unless you have verified behavior."
+        )
+
+    step("Instance URL & API token")
+    existing_inst = existing.get("INSTANCE_DOMAIN", "")
+    inst_mode = tui.select(
+        "INSTANCE_DOMAIN — where is your PotatoMesh web UI?",
+        [
+            (f"Local default ({_LOCAL_INSTANCE_DOMAIN})", "local"),
+            ("Remote or custom URL", "custom"),
+        ],
+        default_value=(
+            "local"
+            if _instance_domain_prefers_local_default(existing_inst)
+            else "custom"
+        ),
+        hint="The ingestor POSTs decoded mesh events to this PotatoMesh server (same URL you open in a browser).",
+    )
+    if inst_mode == "local":
+        instance = _LOCAL_INSTANCE_DOMAIN
+    else:
+        instance = tui.text(
+            "INSTANCE_DOMAIN (full URL or hostname; https:// added if no scheme)",
+            existing_inst,
+            hint="Use a full URL when in doubt; a bare hostname gets https:// prepended by the ingestor.",
+        )
+    token = tui.text(
+        "API_TOKEN",
+        existing.get("API_TOKEN", ""),
+        hint="Shared secret required on ingest HTTP requests; must match API_TOKEN on the web server.",
+    )
+
+    step("Debug & energy saving")
+    debug = (
+        "1"
+        if tui.confirm(
+            "Enable DEBUG=1?",
+            existing.get("DEBUG", "0") == "1",
+            hint="Extra ingestor logging and debug-only file traces; useful for troubleshooting, noisy otherwise.",
+        )
+        else "0"
+    )
+    energy = (
+        "1"
+        if tui.confirm(
+            "Enable ENERGY_SAVING=1?",
+            existing.get("ENERGY_SAVING", "0") == "1",
+            hint="Disconnects and sleeps on a schedule to cut CPU/BLE use; may miss brief traffic between wakeups.",
+        )
+        else "0"
+    )
+
+    merged: dict[str, str] = {
+        "PROVIDER": provider,
+        "CONNECTION": connection,
+        "ALLOWED_CHANNELS": allowed,
+        "HIDDEN_CHANNELS": hidden,
+        "INSTANCE_DOMAIN": instance,
+        "API_TOKEN": token,
+        "DEBUG": debug,
+        "ENERGY_SAVING": energy,
+    }
+
+    step("Write .env file")
+    if not tui.confirm(
+        f"Write configuration to {path}?",
+        True,
+        hint="Only wizard-managed keys are updated; other lines in the file stay unchanged.",
+    ):
+        tui.print_aborted()
+        return 0
+
+    env_file.merge_write_env(path, merged)
+    tui.print_saved(path, connection_kind(connection))
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except KeyboardInterrupt:
+        tui.print_cancelled()
+        raise SystemExit(130) from None
diff --git a/data/mesh_env/branding.py b/data/mesh_env/branding.py
new file mode 100644
index 00000000..cec8ffef
--- /dev/null
+++ b/data/mesh_env/branding.py
@@ -0,0 +1,50 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Wordmark and copy for the env wizard (pyfiglet + static fallback)."""
+
+from __future__ import annotations
+
+# Hand-drawn fallback (~62 columns); used when pyfiglet is unavailable.
+POTATO_MESH_ASCII = r"""
+  ____       _        _     __  __                     _
+ |  _ \ ___ | |_ __ _| |   |  \/  | _____   _____  ___| |__
+ | |_) / _ \| __/ _` | |   | |\/| |/ _ \ \ / / _ \/ __| '_ \
+ |  __/ (_) | || (_| | |   | |  | | (_) \ V /  __/ (__| | | |
+ |_|   \___/ \__\__,_|_|   |_|  |_|\___/ \_/ \___|\___|_| |_|
+""".strip()
+
+TAGLINE = "ingestor · env wizard"
+
+_DEFAULT_PHRASE = "Potato Mesh"
+
+# Prefer readable fonts that stay mostly ASCII; order is try-first.
+_FIGLET_FONT_PREFERENCE = ("slant", "small", "standard", "big")
+
+
+def render_wordmark(phrase: str | None = None) -> str:
+    """Return multi-line ASCII art for *phrase* via pyfiglet, or :data:`POTATO_MESH_ASCII`."""
+
+    label = (phrase or _DEFAULT_PHRASE).strip() or _DEFAULT_PHRASE
+    try:
+        from pyfiglet import Figlet
+    except ImportError:
+        return POTATO_MESH_ASCII
+
+    for font in _FIGLET_FONT_PREFERENCE:
+        try:
+            return Figlet(font=font).renderText(label).rstrip("\n")
+        except Exception:
+            continue
+    return POTATO_MESH_ASCII
diff --git a/data/mesh_env/connection_parse.py b/data/mesh_env/connection_parse.py
new file mode 100644
index 00000000..ae551532
--- /dev/null
+++ b/data/mesh_env/connection_parse.py
@@ -0,0 +1,92 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Parse connection targets; mirrors ``mesh_ingestor.connection`` for wizard-only use."""
+
+from __future__ import annotations
+
+import re
+
+DEFAULT_TCP_PORT: int = 4403
+
+BLE_ADDRESS_RE = re.compile(
+    r"^(?:"
+    r"(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}|"
+    r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
+    r")$"
+)
+
+
+def parse_ble_target(value: str) -> str | None:
+    if not value:
+        return None
+    value = value.strip()
+    if not value:
+        return None
+    if BLE_ADDRESS_RE.fullmatch(value):
+        return value.upper()
+    return None
+
+
+def parse_tcp_target(value: str) -> tuple[str, int] | None:
+    if not value:
+        return None
+    value = value.strip()
+    if not value:
+        return None
+    if "://" in value:
+        value = value.split("://", 1)[1]
+    if value.startswith("["):
+        bracket_end = value.find("]")
+        if bracket_end == -1:
+            return None
+        host = value[1:bracket_end]
+        rest = value[bracket_end + 1 :]
+        if rest.startswith(":"):
+            try:
+                port = int(rest[1:])
+            except ValueError:
+                return None
+            if not (1 <= port <= 65535):
+                return None
+        else:
+            port = DEFAULT_TCP_PORT
+        if not host:
+            return None
+        return host, port
+    if value.count(":") != 1:
+        return None
+    host, _, port_str = value.partition(":")
+    if not host:
+        return None
+    try:
+        port = int(port_str)
+    except ValueError:
+        return None
+    if not (1 <= port <= 65535):
+        return None
+    return host, port
+
+
+def connection_kind(target: str) -> str:
+    """Return ``"ble"``, ``"tcp"``, or ``"serial"`` for a non-empty target string."""
+
+    t = (target or "").strip()
+    if not t:
+        return "serial"
+    if parse_ble_target(t):
+        return "ble"
+    if parse_tcp_target(t):
+        return "tcp"
+    return "serial"
diff --git a/data/mesh_env/devices.py b/data/mesh_env/devices.py
new file mode 100644
index 00000000..27e1ef53
--- /dev/null
+++ b/data/mesh_env/devices.py
@@ -0,0 +1,58 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Serial path discovery and BLE scanning (wizard-only; patterns match ``mesh_ingestor.connection``)."""
+
+from __future__ import annotations
+
+import glob
+
+DEFAULT_SERIAL_PATTERNS: tuple[str, ...] = (
+    "/dev/ttyACM*",
+    "/dev/ttyUSB*",
+    "/dev/tty.usbmodem*",
+    "/dev/tty.usbserial*",
+    "/dev/cu.usbmodem*",
+    "/dev/cu.usbserial*",
+)
+
+
+def list_serial_paths() -> list[str]:
+    """Return deduplicated serial device paths (same glob rules as the ingestor)."""
+
+    candidates: list[str] = []
+    seen: set[str] = set()
+    for pattern in DEFAULT_SERIAL_PATTERNS:
+        for path in sorted(glob.glob(pattern)):
+            if path not in seen:
+                candidates.append(path)
+                seen.add(path)
+    if "/dev/ttyACM0" not in seen:
+        candidates.append("/dev/ttyACM0")
+    return candidates
+
+
+async def scan_ble_devices(timeout: float = 8.0) -> list[tuple[str, str]]:
+    """Return ``(name_or_unknown, address)`` for discovered BLE peripherals."""
+
+    from bleak import BleakScanner
+
+    devices = await BleakScanner.discover(timeout=timeout)
+    out: list[tuple[str, str]] = []
+    for d in devices:
+        name = (d.name or "").strip() or "(no name)"
+        addr = getattr(d, "address", None) or str(d)
+        out.append((name, addr))
+    out.sort(key=lambda x: (x[0].lower(), x[1]))
+    return out
diff --git a/data/mesh_env/env_file.py b/data/mesh_env/env_file.py
new file mode 100644
index 00000000..53131e92
--- /dev/null
+++ b/data/mesh_env/env_file.py
@@ -0,0 +1,154 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Read and write ``.env`` files while preserving unmanaged lines."""
+
+from __future__ import annotations
+
+import os
+import re
+import tempfile
+from pathlib import Path
+
+_MANAGED_KEYS: frozenset[str] = frozenset(
+    {
+        "PROVIDER",
+        "CONNECTION",
+        "ALLOWED_CHANNELS",
+        "HIDDEN_CHANNELS",
+        "INSTANCE_DOMAIN",
+        "API_TOKEN",
+        "DEBUG",
+        "ENERGY_SAVING",
+    }
+)
+
+_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
+
+# Dropped from preserved text so re-running the wizard does not stack duplicate headers.
+_MANAGED_BLOCK_HEADER = "# --- potato-mesh ingestor (mesh_env wizard) ---"
+
+
+def managed_keys() -> frozenset[str]:
+    return _MANAGED_KEYS
+
+
+def parse_env_lines(text: str) -> dict[str, str]:
+    """Parse ``KEY=value`` assignments; ignores export prefix and strips quotes."""
+
+    result: dict[str, str] = {}
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line.startswith("export "):
+            line = line[7:].lstrip()
+        m = _KEY_RE.match(line)
+        if not m:
+            continue
+        key, val = m.group(1), m.group(2).strip()
+        if val.startswith('"') and val.endswith('"') and len(val) >= 2:
+            val = val[1:-1].replace('\\"', '"')
+        elif val.startswith("'") and val.endswith("'") and len(val) >= 2:
+            val = val[1:-1]
+        result[key] = val
+    return result
+
+
+def load_managed_from_file(path: Path) -> dict[str, str]:
+    if not path.is_file():
+        return {}
+    return {
+        k: v
+        for k, v in parse_env_lines(
+            path.read_text(encoding="utf-8", errors="replace")
+        ).items()
+        if k in _MANAGED_KEYS
+    }
+
+
+def merge_write_env(path: Path, values: dict[str, str]) -> None:
+    """Drop prior managed-key lines from *path* and append a block with *values*.
+
+    The wizard banner comment is not preserved from the old file so repeat runs do
+    not accumulate duplicate headers.
+    """
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    existing = ""
+    if path.is_file():
+        existing = path.read_text(encoding="utf-8", errors="replace")
+
+    kept_lines: list[str] = []
+    for raw_line in existing.splitlines():
+        stripped = raw_line.strip()
+        if stripped.startswith("#") or not stripped:
+            if stripped == _MANAGED_BLOCK_HEADER:
+                continue
+            kept_lines.append(raw_line)
+            continue
+        line = stripped
+        if line.startswith("export "):
+            line = line[7:].lstrip()
+        m = _KEY_RE.match(line)
+        if m and m.group(1) in _MANAGED_KEYS:
+            continue
+        kept_lines.append(raw_line)
+
+    while kept_lines and kept_lines[-1].strip() == "":
+        kept_lines.pop()
+
+    block_lines = [
+        "",
+        _MANAGED_BLOCK_HEADER,
+    ]
+    order = (
+        "PROVIDER",
+        "CONNECTION",
+        "ALLOWED_CHANNELS",
+        "HIDDEN_CHANNELS",
+        "INSTANCE_DOMAIN",
+        "API_TOKEN",
+        "DEBUG",
+        "ENERGY_SAVING",
+    )
+    for key in order:
+        if key not in _MANAGED_KEYS or key not in values:
+            continue
+        val = values[key]
+        if val is None:
+            continue
+        sval = str(val)
+        if re.search(r"[\s#\"']", sval) or sval == "":
+            esc = sval.replace("\\", "\\\\").replace('"', '\\"')
+            block_lines.append(f'{key}="{esc}"')
+        else:
+            block_lines.append(f"{key}={sval}")
+
+    out_text = "\n".join(kept_lines)
+    if kept_lines:
+        out_text += "\n"
+    out_text += "\n".join(block_lines) + "\n"
+
+    fd, tmp = tempfile.mkstemp(prefix=".env.", dir=str(path.parent), text=True)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            fh.write(out_text)
+        os.replace(tmp, path)
+    finally:
+        if os.path.exists(tmp):
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
diff --git a/data/mesh_env/meshcore_probe.py b/data/mesh_env/meshcore_probe.py
new file mode 100644
index 00000000..d3bcf20f
--- /dev/null
+++ b/data/mesh_env/meshcore_probe.py
@@ -0,0 +1,84 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Short MeshCore sessions to query channel names via ``get_channel`` (no mesh_ingestor imports)."""
+
+from __future__ import annotations
+
+import asyncio
+
+from .connection_parse import parse_ble_target, parse_tcp_target
+
+_DEFAULT_BAUDRATE = 115200
+_MAX_CHANNEL_PROBE = 32
+
+
+def _make_connection(target: str, baudrate: int = _DEFAULT_BAUDRATE):
+    from meshcore import BLEConnection, SerialConnection, TCPConnection
+
+    ble_addr = parse_ble_target(target)
+    if ble_addr:
+        return BLEConnection(address=ble_addr)
+    tcp_target = parse_tcp_target(target)
+    if tcp_target:
+        host, port = tcp_target
+        return TCPConnection(host, port)
+    return SerialConnection(target, baudrate)
+
+
+async def _probe_async(target: str) -> tuple[list[tuple[int, str]], str | None]:
+    from meshcore import MeshCore
+    from meshcore.events import EventType
+
+    rows: list[tuple[int, str]] = []
+    err: str | None = None
+    mc = None
+    try:
+        cx = _make_connection(target.strip(), _DEFAULT_BAUDRATE)
+        mc = MeshCore(cx)
+        res = await mc.connect()
+        if res is None:
+            return [], "MeshCore node did not complete the appstart handshake."
+
+        for idx in range(_MAX_CHANNEL_PROBE):
+            try:
+                evt = await mc.commands.get_channel(idx)
+            except Exception as exc:
+                err = str(exc)
+                break
+            if evt.type == EventType.ERROR:
+                continue
+            if evt.type != EventType.CHANNEL_INFO:
+                continue
+            payload = evt.payload or {}
+            name = (payload.get("channel_name") or "").strip()
+            if name:
+                rows.append((int(payload.get("channel_idx", idx)), name))
+
+        rows.sort(key=lambda x: x[0])
+        return rows, err
+    except Exception as exc:
+        return [], str(exc)
+    finally:
+        if mc is not None:
+            try:
+                await mc.disconnect()
+            except Exception:
+                pass
+
+
+def probe_channels(target: str) -> tuple[list[tuple[int, str]], str | None]:
+    """Run :func:`_probe_async` in a fresh event loop (wizard is synchronous at top level)."""
+
+    return asyncio.run(_probe_async(target))
diff --git a/data/mesh_env/meshtastic_probe.py b/data/mesh_env/meshtastic_probe.py
new file mode 100644
index 00000000..ea349e0c
--- /dev/null
+++ b/data/mesh_env/meshtastic_probe.py
@@ -0,0 +1,200 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Short Meshtastic connections to list channel index/name pairs (no mesh_ingestor imports)."""
+
+from __future__ import annotations
+
+from typing import Any, Iterable, Iterator
+
+from .connection_parse import parse_ble_target, parse_tcp_target
+
+try:
+    from meshtastic.protobuf import channel_pb2
+except Exception:
+    channel_pb2 = None
+
+_ROLE_PRIMARY = 1
+_ROLE_SECONDARY = 2
+if channel_pb2 is not None:
+    try:
+        _ROLE_PRIMARY = int(channel_pb2.Channel.Role.PRIMARY)
+        _ROLE_SECONDARY = int(channel_pb2.Channel.Role.SECONDARY)
+    except Exception:
+        pass
+
+
+def _iter_channel_objects(channels_obj: Any) -> Iterator[Any]:
+    if channels_obj is None:
+        return iter(())
+
+    if isinstance(channels_obj, dict):
+        return iter(channels_obj.values())
+
+    if isinstance(channels_obj, Iterable):
+        return iter(list(channels_obj))
+
+    length_fn = getattr(channels_obj, "__len__", None)
+    getitem = getattr(channels_obj, "__getitem__", None)
+    if callable(length_fn) and callable(getitem):
+        try:
+            length = int(length_fn())
+        except Exception:
+            length = None
+        if length is not None and length >= 0:
+            snapshot = []
+            for index in range(length):
+                try:
+                    snapshot.append(getitem(index))
+                except Exception:
+                    break
+            return iter(snapshot)
+
+    return iter(())
+
+
+def _extract_channel_name(settings_obj: Any) -> str | None:
+    if settings_obj is None:
+        return None
+    if isinstance(settings_obj, dict):
+        candidate = settings_obj.get("name")
+    else:
+        candidate = getattr(settings_obj, "name", None)
+    if isinstance(candidate, str):
+        candidate = candidate.strip()
+        if candidate:
+            return candidate
+    return None
+
+
+def _normalize_role(role: Any) -> int | None:
+    if isinstance(role, int):
+        return role
+    if isinstance(role, str):
+        value = role.strip().upper()
+        if value == "PRIMARY":
+            return _ROLE_PRIMARY
+        if value == "SECONDARY":
+            return _ROLE_SECONDARY
+        try:
+            return int(value)
+        except ValueError:
+            return None
+    name_attr = getattr(role, "name", None)
+    if isinstance(name_attr, str):
+        return _normalize_role(name_attr)
+    value_attr = getattr(role, "value", None)
+    if isinstance(value_attr, int):
+        return value_attr
+    try:
+        return int(role)  # type: ignore[arg-type]
+    except Exception:
+        return None
+
+
+def _channel_tuple(
+    channel_obj: Any, primary_fallback: str | None
+) -> tuple[int, str] | None:
+    role_value = _normalize_role(getattr(channel_obj, "role", None))
+    if role_value == _ROLE_PRIMARY:
+        channel_index = 0
+        channel_name = _extract_channel_name(getattr(channel_obj, "settings", None))
+        if channel_name is None:
+            channel_name = primary_fallback
+    elif role_value == _ROLE_SECONDARY:
+        raw_index = getattr(channel_obj, "index", None)
+        try:
+            channel_index = int(raw_index)
+        except Exception:
+            channel_index = None
+        channel_name = _extract_channel_name(getattr(channel_obj, "settings", None))
+    else:
+        return None
+
+    if not isinstance(channel_index, int):
+        return None
+    if not isinstance(channel_name, str) or not channel_name:
+        return None
+    return channel_index, channel_name
+
+
+def extract_channel_rows(
+    iface: Any, primary_fallback: str | None
+) -> list[tuple[int, str]]:
+    local_node = getattr(iface, "localNode", None)
+    channels_obj = getattr(local_node, "channels", None) if local_node else None
+    channel_entries: list[tuple[int, str]] = []
+    seen_indices: set[int] = set()
+    for candidate in _iter_channel_objects(channels_obj):
+        result = _channel_tuple(candidate, primary_fallback)
+        if result is None:
+            continue
+        index, name = result
+        if index in seen_indices:
+            continue
+        channel_entries.append((index, name))
+        seen_indices.add(index)
+    channel_entries.sort(key=lambda x: x[0])
+    return channel_entries
+
+
+def open_meshtastic_interface(target: str):
+    """Return a connected Meshtastic interface for *target* (serial path, BLE MAC/UUID, or host:port)."""
+
+    from meshtastic.serial_interface import SerialInterface
+    from meshtastic.tcp_interface import TCPInterface
+
+    t = (target or "").strip()
+    ble = parse_ble_target(t)
+    if ble:
+        try:
+            from meshtastic.ble_interface import BLEInterface
+        except Exception as exc:
+            raise RuntimeError(
+                "BLE requested but meshtastic BLE extras are not available. "
+                "Install meshtastic with the 'ble' extra."
+            ) from exc
+        return BLEInterface(address=ble)
+
+    tcp = parse_tcp_target(t)
+    if tcp:
+        host, port = tcp
+        return TCPInterface(hostname=host, portNumber=port)
+
+    return SerialInterface(devPath=t)
+
+
+def probe_channels(
+    target: str, *, channel_fallback: str | None = None
+) -> tuple[list[tuple[int, str]], str | None]:
+    """Connect, wait for config, return ``(rows, error)``."""
+
+    iface = None
+    try:
+        iface = open_meshtastic_interface(target)
+        wait = getattr(iface, "waitForConfig", None)
+        if callable(wait):
+            wait()
+        rows = extract_channel_rows(iface, channel_fallback)
+        return rows, None
+    except Exception as exc:
+        return [], str(exc)
+    finally:
+        if iface is not None:
+            close = getattr(iface, "close", None)
+            if callable(close):
+                try:
+                    close()
+                except Exception:
+                    pass
diff --git a/data/mesh_env/tui.py b/data/mesh_env/tui.py
new file mode 100644
index 00000000..839ed492
--- /dev/null
+++ b/data/mesh_env/tui.py
@@ -0,0 +1,384 @@
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Rich framing + questionary prompts (TTY); plain fallback when not a terminal.
+
+Set ``MESH_ENV_FORCE_RICH=1`` if the banner/panels do not appear but your terminal
+supports ANSI (some IDEs mis-report ``isatty`` on stdout).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+import questionary
+from questionary import Choice, Style
+
+from .branding import TAGLINE, render_wordmark
+
+# Prompt theme aligned with Rich panels (cyan / gold / neutral grays).
+_Q_STYLE = Style(
+    [
+        ("qmark", "fg:#5ee7b8 bold"),
+        ("question", "bold fg:#e8eaed"),
+        ("answer", "fg:#5ee7b8 bold"),
+        ("pointer", "fg:#5ee7b8 bold"),
+        ("highlighted", "fg:#ffd88a bold"),
+        ("selected", "fg:#7dd3fc"),
+        ("instruction", "fg:#9ca3af"),
+        ("text", "fg:#d1d5db"),
+    ]
+)
+
+_Q_KW: dict[str, Any] = {"style": _Q_STYLE}
+
+_rich_console: Any | None = None
+_rich_err: Any | None = None
+
+
+def _env_truthy(name: str) -> bool:
+    return os.environ.get(name, "").strip().lower() in ("1", "true", "yes")
+
+
+def _rich_frames() -> bool:
+    """Use Rich panels/rules when stdout can render (not tied to stdin)."""
+
+    if _env_truthy("MESH_ENV_FORCE_RICH"):
+        return True
+    return sys.stdout.isatty()
+
+
+def _stderr_styled() -> bool:
+    if _env_truthy("MESH_ENV_FORCE_RICH"):
+        return True
+    return sys.stderr.isatty()
+
+
+def _c() -> Any:
+    global _rich_console
+    if _rich_console is None:
+        from rich.console import Console
+
+        _rich_console = Console(
+            highlight=False,
+            soft_wrap=True,
+            force_terminal=_env_truthy("MESH_ENV_FORCE_RICH"),
+        )
+    return _rich_console
+
+
+def _cerr() -> Any:
+    global _rich_err
+    if _rich_err is None:
+        from rich.console import Console
+
+        _rich_err = Console(
+            highlight=False,
+            stderr=True,
+            soft_wrap=True,
+            force_terminal=_env_truthy("MESH_ENV_FORCE_RICH"),
+        )
+    return _rich_err
+
+
+def is_interactive() -> bool:
+    """True when questionary can use the full-screen prompt toolkit UI."""
+
+    return sys.stdin.isatty() and sys.stdout.isatty()
+
+
+def show_welcome(
+    env_path: Path,
+    *,
+    profile_name: str | None = None,
+    is_new_file: bool = False,
+) -> None:
+    """Banner for the env wizard; *profile_name* / *is_new_file* clarify which file is being edited."""
+
+    from rich.markup import escape
+
+    path_esc = escape(str(env_path))
+    if not _rich_frames():
+        if profile_name:
+            act = "Creating" if is_new_file else "Editing"
+            print(f"PotatoMesh env wizard — {act} profile {profile_name!r}")
+            print(f"Env file: {env_path}")
+        else:
+            print(f"PotatoMesh env wizard → {env_path}")
+        return
+
+    from rich.align import Align
+    from rich import box
+    from rich.panel import Panel
+    from rich.text import Text
+
+    art = Text(render_wordmark(), style="bold #ffd88a")
+    inner = Align.center(art)
+    sub_lines: list[str] = []
+    if profile_name:
+        act = "Creating" if is_new_file else "Editing"
+        sub_lines.append(
+            f"[bold white]{escape(act)} profile[/] [bold #5ee7b8]{escape(profile_name)}[/]"
+        )
+    sub_lines.append(f"[dim]{path_esc}[/]")
+    panel = Panel.fit(
+        inner,
+        title="[bold bright_cyan]potato-mesh[/] [dim]│[/] [white]" + TAGLINE + "[/]",
+        subtitle="\n".join(sub_lines),
+        subtitle_align="center",
+        border_style="bright_cyan",
+        box=box.ROUNDED,
+        padding=(1, 3),
+    )
+    _c().print()
+    _c().print(Align.center(panel))
+    _c().print()
+
+
+def step_header(step: int, total: int, title: str) -> None:
+    if not _rich_frames():
+        print(f"\n--- {step}/{total} {title} ---")
+        return
+
+    from rich.rule import Rule
+
+    _c().print()
+    _c().print(
+        Rule(
+            f"[dim]Step {step} of {total}[/]  [bold white]{title}[/]",
+            style="bright_cyan",
+            align="left",
+        )
+    )
+
+
+def print_info(msg: str) -> None:
+    if not _rich_frames():
+        print(msg)
+        return
+    _c().print(msg)
+
+
+def print_dim(msg: str) -> None:
+    if not _rich_frames():
+        print(msg)
+        return
+    _c().print(f"[dim]{msg}[/]")
+
+
+def prompt_hint(text: str | None) -> None:
+    """Short dim explanation shown before a prompt (informational, not a question)."""
+
+    t = (text or "").strip()
+    if not t:
+        return
+    print_dim(t)
+
+
+def print_warning(msg: str) -> None:
+    if not _rich_frames():
+        print(msg)
+        return
+
+    from rich import box
+    from rich.panel import Panel
+
+    _c().print(
+        Panel(
+            msg,
+            title="[bold yellow]Note[/]",
+            border_style="yellow",
+            box=box.ROUNDED,
+            padding=(0, 1),
+        )
+    )
+
+
+def print_error(msg: str) -> None:
+    if not _stderr_styled():
+        print(msg, file=sys.stderr)
+        return
+    _cerr().print(f"[bold red]{msg}[/]")
+
+
+def print_aborted() -> None:
+    if not _rich_frames():
+        print("Aborted.")
+        return
+    _c().print("\n[dim]Aborted — no file was written.[/]")
+
+
+def print_cancelled() -> None:
+    if not _stderr_styled():
+        print("\nCancelled.", file=sys.stderr)
+        return
+    _cerr().print("\n[yellow]Cancelled.[/yellow]")
+
+
+def print_saved(path: Path, connection_kind_label: str) -> None:
+    if not _rich_frames():
+        print(f"Wrote {path}")
+        print(f"Connection kind: {connection_kind_label}")
+        return
+
+    from rich import box
+    from rich.panel import Panel
+
+    from rich.align import Align
+
+    body = (
+        f"[bold green]Configuration saved.[/bold green]\n\n"
+        f"[dim]Env file[/dim]     [bold white]{path}[/bold white]\n"
+        f"[dim]Connection[/dim]  [white]{connection_kind_label}[/white]"
+    )
+    panel = Panel.fit(
+        body,
+        title="[bold green]Done[/]",
+        border_style="green",
+        box=box.ROUNDED,
+        padding=(1, 2),
+    )
+    _c().print()
+    _c().print(Align.center(panel))
+    _c().print()
+
+
+def _fallback_text(message: str, default: str = "") -> str:
+    dhint = f" [{default}]" if default else ""
+    raw = input(f"{message}{dhint}: ").strip()
+    return raw if raw else default
+
+
+def _fallback_confirm(message: str, default: bool = True) -> bool:
+    d = "Y/n" if default else "y/N"
+    raw = input(f"{message} ({d}): ").strip().lower()
+    if not raw:
+        return default
+    return raw in ("y", "yes", "1", "true")
+
+
+def text(message: str, default: str = "", *, hint: str | None = None) -> str:
+    """Prompt for a single line of text.
+
+    *hint* is printed in dim style before the prompt when non-empty.
+    """
+
+    prompt_hint(hint)
+    if not is_interactive():
+        return _fallback_text(message, default)
+    r = questionary.text(message, default=default or "", **_Q_KW).unsafe_ask()
+    return r.strip() if isinstance(r, str) else default
+
+
+def confirm(message: str, default: bool = True, *, hint: str | None = None) -> bool:
+    """Yes/no prompt; *hint* is dim informational text before the question."""
+
+    prompt_hint(hint)
+    if not is_interactive():
+        return _fallback_confirm(message, default)
+    r = questionary.confirm(message, default=default, **_Q_KW).unsafe_ask()
+    return bool(r)
+
+
+def select(
+    message: str,
+    choices: list[tuple[str, Any]],
+    default_value: Any | None = None,
+    *,
+    hint: str | None = None,
+) -> Any | None:
+    """Return the *value* of the selected choice (second element of each tuple).
+
+    *hint* is dim informational text before the list.
+    """
+
+    prompt_hint(hint)
+    if not choices:
+        return None
+    default_idx = 0
+    if default_value is not None:
+        for i, (_, val) in enumerate(choices):
+            if val == default_value:
+                default_idx = i
+                break
+
+    if not is_interactive():
+        for i, (title, val) in enumerate(choices):
+            mark = " *" if i == default_idx else ""
+            print(f"  [{i}] {title}{mark}")
+        raw = _fallback_text(f"{message} (number)", str(default_idx))
+        try:
+            idx = int(raw)
+            return choices[idx][1]
+        except (ValueError, IndexError):
+            return choices[default_idx][1]
+
+    qc: list[Choice] = [Choice(title, value=val) for title, val in choices]
+    return questionary.select(
+        message,
+        choices=qc,
+        default=qc[default_idx],
+        **_Q_KW,
+    ).unsafe_ask()
+
+
+def checkbox(
+    message: str,
+    choices: list[tuple[str, Any]],
+    *,
+    prechecked_values: frozenset[str] | None = None,
+    hint: str | None = None,
+) -> list[Any]:
+    """Return list of selected values (may be empty).
+
+    *prechecked_values* are matched case-insensitively against each choice *value*.
+    *hint* is dim informational text before the checklist.
+    """
+
+    prompt_hint(hint)
+    if not choices:
+        return []
+    pred_cf = prechecked_values or frozenset()
+
+    def _checked(val: Any) -> bool:
+        return str(val).casefold() in pred_cf
+
+    if not is_interactive():
+        for i, (title, val) in enumerate(choices):
+            mark = " [x]" if _checked(val) else ""
+            print(f"  [{i}] {title}{mark}")
+        default_nums = ",".join(
+            str(i) for i, (_, val) in enumerate(choices) if _checked(val)
+        )
+        raw = _fallback_text(
+            f"{message} (comma-separated numbers, empty=none)", default_nums
+        ).replace(" ", "")
+        if not raw:
+            return []
+        out: list[Any] = []
+        for part in raw.split(","):
+            if not part:
+                continue
+            try:
+                out.append(choices[int(part)][1])
+            except (ValueError, IndexError):
+                pass
+        return out
+
+    qc = [Choice(title, value=val, checked=_checked(val)) for title, val in choices]
+    r = questionary.checkbox(message, choices=qc, **_Q_KW).unsafe_ask()
+    return list(r)
diff --git a/data/potato_mesh_env.sh b/data/potato_mesh_env.sh
new file mode 100644
index 00000000..c2c002e8
--- /dev/null
+++ b/data/potato_mesh_env.sh
@@ -0,0 +1,62 @@
+#!/usr/bin/env bash
+# Copyright © 2025-26 l5yth & contributors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Shared helpers for mesh.sh and config.sh (single copy for tooling / duplication limits).
+
+# MODE is "mesh" (profile or default only; invalid/extra args exit) or "config" (profile, explicit PATH, or default).
+# Pass script arguments after REPO: potato_mesh_resolve_env_file MODE REPO "$@"
+# Sets _env_file and _potato_mesh_env_shift (how many leading args to shift from the caller).
+potato_mesh_resolve_env_file() {
+  local _pm_mode="$1"
+  local _pm_repo="$2"
+  shift 2
+
+  if [[ $# -gt 0 ]] && [[ "${1}" =~ ^[a-zA-Z0-9][a-zA-Z0-9_-]*$ ]]; then
+    _env_file="${_pm_repo}/.env-${1}"
+    _potato_mesh_env_shift=1
+    return 0
+  fi
+  if [[ "${_pm_mode}" == "config" ]] && [[ $# -gt 0 ]] && [[ "${1}" != -* ]]; then
+    _env_file="${1}"
+    _potato_mesh_env_shift=1
+    return 0
+  fi
+  if [[ "${_pm_mode}" == "mesh" ]] && [[ $# -gt 0 ]]; then
+    echo "mesh.sh: invalid profile name (use letters, digits, underscores, hyphens): ${1}" >&2
+    echo "Usage: mesh.sh [profile]" >&2
+    exit 2
+  fi
+  _env_file="${_pm_repo}/.env"
+  _potato_mesh_env_shift=0
+}
+
+potato_mesh_source_env_if_exists() {
+  local _pm_env_path="$1"
+  if [[ -f "${_pm_env_path}" ]]; then
+    set -a
+    # shellcheck disable=SC1090
+    source "${_pm_env_path}"
+    set +a
+  fi
+}
+
+potato_mesh_venv_and_requirements() {
+  local _pm_req="$1"
+  python -m venv .venv
+  # shellcheck disable=SC1091
+  source .venv/bin/activate
+  pip install -U pip
+  pip install -r "${_pm_req}"
+}
diff --git a/data/requirements.txt b/data/requirements.txt
index 6aadee50..00ce50a6 100644
--- a/data/requirements.txt
+++ b/data/requirements.txt
@@ -3,6 +3,9 @@ meshtastic>=2.5.0
 meshcore>=2.3.5
 bleak>=0.21.0
 protobuf>=5.27.2
+questionary>=2.0.0
+rich>=13.0.0
+pyfiglet>=1.0.0
 
 # Development dependencies (optional)
 black>=24.8.0