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>

6 files changed, 234 insertions, 4 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
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index 46ad68e..73d6f72 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -59,6 +59,7 @@ class FakeNode:
                          "pinned_via": "code"}]
         self.requests: list[dict] = []
         self.role = ""                       # "operator": this key was paired here
+        self.ack: dict = {}                  # the sealed configuration, opened
         self.admin_ops: dict[str, dict] = {}
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
@@ -74,7 +75,7 @@ class FakeNode:
 
     def session(self) -> GroupSession:
         return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
-                            {}, self.identity, self.nonce, self.role)
+                            dict(self.ack), self.identity, self.nonce, self.role)
 
     def request_device(self, code: str) -> Identity:
         """Another machine files a request with `code`, as it would before its proof."""
diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py
index e0b5518..e06fb9c 100644
--- a/packages/meshbay-cli/tests/test_cli_actions.py
+++ b/packages/meshbay-cli/tests/test_cli_actions.py
@@ -457,6 +457,59 @@ def test_rm_deletes_only_what_this_account_may_and_asks_first(node, monkeypatch,
     assert not node.files
 
 
+def test_search_finds_across_groups_once_per_file(hub, monkeypatch, capsys):
+    from fake_node import FakeNode
+    from meshbay_cli.client.session import DeviceLinkPending
+    family, club, hidden = FakeNode(), FakeNode(), FakeNode()
+    hub.groups += [{**CLUB, "id": "g2"}, {**CLUB, "id": "g3", "name": "Album"},
+                   {**CLUB, "id": "g4", "name": "Away"}, {**CLUB, "id": "g5", "name": "New"}]
+    nodes = {"g1": family, "g2": club, "g3": hidden}
+    hidden.ack["search_listed"] = False
+    club.ack["is_node_admin"] = True
+
+    async def open_group(account, group):
+        if group["id"] == "g4":
+            from meshbay_cli.errors import NODE, CliError
+            raise CliError(NODE, "No node serving it is online.")
+        if group["id"] == "g5":
+            raise DeviceLinkPending("ABCD-EFGH", 0)
+        return nodes[group["id"]].session()
+    monkeypatch.setattr(pool, "open_group", open_group)
+    _signed_in(hub, capsys)
+
+    family.add("share/Holidays 2024", "Beach.JPG", b"same bytes")
+    club.add("share/Holidays 2024", "Beach.JPG", b"same bytes")     # the same file
+    track = family.add("share/music", "track.flac", b"other")
+    track.update(type="audio", artist="The Holiday Band")
+    family.add("share", "notes 100%.txt", b"n")
+    hidden.add("share", "holiday-secret.jpg", b"s")
+
+    assert main("search", "HOLIDAY") == 4               # two groups not searched
+    out = capsys.readouterr()
+    # The beach photo once, from the reader's own node first; the track by its tag.
+    assert out.out.splitlines() == [
+        "Club%2FPhotos@bob-test/share/Holidays 2024/Beach.JPG",
+        f"Family@{USERNAME}/share/music/track.flac"]
+    assert "holiday-secret" not in out.out
+    assert "not searched: Away@bob-test" in out.err
+    assert "ABCD-EFGH" in out.err
+
+    assert main("--json", "search", "beach", "2024") == 4
+    [hit] = json.loads(capsys.readouterr().out)
+    assert hit["also"] == [f"Family@{USERNAME}/share/Holidays 2024/Beach.JPG"]
+
+    assert main("search", "-l", "-t", "audio", "holiday") == 4
+    line = capsys.readouterr().out.strip()
+    assert line.startswith("audio") and line.endswith("track.flac")
+    assert main("search", "100%") == 4
+    assert capsys.readouterr().out.strip() == f"Family@{USERNAME}/share/notes 100%25.txt"
+    assert main("search", "nothing-like-this") == 4
+
+    hub.groups = [g for g in hub.groups if g["id"] in ("g1", "g2")]
+    assert main("search", "nothing-like-this") == 3
+    assert main("search", "beach") == 0
+
+
 def test_every_action_is_run_here():
     """Runs last in this module: every action the parser knows was run above."""
     assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}"
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index 988fb38..8ce3ebc 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -888,3 +888,31 @@ def test_a_member_joins_links_devices_shares_with_browsers_and_leaves(world):
     assert bob(world, "bob1", "ls", f"{group}/share").returncode == 0
     for home in ("bob1", "bob2", "bob3"):
         assert bob(world, home, "logout").returncode == 0
+
+
+def test_search_finds_a_file_and_get_takes_its_path(world, tmp_path):
+    """The path `search` prints is the one `get` takes, accents, spaces, an
+    apostrophe and a group name holding a / and an @ included."""
+    deadline = time.monotonic() + 90
+    while True:
+        r = cli(world, "--hub", world["hub_url"], "-u", USERNAME, "login",
+                "--password-file", "-", stdin=PASSPHRASE + "\n", home="cli-search")
+        if r.returncode == 0:
+            break
+        assert "Too many attempts" in r.stderr and time.monotonic() < deadline, r.stderr
+        time.sleep(5)
+    r = cli(world, "search", "PLAGE", "ouest", home="cli-search")
+    # After a test that withdrew the node's copy, this machine is a new device:
+    # search names the group and its code instead of waiting (exit 4).
+    if r.returncode == 4 and "waits to be approved" in r.stderr:
+        code = re.search(r"\(([0-9A-Z]{4}-[0-9A-Z]{4})\)", r.stderr).group(1)
+        asyncio.run(world["app"].approve(code))
+        r = cli(world, "search", "PLAGE", "ouest", home="cli-search")
+    assert r.returncode == 0, r.stderr
+    [path] = r.stdout.splitlines()
+    assert path == f"{_handle()}/share/2024/été à la mer/plage de l'ouest.jpg"
+    r = cli(world, "get", path, "-o", str(tmp_path) + "/", home="cli-search")
+    assert r.returncode == 0, r.stderr
+    assert (tmp_path / "plage de l'ouest.jpg").stat().st_size == 3000
+    r = cli(world, "search", "no-such-thing-anywhere", home="cli-search")
+    assert r.returncode == 3 and "Nothing found" in r.stderr