5 files changed, 912 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py
new file mode 100644
index 0000000..becd195
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/__init__.py
@@ -0,0 +1 @@
+"""The interactive mode: `meshbay-cli` with no argument."""
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/app.py b/packages/meshbay-cli/src/meshbay_cli/tui/app.py
new file mode 100644
index 0000000..dbb773c
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/app.py
@@ -0,0 +1,76 @@
+"""
+`meshbay-cli` with no argument: a full-screen terminal interface.
+
+Textual, because it runs on the same asyncio loop as aiortc and httpx rather
+than beside them, and on Linux terminals and Windows' consoles alike. Signed in
+already, it opens on the two panes; otherwise on the sign-in screen first.
+
+Nothing is printed while it runs — a line on stderr would tear the screen — so
+the log goes to a file in the CLI's data directory instead.
+"""
+
+import logging
+
+from textual.app import App
+
+from ..client.account import Account, no_passphrase
+from ..client.hub import Hub
+from ..client.store import data_dir
+from ..commands.context import Context
+from ..errors import CliError
+from .screens import DeviceCode, MainScreen, SignInScreen
+
+
+class MeshBayApp(App):
+    TITLE = "MeshBay"
+
+    def __init__(self, ctx: Context):
+        super().__init__()
+        self.ctx = ctx
+        self.hub = ctx.hub
+
+    async def use_hub(self, url: str) -> None:
+        if url != self.hub.url:
+            await self.hub.aclose()
+            self.hub = Hub(url)
+            self.ctx.hub = self.hub
+            self.ctx.hub_url = url
+
+    def on_mount(self) -> None:
+        account = None
+        if self.ctx.store.peek().get("hub") == self.hub.url:
+            try:
+                account = Account.open(self.hub, None, no_passphrase, self.ctx.store)
+            except CliError as e:
+                self.notify(e.message, severity="warning", markup=False)
+        if account is not None and account.remembered:
+            self.start(account)
+        else:
+            self.push_screen(SignInScreen())
+
+    def start(self, account: Account) -> None:
+        account.on_device_code = DeviceCode(self)
+        if isinstance(self.screen, SignInScreen):
+            self.switch_screen(MainScreen(account))
+        else:
+            self.push_screen(MainScreen(account))
+
+
+def _log_to_file() -> None:
+    root = logging.getLogger()
+    for h in list(root.handlers):
+        root.removeHandler(h)
+    try:
+        directory = data_dir()
+        directory.mkdir(parents=True, exist_ok=True)
+        handler = logging.FileHandler(directory / "interactive.log", encoding="utf-8")
+    except OSError:
+        handler = logging.NullHandler()
+    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s"))
+    root.addHandler(handler)
+
+
+async def run_interactive(ctx: Context) -> int:
+    _log_to_file()
+    await MeshBayApp(ctx).run_async()
+    return 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/panes.py b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py
new file mode 100644
index 0000000..d1e3dd0
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/panes.py
@@ -0,0 +1,372 @@
+"""
+The two panes: this machine's folders on the left, the groups on the right.
+
+Both draw rows the same way — folders first, then files, each sorted by the
+pane's key — and both mark a selection with Space. Names are someone else's
+text on the right and may be on the left: every name is escaped (`safe`) and
+handed to the table as plain text, never as markup.
+"""
+
+import asyncio
+import os
+import stat
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from rich.text import Text
+from textual.app import ComposeResult
+from textual.binding import Binding
+from textual.containers import Vertical
+from textual.widgets import DataTable, Input, Static
+
+from ..client.groups import my_groups
+from ..client.paths import handle
+from ..client.tree import Tree
+from ..errors import CliError
+from ..output import date, human_size, safe
+
+PARENT = ".."
+SORTS = ("name", "size", "date")
+SIZE_WIDTH, DATE_WIDTH = 9, 16
+# Cell padding on both sides of three columns, and the vertical scrollbar.
+CHROME = 2 * 3 + 2
+
+
+@dataclass
+class Row:
+    key: str
+    name: str
+    is_dir: bool
+    size: int | None = None
+    when: float | None = None
+    note: str = ""                       # "down", "read-only", ...
+    dim: bool = False
+    data: dict = field(default_factory=dict)
+
+
+class Pane(Vertical, can_focus=False):
+    """A titled table with marks, a sort key and a filter."""
+
+    BINDINGS = [
+        Binding("space", "mark", "Select"),
+        Binding("backspace", "up", "Up"),
+        Binding("slash", "filter", "Filter"),
+        Binding("s", "sort", "Sort"),
+        Binding("r", "refresh", "Refresh"),
+    ]
+
+    DEFAULT_CSS = """
+    Pane { width: 1fr; border: round $panel; }
+    Pane:focus-within { border: round $accent; }
+    Pane > .title { height: 1; padding: 0 1; background: $panel; }
+    Pane > DataTable { height: 1fr; }
+    Pane > Input { display: none; }
+    Pane > Input.shown { display: block; }
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.rows: list[Row] = []
+        self.marks: set[str] = set()
+        self.sort = "name"
+        self.filter_text = ""
+        self.name_width = 30
+        self.show_date = True
+
+    def compose(self) -> ComposeResult:
+        yield Static("", classes="title", markup=False)
+        table = DataTable(cursor_type="row", zebra_stripes=False)
+        self._add_columns(table)
+        yield table
+        yield Input(placeholder="filter — Enter to apply, Esc to clear")
+
+    def _add_columns(self, table: DataTable) -> None:
+        table.add_column("Name", key="name", width=self.name_width)
+        table.add_column(Text("Size", justify="right"), key="size", width=SIZE_WIDTH)
+        if self.show_date:
+            table.add_column("Date", key="date", width=DATE_WIDTH)
+
+    def on_resize(self) -> None:
+        """
+        The name takes what the size and the date leave, never more: a long name
+        with no spaces in it must not push them out of the pane. In a pane too
+        narrow for all three (two panes in an 80-column terminal), the date goes.
+        """
+        self._draw_title()
+        inner = self.table.size.width
+        show_date = inner - SIZE_WIDTH - DATE_WIDTH - CHROME >= 16
+        width = max(8, inner - SIZE_WIDTH - (DATE_WIDTH if show_date else 0)
+                    - (CHROME if show_date else CHROME - 2))
+        if (width, show_date) != (self.name_width, self.show_date):
+            self.name_width, self.show_date = width, show_date
+            self.table.clear(columns=True)
+            self._add_columns(self.table)
+            self.redraw()
+
+    @property
+    def table(self) -> DataTable:
+        return self.query_one(DataTable)
+
+    def set_title(self, text: str) -> None:
+        self._title = text
+        self._draw_title()
+
+    def _draw_title(self) -> None:
+        """One line; a long path loses its beginning, which is the part known."""
+        text = getattr(self, "_title", "")
+        room = max(8, self.size.width - 4)
+        if len(text) > room:
+            text = "…" + text[-(room - 1):]
+        self.query_one(".title", Static).update(Text(text, no_wrap=True))
+
+    # ── Drawing ──────────────────────────────────────────────────────────────
+
+    def show(self, rows: list[Row], *, keep_cursor: str | None = None) -> None:
+        self.rows = rows
+        self.marks &= {r.key for r in rows}
+        self.redraw(keep_cursor=keep_cursor)
+
+    def _ordered(self) -> list[Row]:
+        def k(r: Row):
+            if self.sort == "size":
+                return (-(r.size or 0), r.name.lower())
+            if self.sort == "date":
+                return (-(r.when or 0), r.name.lower())
+            return (r.name.lower(), r.name)
+        visible = [r for r in self.rows if r.key != PARENT
+                   and self.filter_text.lower() in r.name.lower()]
+        parent = [r for r in self.rows if r.key == PARENT]
+        return (parent + sorted((r for r in visible if r.is_dir), key=k)
+                + sorted((r for r in visible if not r.is_dir), key=k))
+
+    def redraw(self, *, keep_cursor: str | None = None) -> None:
+        table = self.table
+        current = keep_cursor or self.cursor_key()
+        table.clear()
+        for r in self._ordered():
+            mark = "* " if r.key in self.marks else "  "
+            name = Text(mark + safe(r.name) + ("/" if r.is_dir and r.key != PARENT else ""),
+                        style="dim" if r.dim else ("bold" if r.key in self.marks else ""),
+                        no_wrap=True, overflow="ellipsis")
+            if r.note:
+                name.append(f"  ({r.note})", style="italic dim")
+            size = Text("" if r.is_dir or r.size is None else human_size(r.size),
+                        justify="right")
+            cells = [name, size]
+            if self.show_date:
+                cells.append(Text(date(int(r.when)) if r.when else ""))
+            table.add_row(*cells, key=r.key)
+        keys = [r.key for r in self._ordered()]
+        if current in keys:
+            table.move_cursor(row=keys.index(current))
+
+    def cursor_key(self) -> str | None:
+        table = self.table
+        if table.row_count == 0:
+            return None
+        try:
+            return table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value
+        except Exception:
+            return None
+
+    def row(self, key: str | None) -> Row | None:
+        return next((r for r in self.rows if r.key == key), None)
+
+    def selection(self) -> list[Row]:
+        """The marked rows, or the one under the cursor."""
+        if self.marks:
+            return [r for r in self.rows if r.key in self.marks]
+        r = self.row(self.cursor_key())
+        return [r] if r and r.key != PARENT else []
+
+    # ── Actions ──────────────────────────────────────────────────────────────
+
+    def action_mark(self) -> None:
+        key = self.cursor_key()
+        if key is None or key == PARENT:
+            return
+        self.marks ^= {key}
+        row = self.table.cursor_row
+        self.redraw(keep_cursor=key)
+        self.table.move_cursor(row=min(row + 1, self.table.row_count - 1))
+
+    def action_sort(self) -> None:
+        self.sort = SORTS[(SORTS.index(self.sort) + 1) % len(SORTS)]
+        self.app.notify(f"Sorted by {self.sort}", timeout=2)
+        self.redraw()
+
+    def action_filter(self) -> None:
+        box = self.query_one(Input)
+        box.add_class("shown")
+        box.value = self.filter_text
+        box.focus()
+
+    def on_input_changed(self, event: Input.Changed) -> None:
+        self.filter_text = event.value
+        self.redraw()
+
+    def on_input_submitted(self, event: Input.Submitted) -> None:
+        self.table.focus()
+
+    def key_escape(self) -> None:
+        box = self.query_one(Input)
+        if box.has_class("shown"):
+            box.value = ""
+            box.remove_class("shown")
+            self.filter_text = ""
+            self.redraw()
+            self.table.focus()
+
+    def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
+        row = self.row(event.row_key.value)
+        if row is None:
+            return
+        if row.key == PARENT:
+            self.action_up()
+        elif row.is_dir:
+            self.enter(row)
+
+    # Subclasses
+    def enter(self, row: Row) -> None: ...
+    def action_up(self) -> None: ...
+    def action_refresh(self) -> None: ...
+
+
+class LocalPane(Pane):
+    def __init__(self, start: Path, **kwargs):
+        super().__init__(**kwargs)
+        self.path = start.resolve()
+
+    def on_mount(self) -> None:
+        self.load()
+
+    def load(self, keep: str | None = None) -> None:
+        self.set_title(f"Local: {self.path}")
+        self.run_worker(self._load(keep), exclusive=True, group="local")
+
+    async def _load(self, keep: str | None) -> None:
+        try:
+            rows = await asyncio.to_thread(_scan, self.path)
+        except OSError as e:
+            self.app.notify(f"{self.path}: {e.strerror}", severity="error", markup=False)
+            rows = []
+        if self.path.parent != self.path:
+            rows.insert(0, Row(PARENT, PARENT, True))
+        self.show(rows, keep_cursor=keep)
+
+    def enter(self, row: Row) -> None:
+        self.path = self.path / row.name
+        self.marks.clear()
+        self.load()
+
+    def action_up(self) -> None:
+        if self.path.parent != self.path:
+            came_from = self.path.name
+            self.path = self.path.parent
+            self.marks.clear()
+            self.load(keep=came_from)
+
+    def action_refresh(self) -> None:
+        # No position captured now: the cursor may move while the folder is read,
+        # and the redraw keeps wherever it is when the rows arrive.
+        self.load()
+
+
+def _scan(path: Path) -> list[Row]:
+    rows = []
+    with os.scandir(path) as it:
+        for e in it:
+            try:
+                st = e.stat()
+            except OSError:
+                continue
+            is_dir = stat.S_ISDIR(st.st_mode)
+            if not is_dir and not stat.S_ISREG(st.st_mode):
+                continue
+            rows.append(Row(e.name, e.name, is_dir, None if is_dir else st.st_size,
+                            st.st_mtime))
+    return rows
+
+
+class RemotePane(Pane):
+    """Groups, then a group's roots and folders — the Files tab, never merged."""
+
+    def __init__(self, screen, **kwargs):
+        super().__init__(**kwargs)
+        self.main = screen
+        self.groups: list[dict] = []
+        self.group: dict | None = None       # None: the list of groups
+        self.path = ""                       # inside the group: root/folders
+
+    def on_mount(self) -> None:
+        self.load()
+
+    @property
+    def location(self) -> str:
+        if self.group is None:
+            return "Your groups"
+        return f"{handle(self.group)}/{self.path}".rstrip("/")
+
+    def load(self, keep: str | None = None) -> None:
+        self.set_title(f"{safe(self.location)}  (loading…)")
+        self.run_worker(self._load(keep), exclusive=True, group="remote")
+
+    async def _load(self, keep: str | None) -> None:
+        try:
+            if self.group is None:
+                self.groups = await my_groups(self.main.account)
+                rows = [Row(g["id"], handle(g), True,
+                            note="" if g.get("node_online") else "down",
+                            dim=not g.get("node_online"), data=g)
+                        for g in self.groups]
+            else:
+                tree = await self.main.tree(self.group)
+                rows = [Row(PARENT, PARENT, True)] + _tree_rows(tree, self.path)
+        except CliError as e:
+            self.set_title(safe(self.location))
+            self.app.notify(e.message, severity="error", markup=False, timeout=8)
+            if self.group is not None:
+                self.group, self.path = None, ""
+                self.load()
+            return
+        self.set_title(safe(self.location))
+        self.show(rows, keep_cursor=keep)
+
+    def enter(self, row: Row) -> None:
+        self.marks.clear()
+        if self.group is None:
+            self.group, self.path = row.data, ""
+        else:
+            self.path = row.data["path"]
+        self.load()
+
+    def action_up(self) -> None:
+        if self.group is None:
+            return
+        self.marks.clear()
+        if not self.path:
+            came_from, self.group = self.group["id"], None
+        else:
+            came_from = self.path
+            self.path = self.path.rpartition("/")[0]
+        self.load(keep=came_from)
+
+    def action_refresh(self) -> None:
+        if self.group is not None:
+            self.main.forget_tree(self.group)
+        self.load()
+
+
+def _tree_rows(tree: Tree, path: str) -> list[Row]:
+    rows = []
+    for item in tree.children(path):
+        note = ""
+        if item.root is not None:
+            note = "" if item.root.get("writable") else "read-only"
+            if not item.available:
+                note = "unavailable"
+        rows.append(Row(item.path, item.name, item.is_dir,
+                        None if item.is_dir else item.size,
+                        item.added_at or None, note=note, dim=not item.available,
+                        data={"path": item.path, "entry": item.entry}))
+    return rows
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/screens.py b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
new file mode 100644
index 0000000..9d19431
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/screens.py
@@ -0,0 +1,354 @@
+"""
+The sign-in screen, and the main screen: two panes and the transfers.
+
+F5 copies the selection to the other pane's folder — a download or an upload,
+with the rules `get` and `put` follow: an upload is decided whole before it
+starts (names the node refuses stop it, names already there are asked about
+once for the batch), nothing on either side is replaced without being asked,
+and a name from a group is made writable here before it is written.
+"""
+
+import asyncio
+import contextlib
+import threading
+from pathlib import Path
+
+from rich.text import Text
+from textual.app import ComposeResult
+from textual.binding import Binding
+from textual.containers import Horizontal, Vertical
+from textual.screen import Screen
+from textual.widgets import Button, Footer, Input, Static
+
+from ..client.account import Account
+from ..client.names import UnsafeName, local_path
+from ..client.paths import handle
+from ..client.pool import Pool
+from ..client.store import FILE_KEY_NOTICE, KEY_FILE
+from ..client.transfers import PART_SUFFIX, TransferError, create_dir, download, upload
+from ..client.tree import Tree
+from ..client.upload_plan import plan_upload, refused_names, taken_names
+from ..errors import CliError
+from ..output import safe
+from .panes import LocalPane, RemotePane
+from .widgets import Choice, TransferStrip
+
+PER_GROUP = 2
+
+
+class SignInScreen(Screen):
+    DEFAULT_CSS = """
+    SignInScreen { align: center middle; }
+    SignInScreen > Vertical { width: 64; height: auto; border: thick $accent;
+                              padding: 1 2; background: $surface; }
+    SignInScreen Input { margin-bottom: 1; }
+    SignInScreen #error { color: $error; height: auto; }
+    """
+
+    def compose(self) -> ComposeResult:
+        ctx = self.app.ctx
+        with Vertical():
+            yield Static(Text("Sign in to MeshBay"))
+            yield Input(value=ctx.hub_url, placeholder="hub", id="hub")
+            yield Input(value=ctx.username or ctx.stored_username or "",
+                        placeholder="username", id="user")
+            yield Input(placeholder="passphrase", password=True, id="pass")
+            yield Button("Sign in", id="go", variant="primary")
+            yield Static("", id="error", markup=False)
+
+    def on_mount(self) -> None:
+        empty = next((i for i in self.query(Input) if not i.value), None)
+        (empty or self.query_one("#pass", Input)).focus()
+
+    def on_input_submitted(self, event: Input.Submitted) -> None:
+        self._submit()
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        self._submit()
+
+    def _submit(self) -> None:
+        hub = self.query_one("#hub", Input).value.strip().rstrip("/")
+        user = self.query_one("#user", Input).value.strip()
+        passphrase = self.query_one("#pass", Input).value
+        if not (hub and user and passphrase):
+            self.query_one("#error", Static).update("Hub, username and passphrase.")
+            return
+        self.query_one("#error", Static).update("Signing in…")
+        self.run_worker(self._sign_in(hub, user, passphrase), exclusive=True)
+
+    async def _sign_in(self, hub_url: str, user: str, passphrase: str) -> None:
+        app = self.app
+        try:
+            await app.use_hub(hub_url)
+            account = Account(app.hub, user, lambda: passphrase)
+            await account.hub.check_version()
+            await account.sign_in(remember=True)
+            kind = await account.remember(app.ctx.store)
+        except CliError as e:
+            self.query_one("#error", Static).update(e.message)
+            return
+        if kind == KEY_FILE:
+            app.notify(FILE_KEY_NOTICE, timeout=10, markup=False)
+        app.start(account)
+
+
+class MainScreen(Screen):
+    BINDINGS = [
+        Binding("f5", "copy", "Copy"),
+        Binding("q", "app.quit", "Quit"),
+        Binding("tab", "app.focus_next", "Switch", show=False),
+    ]
+
+    DEFAULT_CSS = """
+    MainScreen > Horizontal { height: 1fr; }
+    """
+
+    def __init__(self, account: Account):
+        super().__init__()
+        self.account = account
+        self.pool = Pool(account)
+        self._trees: dict[str, Tree] = {}
+        self._gates: dict[str, asyncio.Semaphore] = {}
+        self._watching: set[str] = set()
+
+    def compose(self) -> ComposeResult:
+        with Horizontal():
+            yield LocalPane(Path.cwd(), id="local")
+            yield RemotePane(self, id="remote")
+        yield TransferStrip(id="transfers")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self.query_one(LocalPane).table.focus()
+
+    @property
+    def local(self) -> LocalPane:
+        return self.query_one(LocalPane)
+
+    @property
+    def remote(self) -> RemotePane:
+        return self.query_one(RemotePane)
+
+    @property
+    def strip(self) -> TransferStrip:
+        return self.query_one(TransferStrip)
+
+    # ── Groups ───────────────────────────────────────────────────────────────
+
+    async def tree(self, group: dict) -> Tree:
+        """The group's index, asked once and then kept current by the node's pushes."""
+        if group["id"] not in self._trees:
+            session = await self.pool.session(group)
+            self._trees[group["id"]] = Tree.from_index(await session.index())
+            if group["id"] not in self._watching:
+                self._watching.add(group["id"])
+                session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g))
+        return self._trees[group["id"]]
+
+    def forget_tree(self, group: dict) -> None:
+        self._trees.pop(group["id"], None)
+
+    def _changed(self, group: dict) -> None:
+        """Something changed in the group: the index is asked for again, and the
+        pane redrawn if it is showing that group."""
+        self.forget_tree(group)
+        remote = self.remote
+        if remote.group is not None and remote.group["id"] == group["id"]:
+            remote.load()
+
+    # ── F5 ───────────────────────────────────────────────────────────────────
+
+    def action_copy(self) -> None:
+        focused = self.app.focused
+        if focused is not None and self.remote in focused.ancestors_with_self:
+            self.run_worker(self._download(), group="copy")
+        else:
+            self.run_worker(self._upload(), group="copy")
+
+    def _gate(self, group: dict) -> asyncio.Semaphore:
+        return self._gates.setdefault(group["id"], asyncio.Semaphore(PER_GROUP))
+
+    async def _ask(self, message: str, options: list[tuple[str, str]]) -> str:
+        return await self.app.push_screen(Choice(message, options), wait_for_dismiss=True)
+
+    async def _download(self) -> None:
+        remote, local = self.remote, self.local
+        rows = remote.selection()
+        if remote.group is None or not rows:
+            self.app.notify("Open a group and choose what to download.", timeout=4)
+            return
+        group, dest = remote.group, local.path
+        try:
+            tree = await self.tree(group)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        plan: list[tuple[dict, Path]] = []
+        dirs: list[Path] = []
+        problems: list[str] = []
+        for row in rows:
+            path = row.data["path"]
+            top = path.rsplit("/", 1)[-1]
+            if not tree.root_available(path):
+                problems.append(f"{top}: its folder on the node is unavailable")
+                continue
+            if not row.is_dir:
+                try:
+                    plan.append((row.data["entry"], local_path(dest, [top])[0]))
+                except UnsafeName as e:
+                    problems.append(str(e))
+                continue
+            for folder, items in tree.walk(path):
+                rel = [top] + [p for p in folder[len(path):].strip("/").split("/") if p]
+                try:
+                    dirs.append(local_path(dest, rel)[0])
+                    plan += [(i.entry, local_path(dest, rel + [i.name])[0])
+                             for i in items if not i.is_dir]
+                except UnsafeName as e:
+                    problems.append(str(e))
+        for p in problems:
+            self.app.notify(p, severity="warning", markup=False)
+        existing = [t for _, t in plan if t.exists()
+                    and not t.with_name(t.name + PART_SUFFIX).exists()]
+        overwrite = False
+        if existing:
+            names = ", ".join(safe(t.name) for t in existing[:5]) + (
+                f" and {len(existing) - 5} more" if len(existing) > 5 else "")
+            answer = await self._ask(
+                f"{len(existing)} file(s) already in {dest}: {names}.",
+                [("skip", "Skip them"), ("overwrite", "Overwrite"), ("cancel", "Cancel")])
+            if answer == "cancel":
+                return
+            overwrite = answer == "overwrite"
+            if not overwrite:
+                plan = [(e, t) for e, t in plan if t not in existing]
+        for d in dirs:
+            await asyncio.to_thread(d.mkdir, parents=True, exist_ok=True)
+        remote.marks.clear()
+        remote.redraw()
+        session = await self.pool.session(group)
+        await asyncio.gather(*(self._one_download(group, session, e, t, overwrite)
+                               for e, t in plan))
+        local.action_refresh()
+
+    async def _one_download(self, group, session, entry: dict, target: Path,
+                            overwrite: bool) -> None:
+        job = self.strip.add("↓", entry["name"], int(entry["size"]))
+        async with self._gate(group):
+            self.strip.update_job(job, state="running")
+            try:
+                await download(
+                    session, entry, target, overwrite=overwrite,
+                    progress=lambda n: self.strip.update_job(job, done=job.done + n),
+                    on_queued=lambda n: self.strip.update_job(job, state="queued",
+                                                              note=f"({n} ahead)"))
+                self.strip.update_job(job, state="done",
+                                      note="" if target.name == entry["name"]
+                                      else f"saved as {target.name}")
+            except TransferError as e:
+                self.strip.update_job(job, state="failed", note=e.message)
+
+    async def _upload(self) -> None:
+        remote, local = self.remote, self.local
+        rows = local.selection()
+        if not rows:
+            return
+        if remote.group is None or not remote.path:
+            self.app.notify("Open a folder of a group on the right to upload into.",
+                            timeout=4)
+            return
+        group, folder = remote.group, remote.path
+        try:
+            tree = await self.tree(group)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        root = tree.roots.get(folder.split("/", 1)[0], {})
+        if not root.get("writable") or not root.get("available", True):
+            self.app.notify(f"{safe(handle(group))}/{safe(folder)} does not accept "
+                            "uploads (read-only or unavailable).", severity="error",
+                            markup=False)
+            return
+        try:
+            dirs, files, problems = plan_upload([str(local.path / r.name) for r in rows],
+                                                recursive=True)
+        except CliError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        problems += refused_names([f.rel[-1] for f in files], dirs)
+        if problems:
+            await self._ask("Nothing was sent:\n" + "\n".join(safe(p) for p in problems),
+                            [("ok", "OK")])
+            return
+        tops = sorted({d[0] for d in dirs} | {f.rel[0] for f in files})
+        taken = taken_names(tree, folder, tops)
+        if taken:
+            answer = await self._ask(
+                f"Already in {safe(folder)}: {', '.join(safe(t) for t in sorted(taken))}. "
+                "The node never replaces a file.",
+                [("skip", "Skip them"), ("cancel", "Cancel")])
+            if answer != "skip":
+                return
+            dirs = [d for d in dirs if d[0] not in taken]
+            files = [f for f in files if f.rel[0] not in taken]
+        local.marks.clear()
+        local.redraw()
+        session = await self.pool.session(group)
+        remote_dir = {(): folder}
+        try:
+            for d in dirs:
+                remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])],
+                                                        d[-1])
+        except TransferError as e:
+            self.app.notify(e.message, severity="error", markup=False)
+            return
+        await asyncio.gather(*(self._one_upload(group, session, f,
+                                                remote_dir[tuple(f.rel[:-1])])
+                               for f in files))
+        self._changed(group)
+
+    async def _one_upload(self, group, session, item, folder: str) -> None:
+        name = item.rel[-1]
+        job = self.strip.add("↑", name, item.size)
+        async with self._gate(group):
+            self.strip.update_job(job, state="running")
+            lock = threading.Lock()
+            try:
+                with open(item.local, "rb") as fh:
+                    def read(offset: int, length: int) -> bytes:
+                        with lock:
+                            fh.seek(offset)
+                            return fh.read(length)
+                    stored = await upload(
+                        session, read, item.size, name, folder,
+                        progress=lambda n: self.strip.update_job(job, done=job.done + n),
+                        on_queued=lambda n: self.strip.update_job(job, state="queued",
+                                                                  note=f"({n} ahead)"))
+                self.strip.update_job(job, state="done", note="" if stored == name
+                                      else f"stored as {stored}")
+            except OSError as e:
+                self.strip.update_job(job, state="failed", note=e.strerror or str(e))
+            except TransferError as e:
+                self.strip.update_job(job, state="failed", note=e.message)
+
+    async def on_unmount(self) -> None:
+        with contextlib.suppress(Exception):
+            await self.pool.close()
+
+
+class DeviceCode:
+    """`account.on_device_code` for the interactive mode: the code in a
+    notification that stays until the approval lands; the node is retried meanwhile."""
+
+    def __init__(self, app):
+        self.app = app
+
+    def __call__(self, group: dict, pending, *, first: bool) -> bool:
+        if first:
+            self.app.notify(
+                f"This machine is not yet a device of yours on the node hosting "
+                f"{safe(handle(group))}. Approve it from the MeshBay application or a "
+                f"browser: the group's Settings → \"Your devices on this node\" → "
+                f"enter {pending.link_code}. Waiting…",
+                title="Approve this machine", timeout=600, markup=False)
+        return True
diff --git a/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py
new file mode 100644
index 0000000..66fca4e
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/tui/widgets.py
@@ -0,0 +1,109 @@
+"""A question with a few answers, and the strip that shows transfers."""
+
+import time
+from dataclasses import dataclass
+
+from rich.text import Text
+from textual.app import ComposeResult
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import Button, Static
+
+from ..output import human_size, safe
+
+
+class Choice(ModalScreen[str]):
+    """One question, asked once for a whole batch — never once per file."""
+
+    DEFAULT_CSS = """
+    Choice { align: center middle; }
+    Choice > Vertical { width: 72; height: auto; max-height: 80%;
+                        border: thick $accent; background: $surface; padding: 1 2; }
+    Choice Static { height: auto; margin-bottom: 1; }
+    Choice Horizontal { height: auto; align: center middle; }
+    Choice Button { margin: 0 1; }
+    """
+    BINDINGS = [("escape", "dismiss('cancel')", "Cancel")]
+
+    def __init__(self, message: str, options: list[tuple[str, str]]):
+        super().__init__()
+        self.message = message
+        self.options = options
+
+    def compose(self) -> ComposeResult:
+        with Vertical():
+            yield Static(Text(self.message))
+            with Horizontal():
+                for value, label in self.options:
+                    yield Button(label, id=f"choice-{value}")
+
+    def on_mount(self) -> None:
+        self.query(Button).first().focus()
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        self.dismiss(event.button.id.removeprefix("choice-"))
+
+
+@dataclass
+class Job:
+    arrow: str
+    name: str
+    size: int
+    done: int = 0
+    state: str = "waiting"        # waiting, queued, running, done, failed
+    note: str = ""
+    started: float = 0.0
+
+
+class TransferStrip(Static):
+    """The running transfers, then the latest finished ones."""
+
+    DEFAULT_CSS = """
+    TransferStrip { height: 6; border-top: solid $panel; padding: 0 1; }
+    """
+    SHOWN = 5
+
+    def __init__(self, **kwargs):
+        super().__init__("", markup=False, **kwargs)
+        self.jobs: list[Job] = []
+
+    def on_mount(self) -> None:
+        self.refresh_text()
+
+    def add(self, arrow: str, name: str, size: int) -> Job:
+        job = Job(arrow, name, size)
+        self.jobs.append(job)
+        self.refresh_text()
+        return job
+
+    def update_job(self, job: Job, **fields) -> None:
+        for k, v in fields.items():
+            setattr(job, k, v)
+        if job.state == "running" and not job.started:
+            job.started = time.monotonic()
+        self.refresh_text()
+
+    def refresh_text(self) -> None:
+        active = [j for j in self.jobs if j.state in ("running", "queued", "waiting")]
+        ended = [j for j in self.jobs if j.state in ("done", "failed")]
+        lines = [self._line(j) for j in active[:self.SHOWN]]
+        if len(active) > self.SHOWN:
+            lines.append(f"  … and {len(active) - self.SHOWN} more waiting")
+        for j in reversed(ended[-(self.SHOWN - len(lines)):] if len(lines) < self.SHOWN
+                          else []):
+            lines.append(self._line(j))
+        self.update(Text("\n".join(lines) or "No transfers."))
+
+    def _line(self, j: Job) -> str:
+        name = safe(j.name)
+        if j.state == "running":
+            pct = 100 * j.done // j.size if j.size else 100
+            rate = j.done / max(time.monotonic() - j.started, 0.001)
+            return f"{j.arrow} {name}  {pct}%  {human_size(int(rate))}/s"
+        if j.state == "queued":
+            return f"{j.arrow} {name}  waiting for a free slot on the node {j.note}"
+        if j.state == "waiting":
+            return f"{j.arrow} {name}  waiting"
+        if j.state == "done":
+            return f"{j.arrow} {name}  done{(' — ' + j.note) if j.note else ''}"
+        return f"{j.arrow} {name}  FAILED: {j.note}"