feat(cli): search across groups

search dials every group through the pool and matches each index here, on
the Search page's fields (name, folder, title, artist, album), every word
required. A group left out of Search by its operator (search_listed in the
sealed ack) is skipped before its index is asked for. A file several groups
hold is one result, keyed by its content hash, printed from the reader's own
node first, as a path get takes. A group that cannot be searched is named
and the others go on; one waiting for this machine's approval names its code
instead of holding everything up.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

3 files changed, 151 insertions, 3 deletions

diff --git a/packages/meshbay-cli/src/meshbay_cli/__main__.py b/packages/meshbay-cli/src/meshbay_cli/__main__.py
index a29b528..dbb5a19 100644
--- a/packages/meshbay-cli/src/meshbay_cli/__main__.py
+++ b/packages/meshbay-cli/src/meshbay_cli/__main__.py
@@ -25,13 +25,14 @@ from .commands import (
     ls,
     put,
     rm,
+    search,
     trust,
 )
 from .commands.context import Context
 from .errors import INTERRUPTED, USAGE, CliError
 
 ACTIONS = {"login": login, "logout": logout, "groups": groups, "invitations": invitations,
-           "join": join, "decline": decline, "leave": leave, "ls": ls, "get": get,
+           "join": join, "decline": decline, "leave": leave, "ls": ls, "search": search, "get": get,
            "put": put, "rm": rm, "devices": devices, "trust": trust,
            "browser-access": browser_access}
 
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/search.py b/packages/meshbay-cli/src/meshbay_cli/commands/search.py
new file mode 100644
index 0000000..dbd0db1
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/search.py
@@ -0,0 +1,145 @@
+"""
+`search`: files across every group of the account, by name, folder or tags.
+
+Each group's node is dialled (six negotiations at most, the pool's rule) and its
+index matched here, as the application's Search does: the node serves the same
+index to Search and to the group page. A group its operator left out of Search
+(`search_listed`, in the sealed ack) is skipped before its index is asked for —
+a listing preference, not a protection (docs/MESHBAY_DESIGN.md §9.11). A file
+several groups hold is one result: entries are keyed by their content hash,
+and every place it can be fetched from is kept. A group whose node does not
+answer is reported and costs the others nothing (exit 4).
+"""
+
+import asyncio
+import json
+import sys
+
+from meshbay_common.paths import fold
+
+from ..client.groups import my_groups
+from ..client.paths import encode_name, handle
+from ..client.pool import Pool
+from ..client.session import DeviceLinkPending
+from ..client.tree import Tree
+from ..errors import NOT_FOUND, PARTIAL, USAGE, CliError
+from ..output import date, err, human_size, safe, sentence
+from .context import Context
+
+HELP = "find files in all your groups by name, folder, title, artist or album"
+
+TYPES = ("video", "audio", "image", "document", "archive", "other")
+# What a word is looked for in: the Search page's fields.
+FIELDS = ("name", "path", "display_title", "artist", "album")
+
+
+def add_parser(sub, parent) -> None:
+    p = sub.add_parser("search", parents=[parent], help=HELP, description=sentence(HELP)
+                       + ". Every word must be found, without regard to case. A file "
+                       "several groups hold is listed once. Groups their operator "
+                       "left out of Search are not searched.")
+    p.add_argument("words", nargs="+", metavar="WORD")
+    p.add_argument("-t", "--type", choices=TYPES, help="only files of this kind")
+    p.add_argument("-l", dest="long", action="store_true",
+                   help="type, size and date, and how many other groups hold it")
+
+
+def matches(entry: dict, words: list[str]) -> bool:
+    text = [fold(str(entry.get(f) or "")) for f in FIELDS]
+    return all(any(w in t for t in text) for w in words)
+
+
+def remote_path(group: dict, entry: dict) -> str:
+    """The path `get` takes: the group as `groups` prints it, then the file."""
+    parts = [p for p in entry["path"].split("/") if p] + [entry["name"]]
+    return "/".join([handle(group)] + [encode_name(p) for p in parts])
+
+
+async def run(ctx: Context) -> int:
+    await ctx.hub.check_version()
+    account = ctx.account()
+    # A group that needs this machine approved is reported, not waited for:
+    # a search does not hold every other group's results behind one code.
+    account.on_device_code = lambda group, pending, *, first: False
+    pool = Pool(account)
+    try:
+        return await _run(ctx, account, pool)
+    finally:
+        await pool.close()
+        await account.close()
+
+
+async def _search_group(pool: Pool, group: dict, words: list[str], kind: str | None):
+    """(matching entries, own node) — or None when the group is left out of Search."""
+    session = await pool.session(group)
+    if session.ack.get("search_listed") is False:
+        return None
+    tree = Tree.from_index(await session.index())
+    found = [e for entries in tree.files.values() for e in entries
+             if tree.root_available(e["path"]) and (kind is None or e.get("type") == kind)
+             and matches(e, words)]
+    return found, bool(session.ack.get("is_node_admin"))
+
+
+async def _run(ctx: Context, account, pool: Pool) -> int:
+    a = ctx.args
+    words = [fold(w) for w in " ".join(a.words).split() if w]
+    if not words:
+        raise CliError(USAGE, "Say what to look for.")
+    groups = [g for g in await my_groups(account) if g.get("hosted", True)]
+
+    async def one(group):
+        try:
+            return group, await _search_group(pool, group, words, a.type), None
+        except DeviceLinkPending as e:
+            return group, None, f"this machine waits to be approved there ({e.link_code})"
+        except CliError as e:
+            return group, None, e.message
+
+    results = await asyncio.gather(*(one(g) for g in groups))
+
+    # One result per content hash; its first place is the reader's own node if
+    # one holds it, then the group names in order — stable from run to run.
+    by_id: dict[str, dict] = {}
+    unreachable = []
+    for group, found, problem in sorted(results, key=lambda r: (r[0]["name"].lower(),
+                                                                r[0].get("owner_username"))):
+        if problem is not None:
+            unreachable.append(f"{handle(group)}: {problem}")
+            continue
+        if found is None:
+            continue
+        entries, own = found
+        for e in entries:
+            hit = by_id.setdefault(e["id"], {"entry": e, "places": []})
+            place = remote_path(group, e)
+            if own:
+                hit["places"].insert(0, place)
+            else:
+                hit["places"].append(place)
+
+    hits = sorted(by_id.values(), key=lambda h: (fold(h["entry"]["name"]), h["places"][0]))
+    if a.json:
+        json.dump([{"name": h["entry"]["name"], "id": h["entry"]["id"],
+                    "type": h["entry"].get("type"), "size": h["entry"].get("size"),
+                    "added_at": h["entry"].get("added_at"), "path": h["places"][0],
+                    "also": h["places"][1:]} for h in hits],
+                  sys.stdout, indent=1, ensure_ascii=False)
+        print()
+    else:
+        for h in hits:
+            e, path = h["entry"], safe(h["places"][0])
+            if a.long:
+                others = len(h["places"]) - 1
+                note = f"  (+{others} other group{'s' if others != 1 else ''})" if others else ""
+                print(f"{e.get('type') or '':<9}{human_size(int(e.get('size') or 0)):>10}  "
+                      f"{date(int(e.get('added_at') or 0)):<16}  {path}{note}")
+            else:
+                print(path)
+    for u in unreachable:
+        err(f"meshbay-cli: not searched: {safe(u)}")
+    if not hits:
+        if not a.quiet and not a.json:
+            err("Nothing found.")
+        return PARTIAL if unreachable else NOT_FOUND
+    return PARTIAL if unreachable else 0
diff --git a/packages/meshbay-cli/src/meshbay_cli/manpage.py b/packages/meshbay-cli/src/meshbay_cli/manpage.py
index 8316fc5..3fa9550 100644
--- a/packages/meshbay-cli/src/meshbay_cli/manpage.py
+++ b/packages/meshbay-cli/src/meshbay_cli/manpage.py
@@ -44,8 +44,8 @@ EXIT_CODES = {
 DESCRIPTION = r"""
 .B meshbay-cli
 is a member's view of their MeshBay groups from a terminal: sign in, join and
-leave groups, list them and their folders, download, upload and delete files,
-and approve this account's other devices.
+leave groups, list them and their folders, search them all at once, download,
+upload and delete files, and approve this account's other devices.
 It talks to the hub only to sign in and to find a group's nodes; files go
 directly between this machine and the node, end-to-end encrypted, never
 through the hub.
@@ -298,6 +298,8 @@ meshbay\-cli get 'Family/music/*.flac'
 meshbay\-cli put \-R ~/scans Family/documents
 tar c notes | meshbay\-cli put \- Family/documents/notes.tar
 meshbay\-cli ls \-\-json \-R / | jq '.[] | .path'
+meshbay\-cli search \-t audio holiday
+meshbay\-cli search beach 2024 | head \-1 | xargs \-d '\\n' meshbay\-cli get
 meshbay\-cli rm 'Family/documents/draft*.odt'
 meshbay\-cli devices approve Family@bob ZX4P\-7QRT
 meshbay\-cli leave \-\-yes Club@carol